Reject pickle and copy on client classes

DqliteConnection / ConnectionPool / ClusterClient / DqliteProtocol
each hold loop-bound state that cannot survive serialization:
sockets, asyncio.Lock / Queue / Event, weak cursor sets, single-
flight slot maps. ConnectionPool / ClusterClient pickle SILENTLY in
Python 3.10+ (asyncio primitives became pickleable) — producing a
"live"-looking duplicate detached from any running loop. Any use
yields opaque corruption.

Add ``__reduce__`` raising a clear driver-level TypeError to each
class, naming the class and the unpickleable internal state.
Symmetric with the existing dbapi Connection / Cursor guards;
applies uniformly to pickle, copy.copy, and copy.deepcopy.

Author: antoineleclair

src/dqliteclient/cluster.py

@@ -5,7 +5,7 @@
 import logging
 import random
 from collections.abc import Callable, Iterable
-from typing import Final
+from typing import Final, NoReturn
 
 from dqliteclient.connection import DqliteConnection, _parse_address, _validate_timeout
 from dqliteclient.exceptions import (
@@ -161,6 +161,21 @@ def from_addresses(
         store = MemoryNodeStore(addresses)
         return cls(store, timeout=timeout, redirect_policy=redirect_policy)
 
+    def __reduce__(self) -> NoReturn:
+        # Holds a per-client single-flight slot map keyed by loop-bound
+        # asyncio.Task instances and a NodeStore that may itself hold
+        # mutable address state. Pickling produces a duplicate detached
+        # from any loop and from the original NodeStore's lifecycle —
+        # any use yields opaque corruption. Surface a clear
+        # driver-level TypeError instead. Symmetric with the
+        # ConnectionPool / DqliteConnection guards.
+        raise TypeError(
+            f"cannot pickle {type(self).__name__!r} object — holds a "
+            f"loop-bound single-flight slot map and a NodeStore "
+            f"reference; reconstruct from configuration in the target "
+            f"process instead."
+        )
+
     def _check_redirect(self, address: str) -> None:
         """Reject leader-redirect targets that fail the configured policy."""
         if self._redirect_policy is None:

src/dqliteclient/connection.py

@@ -10,7 +10,7 @@
 from collections.abc import AsyncIterator, Awaitable, Callable, Mapping, Sequence
 from contextlib import asynccontextmanager
 from types import TracebackType
-from typing import Any, Final
+from typing import Any, Final, NoReturn
 
 from dqliteclient.exceptions import (
     DataError,
@@ -787,6 +787,22 @@ def __repr__(self) -> str:
         state = "connected" if self._protocol is not None else "disconnected"
         return f"<DqliteConnection address={self._address!r} database={self._database!r} {state}>"
 
+    def __reduce__(self) -> NoReturn:
+        # ``DqliteConnection`` holds a live socket, an event-loop-bound
+        # asyncio.Lock, and a WeakSet of registered cursors — none of
+        # which survive serialization. Surface a clear driver-level
+        # TypeError instead of leaking the underlying ``cannot pickle
+        # '_thread.lock'`` (or, worse, the cryptic
+        # ``Can't pickle local object 'WeakSet.__init__.<locals>._remove'``
+        # post-connect). Symmetric with the dbapi-layer guards on
+        # Connection / Cursor.
+        raise TypeError(
+            f"cannot pickle {type(self).__name__!r} object — holds a "
+            f"live socket, loop-bound asyncio.Lock, and weak cursor "
+            f"refs; reconstruct from configuration in the target "
+            f"process instead."
+        )
+
     @property
     def in_transaction(self) -> bool:
         """Check if a transaction is active.

src/dqliteclient/pool.py

@@ -6,7 +6,7 @@
 from collections.abc import AsyncIterator, Sequence
 from contextlib import asynccontextmanager
 from types import TracebackType
-from typing import Any, Final
+from typing import Any, Final, NoReturn
 
 from dqliteclient.cluster import ClusterClient
 from dqliteclient.connection import (
@@ -280,6 +280,21 @@ def __repr__(self) -> str:
             f"max_size={self._max_size}, {state})"
         )
 
+    def __reduce__(self) -> "NoReturn":
+        # ``asyncio.Queue`` and ``asyncio.Lock`` became pickleable in
+        # Python 3.10+, so a naive ``pickle.dumps(pool)`` SILENTLY
+        # produces a "live"-looking duplicate — detached from any
+        # running loop, with fresh internal locks and queue. Any use
+        # of the duplicate yields opaque corruption. Surface a clear
+        # driver-level TypeError instead. Symmetric with the dbapi
+        # Connection / Cursor guards.
+        raise TypeError(
+            f"cannot pickle {type(self).__name__!r} object — holds "
+            f"loop-bound asyncio.Queue / asyncio.Lock / asyncio.Event "
+            f"and live worker tasks; reconstruct from configuration "
+            f"in the target process instead."
+        )
+
     async def initialize(self) -> None:
         """Initialize the pool with minimum connections.
 

src/dqliteclient/protocol.py

@@ -5,7 +5,7 @@
 import secrets
 import sys
 from collections.abc import Sequence
-from typing import Any, Final
+from typing import Any, Final, NoReturn
 
 from dqliteclient.exceptions import DqliteConnectionError, OperationalError, ProtocolError
 from dqlitewire import (
@@ -161,6 +161,20 @@ def __init__(
         # a latency-SLO boundary from server-induced amplification.
         self._trust_server_heartbeat = trust_server_heartbeat
 
+    def __reduce__(self) -> NoReturn:
+        # Wraps a live ``asyncio.StreamReader`` / ``StreamWriter``
+        # (loop-bound), a ``MessageDecoder`` with internal buffer
+        # state, and per-stream cap counters that mean nothing
+        # post-deserialise. Surface a clear driver-level TypeError
+        # instead of leaking the underlying ``cannot pickle
+        # 'asyncio.streams.StreamReader'``.
+        raise TypeError(
+            f"cannot pickle {type(self).__name__!r} object — wraps "
+            f"loop-bound StreamReader / StreamWriter and a stateful "
+            f"MessageDecoder; reconstruct from a fresh wire handshake "
+            f"in the target process instead."
+        )
+
     @property
     def is_wire_coherent(self) -> bool:
         """True if the decoder buffer has not been poisoned.

tests/test_pickle_guards.py

@@ -0,0 +1,91 @@
+"""Pin: client-layer classes raise a clear ``TypeError`` on pickle /
+copy / deepcopy. Symmetric with the dbapi's existing pickle guards
+on Connection / Cursor.
+
+Without explicit ``__reduce__`` raises:
+- ``ConnectionPool`` / ``ClusterClient`` SILENTLY pickle (asyncio
+  primitives became pickleable in 3.10+) — producing a
+  "live"-looking duplicate detached from any loop. Any use yields
+  opaque corruption.
+- ``DqliteConnection`` post-``connect()`` raises a cryptic
+  ``AttributeError("Can't pickle local object
+  'WeakSet.__init__.<locals>._remove'")`` from the cursor-tracking
+  weak set.
+- ``DqliteProtocol`` raises an opaque error from wrapped
+  StreamReader / StreamWriter.
+
+Mirror the ISSUE-791 pattern: every class raises a
+driver-level ``TypeError`` naming the specific class.
+"""
+
+from __future__ import annotations
+
+import copy
+import pickle
+
+import pytest
+
+from dqliteclient.cluster import ClusterClient
+from dqliteclient.connection import DqliteConnection
+from dqliteclient.node_store import MemoryNodeStore
+from dqliteclient.pool import ConnectionPool
+
+
+class TestDqliteConnectionPickleGuard:
+    def test_pickle_raises(self) -> None:
+        conn = DqliteConnection("localhost:9001")
+        with pytest.raises(TypeError, match="DqliteConnection"):
+            pickle.dumps(conn)
+
+    def test_copy_copy_raises(self) -> None:
+        conn = DqliteConnection("localhost:9001")
+        with pytest.raises(TypeError, match="DqliteConnection"):
+            copy.copy(conn)
+
+    def test_copy_deepcopy_raises(self) -> None:
+        conn = DqliteConnection("localhost:9001")
+        with pytest.raises(TypeError, match="DqliteConnection"):
+            copy.deepcopy(conn)
+
+
+class TestConnectionPoolPickleGuard:
+    def test_pickle_raises(self) -> None:
+        pool = ConnectionPool(addresses=["localhost:9001"])
+        with pytest.raises(TypeError, match="ConnectionPool"):
+            pickle.dumps(pool)
+
+    def test_copy_copy_raises(self) -> None:
+        pool = ConnectionPool(addresses=["localhost:9001"])
+        with pytest.raises(TypeError, match="ConnectionPool"):
+            copy.copy(pool)
+
+    def test_copy_deepcopy_raises(self) -> None:
+        pool = ConnectionPool(addresses=["localhost:9001"])
+        with pytest.raises(TypeError, match="ConnectionPool"):
+            copy.deepcopy(pool)
+
+
+class TestClusterClientPickleGuard:
+    def test_pickle_raises(self) -> None:
+        cluster = ClusterClient(node_store=MemoryNodeStore(["localhost:9001"]))
+        with pytest.raises(TypeError, match="ClusterClient"):
+            pickle.dumps(cluster)
+
+    def test_copy_copy_raises(self) -> None:
+        cluster = ClusterClient(node_store=MemoryNodeStore(["localhost:9001"]))
+        with pytest.raises(TypeError, match="ClusterClient"):
+            copy.copy(cluster)
+
+
+class TestDqliteProtocolPickleGuard:
+    def test_pickle_raises(self) -> None:
+        import asyncio
+        from unittest.mock import MagicMock
+
+        from dqliteclient.protocol import DqliteProtocol
+
+        reader = MagicMock(spec=asyncio.StreamReader)
+        writer = MagicMock(spec=asyncio.StreamWriter)
+        proto = DqliteProtocol(reader=reader, writer=writer)
+        with pytest.raises(TypeError, match="DqliteProtocol"):
+            pickle.dumps(proto)