Cluster + protocol robustness: validation, anchored substrings, IPv6 zones

Six client-layer correctness gaps closed in one bundle:

- protocol.interrupt() drain now treats ResultResponse and
  EmptyResponse as terminal. The EXEC / EXEC_SQL done-callback
  emits ResultResponse before the EmptyResponse acknowledgement
  when the interrupted statement is on the EXEC side; without this
  branch a cancel landing on EXEC poisoned the wire with "Expected
  EmptyResponse" ProtocolError.

- protocol.prepare() now cross-validates StmtResponse.db_id against
  the requested db_id. Mismatch is a ProtocolError that invalidates
  the connection — silent registry drift would route writes to the
  wrong DB.

- NO_TRANSACTION_MESSAGE_SUBSTRINGS: drop the bare "cannot rollback"
  token. The remaining "no transaction is active" anchor covers
  both upstream wordings ("cannot rollback - no transaction is
  active" and "cannot commit - no transaction is active") and is
  not subject to the DQLITE_ERROR=1 / SQLITE_ERROR=1 collision —
  any unrelated code-1 message containing "cannot rollback" was
  previously triggering silent-swallow.

- MemoryNodeStore now strips/dedups/rejects empty seed entries.
  A typoed "localhost:9001\n" used to leak ValueError through
  find_leader's narrow exception filter; duplicates inflated the
  probe count and the per-node error lines.

- _parse_address rejects pathological IPv6 zone-id shapes (empty,
  whitespace-containing, slash-containing) at the strict-bracket
  parser instead of letting them slip past to the regex-fallback's
  "not a valid hostname or IP literal" diagnostic.

- README documents fork-after-init contract (gunicorn --preload,
  multiprocessing, Celery prefork pool guidance); pool.initialize()
  docstring documents the parallel-warm-up shape and the
  leader-side serialisation cost.

Tests updated for the anchored-substring contract.

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

Author: antoineleclair

README.md

@@ -43,6 +43,41 @@ async with pool.acquire() as conn:
     rows = await conn.fetch("SELECT 1")
 ```
 
+The pool issues `min_size` connection handshakes in parallel during
+`initialize()`. All initial connects target whichever node the
+cluster client identifies as the current leader, so the leader
+serialises its incoming-connection acceptance — `min_size=N` does
+NOT speed up startup linearly with N. A balanced default
+(`min_size=1` or low single digits) keeps cold-start latency
+predictable; raise it only when steady-state concurrency demands
+warm connections at engine startup.
+
+## Forking and multiprocessing
+
+Connections, pools, and the cluster client are not safe to use
+across `os.fork()`. The library detects fork-after-init and raises
+`InterfaceError` from any operation in the child process; the
+inherited TCP socket would otherwise be shared with the parent
+(writes would interleave on the wire) and asyncio primitives are
+bound to the parent's event loop.
+
+Common deployment patterns that fork after import:
+
+- **gunicorn** with `--preload`: workers inherit pools created in
+  the parent. Move pool creation into a per-worker `post_fork`
+  hook (gunicorn `post_fork` config) instead of the module top
+  level.
+- **multiprocessing**: child processes must reconstruct
+  connections / pools from configuration (addresses, database
+  name) rather than receive a parent-built object.
+- **Celery prefork pool**: each worker process must create its
+  own pool inside the worker init signal, not at module load.
+
+The fork detection is best-effort (pid mismatch); silent
+double-FIN on the parent's socket is a real risk if the guard is
+bypassed (e.g. via `__new__` to skip `__init__`). Just create the
+pool / connection in the child process you intend to use it from.
+
 ## Layering
 
 `dqlite-client` is the low-level async wire client. Most applications

src/dqliteclient/connection.py

@@ -710,6 +710,26 @@ def parse_address(address: str) -> tuple[str, int]:
             # by non-hex characters intact, so the no-encoding path
             # round-trips byte-for-byte.
             host = host[:zone_sep] + unquote(host[zone_sep:])
+            # Reject pathological zone-id shapes here (rather than
+            # letting them slip past the IPv6 parse and fall through
+            # to the regex-fallback "not a valid hostname or IP
+            # literal" diagnostic, which is a different code path
+            # with a different error message). Empty zone, embedded
+            # whitespace, or path-like characters are not valid
+            # RFC 6874 zone identifiers.
+            zone = host[zone_sep + 1 :]
+            if not zone:
+                raise ValueError(
+                    f"Bracket syntax in {address!r} has an empty IPv6 zone "
+                    f"identifier (after '%'); supply a zone like '%eth0' "
+                    f"or remove the '%'"
+                )
+            if any(c.isspace() or c == "/" for c in zone):
+                raise ValueError(
+                    f"Bracket syntax in {address!r} has an invalid IPv6 "
+                    f"zone identifier {zone!r}; zone IDs must not contain "
+                    f"whitespace or '/'"
+                )
 
         # Strict bracket discipline: validate the contents are an
         # IPv6 literal. ``ipaddress.ip_address`` does not accept

src/dqliteclient/node_store.py

@@ -97,9 +97,32 @@ def __init__(
             raise TypeError("Pass only one of 'addresses' or 'initial_addresses'")
         seed = addresses if addresses is not None else initial_addresses
         if seed:
+            # Strip leading/trailing whitespace and reject empty
+            # entries up front. Without this, a typoed seed like
+            # ``["localhost:9001\n"]`` (e.g. read from a config
+            # file) leaks ``ValueError`` through the sweep's narrow
+            # exception filter and surfaces deep inside
+            # ``find_leader``. Deduplicate while preserving order
+            # so a config with the same address twice doesn't
+            # double the probe count and double the per-node
+            # error lines in the failure-aggregate message.
+            seen: set[str] = set()
+            unique: list[str] = []
+            for raw in seed:
+                if not isinstance(raw, str):
+                    raise TypeError(
+                        f"NodeStore addresses must be 'host:port' strings, got {type(raw).__name__}"
+                    )
+                addr = raw.strip()
+                if not addr:
+                    raise ValueError("NodeStore addresses must be non-empty 'host:port' strings")
+                if addr in seen:
+                    continue
+                seen.add(addr)
+                unique.append(addr)
             self._nodes: tuple[NodeInfo, ...] = tuple(
                 NodeInfo(node_id=i + 1, address=addr, role=NodeRole.VOTER)
-                for i, addr in enumerate(seed)
+                for i, addr in enumerate(unique)
             )
         else:
             self._nodes = ()

src/dqliteclient/pool.py

@@ -315,6 +315,17 @@ async def initialize(self) -> None:
         succeeded — they leak as orphaned transports. Use
         ``return_exceptions=True`` so every task resolves, then close
         survivors explicitly before re-raising the first failure.
+
+        Warm-up shape: this issues ``min_size`` connection handshakes
+        IN PARALLEL via ``asyncio.gather``. Every initial connect
+        targets the current leader (after cluster discovery), so the
+        leader-side handshake acceptance is serialised and a large
+        ``min_size`` does NOT linearly speed up startup. Diverges from
+        Go-dqlite's pool which warms lazily; the parallel-warm shape
+        is a deliberate trade-off: predictable cold-start memory at
+        the cost of leader-side serialisation. Keep ``min_size`` low
+        (single digits) unless steady-state concurrency demands warm
+        connections at engine startup.
         """
         if os.getpid() != self._creator_pid:
             raise InterfaceError(

src/dqliteclient/protocol.py

@@ -336,6 +336,20 @@ async def prepare(self, db_id: int, sql: str) -> tuple[int, int]:
                 f"Expected StmtResponse, got {type(response).__name__}{self._addr_suffix()}"
             )
 
+        # Defense-in-depth: confirm the server's StmtResponse echoes
+        # the db_id we asked it to prepare against. A mismatch would
+        # mean the server's prepared-statement registry has drifted
+        # from the client's view, and any future exec/finalize
+        # against the returned stmt_id would target a different
+        # database. Surface this as a ProtocolError so the
+        # connection is invalidated rather than silently routing
+        # writes against the wrong DB.
+        if response.db_id != db_id:
+            raise ProtocolError(
+                f"StmtResponse db_id {response.db_id} does not match "
+                f"requested db_id {db_id}{self._addr_suffix()}"
+            )
+
         return response.stmt_id, response.num_params
 
     async def finalize(self, db_id: int, stmt_id: int) -> None:
@@ -418,6 +432,18 @@ async def interrupt(self, db_id: int) -> None:
             response = await self._read_response(deadline=deadline)
             if isinstance(response, EmptyResponse):
                 return
+            # ``ResultResponse`` is the EXEC-side terminal: when the
+            # interrupted statement was an EXEC / EXEC_SQL whose
+            # done-callback emitted RESULT before the INTERRUPT
+            # took effect, the response queue holds the RESULT
+            # before the EmptyResponse acknowledgement. Treat it as
+            # equivalent to EmptyResponse — the wire is in a
+            # coherent state and the interrupt has been honoured.
+            # Without this branch, a cancel landing on an EXEC
+            # path would hit the "Expected EmptyResponse" arm and
+            # poison the wire.
+            if isinstance(response, ResultResponse):
+                return
             if isinstance(response, FailureResponse):
                 raise OperationalError(response.code, self._failure_text(response))
             # RowsResponse mid-drain is expected: the server's in-flight

tests/test_no_transaction_substrings_source_of_truth.py

@@ -36,11 +36,14 @@ def test_recogniser_rejects_unrelated_message() -> None:
 
 
 def test_substring_constant_has_expected_values() -> None:
-    """Pin the canonical substring list — both clauses must be
-    present so a server-version drift that drops one but not the
-    other still triggers the recogniser. If a server change requires
-    an update, the integration pin
-    ``test_no_transaction_error_wording.py`` (in dbapi) catches it
-    against a live cluster."""
+    """Pin the canonical substring — the anchored
+    ``"no transaction is active"`` is the single source of truth.
+    Both upstream wordings (``"cannot rollback - no transaction is
+    active"`` and ``"cannot commit - no transaction is active"``)
+    contain this substring; the prior bare ``"cannot rollback"``
+    token was removed because it was too permissive (any unrelated
+    SQLite error or DQLITE_ERROR=1 message containing those words
+    could trigger the silent-swallow path). The integration pin
+    ``test_no_transaction_error_wording.py`` (in dbapi) catches a
+    server-version drift against a live cluster."""
     assert "no transaction is active" in NO_TRANSACTION_MESSAGE_SUBSTRINGS
-    assert "cannot rollback" in NO_TRANSACTION_MESSAGE_SUBSTRINGS

tests/test_pool_reset_no_tx_preserves_slot.py

@@ -35,12 +35,22 @@ def test_classifier_matches_no_tx_active_with_sqlite_error_code(self) -> None:
         exc = OperationalError(1, "cannot rollback - no transaction is active")
         assert _is_no_tx_rollback_error(exc) is True
 
-    def test_classifier_matches_cannot_rollback_wording(self) -> None:
-        # The substring "cannot rollback" alone is also accepted, since
-        # the C server has historically used both wordings.
-        exc = OperationalError(1, "cannot rollback")
+    def test_classifier_matches_cannot_commit_wording(self) -> None:
+        # Both upstream wordings contain "no transaction is active"
+        # so the anchored substring covers commit and rollback
+        # equally. The bare "cannot rollback" token was previously
+        # also accepted but was too permissive.
+        exc = OperationalError(1, "cannot commit - no transaction is active")
         assert _is_no_tx_rollback_error(exc) is True
 
+    def test_classifier_rejects_bare_cannot_rollback(self) -> None:
+        # A message containing "cannot rollback" without the anchored
+        # "no transaction is active" clause must NOT match — any
+        # unrelated SQLite (or DQLITE_ERROR=1) error happening to
+        # contain those words would otherwise trigger silent-swallow.
+        exc = OperationalError(1, "cannot rollback because the disk is full")
+        assert _is_no_tx_rollback_error(exc) is False
+
     def test_classifier_rejects_no_tx_message_with_wrong_code(self) -> None:
         # SQLITE_BUSY (5) primary code with a coincidental no-tx wording
         # must NOT match — the user's tx might have been partially