fix: gate ReadBuffer public API on poison flag

Issue 026 introduced the poison flag so that once a parse error
desynchronizes the stream, subsequent operations fail fast with
ProtocolError. The check was wired into MessageDecoder.decode() /
decode_continuation() but NOT into ReadBuffer's own public API. A
caller holding a raw ReadBuffer (exported from the dqlitewire
package) could keep calling feed()/read_message()/peek_header()/...
after poisoning and re-desynchronize the stream silently.

Gate every mutating/consuming public method on ReadBuffer on
self._check_poisoned(): feed, read_message, skip_message,
read_bytes, peek_bytes, peek_header. Observers (has_message,
available, is_poisoned, is_skipping) and recovery primitives
(clear, reset, poison) remain total and do not raise.

Two existing TestDecoderPoisonedState tests pushed bytes through
feed() after a poison to assert that the sticky state survived.
That contract is now stricter: feed() itself refuses, which
surfaces the poison earlier. Update the tests to assert
ProtocolError on the feed() call and still require reset() for
recovery.

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

Author: antoineleclair

src/dqlitewire/buffer.py

@@ -92,7 +92,11 @@ def feed(self, data: bytes) -> None:
         If an oversized message is being skipped (after skip_message() returned
         False), incoming bytes are silently discarded until the full oversized
         message has been consumed.
+
+        Raises ``ProtocolError`` if the buffer is poisoned — callers must
+        ``reset()`` (or ``clear()``) before feeding further data.
         """
+        self._check_poisoned()
         if self._skip_remaining > 0:
             discard = min(len(data), self._skip_remaining)
             data = data[discard:]
@@ -146,7 +150,10 @@ def peek_header(self) -> tuple[int, int, int] | None:
         ``max_message_size`` guard the caller is presumably relying on. The
         error message and exception type match ``read_message()`` so the
         consume-side recovery path (``skip_message()``) applies the same way.
+
+        Raises ``ProtocolError`` if the buffer is poisoned.
         """
+        self._check_poisoned()
         available = len(self._data) - self._pos
 
         if available < HEADER_SIZE:
@@ -174,7 +181,9 @@ def read_message(self) -> bytes | None:
         Returns the message data (including header) or None if not enough data.
         Raises ``DecodeError`` if the next buffered header claims a message
         larger than ``max_message_size``. Use ``skip_message()`` to recover.
+        Raises ``ProtocolError`` if the buffer is poisoned.
         """
+        self._check_poisoned()
         available = len(self._data) - self._pos
         if available < HEADER_SIZE:
             return None
@@ -212,7 +221,10 @@ def skip_message(self) -> bool:
         Returns True if a message was fully skipped, False if not enough data
         for a header, a normal-sized message is still incomplete, or an
         oversized message skip is still in progress (check ``is_skipping``).
+
+        Raises ``ProtocolError`` if the buffer is poisoned.
         """
+        self._check_poisoned()
         available = len(self._data) - self._pos
         if available < HEADER_SIZE:
             return False
@@ -244,7 +256,9 @@ def read_bytes(self, n: int) -> bytes | None:
         """Read exactly n bytes from the buffer.
 
         Returns None if not enough data available.
+        Raises ``ProtocolError`` if the buffer is poisoned.
         """
+        self._check_poisoned()
         available = len(self._data) - self._pos
         if available < n:
             return None
@@ -260,7 +274,10 @@ def peek_bytes(self, n: int) -> bytes | None:
         Returns None if fewer than n bytes are available. Symmetric with
         ``read_bytes(n)`` but non-consuming — use this when you need to
         validate the bytes before deciding whether to consume them.
+
+        Raises ``ProtocolError`` if the buffer is poisoned.
         """
+        self._check_poisoned()
         available = len(self._data) - self._pos
         if available < n:
             return None

tests/test_buffer.py

@@ -216,6 +216,57 @@ def test_clear_also_unpoisons(self) -> None:
         buf.feed(b"\x01\x00\x00\x00\x00\x00\x00\x00")
         assert buf.available() == 8
 
+    def test_public_api_honors_poison(self) -> None:
+        """Regression for issue 038.
+
+        Issue 026 introduced the poison flag so that once a decode
+        error has desynchronized the stream, subsequent operations
+        fail fast with ``ProtocolError`` instead of silently reading
+        from an unknown offset. The check was wired into
+        ``MessageDecoder.decode`` / ``decode_continuation`` but NOT
+        into ``ReadBuffer``'s own public API. A caller holding a raw
+        ``ReadBuffer`` (exported from ``dqlitewire`` since
+        ``__init__.py``) could keep calling ``feed()``, ``read_message()``,
+        etc., after a poison was set and re-desynchronize the stream
+        silently.
+
+        Every mutating/consuming public method on ``ReadBuffer`` must
+        raise ``ProtocolError`` when poisoned. Pure observers
+        (``has_message``, ``available``, ``is_poisoned``, ``is_skipping``)
+        and recovery primitives (``clear``, ``reset``, ``poison``)
+        must remain callable.
+        """
+        import pytest
+
+        from dqlitewire.exceptions import DecodeError, ProtocolError
+
+        buf = ReadBuffer()
+        buf.poison(DecodeError("original cause"))
+
+        cases: list[tuple[str, object]] = [
+            ("feed", lambda: buf.feed(b"x" * 8)),
+            ("read_message", lambda: buf.read_message()),
+            ("skip_message", lambda: buf.skip_message()),
+            ("read_bytes", lambda: buf.read_bytes(4)),
+            ("peek_bytes", lambda: buf.peek_bytes(4)),
+            ("peek_header", lambda: buf.peek_header()),
+        ]
+        for name, call in cases:
+            with pytest.raises(ProtocolError, match="poisoned") as ei:
+                call()  # type: ignore[operator]
+            assert isinstance(ei.value.__cause__, DecodeError), name
+
+        # Observers must remain callable on a poisoned buffer.
+        _ = buf.available()
+        _ = buf.has_message()
+        _ = buf.is_poisoned
+        _ = buf.is_skipping
+
+        # Recovery path must still work: reset() unpoisons.
+        buf.reset()
+        assert not buf.is_poisoned
+        buf.feed(b"\x00" * 8)
+
     def test_has_message_is_total_on_oversized(self) -> None:
         """has_message() must not raise — oversized headers surface at consume time.
 

tests/test_codec.py

@@ -1015,8 +1015,11 @@ def test_decode_poisons_on_unknown_message_type(self) -> None:
         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))
+        # Per issue 038, every mutating/consuming entry point on the buffer
+        # and decoder — including feed() — refuses to run on a poisoned
+        # buffer. Recovery requires reset().
+        with pytest.raises(ProtocolError, match="poisoned"):
+            decoder.feed(struct.pack("<IBBH", 0, 0xFE, 0, 0))
         with pytest.raises(ProtocolError, match="poisoned"):
             decoder.decode()
 
@@ -1108,8 +1111,12 @@ def boom(_data: bytes) -> object:
         # 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.
+        # Per issue 038, feed() on a poisoned buffer also raises, so we
+        # cannot push more bytes through without a reset — the poison
+        # check fires directly on decode().
         del decoder.decode_bytes  # type: ignore[attr-defined]
-        decoder.feed(msg)
+        with pytest.raises(ProtocolError, match="poisoned"):
+            decoder.feed(msg)
         with pytest.raises(ProtocolError, match="poisoned"):
             decoder.decode()