feat: split ProtocolError into specific subclasses

antoineleclair · claude · antoineleclair · commit 9db83909fcf2 · 2026-04-16T20:51:23.000-04:00
ProtocolError was raised at six qualitatively different sites —
server-returned FAILURE, stream-desync, handshake-not-done, bad
protocol version, ROWS continuation state-machine misuse, and buffer
poisoning — forcing callers to string-match exception messages to
recover. Add specific subclasses under ProtocolError so callers can
discriminate via the exception class:

- ServerFailure (with structured code/message attributes): peer sent
  a FAILURE response; connection is still usable.
- StreamError: stream is at unknown offset; reconnect required.
- PoisonedError (StreamError): buffer was poisoned by a prior
  failure; __cause__ carries the original exception.
- HandshakeError: handshake not done, already done, or bad version.
- ContinuationError: ROWS continuation state-machine misuse.

Every new subclass inherits from ProtocolError, so existing
`except ProtocolError` catches continue to work unchanged.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/src/dqlitewire/__init__.py b/src/dqlitewire/__init__.py
@@ -54,16 +54,28 @@
     ResponseType,
     ValueType,
 )
-from dqlitewire.exceptions import DecodeError, EncodeError, ProtocolError
+from dqlitewire.exceptions import (
+    ContinuationError,
+    DecodeError,
+    EncodeError,
+    HandshakeError,
+    PoisonedError,
+    ProtocolError,
+    ServerFailure,
+    StreamError,
+)
 
 __all__ = [
+    "ContinuationError",
     "DecodeError",
     "EncodeError",
+    "HandshakeError",
     "MessageDecoder",
     "MessageEncoder",
     "NodeRole",
     "PROTOCOL_VERSION",
     "PROTOCOL_VERSION_LEGACY",
+    "PoisonedError",
     "ProtocolError",
     "ROW_DONE_BYTE",
     "ROW_DONE_MARKER",
@@ -72,6 +84,8 @@
     "ReadBuffer",
     "RequestType",
     "ResponseType",
+    "ServerFailure",
+    "StreamError",
     "ValueType",
     "WriteBuffer",
     "decode_message",
diff --git a/src/dqlitewire/buffer.py b/src/dqlitewire/buffer.py
@@ -1,7 +1,7 @@
 """Buffer utilities for streaming protocol data."""
 
 from dqlitewire.constants import HEADER_SIZE, WORD_SIZE
-from dqlitewire.exceptions import DecodeError, ProtocolError
+from dqlitewire.exceptions import DecodeError, PoisonedError
 
 _COMPACT_THRESHOLD = 4096
 
@@ -134,7 +134,7 @@ def reset(self) -> None:
 
     def _check_poisoned(self) -> None:
         if self._poisoned is not None:
-            raise ProtocolError(
+            raise PoisonedError(
                 "buffer is poisoned; call reset() and reconnect"
             ) from self._poisoned
 
diff --git a/src/dqlitewire/codec.py b/src/dqlitewire/codec.py
@@ -8,7 +8,13 @@
     RequestType,
     ResponseType,
 )
-from dqlitewire.exceptions import DecodeError, ProtocolError
+from dqlitewire.exceptions import (
+    ContinuationError,
+    DecodeError,
+    HandshakeError,
+    ServerFailure,
+    StreamError,
+)
 from dqlitewire.messages.base import Header, Message
 from dqlitewire.messages.requests import (
     AddRequest,
@@ -126,7 +132,7 @@ def __init__(self, version: int = PROTOCOL_VERSION) -> None:
                      (0x86104dd760433fe5) for pre-1.0 dqlite servers.
         """
         if version not in _SUPPORTED_VERSIONS:
-            raise ProtocolError(
+            raise HandshakeError(
                 f"Unsupported protocol version: {version:#x}. "
                 f"Supported: {', '.join(f'{v:#x}' for v in sorted(_SUPPORTED_VERSIONS))}"
             )
@@ -201,7 +207,7 @@ def __init__(
                     this limit raises ``DecodeError``.
         """
         if not is_request and version not in _SUPPORTED_VERSIONS:
-            raise ProtocolError(
+            raise HandshakeError(
                 f"Unsupported protocol version: {version:#x}. "
                 f"Supported: {', '.join(f'{v:#x}' for v in sorted(_SUPPORTED_VERSIONS))}"
             )
@@ -270,7 +276,7 @@ def skip_message(self) -> bool:
         abandon the stream.
         """
         if self._continuation_expected:
-            raise ProtocolError(
+            raise ContinuationError(
                 "Cannot skip a message while a ROWS continuation is "
                 "in progress. Call decode_continuation() to drain, "
                 "or reset() to abandon the stream."
@@ -300,7 +306,7 @@ def decode_continuation(self) -> RowsResponse | None:
         """
         self._buffer._check_poisoned()
         if not self._continuation_expected:
-            raise ProtocolError(
+            raise ContinuationError(
                 "decode_continuation() called but no ROWS continuation "
                 "is in progress. Use decode() for the initial message."
             )
@@ -332,12 +338,10 @@ def decode_continuation(self) -> RowsResponse | None:
             if header.msg_type == ResponseType.FAILURE:
                 self._continuation_expected = False
                 failure = FailureResponse.decode_body(body, schema=header.schema)
-                raise ProtocolError(
-                    f"Server error during ROWS continuation: [{failure.code}] {failure.message}"
-                )
+                raise ServerFailure(failure.code, failure.message)
             if header.msg_type != ResponseType.ROWS:
                 self._continuation_expected = False
-                raise ProtocolError(
+                raise StreamError(
                     f"Expected ROWS continuation (type {ResponseType.ROWS}), "
                     f"got type {header.msg_type}"
                 )
@@ -366,13 +370,13 @@ def decode(self) -> Message | None:
         """
         self._buffer._check_poisoned()
         if self._continuation_expected:
-            raise ProtocolError(
+            raise ContinuationError(
                 "Cannot decode a new message while a ROWS continuation "
                 "is in progress. Call decode_continuation() until "
                 "has_more is False, or call reset() to abandon the stream."
             )
         if not self._handshake_done:
-            raise ProtocolError(
+            raise HandshakeError(
                 "Protocol handshake not yet received. Call decode_handshake() before decode()."
             )
         # read_message may raise DecodeError for an oversized header. That
@@ -421,7 +425,7 @@ def decode_bytes(self, data: bytes) -> Message:
         """
         self._buffer._check_poisoned()
         if not self._handshake_done:
-            raise ProtocolError(
+            raise HandshakeError(
                 "Protocol handshake not yet received. "
                 "Call decode_handshake() before decode_bytes()."
             )
@@ -490,7 +494,7 @@ def decode_handshake(self) -> int | None:
         protocol version" error.
         """
         if self._handshake_done:
-            raise ProtocolError("Handshake already completed")
+            raise HandshakeError("Handshake already completed")
         # Peek first so we only commit on a valid version. An invalid version
         # leaves the bytes in place — a retry is deterministic rather than
         # silently advancing into real message data.
@@ -499,7 +503,7 @@ def decode_handshake(self) -> int | None:
             return None
         version = int.from_bytes(peek, "little")
         if version not in _SUPPORTED_VERSIONS:
-            raise ProtocolError(f"Unsupported protocol version: {version:#x}")
+            raise HandshakeError(f"Unsupported protocol version: {version:#x}")
         # Commit state BEFORE consuming bytes. If the consume is
         # interrupted by an async exception, revert so the peek/commit
         # pair becomes atomic from the caller's perspective.
diff --git a/src/dqlitewire/exceptions.py b/src/dqlitewire/exceptions.py
@@ -1,19 +1,90 @@
-"""Exceptions for dqlite wire protocol."""
+"""Exceptions for dqlite wire protocol.
+
+All exceptions inherit from ``ProtocolError`` so callers can keep a
+single broad ``except ProtocolError`` catch. Callers who need to
+distinguish recoverable server errors from fatal stream desync can
+catch the more specific subclasses below.
+"""
 
 
 class ProtocolError(Exception):
     """Base exception for protocol errors."""
 
-    pass
-
 
 class EncodeError(ProtocolError):
     """Error during message encoding."""
 
-    pass
-
 
 class DecodeError(ProtocolError):
     """Error during message decoding."""
 
-    pass
+
+class StreamError(ProtocolError):
+    """The stream is at an unknown offset; the connection must be rebuilt.
+
+    Callers catching this should close and re-establish the underlying
+    socket. The decoder and buffer are not recoverable — any remaining
+    buffered bytes are at an unknown protocol offset.
+    """
+
+
+class PoisonedError(StreamError):
+    """Raised by ``ReadBuffer._check_poisoned`` when the buffer was
+    previously poisoned by a decode failure or signal interruption.
+
+    The original exception that caused the poisoning is available via
+    ``__cause__``.
+    """
+
+
+class HandshakeError(ProtocolError):
+    """The handshake has not been completed, has been completed with
+    an unsupported version, or an attempt was made to re-perform an
+    already-completed handshake. Caller should call ``decode_handshake()``
+    before attempting to decode messages, and should only call it once.
+    """
+
+
+class ContinuationError(ProtocolError):
+    """The ROWS continuation state machine was used incorrectly — e.g.
+    ``decode()`` was called while a multipart ``RowsResponse`` was in
+    progress, or ``decode_continuation()`` was called when no
+    continuation was pending.
+
+    Recoverable: call the correct method (``decode_continuation()`` or
+    ``reset()``) and proceed.
+    """
+
+
+class ServerFailure(ProtocolError):
+    """The peer returned a FAILURE response.
+
+    Semantically a server-side error (e.g. SQL constraint violation,
+    busy database, or the dqlite-specific "not leader" response) rather
+    than a wire-level corruption. The SQLite error code is exposed
+    programmatically so callers can distinguish recoverable conditions
+    (like ``SQLITE_IOERR_NOT_LEADER``, which callers may handle by
+    redirecting to the cluster leader) from terminal ones.
+
+    Connection state: in the normal ``decode()`` path the server
+    returns a ``FailureResponse`` *object* (not an exception), so no
+    state is disturbed. This exception is raised only from
+    ``decode_continuation()`` when FAILURE arrives mid-stream during
+    a multi-part ``RowsResponse``. In that path the buffer is also
+    poisoned (see issue 083): the remaining continuation frames the
+    caller was expecting will never arrive, and the next decode must
+    start fresh. Callers catching ``ServerFailure`` in that path
+    should treat the decoder as requiring ``reset()`` and reconnect.
+
+    Attributes:
+        code: SQLite error code (or extended dqlite code). Common codes
+            include ``SQLITE_BUSY`` (5), ``SQLITE_ERROR`` (1), and the
+            dqlite-specific ``SQLITE_IOERR_NOT_LEADER``. See the SQLite
+            documentation for the full list.
+        message: Human-readable error message from the server.
+    """
+
+    def __init__(self, code: int, message: str) -> None:
+        super().__init__(f"[{code}] {message}")
+        self.code = code
+        self.message = message
diff --git a/tests/test_codec.py b/tests/test_codec.py
@@ -1120,8 +1120,16 @@ def test_decode_continuation_returns_none_when_no_data(self) -> None:
         assert result is None
 
     def test_decode_continuation_raises_on_failure_response(self) -> None:
-        """decode_continuation should surface server error, not a confusing DecodeError."""
-        from dqlitewire.exceptions import ProtocolError
+        """decode_continuation should surface server error as ServerFailure.
+
+        The exception must be a ServerFailure (subclass of ProtocolError)
+        carrying structured ``code`` and ``message`` attributes — NOT a
+        DecodeError (which would mean the failure body was misinterpreted
+        as row data) and NOT a bare ProtocolError (issue 230 — callers
+        need to distinguish recoverable server errors from fatal stream
+        desync).
+        """
+        from dqlitewire.exceptions import ServerFailure
         from dqlitewire.messages.responses import FailureResponse
 
         failure = FailureResponse(code=266, message="disk I/O error")
@@ -1131,11 +1139,10 @@ def test_decode_continuation_raises_on_failure_response(self) -> None:
         decoder._continuation_expected = True
         decoder.feed(failure_bytes)
 
-        with pytest.raises(ProtocolError, match="disk I/O error") as exc_info:
+        with pytest.raises(ServerFailure, match="disk I/O error") as exc_info:
             decoder.decode_continuation()
-        # Must be a ProtocolError, NOT a DecodeError (which would mean the
-        # failure body was misinterpreted as row data).
-        assert type(exc_info.value) is ProtocolError
+        assert exc_info.value.code == 266
+        assert exc_info.value.message == "disk I/O error"
 
     def test_decode_continuation_raises_on_unexpected_type(self) -> None:
         """decode_continuation should raise ProtocolError for non-ROWS, non-FAILURE type."""
diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py
