Reject use of ClusterClient.find_leader from a forked child process

The single-flight slot map (_find_leader_tasks) holds asyncio.Task
instances bound to the parent's loop. A child that forks mid-sweep
and calls find_leader observes an inherited task and falls through
to await asyncio.shield(<parent-task>) — undefined behavior.

Record os.getpid() in __init__ and raise InterfaceError("ClusterClient
used after fork; reconstruct from configuration in the target
process") at the top of find_leader. Symmetric with the cycle 20
DqliteConnection / ConnectionPool guards.

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

Author: antoineleclair

src/dqliteclient/cluster.py

@@ -3,6 +3,7 @@
 import asyncio
 import contextlib
 import logging
+import os
 import random
 from collections.abc import Callable, Iterable
 from typing import Final, NoReturn
@@ -12,6 +13,7 @@
     ClusterError,
     ClusterPolicyError,
     DqliteConnectionError,
+    InterfaceError,
     OperationalError,
     ProtocolError,
 )
@@ -148,6 +150,13 @@ def __init__(
         # one pool out of widened heartbeats. Bounded to two slots
         # since the flag is a bool.
         self._find_leader_tasks: dict[tuple[bool], asyncio.Task[str]] = {}
+        # Fork-after-init: the slot map holds asyncio.Task instances
+        # bound to the parent's loop. A child that forks mid-sweep
+        # would observe an inherited task and ``await
+        # asyncio.shield(<parent-loop task>)`` — undefined behaviour.
+        # Symmetric with the DqliteConnection / ConnectionPool
+        # pid guards added in cycle 20.
+        self._creator_pid = os.getpid()
 
     @classmethod
     def from_addresses(
@@ -214,6 +223,15 @@ async def find_leader(self, *, trust_server_heartbeat: bool = False) -> str:
         (re-poll the node store between probes) costs an extra await
         per probe to close a window most callers do not exercise.
         """
+        if os.getpid() != self._creator_pid:
+            # Fork-after-init: the slot map holds parent-loop tasks
+            # that the child cannot drive. Surface a clear
+            # InterfaceError instead of letting a sibling task land
+            # at ``await asyncio.shield(<parent-task>)``.
+            raise InterfaceError(
+                "ClusterClient used after fork; reconstruct from configuration "
+                "in the target process."
+            )
         key: tuple[bool] = (trust_server_heartbeat,)
         task = self._find_leader_tasks.get(key)
         if task is None or task.done():

tests/test_cluster_client_fork_guard.py

@@ -0,0 +1,74 @@
+"""Pin: ``ClusterClient.find_leader`` raises a clear ``InterfaceError``
+when called after ``os.fork``.
+
+The single-flight slot map holds ``asyncio.Task`` instances bound to
+the parent's loop. A child that forks mid-sweep and calls
+``find_leader`` would observe an inherited task and fall through to
+``await asyncio.shield(<parent-loop task>)`` — undefined behaviour.
+
+Cycle 20 added pid guards to ``DqliteConnection`` and
+``ConnectionPool``; this extends the same pattern to ``ClusterClient``.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import patch
+
+import pytest
+
+from dqliteclient.cluster import ClusterClient
+from dqliteclient.exceptions import InterfaceError
+from dqliteclient.node_store import MemoryNodeStore
+
+
+@pytest.mark.asyncio
+async def test_find_leader_after_fork_raises_interface_error() -> None:
+    cluster = ClusterClient(node_store=MemoryNodeStore(["localhost:9001"]))
+    fake_parent_pid = cluster._creator_pid + 1
+    cluster._creator_pid = fake_parent_pid
+
+    with (
+        patch("dqliteclient.cluster.os.getpid", return_value=fake_parent_pid + 1),
+        pytest.raises(InterfaceError, match="fork"),
+    ):
+        await cluster.find_leader()
+
+
+@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires os.fork")
+def test_find_leader_actual_fork_raises() -> None:
+    """End-to-end fork: parent stages a ClusterClient; child calls
+    find_leader and reports back."""
+    cluster = ClusterClient(node_store=MemoryNodeStore(["127.0.0.1:9999"]))
+
+    r, w = os.pipe()
+    pid = os.fork()
+    if pid == 0:
+        try:
+            os.close(r)
+            try:
+                import asyncio
+
+                async def run() -> None:
+                    await cluster.find_leader()
+
+                asyncio.run(run())
+                os.write(w, b"NO_RAISE")
+            except InterfaceError as e:
+                os.write(w, b"OK" if "fork" in str(e) else f"WRONG_MSG:{e}".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)
+    assert result == b"OK", f"child reported: {result!r}"