Pin cycle 22's _invalidate cancel-and-detach prior pending_drain (and fix the cause-shadowing bug surfaced by the pin)

Cycle 22 added two new behaviours to ``_invalidate`` that
were never regression-tested:

1. Cancel-and-detach the prior ``_pending_drain`` task
   before overwriting the slot. A second ``_invalidate``
   while the first task was still in-flight previously
   orphaned the first one, recreating "Task was destroyed
   but it is pending" at GC.
2. Try/except around ``loop.create_task`` so a
   ``RuntimeError("Event loop is closed")`` during
   ``engine.dispose()`` does not replace the original
   cancel/cause with a bare RuntimeError.

Adding the test for #2 exposed a sibling bug: the except
clause was ``as cause``, which Python deletes at end-of-
block. The function parameter is also named ``cause``, so
the name was unbound when the later ``if cause is not None
and self._invalidation_cause is None`` check ran on the
closed-loop arm — surfacing as ``UnboundLocalError`` instead
of preserving the original cause as the cycle 22 commit
intended. Rename the except binding to ``schedule_err`` so
the parameter is preserved for the cause-recording branch.

Add a third pin for cycle 22's clear of
``_invalidation_cause`` in ``_close_impl`` so the cached
exception does not pin frame globals/locals across close
→ reconnect cycles (a sibling concern from
ISSUE-1038).

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

Author: antoineleclair

src/dqliteclient/connection.py

@@ -1456,18 +1456,26 @@ async def _bounded_drain() -> None:
                 # close() awaits it.
                 try:
                     self._pending_drain = loop.create_task(_bounded_drain())
-                except RuntimeError as cause:
+                except RuntimeError as schedule_err:
                     # ``loop.create_task`` raises
                     # RuntimeError("Event loop is closed") if the
                     # loop has been stopped — a real shape during
                     # interpreter shutdown / engine.dispose() races.
                     # Don't replace the original cancel/cause with a
                     # bare RuntimeError; log and move on with no
                     # pending drain.
+                    #
+                    # Bind to ``schedule_err`` (NOT ``cause``) so the
+                    # except's ``as`` does not shadow the function
+                    # parameter ``cause`` and leave it unbound for
+                    # the ``cause is not None`` check below — Python
+                    # deletes ``except as`` bindings at end-of-block,
+                    # so reusing the name silently breaks a sibling
+                    # reference further down.
                     logger.debug(
                         "Connection._invalidate: loop.create_task raised %s "
                         "while scheduling _bounded_drain; original cause preserved",
-                        type(cause).__name__,
+                        type(schedule_err).__name__,
                         exc_info=True,
                     )
                     self._pending_drain = None

tests/test_invalidate_cancel_prior_pending_drain.py

