Short-circuit close() / initialize() in forked child to spare inherited FDs

The first fork-guard commit covered the use-after-fork case (acquire,
_check_in_use), but close() and pool.initialize() were left to fail
late. In a child, the inherited connection FDs are shared with the
parent: writer.close() in DqliteConnection.close() or the per-conn
drain inside ConnectionPool.close() would send FIN on sockets the
parent still uses, breaking the parent's connections from the child's
GC sweep.

Pid-check at the top of:

- DqliteConnection.close: flip _protocol/_db_id to None and bail —
  no wire teardown, the FD stays open in the child for the parent.
- ConnectionPool.close: flip _closed=True and bail — no _drain_idle.
- ConnectionPool.initialize: raise InterfaceError up-front so a child
  cannot kick off TCP work against asyncio primitives bound to the
  parent's loop.

Tests cover all three paths plus the existing acquire/check_in_use
short-circuits.

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

Author: antoineleclair

src/dqliteclient/connection.py

@@ -1087,6 +1087,19 @@ async def close(self) -> None:
         # close path has already run under pool ownership.
         if self._pool_released:
             return
+        # 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
+        # touching the wire so the child can clean up its references
+        # quietly, then bail. The pid-aware ``_check_in_use`` would
+        # also raise here, but close() is documented as idempotent and
+        # 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.
+        if os.getpid() != self._creator_pid:
+            self._protocol = None
+            self._db_id = None
+            return
         # 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

src/dqliteclient/pool.py

@@ -316,6 +316,11 @@ async def initialize(self) -> None:
         ``return_exceptions=True`` so every task resolves, then close
         survivors explicitly before re-raising the first failure.
         """
+        if os.getpid() != self._creator_pid:
+            raise InterfaceError(
+                "Pool used after fork; reconstruct from configuration "
+                "in the target process."
+            )
         # Hold the lock across the gather so a second concurrent
         # initialize() call observes _initialized=True after the first
         # completes and returns without re-creating.
@@ -1307,6 +1312,16 @@ async def close(self) -> None:
             if self._close_done is not None:
                 await self._close_done.wait()
             return
+        # Fork-after-init: the inherited connection FDs are shared with
+        # the parent. Draining and writer.close() in the child would
+        # send FIN on sockets the parent still uses. Flip the closed
+        # flag so the child's references can be GC'd quietly without
+        # touching the wire. The child cannot acquire new connections
+        # either way (pid-aware ``acquire`` rejects). Symmetric with
+        # ``DqliteConnection.close``'s fork short-circuit.
+        if os.getpid() != self._creator_pid:
+            self._closed = True
+            return
         # Publish the drain-done event BEFORE flipping the closed flag
         # so any second caller observing ``_closed=True`` is guaranteed
         # to see a valid ``_close_done`` to wait on. Under single-task

tests/test_connection_after_fork_raises.py

@@ -91,3 +91,94 @@ async def run() -> None:
 
     result = _run_in_child(child_check)
     assert result == b"OK", f"child reported: {result!r}"
+
+
+@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires os.fork")
+def test_connection_pool_initialize_after_fork_raises_interface_error() -> None:
+    pool = ConnectionPool(addresses=["127.0.0.1:9999"])
+
+    def child_check() -> None:
+        async def run() -> None:
+            await pool.initialize()
+
+        asyncio.run(run())
+
+    result = _run_in_child(child_check)
+    assert result == b"OK", f"child reported: {result!r}"
+
+
+@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires os.fork")
+def test_dqlite_connection_close_after_fork_short_circuits() -> None:
+    """``close()`` in the child must not touch the inherited socket
+    (which is shared with the parent — sending FIN would close it
+    for the parent too). Short-circuits to a quiet local-state flip."""
+    conn = DqliteConnection("127.0.0.1:9999")
+
+    r, w = os.pipe()
+    pid = os.fork()
+    if pid == 0:
+        try:
+            os.close(r)
+            try:
+
+                async def run() -> None:
+                    await conn.close()
+
+                asyncio.run(run())
+                # close() did not raise — the child's local state is
+                # marked closed without touching the wire.
+                os.write(w, b"OK")
+            except Exception as e:  # noqa: BLE001
+                os.write(w, f"WRONG:{type(e).__name__}:{e}".encode())
+            finally:
+                os.close(w)
+        finally:
+            os._exit(0)
+    os.close(w)
+    result = b""
+    while True:
+        chunk = os.read(r, 4096)
+        if not chunk:
+            break
+        result += chunk
+    os.close(r)
+    os.waitpid(pid, 0)
+    assert result == b"OK", f"child reported: {result!r}"
+
+
+@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires os.fork")
+def test_connection_pool_close_after_fork_short_circuits() -> None:
+    """``pool.close()`` in the child must not drain inherited connection
+    FDs — those would close sockets the parent still uses. Short-
+    circuits to a quiet local-state flip."""
+    pool = ConnectionPool(addresses=["127.0.0.1:9999"])
+
+    r, w = os.pipe()
+    pid = os.fork()
+    if pid == 0:
+        try:
+            os.close(r)
+            try:
+
+                async def run() -> None:
+                    await pool.close()
+                    assert pool._closed is True
+
+                asyncio.run(run())
+                os.write(w, b"OK")
+            except Exception as e:  # noqa: BLE001
+                os.write(w, f"WRONG:{type(e).__name__}:{e}".encode())
+            finally:
+                os.close(w)
+        finally:
+            os._exit(0)
+    os.close(w)
+    result = b""
+    while True:
+        chunk = os.read(r, 4096)
+        if not chunk:
+            break
+        result += chunk
+    os.close(r)
+    os.waitpid(pid, 0)
+    assert result == b"OK", f"child reported: {result!r}"