Let cancellation escape pool._drain_idle

The helper caught ``BaseException`` and silently discarded it. That
included ``asyncio.CancelledError``, so ``pool.close()`` invoked
inside ``asyncio.timeout()`` or a ``TaskGroup`` would hang past its
deadline while drain kept running — the cancellation was absorbed.

Narrow the catch to ``Exception`` and log the address+error at DEBUG.
Transport-level close failures (``BrokenPipeError``, ``OSError``,
our own ``DqliteConnectionError``) are still absorbed so drain
finishes the remaining idle connections; cancellation and
interpreter-exit signals now propagate. ``_release_reservation``
stays inside the ``finally``, so the per-iteration reservation
invariant is preserved under both paths.

Add regression tests for both angles: ``CancelledError`` from a
connection's close escapes ``_drain_idle`` and still decrements
``_size``; an ``OSError`` is absorbed, subsequent connections are
drained, and a DEBUG log record names the failure.

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

Author: antoineleclair

src/dqliteclient/pool.py

@@ -264,8 +264,18 @@ async def _drain_idle(self) -> None:
                 break
             try:
                 await conn.close()
-            except BaseException:
-                pass
+            except Exception as exc:
+                # Transport-level failures (BrokenPipeError, OSError, our
+                # own DqliteConnectionError) are absorbed so drain can
+                # finish the remaining connections. CancelledError /
+                # KeyboardInterrupt / SystemExit propagate — swallowing
+                # them used to break structured concurrency (``asyncio.
+                # timeout`` around ``pool.close()`` would silently hang).
+                logger.debug(
+                    "pool: close() on idle connection %r failed: %r",
+                    getattr(conn, "_address", "?"),
+                    exc,
+                )
             finally:
                 await self._release_reservation()
 

tests/test_pool.py

@@ -1,5 +1,6 @@
 """Tests for connection pooling."""
 
+import asyncio
 import contextlib
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -1243,3 +1244,62 @@ def test_unexpected_exception_is_not_swallowed(self) -> None:
         conn = self._make_conn(transport=transport, reader=reader)
         with pytest.raises(ValueError, match="broken mock"):
             _socket_looks_dead(conn)
+
+
+class TestDrainIdleCancellation:
+    """``_drain_idle`` must not swallow ``CancelledError`` or any other
+    ``BaseException``. The reservation release on each iteration still runs,
+    so cancellation is clean — but the cancel signal itself must propagate.
+
+    A close() failure for a non-cancellation reason should be absorbed so
+    drain can finish the remaining connections, but logged for diagnostics.
+    """
+
+    async def test_cancelled_error_in_close_propagates(self) -> None:
+        """An async close() raising ``CancelledError`` must escape ``_drain_idle``.
+        Previously the helper's ``except BaseException: pass`` swallowed it,
+        breaking structured concurrency (``asyncio.timeout`` / ``TaskGroup``).
+        """
+        pool = ConnectionPool(["localhost:9001"])
+
+        bad_conn = MagicMock()
+
+        async def cancel_close() -> None:
+            raise asyncio.CancelledError("outer timeout")
+
+        bad_conn.close = cancel_close
+        await pool._pool.put(bad_conn)
+        pool._size = 1
+
+        with pytest.raises(asyncio.CancelledError, match="outer timeout"):
+            await pool._drain_idle()
+
+        # Reservation must still be released on the cancellation path so
+        # the pool is recoverable after the cancel.
+        assert pool._size == 0
+
+    async def test_ordinary_exception_is_absorbed_and_logged(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """An ``OSError`` from close() should be absorbed so drain continues.
+        Behaviour is logged at DEBUG for operator diagnostics.
+        """
+        import logging
+
+        pool = ConnectionPool(["localhost:9001"])
+
+        c1, c2 = MagicMock(), MagicMock()
+        c1._address = "n1:9001"
+        c2._address = "n2:9001"
+        c1.close = AsyncMock(side_effect=OSError("boom"))
+        c2.close = AsyncMock()
+        await pool._pool.put(c1)
+        await pool._pool.put(c2)
+        pool._size = 2
+
+        with caplog.at_level(logging.DEBUG, logger="dqliteclient.pool"):
+            await pool._drain_idle()
+
+        c2.close.assert_awaited_once()
+        assert pool._size == 0
+        assert any("boom" in rec.getMessage() for rec in caplog.records)