Narrow _abort_protocol to suppress only expected drain errors

``contextlib.suppress(BaseException)`` caught CancelledError along with
the TimeoutError from the bounded wait_closed, violating the structured
concurrency contract: an outer ``asyncio.timeout`` scope that cancels
mid-abort would observe the original connect failure instead of its
own cancellation. Narrow to (TimeoutError, OSError) for the expected
slow-drain / already-closed paths and DEBUG-log anything else.

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

Author: antoineleclair

src/dqliteclient/connection.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import contextlib
+import logging
 import math
 from collections.abc import AsyncIterator, Awaitable, Callable, Sequence
 from contextlib import asynccontextmanager
@@ -22,6 +23,8 @@
 from dqlitewire import LEADER_ERROR_CODES as _LEADER_ERROR_CODES
 from dqlitewire.exceptions import EncodeError as _WireEncodeError
 
+logger = logging.getLogger(__name__)
+
 
 def _parse_address(address: str) -> tuple[str, int]:
     """Parse a host:port address string, handling IPv6 brackets."""
@@ -219,8 +222,19 @@ async def _abort_protocol(self) -> None:
             return
         self._protocol = None
         protocol.close()
-        with contextlib.suppress(BaseException):
+        # Narrow the suppression: a bounded wait on the transport drain
+        # can legitimately raise TimeoutError (slow peer) or OSError
+        # (already-closed writer). Anything else — especially
+        # CancelledError from an outer ``asyncio.timeout`` scope — must
+        # propagate so structured-concurrency cancellation semantics
+        # remain intact. DEBUG-log an unexpected Exception for
+        # diagnostics; do not swallow.
+        try:
             await asyncio.wait_for(protocol.wait_closed(), timeout=0.5)
+        except (TimeoutError, OSError):
+            pass
+        except Exception:  # pragma: no cover
+            logger.debug("_abort_protocol: unexpected drain error", exc_info=True)
 
     async def __aenter__(self) -> "DqliteConnection":
         await self.connect()

tests/test_connection.py

@@ -981,3 +981,57 @@ async def task_b():
         )
         assert isinstance(errors[0], InterfaceError)
         assert "transaction" in str(errors[0]).lower()
+
+
+class TestAbortProtocolNarrowSuppression:
+    """_abort_protocol must let CancelledError propagate so an outer
+    ``asyncio.timeout`` scope sees the cancellation instead of being
+    swallowed along with the TimeoutError from the bounded drain.
+    """
+
+    async def test_outer_cancel_propagates_through_abort(self) -> None:
+        import asyncio
+        from unittest.mock import MagicMock
+
+        import pytest
+
+        from dqliteclient.connection import DqliteConnection
+
+        conn = DqliteConnection("localhost:9001", database="x", timeout=1.0)
+        proto = MagicMock()
+        proto.close = MagicMock()
+
+        # wait_closed hangs forever; we wrap the abort in wait_for with
+        # a short outer timeout and assert the outer TimeoutError
+        # propagates (in the previous BaseException-suppressing code
+        # it would have been eaten).
+        async def hang_forever() -> None:
+            await asyncio.sleep(999)
+
+        proto.wait_closed = hang_forever
+        conn._protocol = proto
+
+        with pytest.raises(TimeoutError):
+            await asyncio.wait_for(conn._abort_protocol(), timeout=0.1)
+
+    async def test_timeout_during_drain_is_suppressed(self) -> None:
+        """The bounded wait_closed budget *internally* expiring is
+        expected (slow peer) and must not propagate."""
+        import asyncio
+        from unittest.mock import MagicMock
+
+        from dqliteclient.connection import DqliteConnection
+
+        conn = DqliteConnection("localhost:9001", database="x", timeout=1.0)
+        proto = MagicMock()
+        proto.close = MagicMock()
+
+        async def hang_forever() -> None:
+            await asyncio.sleep(999)
+
+        proto.wait_closed = hang_forever
+        conn._protocol = proto
+
+        # No outer timeout: the internal 0.5s wait_for expires, the
+        # resulting TimeoutError is suppressed, and the call returns.
+        await conn._abort_protocol()