Override DqliteError.__reduce__ to preserve code and raw_message on pickle

cycle-27 commits XP2 ("Thread code and raw_message through leader-
change connect rewrap") and XP3 ("Hoist raw_message to DqliteError
base") added optional ``code`` and ``raw_message`` fields to the
exception hierarchy. The default ``Exception.__reduce__`` returns
``(cls, self.args)`` and reconstructs via ``cls(*args)`` — losing
every field set on the instance after ``Exception.__init__``.

Round-trip a leader-change DqliteConnectionError(code=10250,
raw_message="not leader") through ``pickle.dumps`` /
``pickle.loads`` (or through ``multiprocessing.Queue``,
``ProcessPoolExecutor`` task results, Celery, SA's multiprocess
pool) and you get back ``code=None``, ``raw_message=None``: SA's
``is_disconnect`` code-based classifier on the connect path falls
back to the substring scan that XP2 was added to escape, defeating
the cross-process motivation cited in XP2's commit message.

Override ``__reduce__`` on the base ``DqliteError`` to return the
3-tuple ``(callable, args, state)`` form so pickle reconstructs via
``cls(*args)`` then applies ``self.__dict__`` as state. Every
subclass inherits the discipline automatically. ``OperationalError``
remains pickle-lossless via its existing ``args = (code, message)``
discipline plus the new state-dict overlay.

Pin: lossless round-trip across pickle protocols 2..HIGHEST and
``copy.deepcopy`` for every subclass that accepts ``raw_message=``,
plus the leader-change ``code=10250`` shape for
DqliteConnectionError, plus an end-to-end ``multiprocessing.Queue``
crossing.

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

Author: antoineleclair

src/dqliteclient/exceptions.py

@@ -40,6 +40,29 @@ def __init__(self, *args: object, raw_message: str | None = None) -> None:
         super().__init__(*args)
         self.raw_message = raw_message
 
+    def __reduce__(
+        self,
+    ) -> tuple[type["DqliteError"], tuple[object, ...], dict[str, object]]:
+        # Default ``Exception.__reduce__`` returns ``(cls, self.args)``,
+        # which reconstructs via ``cls(*args)``. That loses every field
+        # set on the instance after ``Exception.__init__`` — most
+        # notably ``raw_message`` (set on the ``DqliteError`` base) and
+        # the ``code`` carried by ``DqliteConnectionError``. The
+        # carriers were added in cycle 27 (XP2 / XP3) precisely so the
+        # wire-level signal would survive cross-process pickling
+        # (``ProcessPoolExecutor``, ``multiprocessing.Queue``, Celery
+        # task results, SA's multiprocess pool); without overriding
+        # ``__reduce__`` the round-trip silently drops them.
+        #
+        # Return the 3-tuple ``(callable, args, state)`` form so pickle
+        # reconstructs via ``cls(*args)`` then applies state via
+        # ``self.__dict__.update(state)`` — preserving every attribute
+        # we set on the instance (``raw_message`` on the base, ``code``
+        # on subclasses, the truncated ``message`` on
+        # ``OperationalError``). All subclasses of ``DqliteError``
+        # inherit this discipline automatically.
+        return (self.__class__, self.args, self.__dict__.copy())
+
 
 class DqliteConnectionError(DqliteError):
     """Error establishing or maintaining connection.

tests/test_dqlite_error_pickle_round_trip_lossless.py

@@ -0,0 +1,117 @@
+"""Pin: every ``DqliteError`` subclass round-trips through pickle /
+deepcopy without losing the ``raw_message`` (base) or ``code``
+(``DqliteConnectionError``) fields.
+
+The fields were added in cycle 27 (XP2 / XP3) so the wire-level
+signal would survive cross-process boundaries — ``ProcessPoolExecutor``,
+``multiprocessing.Queue.put(exception)``, Celery task results, SA's
+multiprocess pool. Without overriding ``__reduce__`` the default
+``Exception`` pickle path (``(cls, self.args)``) silently drops every
+attribute set on the instance after ``Exception.__init__``.
+
+Pin lossless round-trip on every subclass that takes ``raw_message=``
+or ``code=`` so a future regression on the ``__reduce__`` discipline
+fails this test.
+"""
+
+from __future__ import annotations
+
+import copy
+import pickle
+
+import pytest
+
+from dqliteclient.exceptions import (
+    ClusterError,
+    ClusterPolicyError,
+    DataError,
+    DqliteConnectionError,
+    DqliteError,
+    InterfaceError,
+    OperationalError,
+    ProtocolError,
+)
+
+
+@pytest.mark.parametrize("protocol", range(2, pickle.HIGHEST_PROTOCOL + 1))
+def test_dqlite_connection_error_pickle_preserves_code(protocol: int) -> None:
+    e = DqliteConnectionError(
+        "Node host:9001 is no longer leader: not leader",
+        code=10250,
+        raw_message="not leader",
+    )
+    restored = pickle.loads(pickle.dumps(e, protocol=protocol))
+    assert restored.code == 10250
+    assert restored.raw_message == "not leader"
+    assert "Node host:9001" in str(restored)
+
+
+@pytest.mark.parametrize("protocol", range(2, pickle.HIGHEST_PROTOCOL + 1))
+def test_dqlite_connection_error_deepcopy_preserves_code(protocol: int) -> None:
+    e = DqliteConnectionError("leader-flip", code=10506, raw_message="leadership lost")
+    restored = copy.deepcopy(e)
+    assert restored.code == 10506
+    assert restored.raw_message == "leadership lost"
+
+
+def test_dqlite_connection_error_default_construction_pickle_round_trip() -> None:
+    """No-arg / message-only constructions still round-trip cleanly."""
+    e = DqliteConnectionError("Connection refused")
+    restored = pickle.loads(pickle.dumps(e))
+    assert restored.code is None
+    assert restored.raw_message is None
+    assert str(restored) == "Connection refused"
+
+
+@pytest.mark.parametrize(
+    "cls",
+    [DqliteError, DataError, InterfaceError, ClusterError, ClusterPolicyError, ProtocolError],
+)
+def test_subclass_raw_message_round_trips_through_pickle(cls: type) -> None:
+    e = cls("msg", raw_message="server text")
+    restored = pickle.loads(pickle.dumps(e))
+    assert restored.raw_message == "server text"
+    assert str(restored) == "msg"
+
+
+@pytest.mark.parametrize(
+    "cls",
+    [DqliteError, DataError, InterfaceError, ClusterError, ClusterPolicyError, ProtocolError],
+)
+def test_subclass_raw_message_round_trips_through_deepcopy(cls: type) -> None:
+    e = cls("msg", raw_message="server text")
+    restored = copy.deepcopy(e)
+    assert restored.raw_message == "server text"
+
+
+def test_operational_error_pickle_still_lossless_after_dqlite_error_reduce() -> None:
+    """Defence pin: the existing OperationalError pickle contract
+    (cycle-22 truncation+raw_message) must not regress when the base
+    DqliteError implements __reduce__."""
+    payload = "y" * 5000
+    e = OperationalError(19, payload)
+    restored = pickle.loads(pickle.dumps(e))
+    assert restored.raw_message == payload
+    assert restored.code == 19
+    assert len(restored.message) < 1200
+    assert "truncated" in restored.message
+
+
+def test_pickle_round_trip_through_multiprocessing_queue() -> None:
+    """End-to-end: an exception sent through ``multiprocessing.Queue``
+    survives with code and raw_message intact. This is the canonical
+    cross-process surface XP2 was added to plumb."""
+    import multiprocessing
+
+    ctx = multiprocessing.get_context("spawn")
+    q = ctx.Queue()
+    e = DqliteConnectionError(
+        "node failover mid-handshake",
+        code=10250,
+        raw_message="not leader",
+    )
+    q.put(e)
+    restored = q.get(timeout=5)
+    assert isinstance(restored, DqliteConnectionError)
+    assert restored.code == 10250
+    assert restored.raw_message == "not leader"