Assert reset agent always set correctly and is active

Fixed fairly critical issue where the DBAPI connection could be returned to
the connection pool while still in an un-rolled-back state. The reset agent
responsible for rolling back the connection could be corrupted in the case
that the transaction was "closed" without being rolled back or committed,
which can occur in some scenarios when using ORM sessions and emitting
.close() in a certain pattern involving savepoints.   The fix ensures that
the reset agent is always active.

note that the reset agent will go away in 2.0 and the only real
purpose of it is for logging of ROLLBACK.   Apparently with the
SQLite singleton engine in the test suite, there are some strucutral
mismatches in the test fixtures where the reset agent is getting
set differently than the transaction likely due to the same connection
being shared in multiple context, though it's unclear.

Fixes: #5326
Change-Id: If056870ea70a2d9a1749768988d5e023f3061b31

6 files changed, 114 insertions, 5 deletions

diff --git a/doc/build/changelog/unreleased_13/5326.rst b/doc/build/changelog/unreleased_13/5326.rst
new file mode 100644
index 000000000..baebc1890
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/5326.rst
@@ -0,0 +1,12 @@
+.. change::
+    :tags: bug, engine
+    :tickets: 5326
+
+    Fixed fairly critical issue where the DBAPI connection could be returned to
+    the connection pool while still in an un-rolled-back state. The reset agent
+    responsible for rolling back the connection could be corrupted in the case
+    that the transaction was "closed" without being rolled back or committed,
+    which can occur in some scenarios when using ORM sessions and emitting
+    .close() in a certain pattern involving savepoints.   The fix ensures that
+    the reset agent is always active.
+
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index a2066de4a..ed0586cc7 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -689,6 +689,7 @@ class Connection(Connectable):
                 self._autobegin()
             else:
                 self._transaction = RootTransaction(self)
+                self.connection._reset_agent = self._transaction
                 return self._transaction
 
         trans = NestedTransaction(self, self._transaction)
@@ -819,10 +820,27 @@ class Connection(Connectable):
             if trans._is_root:
                 assert trans._parent is trans
                 self._transaction = None
+
+                # test suite w/ SingletonThreadPool will have cases
+                # where _reset_agent is on a different Connection
+                # entirely so we can't assert this here.
+                # if (
+                #    not self._is_future
+                #    and self._still_open_and_connection_is_valid
+                # ):
+                #    assert self.__connection._reset_agent is None
             else:
                 assert trans._parent is not trans
                 self._transaction = trans._parent
 
+                # not doing this assertion for now, however this is how
+                # it would look:
+                # if self._still_open_and_connection_is_valid:
+                #    trans = self._transaction
+                #    while not trans._is_root:
+                #        trans = trans._parent
+                #    assert self.__connection._reset_agent is trans
+
     def _rollback_to_savepoint_impl(
         self, name, context, deactivate_only=False
     ):
@@ -1965,10 +1983,10 @@ class Transaction(object):
         an enclosing transaction.
 
         """
-        if not self._parent.is_active:
-            return
-        if self._parent is self:
+
+        if self._parent.is_active and self._parent is self:
             self.rollback()
+        self.connection._discard_transaction(self)
 
     def rollback(self):
         """Roll back this :class:`.Transaction`.
diff --git a/lib/sqlalchemy/pool/base.py b/lib/sqlalchemy/pool/base.py
index ef4a12248..f20b63cf5 100644
--- a/lib/sqlalchemy/pool/base.py
+++ b/lib/sqlalchemy/pool/base.py
@@ -807,7 +807,15 @@ class _ConnectionFairy(object):
                     ", via agent" if self._reset_agent else "",
                 )
             if self._reset_agent:
-                self._reset_agent.rollback()
+                if not self._reset_agent.is_active:
+                    util.warn(
+                        "Reset agent is not active.  "
+                        "This should not occur unless there was already "
+                        "a connectivity error in progress."
+                    )
+                    pool._dialect.do_rollback(self)
+                else:
+                    self._reset_agent.rollback()
             else:
                 pool._dialect.do_rollback(self)
         elif pool._reset_on_return is reset_commit:
