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

2 files changed, 39 insertions, 5 deletions

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)