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, 40 insertions, 25 deletions

diff --git a/lib/sqlalchemy/pool/base.py b/lib/sqlalchemy/pool/base.py
index a77f65205..d67f32442 100644
--- a/lib/sqlalchemy/pool/base.py
+++ b/lib/sqlalchemy/pool/base.py
@@ -382,7 +382,10 @@ class Pool(log.Identified, event.EventTarget):
                 self._dialect.do_close(connection)
         except BaseException as e:
             self.logger.error(
-                "Exception closing connection %r", connection, exc_info=True
+                f"Exception {'terminating' if terminate else 'closing'} "
+                f"connection %r",
+                connection,
+                exc_info=True,
             )
             if not isinstance(e, Exception):
                 raise
@@ -941,27 +944,32 @@ def _finalize_fairy(
     if is_gc_cleanup:
         assert ref is not None
         _strong_ref_connection_records.pop(ref, None)
-    elif fairy:
-        _strong_ref_connection_records.pop(weakref.ref(fairy), None)
-
-    if is_gc_cleanup:
         assert connection_record is not None
         if connection_record.fairy_ref is not ref:
             return
         assert dbapi_connection is None
         dbapi_connection = connection_record.dbapi_connection
 
+    elif fairy:
+        _strong_ref_connection_records.pop(weakref.ref(fairy), None)
+
     # null pool is not _is_asyncio but can be used also with async dialects
-    dont_restore_gced = (
-        pool._dialect.is_async and not pool._dialect.has_terminate
-    )
+    dont_restore_gced = pool._dialect.is_async
 
     if dont_restore_gced:
         detach = connection_record is None or is_gc_cleanup
-        can_manipulate_connection = ref is None
+        can_manipulate_connection = not is_gc_cleanup
+        can_close_or_terminate_connection = (
+            not pool._dialect.is_async or pool._dialect.has_terminate
+        )
+        requires_terminate_for_close = (
+            pool._dialect.is_async and pool._dialect.has_terminate
+        )
+
     else:
         detach = connection_record is None
-        can_manipulate_connection = True
+        can_manipulate_connection = can_close_or_terminate_connection = True
+        requires_terminate_for_close = False
 
     if dbapi_connection is not None:
         if connection_record and echo:
@@ -992,25 +1000,14 @@ def _finalize_fairy(
                     fairy._pool = pool
                     fairy.detach()
 
-                if can_manipulate_connection:
+                if can_close_or_terminate_connection:
                     if pool.dispatch.close_detached:
                         pool.dispatch.close_detached(dbapi_connection)
 
-                    pool._close_connection(dbapi_connection)
-                else:
-                    message = (
-                        "The garbage collector is trying to clean up "
-                        f"connection {dbapi_connection!r}. This feature is "
-                        "unsupported on asyncio "
-                        'dbapis that lack a "terminate" feature, since no '
-                        "IO can be performed at this stage to "
-                        "reset the connection. Please close out all "
-                        "connections when they are no longer used, calling "
-                        "``close()`` or using a context manager to "
-                        "manage their lifetime."
+                    pool._close_connection(
+                        dbapi_connection,
+                        terminate=requires_terminate_for_close,
                     )
-                    pool.logger.error(message)
-                    util.warn(message)
 
         except BaseException as e:
             pool.logger.error(
@@ -1020,6 +1017,24 @@ def _finalize_fairy(
                 connection_record.invalidate(e=e)
             if not isinstance(e, Exception):
                 raise
+        finally:
+            if detach and is_gc_cleanup and dont_restore_gced:
+                message = (
+                    "The garbage collector is trying to clean up "
+                    f"non-checked-in connection {dbapi_connection!r}, "
+                    f"""which will be {
+                        'dropped, as it cannot be safely terminated'
+                        if not can_close_or_terminate_connection
+                        else 'terminated'
+                    }.  """
+                    "Please ensure that SQLAlchemy pooled connections are "
+                    "returned to "
+                    "the pool explicitly, either by calling ``close()`` "
+                    "or by using appropriate context managers to manage "
+                    "their lifecycle."
+                )
+                pool.logger.error(message)
+                util.warn(message)
 
     if connection_record and connection_record.fairy_ref is not None:
         connection_record.checkin()