@@ -818,7 +826,15 @@ class _ConnectionFairy(object):
                     ", via agent" if self._reset_agent else "",
                 )
             if self._reset_agent:
-                self._reset_agent.commit()
+                if not self._reset_agent.is_active:
+                    util.warn(
+                        "Reset agent is not active.  "
+                        "This should not occur unless there was already "
+                        "a connectivity error in progress."
+                    )
+                    pool._dialect.do_commit(self)
+                else:
+                    self._reset_agent.commit()
             else:
                 pool._dialect.do_commit(self)
 
diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index 72e0fa186..370536ce9 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -1688,6 +1688,8 @@ class ResetOnReturnTest(PoolTestBase):
             def __init__(self, conn):
                 self.conn = conn
 
+            is_active = True
+
             def rollback(self):
                 self.conn.special_rollback()
 
@@ -1719,6 +1721,8 @@ class ResetOnReturnTest(PoolTestBase):
             def __init__(self, conn):
                 self.conn = conn
 
+            is_active = True
+
             def rollback(self):
                 self.conn.special_rollback()
 
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index 1836b2e74..85f124d12 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -635,6 +635,63 @@ class ResetAgentTest(fixtures.TestBase):
             trans.commit()
             assert connection.connection._reset_agent is None
 
+    def test_trans_close(self):
+        with testing.db.connect() as connection:
+            trans = connection.begin()
+            assert connection.connection._reset_agent is trans
+            trans.close()
+            assert connection.connection._reset_agent is None
+
+    def test_trans_reset_agent_broken_ensure(self):
+        eng = testing_engine()
+        conn = eng.connect()
+        trans = conn.begin()
+        assert conn.connection._reset_agent is trans
+        trans.is_active = False
+
+        with expect_warnings("Reset agent is not active"):
+            conn.close()
+
+    def test_trans_commit_reset_agent_broken_ensure(self):
+        eng = testing_engine(options={"pool_reset_on_return": "commit"})
+        conn = eng.connect()
+        trans = conn.begin()
+        assert conn.connection._reset_agent is trans
+        trans.is_active = False
+
+        with expect_warnings("Reset agent is not active"):
+            conn.close()
+
+    @testing.requires.savepoints
+    def test_begin_nested_trans_close(self):
+        with testing.db.connect() as connection:
+            t1 = connection.begin()
+            assert connection.connection._reset_agent is t1
+            t2 = connection.begin_nested()
+            assert connection.connection._reset_agent is t1
+            assert connection._transaction is t2
+            t2.close()
+            assert connection._transaction is t1
+            assert connection.connection._reset_agent is t1
+            t1.close()
+            assert connection.connection._reset_agent is None
+        assert not t1.is_active
+
+    @testing.requires.savepoints
+    def test_begin_nested_trans_rollback(self):
+        with testing.db.connect() as connection:
+            t1 = connection.begin()
+            assert connection.connection._reset_agent is t1
+            t2 = connection.begin_nested()
+            assert connection.connection._reset_agent is t1
+            assert connection._transaction is t2
+            t2.close()
+            assert connection._transaction is t1
+            assert connection.connection._reset_agent is t1
+            t1.rollback()
+            assert connection.connection._reset_agent is None
+        assert not t1.is_active
+
     @testing.requires.savepoints
     def test_begin_nested_close(self):
         with testing.db.connect() as connection:
diff --git a/test/orm/test_transaction.py b/test/orm/test_transaction.py
index 2b32282ba..78a62199a 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -684,6 +684,7 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
             sess.rollback,
         )
 
+    @testing.requires.independent_connections
     @testing.emits_warning(".*previous exception")
     def test_failed_rollback_deactivates_transaction(self):
         # test #4050
@@ -746,6 +747,7 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
         # outermost is active
         eq_(session.transaction._state, _session.ACTIVE)
 
+    @testing.requires.independent_connections
     @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