Reorder DqliteConnection.close fork branch ahead of pool-released short-circuit

Move the fork-detection branch (``if _current_pid != self._creator_pid:``)
to run BEFORE the ``if self._pool_released: return`` short-circuit.
Previously the short-circuit pre-empted the fork-cleanup branch for
connections inherited across a fork while marked pool-released (the
SA pool checkin path), leaving the inherited ``_pending_drain``
asyncio.Task — bound to the parent's event loop — alive in the
child. Child GC then tripped "Task was destroyed but it is pending"
and the coroutine frame retained the inherited writer transport
(and FD) indefinitely.

The reorder applies the same field-nulling already documented in the
fork branch (per the original fix) to the pool-released-cross-fork
case, while preserving the same-process idempotency contract
(``_pool_released=True`` in the same process still no-ops).

Author: antoineleclair

src/dqliteclient/connection.py

@@ -1687,22 +1687,6 @@ async def close(self) -> None:
         is already closed after ``writer.close()`` — the remaining
         wait is best-effort cleanup, not correctness-critical.
         """
-        # Pool-released connections are never in_use for close(); their
-        # close path has already run under pool ownership.
-        if self._pool_released:
-            return
-        # Flip the public predicate even if we early-return below
-        # (fork branch / already-disconnected path). ``closed`` is
-        # the "close() has been called" marker; ``is_connected``
-        # remains the "transport alive" marker.
-        self._closed = True
-        # Detach the unclosed-warning finalizer — orderly shutdown
-        # owns the cleanup; firing the warning now would be a false
-        # positive.
-        self._closed_flag[0] = True
-        if self._finalizer is not None:
-            self._finalizer.detach()
-            self._finalizer = None
         # Fork-after-init: the inherited socket FD is shared with the
         # parent. ``writer.close()`` would send a FIN that the parent
         # still depends on. Flip the local state to closed without
@@ -1712,6 +1696,16 @@ async def close(self) -> None:
         # silent on already-closed inputs — silently no-oping in the
         # child preserves that contract for the GC / __del__ path that
         # commonly drives close in a forked worker.
+        #
+        # NB: this branch runs BEFORE the ``_pool_released`` short-
+        # circuit below. The previous ordering pre-empted the fork
+        # cleanup for pool-released connections inherited across a
+        # fork (SA pool checkin path), leaving the inherited
+        # ``_pending_drain`` Task — bound to the parent's loop —
+        # alive in the child until GC, where ``Task.__del__`` later
+        # tripped "Task was destroyed but it is pending" and the
+        # coroutine frame retained the inherited writer transport
+        # (and FD) indefinitely.
         if _current_pid != self._creator_pid:
             # Drop every reference that crosses the fork boundary so
             # GC in the child doesn't keep parent-loop primitives or
@@ -1724,6 +1718,11 @@ async def close(self) -> None:
             # transaction / savepoint / loop bookkeeping also keeps
             # the child's view of the connection self-consistent —
             # ``in_transaction`` reads False for a closed connection.
+            self._closed = True
+            self._closed_flag[0] = True
+            if self._finalizer is not None:
+                self._finalizer.detach()
+                self._finalizer = None
             self._protocol = None
             self._db_id = None
             self._pending_drain = None
@@ -1734,6 +1733,23 @@ async def close(self) -> None:
             self._has_untracked_savepoint = False
             self._bound_loop_ref = None
             return
+        # Pool-released connections are never in_use for close(); their
+        # close path has already run under pool ownership. The same-
+        # process idempotency contract: subsequent calls no-op.
+        if self._pool_released:
+            return
+        # Flip the public predicate even if we early-return below
+        # (already-disconnected path). ``closed`` is the "close() has
+        # been called" marker; ``is_connected`` remains the
+        # "transport alive" marker.
+        self._closed = True
+        # Detach the unclosed-warning finalizer — orderly shutdown
+        # owns the cleanup; firing the warning now would be a false
+        # positive.
+        self._closed_flag[0] = True
+        if self._finalizer is not None:
+            self._finalizer.detach()
+            self._finalizer = None
         # Run the in-use guard BEFORE the ``_protocol is None``
         # early-return so a concurrent ``connect()`` racing with
         # ``close()`` surfaces as ``InterfaceError`` instead of a silent

tests/test_close_pool_released_fork_branch_runs.py

@@ -0,0 +1,122 @@
+"""Pin: ``DqliteConnection.close()`` on a pool-released connection
+in a forked child runs the fork-cleanup branch (nulling
+``_pending_drain`` and the rest of the parent-loop-bound state)
+rather than taking the ``_pool_released`` short-circuit.
+
+Background: the SA pool's checkin path leaves a conn marked
+``_pool_released=True`` so a follow-up ``close()`` is a same-
+process no-op. The fork branch (per ISSUE-944) nulls inherited
+parent-loop state (``_pending_drain`` Task is the canonical leak
+shape). Previously the ``_pool_released`` short-circuit ran
+BEFORE the fork branch, pre-empting the cleanup — pool-released
+conns inherited across fork kept their parent-loop ``_pending_drain``
+references alive in the child, tripping
+"Task was destroyed but it is pending" warnings at child GC and
+retaining the inherited writer transport (and FD) indefinitely.
+
+The fix reorders ``close()`` so the fork detection runs first,
+regardless of ``_pool_released``. Same-process idempotency
+(``_pool_released=True`` -> no-op) is preserved.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import os
+
+import pytest
+
+import dqliteclient.connection as _conn_mod
+from dqliteclient.connection import DqliteConnection
+
+
+@pytest.mark.asyncio
+async def test_pool_released_close_in_forked_child_nulls_pending_drain(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Set ``_pool_released=True`` and a sentinel
+    ``_pending_drain`` Task, simulate fork by monkeypatching
+    ``_current_pid``, then call ``close()``. The fork branch MUST
+    null ``_pending_drain`` even though the connection is pool-
+    released — pre-empting that with the short-circuit was the
+    documented leak shape.
+    """
+    conn = DqliteConnection("leader:9001")
+    conn._pool_released = True
+    sentinel_task = asyncio.create_task(asyncio.sleep(60))
+    conn._pending_drain = sentinel_task
+    conn._in_transaction = True
+    conn._tx_owner = asyncio.current_task()
+
+    monkeypatch.setattr(_conn_mod, "_current_pid", os.getpid() + 1)
+    try:
+        await conn.close()
+    finally:
+        sentinel_task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await sentinel_task
+
+    # The fork branch must null _pending_drain (matching the symmetric
+    # fork branch nullification). The same set of fields is nulled.
+    assert conn._pending_drain is None, (
+        "DqliteConnection.close() in a forked child with _pool_released=True "
+        "must null _pending_drain — the prior ordering pre-empted the fork "
+        "branch and left the parent-loop Task referenced through child GC."
+    )
+    assert conn._in_transaction is False
+    assert conn._tx_owner is None
+    assert conn._bound_loop_ref is None
+    assert conn._protocol is None
+    assert conn._db_id is None
+    assert conn._closed is True
+
+
+@pytest.mark.asyncio
+async def test_pool_released_close_same_process_is_still_no_op() -> None:
+    """Regression guard: same-process ``close()`` on a pool-released
+    conn must still no-op (the SA pool checkin idempotency contract).
+    The reorder must NOT scrub state when same-process +
+    ``_pool_released=True``.
+    """
+    conn = DqliteConnection("leader:9001")
+    conn._pool_released = True
+    # Plant a sentinel _pending_drain to detect any scrubbing.
+    sentinel_task = asyncio.create_task(asyncio.sleep(60))
+    conn._pending_drain = sentinel_task
+    conn._in_transaction = True
+    try:
+        await conn.close()
+
+        # Same-process + pool-released: nothing must be scrubbed.
+        assert conn._pending_drain is sentinel_task
+        assert conn._in_transaction is True
+    finally:
+        sentinel_task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await sentinel_task
+
+
+@pytest.mark.asyncio
+async def test_non_pool_released_fork_branch_still_works(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Regression guard: the fork branch must continue to fire for
+    connections that are NOT pool-released (the original
+    ISSUE-944 path)."""
+    conn = DqliteConnection("leader:9001")
+    sentinel_task = asyncio.create_task(asyncio.sleep(60))
+    conn._pending_drain = sentinel_task
+    conn._in_transaction = True
+
+    monkeypatch.setattr(_conn_mod, "_current_pid", os.getpid() + 1)
+    try:
+        await conn.close()
+    finally:
+        sentinel_task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await sentinel_task
+
+    assert conn._pending_drain is None
+    assert conn._in_transaction is False
+    assert conn._closed is True