@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug, asyncio
+    :tickets: 8419
+
+    Integrated support for asyncpg's ``terminate()`` method call for cases
+    where the connection pool is recycling a possibly timed-out connection,
+    where a connection is being garbage collected that wasn't gracefully
+    closed, as well as when the connection has been invalidated. This allows
+    asyncpg to abandon the connection without waiting for a response that may
+    incur long timeouts.
@@ -748,6 +748,9 @@ def close(self):
 
         self.await_(self._connection.close())
 
+    def terminate(self):
+        self._connection.terminate()
+
 
 class AsyncAdaptFallback_asyncpg_connection(AsyncAdapt_asyncpg_connection):
     __slots__ = ()
@@ -891,6 +894,7 @@ class PGDialect_asyncpg(PGDialect):
     supports_server_side_cursors = True
 
     supports_unicode_binds = True
+    has_terminate = True
 
     default_paramstyle = "format"
     supports_sane_multi_rowcount = False
@@ -987,6 +991,9 @@ def set_deferrable(self, connection, value):
     def get_deferrable(self, connection):
         return connection.deferrable
 
+    def do_terminate(self, dbapi_connection) -> None:
+        dbapi_connection.terminate()
+
     def create_connect_args(self, url):
         opts = url.translate_connect_args(username="user")
 
 
@@ -231,6 +231,7 @@ class DefaultDialect(interfaces.Dialect):
     CACHING_DISABLED = CACHING_DISABLED
     NO_CACHE_KEY = NO_CACHE_KEY
     NO_DIALECT_SUPPORT = NO_DIALECT_SUPPORT
