Reject use of connection / pool from a forked child process

Fork-after-init is unsupported: the inherited TCP socket is shared
with the parent so writes interleave on the wire, and the connection's
asyncio primitives are bound to a loop the child cannot drive. Without
a guard the child silently corrupts the wire or hangs forever.

Record os.getpid() in DqliteConnection.__init__ and ConnectionPool.__init__
and raise InterfaceError("Connection used after fork; reconstruct from
configuration in the target process") from _check_in_use / acquire when
the running process pid no longer matches the creator pid. Symmetric
with the existing __reduce__ pickle/copy guards.

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

Author: antoineleclair

src/dqliteclient/connection.py

@@ -5,6 +5,7 @@
 import ipaddress
 import logging
 import math
+import os
 import re
 import string
 from collections.abc import AsyncIterator, Awaitable, Callable, Mapping, Sequence
@@ -866,6 +867,13 @@ def __init__(
         # ``_invalidate`` so a subsequent ``close()`` can await it and
         # keep the reader task from outliving the connection.
         self._pending_drain: asyncio.Task[None] | None = None
+        # Fork-after-init is unsupported: the inherited TCP socket
+        # would be shared with the parent and writes would interleave
+        # on the wire, and asyncio primitives bound to the parent's
+        # loop are unusable in the child. Store the creator pid so
+        # cross-fork use raises a clear ``InterfaceError`` from any
+        # public method instead of silent corruption.
+        self._creator_pid = os.getpid()
 
     @property
     def address(self) -> str:
@@ -1237,6 +1245,11 @@ def _ensure_connected(self) -> tuple[DqliteProtocol, int]:
 
     def _check_in_use(self) -> None:
         """Raise on misuse: wrong event loop, concurrent access, or use after pool release."""
+        if os.getpid() != self._creator_pid:
+            raise InterfaceError(
+                "Connection used after fork; reconstruct from configuration "
+                "in the target process."
+            )
         if self._pool_released:
             raise InterfaceError(
                 "This connection has been returned to the pool and can no longer "

src/dqliteclient/pool.py

@@ -3,6 +3,7 @@
 import asyncio
 import contextlib
 import logging
+import os
 from collections.abc import AsyncIterator, Sequence
 from contextlib import asynccontextmanager
 from types import TracebackType
@@ -265,6 +266,13 @@ def __init__(
         self._closed_event: asyncio.Event | None = None
         self._close_done: asyncio.Event | None = None
         self._initialized = False
+        # Fork-after-init is unsupported: pooled connections hold
+        # shared TCP sockets and asyncio primitives bound to the
+        # parent's loop. Store the creator pid so cross-fork
+        # ``acquire`` raises a clear ``InterfaceError`` instead of
+        # silently corrupting the wire by interleaving writes.
+        # Symmetric with ``__reduce__`` and the per-connection guard.
+        self._creator_pid = os.getpid()
 
     def __repr__(self) -> str:
         state = "closed" if self._closed else "open"
@@ -656,6 +664,11 @@ async def _drain_remaining_after_cancel(self) -> None:
     @asynccontextmanager
     async def acquire(self) -> AsyncIterator[DqliteConnection]:
         """Acquire a connection from the pool."""
+        if os.getpid() != self._creator_pid:
+            raise InterfaceError(
+                "Pool used after fork; reconstruct from configuration "
+                "in the target process."
+            )
         if self._closed:
             raise DqliteConnectionError("Pool is closed")
 

tests/test_connection_after_fork_raises.py

@@ -0,0 +1,93 @@
+"""Pin: ``DqliteConnection`` and ``ConnectionPool`` raise
+``InterfaceError`` if used from a child process after ``os.fork``.
+
+Fork-after-init is unsupported: the inherited TCP socket would be
+shared with the parent (writes interleaving on the wire), and asyncio
+primitives bound to the parent's loop are unusable in the child.
+
+The fix records ``os.getpid()`` in ``__init__`` (both classes) and
+``_check_in_use`` / ``acquire`` raise a clear ``InterfaceError``
+("reconstruct from configuration in the target process") on pid
+mismatch — symmetric with the existing ``__reduce__`` pickle guards.
+
+The test does not need a live server: the pid check fires before any
+network work; an unconnected instance is sufficient.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+
+import pytest
+
+from dqliteclient import DqliteConnection
+from dqliteclient.exceptions import InterfaceError
+from dqliteclient.pool import ConnectionPool
+
+
+def _run_in_child(check) -> bytes:
+    r, w = os.pipe()
+    pid = os.fork()
+    if pid == 0:
+        try:
+            os.close(r)
+            try:
+                check()
+                os.write(w, b"NO_RAISE")
+            except InterfaceError as e:
+                msg = str(e)
+                if "fork" in msg and "reconstruct from configuration" in msg:
+                    os.write(w, b"OK")
+                else:
+                    os.write(w, f"WRONG_MSG:{msg}".encode())
+            except Exception as e:  # noqa: BLE001
+                os.write(w, f"WRONG_TYPE:{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)
+    return result
+
+
+@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires os.fork")
+def test_dqlite_connection_used_after_fork_raises_interface_error() -> None:
+    conn = DqliteConnection("127.0.0.1:9999")
+    assert conn._creator_pid == os.getpid()
+
+    def child_check() -> None:
+        # ``_check_in_use`` runs ``asyncio.get_running_loop`` after
+        # the pid check. Drive a loop so the pid mismatch is the
+        # raised error, not the "must be used from async context" one.
+        async def run() -> None:
+            conn._check_in_use()
+
+        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_connection_pool_acquire_after_fork_raises_interface_error() -> None:
+    pool = ConnectionPool(addresses=["127.0.0.1:9999"])
+    assert pool._creator_pid == os.getpid()
+
+    def child_check() -> None:
+        async def run() -> None:
+            async with pool.acquire():
+                pass
+
+        asyncio.run(run())
+
+    result = _run_in_child(child_check)
+    assert result == b"OK", f"child reported: {result!r}"

tests/test_interface_error_messages.py

@@ -31,12 +31,15 @@
 def _make_bound_connection() -> DqliteConnection:
     """A connection in the post-bind state on the running loop, with
     every other ``_check_in_use`` precondition relaxed."""
+    import os as _os
+
     conn = DqliteConnection.__new__(DqliteConnection)
     conn._pool_released = False
     conn._bound_loop = asyncio.get_running_loop()
     conn._in_use = False
     conn._in_transaction = False
     conn._tx_owner = None
+    conn._creator_pid = _os.getpid()
     return conn
 
 
@@ -65,12 +68,15 @@ def test_called_from_sync_context_branch_message_substring() -> None:
     message rather than crashing on ``get_running_loop``'s
     RuntimeError. Run as a SYNC test so the surrounding code does
     not have a running loop."""
+    import os as _os
+
     conn = DqliteConnection.__new__(DqliteConnection)
     conn._pool_released = False
     conn._bound_loop = None
     conn._in_use = False
     conn._in_transaction = False
     conn._tx_owner = None
+    conn._creator_pid = _os.getpid()
     with pytest.raises(InterfaceError, match="from within an async context"):
         conn._check_in_use()