Avoid leaking connections if _can_use_connection fails (#1269)

If _can_use_connection fails (say, because of an asyncio timeout), then we may have a full connection that has been created but would be leaked by this function. (note: `_connect_addr` takes care of cleaning up after itself if it fails partway)

This is particularly possible in the case of pgbouncer, where we may succeed at establishing a connection much quicker than even a trivial call to the backing database would take.

I believe this failure mode was introduced in #987

I'm not certain if we should `await` here or just punt the closing of the other connections to a background task (I don't know how risky `close` is on an established connection).

Author: yuliy-openai

asyncpg/connect_utils.py

@@ -1212,30 +1212,36 @@ async def _connect(*, loop, connection_class, record_class, **kwargs):
     candidates = []
     chosen_connection = None
     last_error = None
-    for addr in addrs:
-        try:
-            conn = await _connect_addr(
-                addr=addr,
-                loop=loop,
-                params=params,
-                config=config,
-                connection_class=connection_class,
-                record_class=record_class,
-            )
-            candidates.append(conn)
-            if await _can_use_connection(conn, target_attr):
-                chosen_connection = conn
-                break
-        except OSError as ex:
-            last_error = ex
-    else:
-        if target_attr == SessionAttribute.prefer_standby and candidates:
-            chosen_connection = random.choice(candidates)
+    try:
+        for addr in addrs:
+            try:
+                conn = await _connect_addr(
+                    addr=addr,
+                    loop=loop,
+                    params=params,
+                    config=config,
+                    connection_class=connection_class,
+                    record_class=record_class,
+                )
+                candidates.append(conn)
+                if await _can_use_connection(conn, target_attr):
+                    chosen_connection = conn
+                    break
+            except OSError as ex:
+                last_error = ex
+        else:
+            if target_attr == SessionAttribute.prefer_standby and candidates:
+                chosen_connection = random.choice(candidates)
+    finally:
 
-    await asyncio.gather(
-        *(c.close() for c in candidates if c is not chosen_connection),
-        return_exceptions=True
-    )
+        async def _close_candidates(conns, chosen):
+            await asyncio.gather(
+                *(c.close() for c in conns if c is not chosen),
+                return_exceptions=True
+            )
+        if candidates:
+            asyncio.create_task(
+                _close_candidates(candidates, chosen_connection))
 
     if chosen_connection:
         return chosen_connection