fix: wrap read_message/skip_message/read_bytes in BaseException poison guard

All three consume methods advance `_pos` then CALL `_maybe_compact()`.
The CALL is a CPython eval-breaker checkpoint — a pending signal
(KeyboardInterrupt, PyThreadState_SetAsyncExc, CancelledError) is
delivered there before _maybe_compact's body runs. If the signal
fires at that boundary, `_pos` has advanced (bytes consumed from
the buffer) but the return value is lost with the unwinding frame,
and no poison fires because _maybe_compact's own try/except never
executes.

In a request-response protocol, silently dropping one response
causes the caller to receive the next response as the answer to
the current request — application-layer data corruption.

Wrap each method's mutation block in `try/except BaseException:
poison; raise`, matching the template from issues 037
(_maybe_compact) and 048 (feed). The DecodeError for oversized
headers stays outside the try block to preserve the non-poisoning
recoverable-via-skip_message contract.

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

Author: antoineleclair

src/dqlitewire/buffer.py

@@ -316,9 +316,17 @@ def read_message(self) -> bytes | None:
         if available < total_size:
             return None
 
-        message = bytes(self._data[self._pos : self._pos + total_size])
-        self._pos += total_size
-        self._maybe_compact()
+        try:
+            message = bytes(self._data[self._pos : self._pos + total_size])
+            self._pos += total_size
+            self._maybe_compact()
+        except BaseException as e:
+            if self._poisoned is None:
+                if isinstance(e, Exception):
+                    self._poisoned = e
+                else:
+                    self._poisoned = RuntimeError(f"read_message interrupted: {type(e).__name__}")
+            raise
 
         return message
 
@@ -362,24 +370,33 @@ def skip_message(self) -> bool:
             raise err
         total_size = HEADER_SIZE + (size_words * WORD_SIZE)
 
-        if total_size <= self._max_message_size:
-            # Normal-sized message: only skip when complete to avoid
-            # stream corruption from partially consumed messages.
-            if available < total_size:
-                return False
-            self._pos += total_size
-        else:
-            # Oversized message: discard what we have and track remaining
-            # bytes to be discarded in subsequent feed() calls.
-            # Cap to max_message_size to prevent amplification attacks where
-            # an 8-byte header claiming a ~32 GiB body would cause that much
-            # legitimate data to be silently discarded.
-            effective_total = min(total_size, self._max_message_size)
-            skip_now = min(effective_total, available)
-            self._pos += skip_now
-            self._skip_remaining = effective_total - skip_now
-
-        self._maybe_compact()
+        # Normal-sized message: only skip when complete to avoid
+        # stream corruption from partially consumed messages.
+        if total_size <= self._max_message_size and available < total_size:
+            return False
+
+        try:
+            if total_size <= self._max_message_size:
+                self._pos += total_size
+            else:
+                # Oversized message: discard what we have and track remaining
+                # bytes to be discarded in subsequent feed() calls.
+                # Cap to max_message_size to prevent amplification attacks
+                # where an 8-byte header claiming a ~32 GiB body would cause
+                # that much legitimate data to be silently discarded.
+                effective_total = min(total_size, self._max_message_size)
+                skip_now = min(effective_total, available)
+                self._pos += skip_now
+                self._skip_remaining = effective_total - skip_now
+            self._maybe_compact()
+        except BaseException as e:
+            if self._poisoned is None:
+                if isinstance(e, Exception):
+                    self._poisoned = e
+                else:
+                    self._poisoned = RuntimeError(f"skip_message interrupted: {type(e).__name__}")
+            raise
+
         return self._skip_remaining == 0
 
     def read_bytes(self, n: int) -> bytes | None:
@@ -393,9 +410,17 @@ def read_bytes(self, n: int) -> bytes | None:
         if available < n:
             return None
 
-        data = bytes(self._data[self._pos : self._pos + n])
-        self._pos += n
-        self._maybe_compact()
+        try:
+            data = bytes(self._data[self._pos : self._pos + n])
+            self._pos += n
+            self._maybe_compact()
+        except BaseException as e:
+            if self._poisoned is None:
+                if isinstance(e, Exception):
+                    self._poisoned = e
+                else:
+                    self._poisoned = RuntimeError(f"read_bytes interrupted: {type(e).__name__}")
+            raise
         return data
 
     def peek_bytes(self, n: int) -> bytes | None:

tests/test_buffer_signal_safety.py

@@ -429,3 +429,88 @@ def test_wire_legal_max_size_still_routes_through_oversized_path(self) -> None:
             "oversized-but-wire-legal DecodeError must remain non-poisoning "
             "to preserve the skip_message() recovery contract"
         )
+
+
+class TestConsumeMethodSignalSafety:
+    """Regression tests for issue 061.
+
+    ``read_message()``, ``skip_message()``, and ``read_bytes()`` each
+    advance ``_pos`` and then call ``_maybe_compact()``. The CALL to
+    ``_maybe_compact`` is a CPython eval-breaker check point — a pending
+    signal is delivered there BEFORE the function body runs. If the
+    signal fires at that boundary, ``_pos`` has advanced (bytes consumed)
+    but ``_maybe_compact``'s own try/except never executes, so no poison
+    fires. The return value is lost with the unwinding frame.
+
+    The fix wraps the mutation block in ``try/except BaseException``
+    matching the template from issues 037 (``_maybe_compact``) and
+    048 (``feed``).
+    """
+
+    def test_torn_read_message_leaves_buffer_poisoned(self) -> None:
+        buf = ReadBuffer()
+        buf.feed(b"\x01\x00\x00\x00\x00\x00\x00\x00" + b"\x00" * 8)
+
+        tracer = _raise_on_source_match("read_message", "self._maybe_compact()")
+
+        sys.settrace(tracer)
+        try:
+            with contextlib.suppress(KeyboardInterrupt):
+                buf.read_message()
+        finally:
+            sys.settrace(None)
+
+        assert buf.is_poisoned, (
+            "read_message must poison when a BaseException escapes its mutation block"
+        )
+        with pytest.raises(ProtocolError, match="poisoned"):
+            buf.read_message()
+
+    def test_torn_read_bytes_leaves_buffer_poisoned(self) -> None:
+        buf = ReadBuffer()
+        buf.feed(b"\x00" * 32)
+
+        tracer = _raise_on_source_match("read_bytes", "self._maybe_compact()")
+
+        sys.settrace(tracer)
+        try:
+            with contextlib.suppress(KeyboardInterrupt):
+                buf.read_bytes(8)
+        finally:
+            sys.settrace(None)
+
+        assert buf.is_poisoned, (
+            "read_bytes must poison when a BaseException escapes its mutation block"
+        )
+
+    def test_torn_skip_message_leaves_buffer_poisoned(self) -> None:
+        buf = ReadBuffer()
+        buf.feed(b"\x01\x00\x00\x00\x00\x00\x00\x00" + b"\x00" * 8)
+
+        tracer = _raise_on_source_match("skip_message", "self._maybe_compact()")
+
+        sys.settrace(tracer)
+        try:
+            with contextlib.suppress(KeyboardInterrupt):
+                buf.skip_message()
+        finally:
+            sys.settrace(None)
+
+        assert buf.is_poisoned, (
+            "skip_message must poison when a BaseException escapes its mutation block"
+        )
+
+    def test_happy_paths_still_work(self) -> None:
+        """Sanity: without any interrupt, all three methods work normally."""
+        buf = ReadBuffer()
+        buf.feed(b"\x01\x00\x00\x00\x00\x00\x00\x00" + b"\x00" * 8)
+        msg = buf.read_message()
+        assert msg is not None and not buf.is_poisoned
+
+        buf.feed(b"\x00" * 32)
+        data = buf.read_bytes(8)
+        assert data is not None and len(data) == 8
+
+        buf.feed(b"\x01\x00\x00\x00\x00\x00\x00\x00" + b"\x00" * 8)
+        skipped = buf.skip_message()
+        assert skipped is True and not buf.is_poisoned