True connection floor for the pool

asyncpg/_testbase/__init__.py

@@ -267,6 +267,7 @@ def _shutdown_cluster(cluster):
 
 
 def create_pool(dsn=None, *,
+                init_size=None,
                 min_size=10,
                 max_size=10,
                 max_queries=50000,
@@ -281,6 +282,7 @@ def create_pool(dsn=None, *,
                 **connect_kwargs):
     return pool_class(
         dsn,
+        init_size=init_size,
         min_size=min_size,
         max_size=max_size,
         max_queries=max_queries,
@@ -366,10 +368,16 @@ def setUp(self):
         self._pools = []
 
     def tearDown(self):
-        super().tearDown()
+        maintenance_tasks = []
         for pool in self._pools:
             pool.terminate()
+            if pool._maintenance_task is not None:
+                maintenance_tasks.append(pool._maintenance_task)
+        if maintenance_tasks:
+            self.loop.run_until_complete(asyncio.gather(
+                *maintenance_tasks, return_exceptions=True))
         self._pools = []
+        super().tearDown()
 
     def create_pool(self, pool_class=pg_pool.Pool,
                     connection_class=pg_connection.Connection, **kwargs):

asyncpg/connect_utils.py

@@ -1158,7 +1158,16 @@ async def __connect_addr(
     else:
         connector = loop.create_connection(proto_factory, *addr)
 
-    tr, pr = await connector
+    try:
+        tr, pr = await connector
+    except (Exception, asyncio.CancelledError):
+        # The protocol can exist before create_connection() returns.  If
+        # that operation is cancelled, nobody will await authentication.
+        if not connected.done():
+            connected.cancel()
+        elif not connected.cancelled():
+            connected.exception()
+        raise
 
     try:
         await connected

asyncpg/connection.py

@@ -54,6 +54,7 @@ class Connection(metaclass=ConnectionMeta):
                  '_intro_query', '_reset_query', '_proxy',
                  '_stmt_exclusive_section', '_config', '_params', '_addr',
                  '_log_listeners', '_termination_listeners', '_cancellations',
+                 '_pool_holder',
                  '_source_traceback', '_query_loggers', '__weakref__')
 
     def __init__(self, protocol, transport, loop,
@@ -105,6 +106,7 @@ def __init__(self, protocol, transport, loop,
 
         self._reset_query = None
         self._proxy = None
+        self._pool_holder = None
 
         # Used to serialize operations that might involve anonymous
         # statements.  Specifically, we want to make the following
@@ -134,6 +136,10 @@ def __del__(self):
 
             warnings.warn(msg, ResourceWarning)
             if not self._loop.is_closed():
+                # A weak holder reference may still be live during GC.
+                # Finalization must not notify the pool and restart
+                # maintenance.
+                self._pool_holder = None
                 self.terminate()
 
     async def add_listener(self, channel, callback):
@@ -1599,10 +1605,11 @@ def _cleanup(self):
         # Free the resources associated with this connection.
         # This must be called when a connection is terminated.
 
-        if self._proxy is not None:
-            # Connection is a member of a pool, so let the pool
-            # know that this connection is dead.
-            self._proxy._holder._release_on_close()
+        if self._pool_holder is not None:
+            # Idle connections have no proxy, but still belong to a holder.
+            holder, self._pool_holder = self._pool_holder(), None
+            if holder is not None and holder._con is self:
+                holder._release_on_close()
 
         self._mark_stmts_as_closed()
         self._listeners.clear()