fix: add poisoned-state flag to ReadBuffer and MessageDecoder

After a parse failure consumes bytes from the buffer but leaves the
stream at an unknown offset, continuing to decode silently desynchronizes
the stream. Go's protocol.Protocol guards against this with a sticky
netErr field; this adds the equivalent for the Python codec.

- ReadBuffer gains poison(), is_poisoned, reset(), and an internal
  _check_poisoned() helper. The flag is sticky (first error wins, never
  overwritten) and cleared only by reset(); plain clear() does not
  un-poison so callers cannot accidentally reuse a dead decoder.

- MessageDecoder forwards is_poisoned and reset(). decode() and
  decode_continuation() check poison at entry and wrap body parsing
  in a try/except that poisons on ANY exception (not just DecodeError
  / ProtocolError) before re-raising. The broader catch is intentional:
  decode_body implementations can raise struct.error, ValueError,
  UnicodeDecodeError, IndexError, etc., and any of those after the
  bytes have been consumed means the stream is desynchronized.

- Oversized-header errors from read_message() are deliberately NOT
  poisoned: the bytes have not been consumed and skip_message() is
  the documented recovery path. Only errors from already-consumed
  bytes mark the decoder dead.

- Server-side decoders' reset() also clears handshake state so a
  reconnect starts from a clean "handshake not yet received" state.

decode_handshake() is intentionally NOT poison-checked here — its
own bytes-leak-on-failure path is the subject of issue 027.

Tests cover: unknown message type poisoning, reset un-poisons,
oversized stays recoverable, non-protocol exceptions also poison,
decode_continuation wrong-type poisons, first-error-wins idempotency,
request-decoder reset clears handshake state.

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

Author: antoineleclair

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
+from dqlitewire.exceptions import DecodeError, ProtocolError
 
 
 class WriteBuffer:
@@ -46,6 +46,36 @@ def __init__(self, max_message_size: int = DEFAULT_MAX_MESSAGE_SIZE) -> None:
         self._pos = 0
         self._max_message_size = max_message_size
         self._skip_remaining = 0
+        self._poisoned: Exception | None = None
+
+    @property
+    def is_poisoned(self) -> bool:
+        """True if a mid-stream error has marked this buffer unrecoverable."""
+        return self._poisoned is not None
+
+    def poison(self, error: Exception) -> None:
+        """Mark the buffer as unrecoverable.
+
+        Call this when a decode error means the stream offset is no longer
+        trustworthy (parsing consumed bytes but failed afterwards). Once
+        poisoned, every public method raises ``ProtocolError`` until
+        ``reset()`` is called.
+        """
+        if self._poisoned is None:
+            self._poisoned = error
+
+    def reset(self) -> None:
+        """Clear buffer state and un-poison. Use after a reconnect."""
+        self._data.clear()
+        self._pos = 0
+        self._skip_remaining = 0
+        self._poisoned = None
+
+    def _check_poisoned(self) -> None:
+        if self._poisoned is not None:
+            raise ProtocolError(
+                "buffer is poisoned; call reset() and reconnect"
+            ) from self._poisoned
 
     def feed(self, data: bytes) -> None:
         """Add received data to the buffer.

src/dqlitewire/codec.py

@@ -154,6 +154,28 @@ def version(self) -> int | None:
         """Protocol version from handshake, or None if not yet received."""
         return self._version
 
+    @property
+    def is_poisoned(self) -> bool:
+        """True if the decoder has hit an unrecoverable mid-stream error.
+
+        Once poisoned, every ``decode*`` method raises ``ProtocolError`` until
+        ``reset()`` is called. This protects callers from silently continuing
+        to decode from an unknown offset after a parse failure desynchronized
+        the stream.
+        """
+        return self._buffer.is_poisoned
+
+    def reset(self) -> None:
+        """Reset decoder state to a fresh, un-poisoned condition.
+
+        Clears the read buffer and, for server-side decoders, returns the
+        handshake state to "not yet received". Use this after a reconnect.
+        """
+        self._buffer.reset()
+        if self._is_request:
+            self._handshake_done = False
+            self._version = None
+
     def feed(self, data: bytes) -> None:
         """Feed data to the decoder."""
         self._buffer.feed(data)
@@ -198,42 +220,67 @@ def decode_continuation(
 
         Returns None if no complete message is available.
         """
+        self._buffer._check_poisoned()
         data = self._buffer.read_message()
         if data is None:
             return None
 