+    has_terminate = False
 
     @util.deprecated_params(
         convert_unicode=(
@@ -684,6 +685,9 @@ def do_rollback(self, dbapi_connection):
     def do_commit(self, dbapi_connection):
         dbapi_connection.commit()
 
+    def do_terminate(self, dbapi_connection):
+        self.do_close(dbapi_connection)
+
     def do_close(self, dbapi_connection):
         dbapi_connection.close()
 
 
@@ -583,6 +583,23 @@ def do_commit(self, dbapi_connection):
 
         raise NotImplementedError()
 
+    def do_terminate(self, dbapi_connection):
+        """Provide an implementation of ``connection.close()`` that tries as
+        much as possible to not block, given a DBAPI
+        connection.
+
+        In the vast majority of cases this just calls .close(), however
+        for some asyncio dialects may call upon different API features.
+
+        This hook is called by the :class:`_pool.Pool`
+        when a connection is being recycled or has been invalidated.
+
+        .. versionadded:: 1.4.41
+
+        """
+
+        raise NotImplementedError()
+
     def do_close(self, dbapi_connection):
         """Provide an implementation of ``connection.close()``, given a DBAPI
         connection.
 
@@ -36,13 +36,17 @@ class _ConnDialect(object):
     """
 
     is_async = False
+    has_terminate = False
 
     def do_rollback(self, dbapi_connection):
         dbapi_connection.rollback()
 
     def do_commit(self, dbapi_connection):
         dbapi_connection.commit()
 
+    def do_terminate(self, dbapi_connection):
+        dbapi_connection.close()
+
     def do_close(self, dbapi_connection):
         dbapi_connection.close()
 
@@ -240,11 +244,17 @@ def _should_wrap_creator(self, creator):
         else:
             return lambda crec: creator()
 
-    def _close_connection(self, connection):
-        self.logger.debug("Closing connection %r", connection)
-
+    def _close_connection(self, connection, terminate=False):
+        self.logger.debug(
+            "%s connection %r",
+            "Hard-closing" if terminate else "Closing",
+            connection,
+        )
         try:
-            self._dialect.do_close(connection)
+            if terminate:
+                self._dialect.do_terminate(connection)
+            else:
+                self._dialect.do_close(connection)
         except Exception:
             self.logger.error(
                 "Exception closing connection %r", connection, exc_info=True
@@ -584,7 +594,7 @@ def invalidate(self, e=None, soft=False):
         if soft:
             self._soft_invalidate_time = time.time()
         else:
-            self.__close()
+            self.__close(terminate=True)
             self.dbapi_connection = None
 
     def get_connection(self):
@@ -630,7 +640,7 @@ def get_connection(self):
             recycle = True
 
         if recycle:
-            self.__close()
+            self.__close(terminate=True)
             self.info.clear()
 
             self.__connect()
@@ -643,11 +653,13 @@ def _is_hard_or_soft_invalidated(self):
             or (self._soft_invalidate_time > self.starttime)
         )
 
-    def __close(self):
+    def __close(self, terminate=False):
         self.finalize_callback.clear()
         if self.__pool.dispatch.close:
             self.__pool.dispatch.close(self.dbapi_connection, self)
-        self.__pool._close_connection(self.dbapi_connection)
+        self.__pool._close_connection(
+            self.dbapi_connection, terminate=terminate
+        )
         self.dbapi_connection = None
 
     def __connect(self):
@@ -709,7 +721,9 @@ def _finalize_fairy(
         dbapi_connection = connection_record.dbapi_connection
 
     # null pool is not _is_asyncio but can be used also with async dialects
-    dont_restore_gced = pool._dialect.is_async
+    dont_restore_gced = (
+        pool._dialect.is_async and not pool._dialect.has_terminate
+    )
 
     if dont_restore_gced:
         detach = not connection_record or ref
@@ -751,8 +765,10 @@ def _finalize_fairy(
                 else:
                     message = (
                         "The garbage collector is trying to clean up "
-                        "connection %r. This feature is unsupported on async "
-                        "dbapi, since no IO can be performed at this stage to "
+                        "connection %r. This feature is unsupported on "
+                        "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 "
 
@@ -468,7 +468,7 @@ def _test_queuepool(self, q, dispose=True):
                 "Connection %r checked out from pool",
                 "Connection %r being returned to pool%s",
                 "Connection %s rollback-on-return",
-                "Closing connection %r",
+                "%s connection %r",
             ]
             + (["Pool disposed. %s"] if dispose else []),
         )
 
@@ -92,10 +92,13 @@ def _queuepool_fixture(self, **kw):
     def _queuepool_dbapi_fixture(self, **kw):
         dbapi = MockDBAPI()
         _is_asyncio = kw.pop("_is_asyncio", False)
+        _has_terminate = kw.pop("_has_terminate", False)
         p = pool.QueuePool(creator=lambda: dbapi.connect("foo.db"), **kw)
         if _is_asyncio:
             p._is_asyncio = True
             p._dialect = _AsyncConnDialect()
+        if _has_terminate:
+            p._dialect.has_terminate = True
         return dbapi, p
 
 
@@ -468,8 +471,10 @@ def checkout(*arg, **kw):
 
         return p, canary
 
-    def _checkin_event_fixture(self, _is_asyncio=False):
-        p = self._queuepool_fixture(_is_asyncio=_is_asyncio)
+    def _checkin_event_fixture(self, _is_asyncio=False, _has_terminate=False):
+        p = self._queuepool_fixture(
+            _is_asyncio=_is_asyncio, _has_terminate=_has_terminate
+        )
         canary = []
 
         @event.listens_for(p, "checkin")
@@ -744,9 +749,13 @@ def test_invalidate_event_exception(self):
         assert canary.call_args_list[0][0][0] is dbapi_con
         assert canary.call_args_list[0][0][2] is exc
 
-    @testing.combinations((True, testing.requires.python3), (False,))
-    def test_checkin_event_gc(self, detach_gced):
-        p, canary = self._checkin_event_fixture(_is_asyncio=detach_gced)
+    @testing.combinations((True,), (False,), argnames="is_asyncio")
+    @testing.combinations((True,), (False,), argnames="has_terminate")
+    @testing.requires.python3
+    def test_checkin_event_gc(self, is_asyncio, has_terminate):
+        p, canary = self._checkin_event_fixture(
+            _is_asyncio=is_asyncio, _has_terminate=has_terminate
+        )
 
         c1 = p.connect()
 
@@ -756,6 +765,8 @@ def test_checkin_event_gc(self, detach_gced):
         del c1
         lazy_gc()
 
+        detach_gced = is_asyncio and not has_terminate
+
         if detach_gced:
             # "close_detached" is not called because for asyncio the
             # connection is just lost.
Original file line number	Diff line number	Diff line change
`@@ -468,7 +468,7 @@ def _test_queuepool(self, q, dispose=True):`
`468`	`468`	`"Connection %r checked out from pool",`
`469`	`469`	`"Connection %r being returned to pool%s",`
`470`	`470`	`"Connection %s rollback-on-return",`
`471`		`- "Closing connection %r",`
	`471`	`+ "%s connection %r",`
`472`	`472`	`]`
`473`	`473`	`+ (["Pool disposed. %s"] if dispose else []),`
`474`	`474`	`)`