do not return asyncio connections to the pool under gc

Repaired a regression caused by the fix for :ticket:`8419` which caused
asyncpg connections to be reset (i.e. transaction ``rollback()`` called)
and returned to the pool normally in the case that the connection were not
explicitly returned to the connection pool and was instead being
intercepted by Python garbage collection, which would fail if the garbage
collection operation were being called outside of the asyncio event loop,
leading to a large amount of stack trace activity dumped into logging
and standard output.

The correct behavior is restored, which is that all asyncio connections
that are garbage collected due to not being explicitly returned to the
connection pool are detached from the pool and discarded, along with a
warning, rather than being returned the pool, as they cannot be reliably
reset. In the case of asyncpg connections, the asyncpg-specific
``terminate()`` method will be used to end the connection more gracefully
within this process as opposed to just dropping it.

This change includes a small behavioral change that is hoped to be useful
for debugging asyncio applications, where the warning that's emitted in the
case of asyncio connections being unexpectedly garbage collected has been
made slightly more aggressive by moving it outside of a ``try/except``
block and into a ``finally:`` block, where it will emit unconditionally
regardless of whether the detach/termination operation succeeded or not. It
will also have the effect that applications or test suites which promote
Python warnings to exceptions will see this as a full exception raise,
whereas previously it was not possible for this warning to actually
propagate as an exception. Applications and test suites which need to
tolerate this warning in the interim should adjust the Python warnings
filter to allow these warnings to not raise.

The behavior for traditional sync connections remains unchanged, that
garbage collected connections continue to be returned to the pool normally
without emitting a warning. This will likely be changed in a future major
release to at least emit a similar warning as is emitted for asyncio
drivers, as it is a usage error for pooled connections to be intercepted by
garbage collection without being properly returned to the pool.

Fixes: #9237
Change-Id: Ib35cfb2e628f2eb2da6d2b65674702556f55603a

1 files changed, 60 insertions, 23 deletions

diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index 4fddcc871..6730d7012 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -23,6 +23,7 @@ from sqlalchemy.testing import assert_raises_context_ok
 from sqlalchemy.testing import assert_warns_message
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import expect_raises
+from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
 from sqlalchemy.testing import is_none
@@ -456,6 +457,13 @@ class PoolEventsTest(PoolTestBase):
         )
         canary = []
 
+        @event.listens_for(p, "reset")
+        def reset(conn, rec, state):
+            canary.append(
+                f"""reset_{'rollback_ok'
+                if state.asyncio_safe else 'no_rollback'}"""
+            )
+
         @event.listens_for(p, "checkin")
         def checkin(*arg, **kw):
             canary.append("checkin")
@@ -668,7 +676,7 @@ class PoolEventsTest(PoolTestBase):
         c1 = p.connect()
         eq_(canary, [])
         c1.close()
-        eq_(canary, ["checkin"])
+        eq_(canary, ["reset_rollback_ok", "checkin"])
 
     def test_reset_event(self):
         p, canary = self._reset_event_fixture()
@@ -728,11 +736,13 @@ class PoolEventsTest(PoolTestBase):
         assert canary.call_args_list[0][0][0] is dbapi_con
         assert canary.call_args_list[0][0][2] is exc
 
-    @testing.combinations((True,), (False,), argnames="is_asyncio")
-    @testing.combinations((True,), (False,), argnames="has_terminate")
+    @testing.variation("is_asyncio", [True, False])
+    @testing.variation("has_terminate", [True, False])
     def test_checkin_event_gc(self, is_asyncio, has_terminate):
+        """tests for #8419, which have been modified for 2.0 in #9237"""
+
         p, canary = self._checkin_event_fixture(
-            _is_asyncio=is_asyncio, _has_terminate=has_terminate
+            _is_asyncio=bool(is_asyncio), _has_terminate=bool(has_terminate)
         )
 
         c1 = p.connect()
@@ -740,18 +750,38 @@ class PoolEventsTest(PoolTestBase):
         dbapi_connection = weakref.ref(c1.dbapi_connection)
 
         eq_(canary, [])
-        del c1
-        lazy_gc()
 
-        detach_gced = is_asyncio and not has_terminate
+        if is_asyncio:
+            if has_terminate:
+                with expect_warnings(
+                    "The garbage collector is trying to clean up.*which will "
+                    "be terminated."
+                ):
+                    del c1
+                    lazy_gc()
+            else:
+                with expect_warnings(
+                    "The garbage collector is trying to clean up.*which will "
+                    "be dropped, as it cannot be safely terminated."
+                ):
+                    del c1
+                    lazy_gc()
+        else:
+            del c1
+            lazy_gc()
+
+        detach_gced = is_asyncio
 
         if detach_gced:
-            # "close_detached" is not called because for asyncio the
-            # connection is just lost.
-            eq_(canary, ["detach"])
+            if has_terminate:
+                eq_(canary, ["reset_no_rollback", "detach", "close_detached"])
+            else:
+                # "close_detached" is not called because for asyncio without
+                # terminate the connection is just lost.
+                eq_(canary, ["reset_no_rollback", "detach"])
 
         else:
-            eq_(canary, ["checkin"])
+            eq_(canary, ["reset_rollback_ok", "checkin"])
 
         gc_collect()
         if detach_gced:
@@ -769,10 +799,13 @@ class PoolEventsTest(PoolTestBase):
         eq_(canary, [])
 
         c1.close()
-        eq_(canary, ["checkin"])
+        eq_(canary, ["reset_rollback_ok", "checkin"])
 
         c2.close()
-        eq_(canary, ["checkin", "checkin"])
+        eq_(
+            canary,
+            ["reset_rollback_ok", "checkin", "reset_rollback_ok", "checkin"],
+        )
 
     def test_listen_targets_scope(self):
         canary = []
@@ -1686,28 +1719,32 @@ class QueuePoolTest(PoolTestBase):
                 raise tsa.exc.DisconnectionError()
 
         conn = pool.connect()
-        old_dbapi_conn = conn.dbapi_connection
+        normally_closed_dbapi_conn = conn.dbapi_connection
         conn.close()
 
-        eq_(old_dbapi_conn.mock_calls, [call.rollback()])
+        eq_(normally_closed_dbapi_conn.mock_calls, [call.rollback()])
 
-        old_dbapi_conn.boom = "yes"
+        normally_closed_dbapi_conn.boom = "yes"
 
         conn = pool.connect()
-        dbapi_conn = conn.dbapi_connection
+
+        # normally closed conn was checked out again but had a problem,
+        # so was replaced
+        eq_(
+            normally_closed_dbapi_conn.mock_calls,
+            [call.rollback(), call.close()],
+        )
+
+        not_closed_dbapi_conn = conn.dbapi_connection
         del conn
         gc_collect()
 
         if detach_gced:
             # new connection was detached + abandoned on return
-            eq_(dbapi_conn.mock_calls, [])
+            eq_(not_closed_dbapi_conn.mock_calls, [])
         else:
             # new connection reset and returned to pool
-            eq_(dbapi_conn.mock_calls, [call.rollback()])
-
-        # old connection was just closed - did not get an
-        # erroneous reset on return
-        eq_(old_dbapi_conn.mock_calls, [call.rollback(), call.close()])
+            eq_(not_closed_dbapi_conn.mock_calls, [call.rollback()])
 
     @testing.requires.timing_intensive
     def test_recycle_pool_no_race(self):