Clear _bound_loop on _invalidate; clear _pool_released before pool-side close; narrow close_impl pending-drain suppress

Three client-side correctness fixes:

- ``_invalidate`` cleared every transaction-state field but did
  not clear ``self._bound_loop``. After invalidation on loop A,
  a reconnect on loop B was rejected by ``_check_in_use``'s
  cross-loop guard even though every other state field was
  scrubbed and the connection was otherwise ready for fresh
  use. Three of the four lifecycle scrub sites already cleared
  ``_bound_loop``; this completes the parity.

- ``ConnectionPool.acquire``'s liveness short-circuit and
  ``_drain_idle`` both called ``conn.close()`` on connections
  carrying ``_pool_released=True`` (set by the prior ``_release``).
  ``DqliteConnection.close()`` early-returns when ``_pool_released``
  is True so user-side close on a checked-in conn is a no-op —
  but the same guard silently absorbed the pool-side cleanup
  close, leaking writer / reader pairs on FIN-seen idle conns
  and on ``pool.close()`` cascades. Clear the flag immediately
  before the close call at both sites.

- ``DqliteConnection._close_impl``'s pending-drain await wrapped
  the await in ``contextlib.suppress(BaseException)``. The
  in-source comment claimed parity with ``connect()``'s
  pending-retire path, but connect actually uses NARROW catches
  (CancelledError + Exception). asyncio cancellation gets
  re-delivered at the next await boundary; signal-driven
  KeyboardInterrupt / SystemExit are one-shot and were lost
  forever if absorbed here. Narrow the suppress to
  ``(Exception, asyncio.CancelledError)``.

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

Author: antoineleclair

src/dqliteclient/connection.py

@@ -1217,15 +1217,17 @@ async def _close_impl(self) -> None:
             # the second caller's resumption — verified by code
             # review, not coverage.
             #
-            # Suppress ``BaseException`` (not just ``Exception``) so a
-            # ``CancelledError`` delivered during ``await pending``
-            # does not propagate out of close() before the protocol
-            # tear-down at lines below runs — leaking ``_protocol``.
-            # Symmetric with ``connect()``'s pending-retire path which
-            # already uses ``BaseException``. The user's outer cancel
-            # gets re-delivered at the next await boundary inside
-            # close (e.g., ``writer.wait_closed`` below).
-            with contextlib.suppress(BaseException):
+            # Narrow suppress to ``(Exception, asyncio.CancelledError)``
+            # so KeyboardInterrupt and SystemExit propagate. Cycle 23
+            # noted that the previous ``BaseException`` suppress was
+            # mis-justified as "symmetric with connect()" — connect's
+            # pending-retire path actually uses NARROW catches
+            # (``except asyncio.CancelledError`` + ``except Exception``).
+            # asyncio cancellation gets re-delivered at the next await
+            # boundary (Python 3.11+ ``Task.cancelling()`` re-arm
+            # contract), but signal-driven KI/SE are one-shot and
+            # would be lost forever if absorbed here.
+            with contextlib.suppress(Exception, asyncio.CancelledError):
                 await pending
         # Mirror ``_invalidate``'s atomic clear of the transaction
         # bookkeeping. Without this, a raw ``BEGIN`` followed by an
@@ -1489,6 +1491,14 @@ async def _bounded_drain() -> None:
         self._savepoint_stack.clear()
         self._savepoint_implicit_begin = False
         self._has_untracked_savepoint = False
+        # Clear the loop binding so a subsequent ``connect()`` after
+        # invalidation can bind to a different loop without tripping
+        # the ``_check_in_use`` cross-loop guard. ``_close_impl``
+        # already clears this; without it here, an invalidated
+        # connection (where every other state field has been
+        # scrubbed) is still rejected as "bound to a different
+        # event loop" on a fresh-loop reconnect.
+        self._bound_loop = None
         # Preserve the FIRST cause: ``_ensure_connected`` raises a
         # synthetic ``DqliteConnectionError("Not connected")`` chained
         # from ``self._invalidation_cause`` whenever an

src/dqliteclient/pool.py

@@ -601,6 +601,15 @@ async def _drain_idle(self) -> None:
                     # Verified by code review, not coverage.
                     break
                 try:
+                    # Clear ``_pool_released`` BEFORE close so the
+                    # close path actually runs. The release flag is
+                    # set when the conn was put back into the queue
+                    # in ``_release``; ``DqliteConnection.close()``
+                    # early-returns on True (user-side close on a
+                    # checked-in conn must be a no-op). Without this
+                    # clear, the pool's drain-side close would be
+                    # silently absorbed and the writer would leak.
+                    conn._pool_released = False
                     # Shield each per-connection close against an outer
                     # ``asyncio.timeout(pool.close())``. Without the shield,
                     # a cancel that lands mid-``wait_closed`` propagates out
@@ -867,6 +876,14 @@ async def acquire(self) -> AsyncIterator[DqliteConnection]:
             # never-connected conn (``_protocol is None``) — it short-
             # circuits to a noop. Shielded so an outer cancel does not
             # leave the writer dangling.
+            #
+            # Clear ``_pool_released`` BEFORE close so the close
+            # actually runs. ``DqliteConnection.close()`` early-returns
+            # when ``_pool_released`` is True (so user-side close on a
+            # checked-in conn is a no-op); without this clear, the
+            # pool-side close on a dead conn we just dequeued would
+            # be silently absorbed and the writer would leak.
+            conn._pool_released = False
             with contextlib.suppress(*_POOL_CLEANUP_EXCEPTIONS):
                 await asyncio.shield(conn.close())
             # Wrap _drain_idle so a cancel mid-drain releases the dead

tests/test_refresh_pid_cache_contract.py

@@ -60,10 +60,11 @@ def test_refresh_pid_cache_is_callable_zero_arg() -> None:
     registration's call shape) is caught."""
     saved = conn_mod._current_pid
     try:
-        result = conn_mod._refresh_pid_cache()
+        # ``register_at_fork`` calls the function with no arguments;
+        # this raises TypeError if a refactor adds a parameter.
+        conn_mod._refresh_pid_cache()
     finally:
         conn_mod._current_pid = saved
-    assert result is None  # PEP 257-style: side-effect function returns None
 
 
 def test_after_in_child_fork_actually_refreshes_cache() -> None: