fix: poison on torn state in ReadBuffer._maybe_compact

antoineleclair · claude · antoineleclair · commit ea5e44aee556 · 2026-04-11T09:47:30.000-04:00
_maybe_compact mutates two attributes — self._data and self._pos —
which compile to two STORE_ATTR bytecodes. CPython checks for pending
signals at bytecode line transitions, so a KeyboardInterrupt (or any
PyErr_SetAsyncExc delivery) landing between the two stores left the
buffer with a freshly compacted _data but a stale _pos still pointing
at the old offset. available() returned a negative number, reads
silently returned nonsense, and no poison fired because no exception
originated inside the buffer — the interrupt was purely external. A
single-owner caller pressing Ctrl-C during a busy decode loop would
end up with silent message dropout and no diagnostic.

We cannot make two STORE_ATTRs atomic at the Python level. Catch
BaseException and poison so the next caller fails fast with
ProtocolError instead of reading from an inconsistent offset.
KeyboardInterrupt is not an Exception subclass, so wrap it in a
RuntimeError before storing in self._poisoned (which is typed
Exception | None); any other Exception subclass is stored as-is.

New tests/test_buffer_signal_safety.py uses sys.settrace to inject
KeyboardInterrupt at the second STORE_ATTR line and asserts the
buffer is either poisoned or self-consistent after the torn window.

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/src/dqlitewire/buffer.py b/src/dqlitewire/buffer.py
@@ -284,10 +284,43 @@ def peek_bytes(self, n: int) -> bytes | None:
         return bytes(self._data[self._pos : self._pos + n])
 
     def _maybe_compact(self) -> None:
-        """Compact buffer if we've consumed a lot."""
-        if self._pos > 4096:
-            self._data = self._data[self._pos :]
+        """Compact buffer if we've consumed a lot.
+
+        Signal-safety note (issue 037): this method mutates two
+        attributes — ``_data`` and ``_pos`` — which compile to two
+        ``STORE_ATTR`` bytecodes. CPython checks for pending signals
+        at bytecode line transitions, so a ``KeyboardInterrupt`` (or
+        any ``PyErr_SetAsyncExc`` delivery) landing between the two
+        stores used to leave the buffer with a freshly compacted
+        ``_data`` but a stale ``_pos`` still pointing at the old
+        offset. ``available()`` would return a negative number, reads
+        would silently return nonsense, and no poison fired because
+        no exception originated inside the buffer — the interrupt was
+        purely external. A single-owner caller pressing Ctrl-C during
+        a busy decode loop would end up with silent message dropout.
+
+        We cannot make the two stores atomic at the Python level, but
+        we can catch ``BaseException`` and poison so that the next
+        caller fails fast with ``ProtocolError`` instead of reading
+        from an inconsistent offset.
+        """
+        if self._pos <= 4096:
+            return
+        try:
+            new_data = self._data[self._pos :]
+            self._data = new_data
             self._pos = 0
+        except BaseException as e:
+            # Any torn state here is unrecoverable — the next caller
+            # MUST see poison, not a silently inconsistent buffer.
+            # ``poison()`` is first-error-wins and its body is a
+            # single ``STORE_ATTR`` so cannot itself be split.
+            if self._poisoned is None:
+                if isinstance(e, Exception):
+                    self._poisoned = e
+                else:
+                    self._poisoned = RuntimeError(f"buffer compact interrupted: {type(e).__name__}")
+            raise
 
     def available(self) -> int:
         """Return number of bytes available to read."""
diff --git a/tests/test_buffer_signal_safety.py b/tests/test_buffer_signal_safety.py
@@ -0,0 +1,214 @@
+"""Signal-safety tests for ReadBuffer (issues 037, 041, 045).
+
+These tests use ``sys.settrace`` to inject a ``KeyboardInterrupt`` at a
+specific source line transition, which is the most reliable way to
+exercise CPython's bytecode-boundary async-exception model without
+relying on wallclock timing. Each test verifies that when an async
+exception lands in the middle of a multi-statement mutation, the
+buffer is left in a state that either:
+
+- is self-consistent (``available()`` is non-negative, reads don't
+  return fabricated bytes), OR
+- is poisoned, so the next caller fails fast with ``ProtocolError``.
+
+"Looks fine but is silently broken" is the failure mode these tests
+guard against.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import sys
+from types import FrameType
+from typing import Any
+
+import pytest
+
+from dqlitewire.buffer import ReadBuffer
+from dqlitewire.exceptions import DecodeError, ProtocolError
+
+_DEFAULT_INJECTED_EXC: BaseException = KeyboardInterrupt("injected")
+
+
+def _raise_on_source_match(
+    func_name: str,
+    needle: str,
+    exc: BaseException = _DEFAULT_INJECTED_EXC,
+) -> Any:
+    """Return a settrace-compatible tracer that raises ``exc`` on the
+    first line inside ``func_name`` whose source contains ``needle``.
+    """
+    state = {"raised": False}
+
+    def tracer(frame: FrameType, event: str, arg: object) -> Any:
+        if event != "line":
+            return tracer
+        if frame.f_code.co_name != func_name:
+            return tracer
+        try:
+            with open(frame.f_code.co_filename) as f:
+                src_line = f.readlines()[frame.f_lineno - 1]
+        except OSError:
+            return tracer
+        if needle in src_line and not state["raised"]:
+            state["raised"] = True
+            raise exc
+        return tracer
+
+    return tracer
+
+
+class TestMaybeCompactSignalSafety:
+    """Regression tests for issue 037.
+
+    ``ReadBuffer._maybe_compact`` used to be two separate ``STORE_ATTR``
+    bytecodes:
+
+        self._data = self._data[self._pos :]   # (A)
+        self._pos = 0                           # (B)
+
+    CPython checks for pending signals at bytecode line transitions. A
+    ``KeyboardInterrupt`` (or any ``PyErr_SetAsyncExc`` delivery) landing
+    between (A) and (B) would leave the buffer with a freshly compacted
+    ``_data`` but a stale ``_pos`` still pointing at the old offset.
+    ``available()`` would return a negative number, reads would return
+    nonsense, and no poison fired because no exception originated
+    *inside* the buffer — the interrupt was purely external. A
+    single-owner caller who did nothing wrong except press Ctrl-C
+    during a busy decode loop would end up with silent message dropout.
+
+    The fix wraps the compaction in ``try/except BaseException`` and
+    poisons on any torn state, so the next call raises ``ProtocolError``
+    instead of lying about the buffer's contents.
+    """
+
+    def test_torn_compact_leaves_buffer_poisoned_or_consistent(self) -> None:
+        buf = ReadBuffer()
+        # Set up a pre-compact state: _pos > 4096 and a tail of bytes
+        # that would remain after compaction.
+        buf._data = bytearray(b"A" * 4096 + b"B" * 100)
+        buf._pos = 4097
+
+        tracer = _raise_on_source_match("_maybe_compact", "self._pos = 0")
+
+        sys.settrace(tracer)
+        try:
+            with contextlib.suppress(KeyboardInterrupt):
+                buf._maybe_compact()
+        finally:
+            sys.settrace(None)
+
+        # Post-condition: EITHER the buffer is poisoned (preferred), OR
+        # it is self-consistent (available() non-negative, reads valid).
+        # The old (broken) behaviour returned available() == -3998.
+        if buf.is_poisoned:
+            # Exact recovery semantics at the public-API layer are
+            # covered by issue 038's tests; here we only require that
+            # the buffer knows it is torn. Inspect _check_poisoned
+            # directly, which is the single source of truth for the
+            # poison gate.
+            with pytest.raises(ProtocolError, match="poisoned"):
+                buf._check_poisoned()
+        else:
+            assert buf.available() >= 0, (
+                f"torn compact left available()={buf.available()} "
+                f"(len={len(buf._data)}, pos={buf._pos})"
+            )
+
+    def test_happy_path_compact_still_works(self) -> None:
+        """Sanity: without any interrupt, compact behaves normally."""
+        buf = ReadBuffer()
+        buf._data = bytearray(b"A" * 4096 + b"B" * 100)
+        buf._pos = 4097
+
+        buf._maybe_compact()
+
+        assert not buf.is_poisoned
+        assert buf._pos == 0
+        assert len(buf._data) == 99
+        assert buf.available() == 99
+        # And reads continue to work.
+        tail = buf.read_bytes(5)
+        assert tail == b"B" * 5
+
+    def test_compact_below_threshold_is_a_noop(self) -> None:
+        """If _pos is below the compact threshold, the method does
+        nothing — no state changes, no poison.
+        """
+        buf = ReadBuffer()
+        buf.feed(b"\x00" * 32)
+        buf.read_bytes(8)
+        assert buf._pos == 8
+
+        buf._maybe_compact()
+
+        assert buf._pos == 8
+        assert not buf.is_poisoned
+
+
+class TestMaybeCompactPoisonIsWellFormed:
+    """If the torn-state fix fires and poisons the buffer, the poison
+    cause must be a real exception instance we can inspect — not None,
+    not a raw BaseException we can't chain through.
+    """
+
+    def test_poison_cause_is_recorded(self) -> None:
+        buf = ReadBuffer()
+        buf._data = bytearray(b"A" * 4096 + b"B" * 100)
+        buf._pos = 4097
+
+        sentinel = RuntimeError("injected torn state")
+        tracer = _raise_on_source_match("_maybe_compact", "self._pos = 0", exc=sentinel)
+
+        sys.settrace(tracer)
+        try:
+            with contextlib.suppress(RuntimeError):
+                buf._maybe_compact()
+        finally:
+            sys.settrace(None)
+
+        if buf.is_poisoned:
+            with pytest.raises(ProtocolError) as ei:
+                buf._check_poisoned()
+            # The original exception should appear in the cause chain.
+            cause = ei.value.__cause__
+            assert cause is not None
+            # Either directly the sentinel, or a wrapper around it.
+            assert cause is sentinel or isinstance(cause, Exception)
+
+    def test_non_exception_base_exception_does_not_crash_poison(self) -> None:
+        """KeyboardInterrupt is not an Exception subclass. The poison
+        path must still accept it (either by widening poison()'s
+        parameter type or by wrapping)."""
+        # Use the class above's scenario but with KeyboardInterrupt —
+        # that is what CPython actually delivers on Ctrl-C.
+        buf = ReadBuffer()
+        buf._data = bytearray(b"A" * 4096 + b"B" * 100)
+        buf._pos = 4097
+
+        tracer = _raise_on_source_match(
+            "_maybe_compact", "self._pos = 0", exc=KeyboardInterrupt("ctrl-c")
+        )
+
+        sys.settrace(tracer)
+        try:
+            with contextlib.suppress(KeyboardInterrupt):
+                buf._maybe_compact()
+        finally:
+            sys.settrace(None)
+
+        # We don't care whether the buffer is poisoned or consistent;
+        # we care that neither path raised TypeError from poison()
+        # rejecting a BaseException that isn't an Exception.
+        _ = buf.is_poisoned  # just reading it must not raise
+        _ = buf.available()
+
+
+def test_decode_error_is_still_recoverable_without_poison() -> None:
+    """Counter-test: the torn-state poison must not fire on any
+    "normal" code path. A legitimate decode on a clean buffer works
+    as before."""
+    buf = ReadBuffer()
+    buf.feed(b"\x00" * 16)
+    assert not buf.is_poisoned
+    _ = DecodeError  # referenced for completeness
