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, 63 insertions, 0 deletions

diff --git a/test/ext/asyncio/test_engine_py3k.py b/test/ext/asyncio/test_engine_py3k.py
index 2eebb433d..bce669e4f 100644
--- a/test/ext/asyncio/test_engine_py3k.py
+++ b/test/ext/asyncio/test_engine_py3k.py
@@ -31,6 +31,7 @@ from sqlalchemy.testing import combinations
 from sqlalchemy.testing import config
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
+from sqlalchemy.testing import eq_regex
 from sqlalchemy.testing import expect_raises
 from sqlalchemy.testing import expect_raises_message
 from sqlalchemy.testing import fixtures
@@ -339,6 +340,68 @@ class AsyncEngineTest(EngineFixture):
 
             is_false(t1 == None)
 
+    @testing.variation("simulate_gc", [True, False])
+    def test_appropriate_warning_for_gced_connection(
+        self, async_engine, simulate_gc
+    ):
+        """test #9237 which builds upon a not really complete solution
+        added for #8419."""
+
+        async def go():
+            conn = await async_engine.connect()
+            await conn.begin()
+            await conn.execute(select(1))
+            pool_connection = await conn.get_raw_connection()
+            return pool_connection
+
+        from sqlalchemy.util.concurrency import await_only
+
+        pool_connection = await_only(go())
+
+        rec = pool_connection._connection_record
+        ref = rec.fairy_ref
+        pool = pool_connection._pool
+        echo = False
+
+        if simulate_gc:
+            # not using expect_warnings() here because we also want to do a
+            # negative test for warnings, and we want to absolutely make sure
+            # the thing here that emits the warning is the correct path
+            from sqlalchemy.pool.base import _finalize_fairy
+
+            with mock.patch.object(
+                pool._dialect,
+                "do_rollback",
+                mock.Mock(side_effect=Exception("can't run rollback")),
+            ), mock.patch("sqlalchemy.util.warn") as m:
+
+                _finalize_fairy(
+                    None, rec, pool, ref, echo, transaction_was_reset=False
+                )
+
+            if async_engine.dialect.has_terminate:
+                expected_msg = (
+                    "The garbage collector is trying to clean up.*which will "
+                    "be terminated."
+                )
+            else:
+                expected_msg = (
+                    "The garbage collector is trying to clean up.*which will "
+                    "be dropped, as it cannot be safely terminated."
+                )
+
+            # [1] == .args, not in 3.7
+            eq_regex(m.mock_calls[0][1][0], expected_msg)
+        else:
+            # the warning emitted by the pool is inside of a try/except:
+            # so it's impossible right now to have this warning "raise".
+            # for now, test by using mock.patch
+
+            with mock.patch("sqlalchemy.util.warn") as m:
+                pool_connection.close()
+
+            eq_(m.mock_calls, [])
+
     def test_clear_compiled_cache(self, async_engine):
         async_engine.sync_engine._compiled_cache["foo"] = "bar"
         eq_(async_engine.sync_engine._compiled_cache["foo"], "bar")