Fix skip_message to not corrupt stream on partial normal-sized messages

skip_message previously advanced past whatever bytes were available,
even for incomplete normal-sized messages. This corrupted the stream
when remaining bytes arrived later. Now only skips complete messages
for normal sizes, reserving partial skip for oversized recovery.

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

Author: antoineleclair

src/dqlitewire/buffer.py

@@ -107,12 +107,13 @@ def read_message(self) -> bytes | None:
     def skip_message(self) -> bool:
         """Skip the current message in the buffer.
 
-        Reads the message size from the header and advances past however many
-        bytes are available (up to the full message size). Useful for recovering
-        after an oversized or malformed message that has_message() rejects.
+        For normal-sized messages (within max_message_size), waits until the
+        full message is available before skipping. For oversized messages that
+        exceed max_message_size, skips whatever bytes are available — this is
+        the recovery path for messages that has_message() rejects.
 
-        Returns True if a message header was found and skipped, False if
-        not enough data for a header.
+        Returns True if a message was fully skipped, False if not enough data
+        for a header or if a normal-sized message is still incomplete.
         """
         available = len(self._data) - self._pos
         if available < HEADER_SIZE:
@@ -121,9 +122,17 @@ def skip_message(self) -> bool:
         size_words = int.from_bytes(self._data[self._pos : self._pos + 4], "little")
         total_size = HEADER_SIZE + (size_words * WORD_SIZE)
 
-        # Skip whatever is available (might be less than total_size for partial messages)
-        skip = min(total_size, available)
-        self._pos += skip
+        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 (recovery path): skip whatever is available
+            # since we'll never buffer the full message anyway.
+            self._pos += min(total_size, available)
+
         self._maybe_compact()
         return True
 

tests/test_buffer.py

@@ -253,6 +253,25 @@ def test_skip_message_allows_oversized_for_recovery(self) -> None:
         # skip_message should succeed (it's the recovery tool for oversized messages)
         assert buf.skip_message() is True
 
+    def test_skip_message_waits_for_complete_normal_sized_message(self) -> None:
+        """skip_message should return False for incomplete normal-sized messages.
+
+        If a message fits within max_message_size but hasn't fully arrived,
+        skip_message must not advance past partial data — doing so would
+        corrupt the stream when the remaining bytes arrive later.
+        """
+        import struct
+
+        buf = ReadBuffer(max_message_size=4096)
+        # Header claiming 5 words (40 bytes body), total 48 bytes — fits in limit
+        header = struct.pack("<IBBH", 5, 0, 0, 0)
+        # Feed only header + 16 bytes of body (incomplete: need 40)
+        buf.feed(header + b"\x00" * 16)
+        # Should return False since message is incomplete but not oversized
+        assert buf.skip_message() is False
+        # Buffer position should not have changed
+        assert buf.available() == 24  # 8 header + 16 partial body
+
     def test_buffer_compaction(self) -> None:
         buf = ReadBuffer()
         # Feed a lot of small messages to trigger compaction