- Fixed bug where internal assertion would fail in the case where

an ``after_rollback()`` handler for a :class:`.Session` incorrectly
adds state to that :class:`.Session` within the handler, and the task
to warn and remove this state (established by :ticket:`2389`) attempts
to proceed.
fixes #3309

3 files changed, 44 insertions, 2 deletions

diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index 6836a15ab..0da20f66e 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -16,6 +16,16 @@
 
     .. change::
         :tags: bug, orm
+        :tickets: 3309
+
+        Fixed bug where internal assertion would fail in the case where
+        an ``after_rollback()`` handler for a :class:`.Session` incorrectly
+        adds state to that :class:`.Session` within the handler, and the task
+        to warn and remove this state (established by :ticket:`2389`) attempts
+        to proceed.
+
+    .. change::
+        :tags: bug, orm
         :pullreq: github:147
 
         Fixed bug where TypeError raised when :meth:`.Query.join` called
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index c47026969..a33b42612 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -404,26 +404,29 @@ class SessionTransaction(object):
             for subtransaction in stx._iterate_parents(upto=self):
                 subtransaction.close()
 
+        boundary = self
         if self._state in (ACTIVE, PREPARED):
             for transaction in self._iterate_parents():
                 if transaction._parent is None or transaction.nested:
                     transaction._rollback_impl()
                     transaction._state = DEACTIVE
+                    boundary = transaction
                     break
                 else:
                     transaction._state = DEACTIVE
 
         sess = self.session
 
-        if self.session._enable_transaction_accounting and \
+        if sess._enable_transaction_accounting and \
                 not sess._is_clean():
+
             # if items were added, deleted, or mutated
             # here, we need to re-restore the snapshot
             util.warn(
                 "Session's state has been changed on "
                 "a non-active transaction - this state "
                 "will be discarded.")
-            self._restore_snapshot(dirty_only=self.nested)
+            boundary._restore_snapshot(dirty_only=boundary.nested)
 
         self.close()
         if self._parent and _capture_exception:
diff --git a/test/orm/test_transaction.py b/test/orm/test_transaction.py
index 538a832a5..6bea5cc7b 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -571,6 +571,35 @@ class SessionTransactionTest(FixtureTest):
         assert u1 in sess
         assert u1 not in sess.deleted
 
+    def test_warning_on_using_inactive_session_rollback_evt(self):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+        sess = Session()
+        u1 = User(id=1, name='u1')
+        sess.add(u1)
+        sess.commit()
+
+        u3 = User(name='u3')
+
+        @event.listens_for(sess, "after_rollback")
+        def evt(s):
+            sess.add(u3)
+
+        sess.add(User(id=1, name='u2'))
+
+        def go():
+            assert_raises(
+                orm_exc.FlushError, sess.flush
+            )
+
+        assert_warnings(go,
+                        ["Session's state has been changed on a "
+                         "non-active transaction - this state "
+                         "will be discarded."],
+                        )
+        assert u3 not in sess
+
     def test_preserve_flush_error(self):
         User = self.classes.User