Deactivate transaction if rollback fails

Fixed regression introduced in 1.2.0b1 due to :ticket:`3934` where the
:class:`.Session` would fail to "deactivate" the transaction, if a
rollback failed (the target issue is when MySQL loses track of a SAVEPOINT).
This would cause a subsequent call to :meth:`.Session.rollback` to raise
an error a second time, rather than completing and bringing the
:class:`.Session` back to ACTIVE.

Fixes: #4050
Change-Id: Id245e8dd3487cb006b2d6631c8bd513b5ce81abe

4 files changed, 120 insertions, 3 deletions

diff --git a/doc/build/changelog/unreleased_12/4050.rst b/doc/build/changelog/unreleased_12/4050.rst
new file mode 100644
index 000000000..762e5f83d
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/4050.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 4050
+
+    Fixed regression introduced in 1.2.0b1 due to :ticket:`3934` where the
+    :class:`.Session` would fail to "deactivate" the transaction, if a
+    rollback failed (the target issue is when MySQL loses track of a SAVEPOINT).
+    This would cause a subsequent call to :meth:`.Session.rollback` to raise
+    an error a second time, rather than completing and bringing the
+    :class:`.Session` back to ACTIVE.
\ No newline at end of file
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 752f182e5..6ecef17f1 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -495,6 +495,7 @@ class SessionTransaction(object):
                     except:
                         rollback_err = sys.exc_info()
                     finally:
+                        transaction._state = DEACTIVE
                         if self.session._enable_transaction_accounting:
                             transaction._restore_snapshot(
                                 dirty_only=transaction.nested)
diff --git a/lib/sqlalchemy/testing/fixtures.py b/lib/sqlalchemy/testing/fixtures.py
index 8cd6e9f9e..bfd98a407 100644
--- a/lib/sqlalchemy/testing/fixtures.py
+++ b/lib/sqlalchemy/testing/fixtures.py
@@ -223,9 +223,9 @@ class RemovesEvents(object):
     def _event_fns(self):
         return set()
 
-    def event_listen(self, target, name, fn):
+    def event_listen(self, target, name, fn, **kw):
         self._event_fns.add((target, name, fn))
-        event.listen(target, name, fn)
+        event.listen(target, name, fn, **kw)
 
     def teardown(self):
         for key in self._event_fns:
diff --git a/test/orm/test_transaction.py b/test/orm/test_transaction.py
index 374b6a9e5..619e99abd 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -12,7 +12,7 @@ from test.orm._fixtures import FixtureTest
 from sqlalchemy import inspect
 
 
-class SessionTransactionTest(FixtureTest):
+class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
     run_inserts = None
     __backend__ = True
 
@@ -478,6 +478,112 @@ class SessionTransactionTest(FixtureTest):
             "transaction.",
             sess.rollback)
 
+    @testing.emits_warning(".*previous exception")
+    def test_failed_rollback_deactivates_transaction(self):
+        # test #4050
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+        session = Session(bind=testing.db)
+
+        rollback_error = testing.db.dialect.dbapi.InterfaceError(
+            "Can't roll back to savepoint")
+
+        def prevent_savepoint_rollback(
+                cursor, statement, parameters, context=None):
+            if "rollback to savepoint" in statement.lower():
+                raise rollback_error
+
+        self.event_listen(
+            testing.db.dialect,
+            "do_execute", prevent_savepoint_rollback)
+
+        with session.transaction:
+            session.add(User(id=1, name='x'))
+
+        session.begin_nested()
+        # raises IntegrityError on flush
+        session.add(User(id=1, name='x'))
+        assert_raises_message(
+            sa_exc.InterfaceError,
+            "Can't roll back to savepoint",
+            session.commit,
+        )
+
+        # rollback succeeds, because the Session is deactivated
+        eq_(session.transaction._state, _session.DEACTIVE)
+        session.rollback()
+
+        # back to normal
+        eq_(session.transaction._state, _session.ACTIVE)
+
+        trans = session.transaction
+
+        # leave the outermost trans
+        session.rollback()
+
+        # trans is now closed
+        eq_(trans._state, _session.CLOSED)
+
+        # outermost transction is new
+        is_not_(session.transaction, trans)
+
+        # outermost is active
+        eq_(session.transaction._state, _session.ACTIVE)
+
+    @testing.emits_warning(".*previous exception")
+    def test_failed_rollback_deactivates_transaction_ctx_integration(self):
+        # test #4050 in the same context as that of oslo.db
+
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+        session = Session(bind=testing.db, autocommit=True)
+
+        evented_exceptions = []
+        caught_exceptions = []
+
+        def canary(context):
+            evented_exceptions.append(context.original_exception)
+
+        rollback_error = testing.db.dialect.dbapi.InterfaceError(
+            "Can't roll back to savepoint")
+
+        def prevent_savepoint_rollback(
+                cursor, statement, parameters, context=None):
+            if "rollback to savepoint" in statement.lower():
+                raise rollback_error
+
+        self.event_listen(testing.db, "handle_error", canary, retval=True)
+        self.event_listen(
+            testing.db.dialect, "do_execute", prevent_savepoint_rollback)
+
+        with session.begin():
+            session.add(User(id=1, name='x'))
+
+        try:
+            with session.begin():
+                try:
+                    with session.begin_nested():
+                        # raises IntegrityError on flush
+                        session.add(User(id=1, name='x'))
+
+                # outermost is the failed SAVEPOINT rollback
+                # from the "with session.begin_nested()"
+                except sa_exc.DBAPIError as dbe_inner:
+                    caught_exceptions.append(dbe_inner.orig)
+                    raise
+        except sa_exc.DBAPIError as dbe_outer:
+            caught_exceptions.append(dbe_outer.orig)
+
+        is_(
+            type(evented_exceptions[0]),
+            testing.db.dialect.dbapi.IntegrityError
+        )
+        eq_(evented_exceptions[1], rollback_error)
+        eq_(len(evented_exceptions), 2)
+        eq_(caught_exceptions, [rollback_error, rollback_error])
+
     def test_no_prepare_wo_twophase(self):
         sess = create_session(bind=testing.db, autocommit=False)