Walk gather's child tasks in initialize's finally so cancel-mid-gather doesn't leak completed conns

``ConnectionPool.initialize`` opened ``min_size`` connections
concurrently via ``asyncio.gather(...)``. The post-gather
result-iteration loop populated the ``successes`` list and
mirrored it into ``unqueued_survivors`` so the finally could
walk and close any conn that didn't make it into the pool
queue.

If the calling task was cancelled WHILE gather was still
running (outer ``asyncio.timeout``, TaskGroup sibling
failure, ``task.cancel()``), gather raised CancelledError
out of the await. Control jumped from the await line
directly to the finally — skipping the result-iteration
loop AND the ``unqueued_survivors = list(successes)``
assignment. The finally's close sweep then iterated the
default empty list, leaking every conn that had completed
before the cancel. Each leaked conn carried a live TCP
socket bound to the leader plus an asyncio
StreamReaderProtocol task pinned by the conn's _protocol
ref — the GC-eventually-warns path with no exception path
to surface the leak in tests.

The ``_size`` accounting was already correct (ISSUE-240
moved that into the finally). The connection-close sweep
now gets the same treatment: build the child tasks
explicitly with ``asyncio.create_task`` so the finally
has a handle on every child regardless of where control
left.

Gated on ``gather_returned``: when gather completed
normally the existing pop-based discipline already
tracked the un-queued tail precisely; walking tasks
again would re-add already-queued conns and double-close.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/dqliteclient/pool.py

@@ -350,13 +350,44 @@ async def initialize(self) -> None:
                 # and closes the unqueued survivors.
                 unqueued = self._min_size
                 unqueued_survivors: list[DqliteConnection] = []
+                # Build child tasks explicitly (not via the ``gather`` *
+                # expression form) so a CancelledError raised out of
+                # ``gather`` itself — outer ``asyncio.timeout`` / a
+                # TaskGroup sibling failure / ``task.cancel()`` —
+                # leaves the finally with a handle on every child's
+                # ``_result``. The expression form discards the
+                # already-completed children's results into the
+                # gather's local frame; once gather raises, those
+                # results are unreachable and the live conns are
+                # orphaned (transports leaked until GC).
+                #
+                # This is the symmetric resource-discipline fix to
+                # the ``_size`` accounting fix (ISSUE-240) — that fix
+                # moved the bookkeeping into the finally; the
+                # connection-close sweep now does the same by
+                # walking the explicit task list.
+                create_tasks: list[asyncio.Task[DqliteConnection]] = [
+                    asyncio.create_task(self._create_connection())
+                    for _ in range(self._min_size)
+                ]
+                # Track whether ``gather`` returned normally; if it
+                # raised CancelledError, the post-gather assignment
+                # to ``unqueued_survivors`` (and the put-loop's pop)
+                # never ran, so the finally must walk the explicit
+                # task list to recover completed children's
+                # results. When gather DID return normally, the
+                # existing pop-based discipline already tracked the
+                # un-queued tail precisely; walking tasks again
+                # would re-add already-queued conns and double-close.
+                gather_returned = False
                 try:
                     # Create min_size connections concurrently so startup
                     # latency doesn't scale with min_size × per-connect RTT.
                     results = await asyncio.gather(
-                        *(self._create_connection() for _ in range(self._min_size)),
+                        *create_tasks,
                         return_exceptions=True,
                     )
+                    gather_returned = True
                     successes: list[DqliteConnection] = []
                     failures: list[BaseException] = []
                     for r in results:
@@ -447,6 +478,32 @@ async def initialize(self) -> None:
                     # but never into the queue. Under a clean success
                     # this list is empty; on partial put-loop cancel it
                     # holds the unqueued tail.
+                    #
+                    # Cancel-during-gather recovery: ``gather``
+                    # propagates CancelledError when its caller is
+                    # cancelled, but children that already completed
+                    # have their results sitting in their tasks'
+                    # ``_result`` slot. The post-gather assignment to
+                    # ``unqueued_survivors`` did not run, so this
+                    # loop iterates an empty list. Walk
+                    # ``create_tasks`` to recover those results.
+                    # Already-failed / not-yet-done / cancelled tasks
+                    # are skipped — only successful results need
+                    # closing here. Only walks the task list if
+                    # ``gather`` raised before returning; once the
+                    # post-gather code ran, the pop-based
+                    # discipline already tracked the un-queued tail
+                    # precisely (walking again would double-close
+                    # conns already in the queue).
+                    if not gather_returned:
+                        for t in create_tasks:
+                            if not t.done() or t.cancelled():
+                                continue
+                            try:
+                                r = t.result()
+                            except BaseException:
+                                continue
+                            unqueued_survivors.append(r)
                     for conn in unqueued_survivors:
                         try:
                             await conn.close()

tests/test_pool_cancellation.py

@@ -169,6 +169,73 @@ async def _flaky_connect(**kwargs: Any) -> _FakeConn:
         )
 
 
+class TestInitializeCancelDuringGatherDoesNotLeakCompletedConns:
+    """An outer cancel landing while ``asyncio.gather`` inside
+    ``initialize`` still has children pending must close the
+    children that already completed. Today the
+    ``unqueued_survivors`` list is populated AFTER gather returns,
+    so a CancelledError raised out of gather skips the assignment
+    and the finally iterates an empty list — leaking every
+    completed conn's transport.
+    """
+
+    async def test_initialize_cancel_during_gather_closes_completed_conns(self) -> None:
+        completed: list[_FakeConn] = []
+        call_idx = [0]
+        completed_event = asyncio.Event()
+
+        async def _half_fast_half_slow(**kwargs: Any) -> _FakeConn:
+            i = call_idx[0]
+            call_idx[0] += 1
+            if i < 5:
+                # Fast path: completes before the timeout fires.
+                c = _FakeConn(name=f"c{i}")
+                completed.append(c)
+                if len(completed) == 5:
+                    completed_event.set()
+                return c
+            # Slow path: hangs forever (cancelled by the timeout).
+            await asyncio.sleep(60)
+            raise RuntimeError("unreachable")
+
+        cluster = MagicMock(spec=ClusterClient)
+        cluster.connect = _half_fast_half_slow
+        pool = ConnectionPool(
+            addresses=["localhost:9001"],
+            min_size=10,
+            max_size=10,
+            timeout=5.0,
+            cluster=cluster,
+        )
+
+        async def _wrapped() -> None:
+            await pool.initialize()
+
+        # Drive the timeout / cancel while the slow children are
+        # still pending. Wait for the fast 5 to complete first so
+        # the leak is deterministic.
+        init_task = asyncio.create_task(_wrapped())
+        await completed_event.wait()
+        await asyncio.sleep(0.01)  # let gather observe the 5 results
+        init_task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await init_task
+
+        # Every conn that completed before the cancel must have
+        # been effectively closed. Today (without the fix), the
+        # unqueued_survivors list is empty in the finally because
+        # the assignment happens AFTER gather returns.
+        unclosed = [c for c in completed if not c.close_effective]
+        assert not unclosed, (
+            f"Cancel-during-gather leaked {len(unclosed)} of {len(completed)} "
+            f"completed conns: {[c.name for c in unclosed]}. The pool's "
+            "finally must walk gather's child tasks (not just the post-gather "
+            "successes list) to reach completed-but-unqueued conns."
+        )
+        # _size must also be drained (the existing ISSUE-240 fix).
+        assert pool._size == 0
+
+
 class TestAcquireCancellationRestoresSize:
     """A caller cancelled while parked in acquire() must not leak the
     pool reservation. If a connection was pulled off the queue and