Pin pool acquire late-winner and initialize unqueued-survivor paths

Two timing-sensitive coverage gaps:

- _put_back_or_release_late_winner: the recovery hook for two race
  paths in acquire (outer cancel arrived just after a sibling
  _release put a conn on the queue, post-wait demux else-arm). Pin
  both arms (queue-has-room → put_nowait, queue-full → close +
  release).

- Initialize unqueued-survivor close-failure log: a successful conn
  that fails put_nowait then fails its own close() must emit a
  DEBUG log without disrupting the rest of the cleanup loop.

Both paths were tested only via integration race scenarios. Adding
focused unit tests on the helpers / specific arms catches a
regression at PR time.

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

Author: antoineleclair

tests/test_pool_acquire_late_winner.py

@@ -0,0 +1,53 @@
+"""``ConnectionPool._put_back_or_release_late_winner`` is the recovery
+hook for two race paths in ``acquire``: an outer cancel that arrives
+just after a sibling ``_release`` puts a conn on the queue, and the
+post-wait demux's else-arm where a timeout snapshot raced a winning
+``get_task``. Both paths route through this helper so the conn is
+not silently dropped.
+
+The helper has two arms:
+- queue not full → ``put_nowait`` and the conn is reusable.
+- queue full → close + release reservation (invariant-violation
+  recovery).
+
+Pin both arms.
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from dqliteclient.pool import ConnectionPool
+
+
+@pytest.mark.asyncio
+async def test_late_winner_put_back_when_queue_has_room() -> None:
+    """The happy path: queue has room, conn is put back and the
+    pool's free-conn counter reflects it."""
+    pool = ConnectionPool(addresses=["h:9001"], min_size=0, max_size=4)
+    fake_conn = MagicMock()
+
+    await pool._put_back_or_release_late_winner(fake_conn)
+
+    assert pool._pool.qsize() == 1
+    queued = pool._pool.get_nowait()
+    assert queued is fake_conn
+
+
+@pytest.mark.asyncio
+async def test_late_winner_close_and_release_on_queue_full() -> None:
+    """The recovery path: queue is full, conn is closed and the
+    reservation released so the pool shrinks rather than leaks."""
+    pool = ConnectionPool(addresses=["h:9001"], min_size=0, max_size=1)
+    # Saturate the queue so put_nowait raises QueueFull.
+    filler = MagicMock()
+    pool._pool.put_nowait(filler)
+
+    fake_conn = MagicMock()
+    fake_conn.close = AsyncMock()
+
+    with patch.object(pool, "_release_reservation", new=AsyncMock()) as rel:
+        await pool._put_back_or_release_late_winner(fake_conn)
+
+    fake_conn.close.assert_awaited_once()
+    rel.assert_awaited_once()

tests/test_pool_initialize_unqueued_survivor_close_log.py

@@ -0,0 +1,56 @@
+"""``ConnectionPool.initialize`` walks ``unqueued_survivors`` after a
+mid-init cancel and closes each. If the close itself raises
+``_POOL_CLEANUP_EXCEPTIONS`` (OSError / DqliteConnectionError), the
+``logger.debug`` emit at ``pool.py:623-627`` is the only forensic
+trail.
+
+Without coverage, a regression that drops the ``except`` clause and
+re-raises during cleanup would supplant the user-visible original
+error and the unqueued-survivor cleanup loop would abort partway
+through, leaking subsequent survivors.
+"""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from dqliteclient.pool import ConnectionPool
+
+
+@pytest.mark.asyncio
+async def test_unqueued_survivor_close_failure_debug_logged(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A successful conn whose subsequent close() raises must
+    emit a DEBUG log without disrupting the rest of the cleanup
+    loop."""
+    pool = ConnectionPool(addresses=["h:9001"], min_size=2, max_size=4)
+
+    # Two conns: one that closes cleanly, one that raises OSError.
+    clean_conn = MagicMock()
+    clean_conn.close = AsyncMock()
+    failing_conn = MagicMock()
+    failing_conn.close = AsyncMock(side_effect=OSError("simulated close failure"))
+
+    # _create_connection returns the two conns in order. After they
+    # succeed, force put_nowait to raise so they end up on
+    # unqueued_survivors.
+    create_calls = iter([clean_conn, failing_conn])
+
+    async def stub_create() -> MagicMock:
+        return next(create_calls)
+
+    with (
+        patch.object(pool, "_create_connection", new=stub_create),
+        patch.object(pool._pool, "put_nowait", side_effect=asyncio.QueueFull),
+        caplog.at_level("DEBUG"),
+        pytest.raises(asyncio.QueueFull),
+    ):
+        await pool.initialize()
+
+    # Both close() calls must have been attempted; the OSError on
+    # the second is logged but does not stop the loop.
+    clean_conn.close.assert_awaited()
+    failing_conn.close.assert_awaited()
+    assert any("unqueued-survivor close error" in r.message for r in caplog.records)