@@ -0,0 +1,149 @@
+"""Pin: cycle 22's cancel-and-detach guard on
+``DqliteConnection._invalidate``.
+
+A second ``_invalidate`` call while a prior bounded-drain task is
+in-flight must cancel the prior task before overwriting
+``self._pending_drain`` with the new one. Without the cancel,
+the first task is orphaned (still running on the loop, no longer
+reachable from ``self``), and ``close()`` awaits only the second
+— recreating the exact "Task was destroyed but it is pending"
+warning the drain mechanism was added to suppress.
+
+Also pins the cycle 22 ``RuntimeError("Event loop is closed")``
+guard around ``loop.create_task``: when the loop is dead the
+fallback sets ``_pending_drain = None`` and preserves the
+original cancel/cause instead of replacing it with a bare
+``RuntimeError``.
+
+And pins the ``_invalidation_cause`` clear on ``_close_impl``
+(cycle 22) so the cached exception does not pin frame
+globals/locals across close → reconnect cycles.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+import pytest
+
+from dqliteclient.connection import DqliteConnection
+
+
+def _make_conn_with_protocol() -> DqliteConnection:
+    """Build a DqliteConnection skeleton sufficient to drive
+    ``_invalidate`` end-to-end without a real wire connection."""
+    conn = DqliteConnection.__new__(DqliteConnection)
+    conn._protocol = None
+    conn._db_id = None
+    conn._pending_drain = None
+    conn._invalidation_cause = None
+    conn._in_use = False
+    conn._in_transaction = False
+    conn._tx_owner = None
+    conn._savepoint_stack = []
+    conn._closed = False  # type: ignore[attr-defined]
+    conn._close_timeout = 0.05
+    conn._address = "test:9001"
+    conn._bound_loop = None
+    conn._pool_released = False
+    return conn
+
+
+class _FakeProtocol:
+    """Minimal stand-in for DqliteProtocol covering the slots
+    ``_invalidate`` reads (``close()``, ``wait_closed()``)."""
+
+    def __init__(self) -> None:
+        self.close_calls = 0
+
+    def close(self) -> None:
+        self.close_calls += 1
+
+    async def wait_closed(self) -> None:
+        # Yield once so the bounded drain has a chance to be
+        # observed in pending state by a sibling _invalidate.
+        await asyncio.sleep(0.5)
+
+
+@pytest.mark.asyncio
+async def test_second_invalidate_cancels_prior_pending_drain() -> None:
+    conn = _make_conn_with_protocol()
+
+    # First invalidate: scheduling a bounded-drain task.
+    conn._protocol = _FakeProtocol()  # type: ignore[assignment]
+    conn._invalidate()
+    first_task = conn._pending_drain
+    assert first_task is not None
+    assert not first_task.done()
+
+    # Re-set the protocol so the second _invalidate's
+    # ``if self._protocol is not None:`` branch runs.
+    conn._protocol = _FakeProtocol()  # type: ignore[assignment]
+    conn._invalidate()
+    second_task = conn._pending_drain
+
+    assert second_task is not None
+    assert second_task is not first_task
+    # The cycle 22 contract: prior task has cancel() scheduled
+    # before the slot is overwritten. ``cancel()`` flips the task
+    # to ``cancelling`` (not yet ``cancelled``) until the task
+    # observes the CancelledError at its next await — pump the
+    # loop so the cancellation lands.
+    assert first_task.cancelling() > 0 or first_task.cancelled() or first_task.done()
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    assert first_task.cancelled() or first_task.done()
+
+
+@pytest.mark.asyncio
+async def test_invalidate_with_closed_loop_preserves_cause(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """``loop.create_task`` raising
+    ``RuntimeError("Event loop is closed")`` must NOT replace the
+    original cancel/cause; cycle 22 added the try/except guard
+    that DEBUG-logs and continues with ``_pending_drain = None``."""
+    conn = _make_conn_with_protocol()
+    conn._protocol = _FakeProtocol()  # type: ignore[assignment]
+
+    # Patch the running loop's create_task to simulate the
+    # closed-loop shape during dispose.
+    loop = asyncio.get_running_loop()
+    real_create_task = loop.create_task
+
+    def _raise_loop_closed(coro: object, **kwargs: object) -> object:
+        # Close the coroutine to suppress "never awaited" warnings.
+        coro.close()  # type: ignore[attr-defined]
+        raise RuntimeError("Event loop is closed")
+
+    loop.create_task = _raise_loop_closed  # type: ignore[assignment]
+    try:
+        caplog.set_level(logging.DEBUG, logger="dqliteclient.connection")
+        conn._invalidate(cause=ValueError("original cause"))
+    finally:
+        loop.create_task = real_create_task
+
+    # ``_pending_drain`` falls back to None instead of leaking the
+    # un-scheduled coroutine; the original cause is preserved on
+    # ``_invalidation_cause`` for downstream chaining.
+    assert conn._pending_drain is None
+    assert isinstance(conn._invalidation_cause, ValueError)
+    assert any("loop.create_task" in r.message for r in caplog.records)
+
+
+@pytest.mark.asyncio
+async def test_close_impl_clears_invalidation_cause() -> None:
+    """Cycle 22 added ``self._invalidation_cause = None`` on
+    ``_close_impl`` so the cached exception does NOT pin
+    frame globals/locals across close → reconnect cycles. A
+    regression that drops the line re-introduces the
+    traceback-pin defect."""
+    conn = _make_conn_with_protocol()
+    # Simulate a prior invalidation having stored a cause.
+    conn._invalidation_cause = ValueError("prior failure")
+    assert conn._invalidation_cause is not None
+
+    await conn._close_impl()
+
+    assert conn._invalidation_cause is None