-        header = Header.decode(data[:HEADER_SIZE])
-        body = data[HEADER_SIZE : HEADER_SIZE + header.size_words * 8]
-
-        if header.msg_type == ResponseType.FAILURE:
-            failure = FailureResponse.decode_body(body, schema=header.schema)
-            raise ProtocolError(
-                f"Server error during ROWS continuation: [{failure.code}] {failure.message}"
+        # Bytes have been consumed — ANY parse failure here leaves the
+        # buffer at an unknown offset and must poison the decoder. Catch
+        # broadly: `decode_body` implementations can raise struct.error,
+        # ValueError, UnicodeDecodeError, IndexError, etc., and all of
+        # them mean the stream is desynchronized.
+        try:
+            header = Header.decode(data[:HEADER_SIZE])
+            body = data[HEADER_SIZE : HEADER_SIZE + header.size_words * 8]
+
+            if header.msg_type == ResponseType.FAILURE:
+                failure = FailureResponse.decode_body(body, schema=header.schema)
+                raise ProtocolError(
+                    f"Server error during ROWS continuation: [{failure.code}] {failure.message}"
+                )
+            if header.msg_type != ResponseType.ROWS:
+                raise ProtocolError(
+                    f"Expected ROWS continuation (type {ResponseType.ROWS}), "
+                    f"got type {header.msg_type}"
+                )
+
+            return RowsResponse.decode_rows_continuation(
+                body, column_names, column_count, max_rows=max_rows
             )
-        if header.msg_type != ResponseType.ROWS:
-            raise ProtocolError(
-                f"Expected ROWS continuation (type {ResponseType.ROWS}), got type {header.msg_type}"
-            )
-
-        return RowsResponse.decode_rows_continuation(
-            body, column_names, column_count, max_rows=max_rows
-        )
+        except Exception as e:
+            self._buffer.poison(e)
+            raise
 
     def decode(self) -> Message | None:
         """Decode the next message from the buffer.
 
         Returns None if no complete message is available.
         Raises ProtocolError if called on a request decoder before decode_handshake().
+        Raises ProtocolError if the decoder is poisoned.
         """
+        self._buffer._check_poisoned()
         if not self._handshake_done:
             raise ProtocolError(
                 "Protocol handshake not yet received. Call decode_handshake() before decode()."
             )
+        # read_message may raise DecodeError for an oversized header. That
+        # error is recoverable via skip_message() — the bytes have not been
+        # consumed — so we deliberately do NOT poison on it.
         data = self._buffer.read_message()
         if data is None:
             return None
 
-        return self.decode_bytes(data)
+        # Bytes have been consumed. ANY parse failure now leaves the
+        # buffer at an unknown offset; poison so subsequent calls fail
+        # fast. Catch broadly: `decode_body` implementations can raise
+        # struct.error, ValueError, UnicodeDecodeError, IndexError, etc.,
+        # and all of them mean the stream is desynchronized.
+        try:
+            return self.decode_bytes(data)
+        except Exception as e:
+            self._buffer.poison(e)
+            raise
 
     def decode_bytes(self, data: bytes) -> Message:
         """Decode a message from bytes.

tests/test_codec.py

@@ -882,3 +882,185 @@ def test_skip_oversized_and_recover(self) -> None:
         decoded = decoder.decode()
         assert isinstance(decoded, FailureResponse)
         assert decoded.code == 42
+
+
+class TestDecoderPoisonedState:
+    """Once the decoder has produced a mid-stream error from already-consumed
+    bytes, subsequent operations must fail fast with a poisoned-state error —
+    the buffer's _pos is in an unknown position and silently continuing would
+    corrupt downstream reads. A reset() method returns the decoder to a clean
+    state after a reconnect.
+    """
+
+    def test_decode_poisons_on_unknown_message_type(self) -> None:
+        """An unknown message type surfaces via decode(), which must poison
+        the decoder so retries fail loudly.
+        """
+        import struct
+
+        from dqlitewire.exceptions import ProtocolError
+
+        decoder = MessageDecoder(is_request=False)
+        # Valid framing, but an unregistered message type (0xFE).
+        header = struct.pack("<IBBH", 0, 0xFE, 0, 0)
+        decoder.feed(header)
+
+        assert decoder.is_poisoned is False
+        with pytest.raises(DecodeError):
+            decoder.decode()
+        assert decoder.is_poisoned is True
+
+        # Subsequent operations fail fast with a ProtocolError from poisoning.
+        # decode() must raise regardless of what is in the buffer.
+        decoder.feed(struct.pack("<IBBH", 0, 0xFE, 0, 0))
+        with pytest.raises(ProtocolError, match="poisoned"):
+            decoder.decode()
+
+    def test_reset_unpoisons_decoder(self) -> None:
+        """reset() clears buffer state and the poison flag so the decoder
+        can be reused after a reconnect.
+        """
+        import struct
+
+        from dqlitewire.exceptions import ProtocolError
+
+        decoder = MessageDecoder(is_request=False)
+        header = struct.pack("<IBBH", 0, 0xFE, 0, 0)
+        decoder.feed(header)
+        with pytest.raises(DecodeError):
+            decoder.decode()
+        assert decoder.is_poisoned is True
+
+        decoder.reset()
+        assert decoder.is_poisoned is False
+
+        # Fresh valid message should decode cleanly.
+        leader = LeaderResponse(node_id=1, address="host:1234").encode()
+        decoder.feed(leader)
+        result = decoder.decode()
+        assert isinstance(result, LeaderResponse)
+        assert result.node_id == 1
+
+        # Sanity: a fresh poison error references a ProtocolError ancestor too.
+        decoder.feed(struct.pack("<IBBH", 0, 0xFE, 0, 0))
+        with pytest.raises(ProtocolError):
+            decoder.decode()
+
+    def test_oversized_header_does_not_poison(self) -> None:
+        """An oversized-header error from the buffer framing layer is
+        recoverable via skip_message(); it must NOT poison because the
+        bytes have not yet been consumed from the buffer.
+        """
+        import struct
+
+        decoder = MessageDecoder(is_request=False)
+        decoder._buffer._max_message_size = 128
+        oversized = struct.pack("<IBBH", 100, 0, 0, 0)  # 808-byte body > 128
+        decoder.feed(oversized)
+
+        with pytest.raises(DecodeError, match="exceeds maximum"):
+            decoder.decode()
+        # NOT poisoned — skip_message() is the documented recovery path.
+        assert decoder.is_poisoned is False
+
+        # Recovery via skip_message should still work.
+        decoder.skip_message()
+        decoder.feed(b"\x00" * decoder._buffer._skip_remaining)
+        assert decoder.is_poisoned is False
+
+        # And a fresh message decodes afterwards.
+        normal = FailureResponse(code=1, message="ok").encode()
+        decoder.feed(normal)
+        assert isinstance(decoder.decode(), FailureResponse)
+
+    def test_poison_catches_non_protocol_exceptions(self) -> None:
+        """Any exception from decode_bytes — not just DecodeError/ProtocolError —
+        must poison the decoder. A `decode_body` implementation could raise
+        struct.error, ValueError, UnicodeDecodeError, IndexError, etc., and
+        all of them mean the bytes have been consumed and the offset is now
+        unknown. Catching only the protocol exception types would let other
+        failures leak through with the buffer in a desynchronized state.
+        """
+        from dqlitewire.exceptions import ProtocolError
+
+        decoder = MessageDecoder(is_request=False)
+        # Build a real message and feed it.
+        msg = LeaderResponse(node_id=1, address="x:1").encode()
+        decoder.feed(msg)
+
+        # Monkey-patch decode_bytes to raise a non-protocol exception, simulating
+        # a bug or unexpected input in a real decode_body implementation.
+        sentinel = ValueError("simulated body parse failure")
+
+        def boom(_data: bytes) -> object:
+            raise sentinel
+
+        decoder.decode_bytes = boom  # type: ignore[method-assign]
+
+        with pytest.raises(ValueError, match="simulated body parse failure"):
+            decoder.decode()
+        assert decoder.is_poisoned is True
+
+        # Subsequent decode() must raise poisoned, even though decode_bytes
+        # is monkey-patched. Restore the real decode_bytes first to make
+        # sure the poison check fires before any decode_bytes call.
+        del decoder.decode_bytes  # type: ignore[attr-defined]
+        decoder.feed(msg)
+        with pytest.raises(ProtocolError, match="poisoned"):
+            decoder.decode()
+
+    def test_decode_continuation_poisons_on_wrong_type(self) -> None:
+        """decode_continuation() must poison the decoder when the next
+        message is not a ROWS continuation. The bytes have been consumed
+        from the buffer; subsequent operations cannot trust the offset.
+        """
+        from dqlitewire.exceptions import ProtocolError
+
+        decoder = MessageDecoder(is_request=False)
+        # Feed a non-ROWS message where decode_continuation expects ROWS.
+        # An EmptyResponse is type 127, body is empty.
+        empty = EmptyResponse().encode()
+        decoder.feed(empty)
+
+        with pytest.raises(ProtocolError, match="Expected ROWS continuation"):
+            decoder.decode_continuation(column_names=["x"], column_count=1)
+        assert decoder.is_poisoned is True
+
+        # Subsequent decode() also fails poisoned.
+        with pytest.raises(ProtocolError, match="poisoned"):
+            decoder.decode()
+
+    def test_poison_is_first_error_wins(self) -> None:
+        """ReadBuffer.poison() must NOT overwrite an existing poison error.
+        First error wins so the original __cause__ remains visible.
+        """
+        from dqlitewire.buffer import ReadBuffer
+
+        buf = ReadBuffer()
+        first = DecodeError("first failure")
+        second = DecodeError("second failure")
+        buf.poison(first)
+        buf.poison(second)
+        assert buf._poisoned is first
+
+    def test_request_decoder_reset_clears_handshake_state(self) -> None:
+        """For a server-side (request) decoder, reset() must also un-do the
+        handshake so a reconnect requires a fresh handshake. Otherwise the
+        decoder would silently accept message bytes as a continuation of
+        the dead session.
+        """
+        decoder = MessageDecoder(is_request=True)
+        # Complete a handshake.
+        decoder.feed(PROTOCOL_VERSION.to_bytes(8, "little"))
+        decoder.decode_handshake()
+        assert decoder._handshake_done is True
+        assert decoder.version == PROTOCOL_VERSION
+
+        decoder.reset()
+        assert decoder._handshake_done is False
+        assert decoder.version is None
+        # And the decoder must refuse decode() until a fresh handshake.
+        from dqlitewire.exceptions import ProtocolError
+
+        with pytest.raises(ProtocolError, match="[Hh]andshake"):
+            decoder.decode()