fix: decode continuations using decode_body (matching C server wire format)

The C dqlite server always writes column_count + column_names in
every ROWS message, including continuations sent after a PART
marker. Python's decode_rows_continuation assumed continuations
omit the column header and jumped straight to row data, causing
parse errors against real server output.

Fix decode_continuation() to call RowsResponse.decode_body()
instead of decode_rows_continuation(), matching the actual wire
format. Remove the now-unnecessary column_names and column_count
parameters since they're in the frame itself. Update all callers
and tests to use the new parameterless API with continuation
frames that include the column header prefix.

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

Author: antoineleclair

src/dqlitewire/codec.py

@@ -244,27 +244,21 @@ def is_skipping(self) -> bool:
         """True if still discarding bytes from an oversized message."""
         return self._buffer.is_skipping
 
-    def decode_continuation(
-        self,
-        column_names: list[str],
-        column_count: int,
-        max_rows: int = RowsResponse.DEFAULT_MAX_ROWS,
-    ) -> RowsResponse | None:
+    def decode_continuation(self) -> RowsResponse | None:
         """Decode a ROWS continuation message from the buffer.
 
-        After receiving a RowsResponse with has_more=True, the server sends
-        additional ROWS messages containing only row data (no column_count or
-        column_names prefix). Use this method instead of decode() for those
-        continuation messages.
-
-        Args:
-            column_names: Column names from the initial RowsResponse.
-            column_count: Number of columns from the initial RowsResponse.
-            max_rows: Maximum rows to decode per message.
+        After receiving a RowsResponse with has_more=True, call this
+        method to decode each subsequent ROWS frame. The C dqlite server
+        sends continuation frames with the same body layout as the
+        initial frame (column_count + column_names + rows + marker),
+        so this method uses ``RowsResponse.decode_body`` — the same
+        decoder used for the initial frame.
 
         Returns None if no complete message is available.
         Raises ProtocolError if no continuation is in progress
         (use ``decode()`` for the initial message).
+        Raises ProtocolError if the server sends a FailureResponse
+        instead of a ROWS continuation (e.g., mid-stream I/O error).
         """
         self._buffer._check_poisoned()
         if not self._continuation_expected:
@@ -276,14 +270,6 @@ def decode_continuation(
         if data is None:
             return None
 
-        # Bytes have been consumed — ANY failure here leaves the
-        # buffer at an unknown offset and must poison the decoder.
-        # Catch BaseException so that signal-delivered KeyboardInterrupt
-        # (issue 045) and unexpected BaseException subclasses also
-        # poison before propagating. `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]
@@ -299,16 +285,11 @@ def decode_continuation(
                     f"got type {header.msg_type}"
                 )
 
-            result = RowsResponse.decode_rows_continuation(
-                body, column_names, column_count, max_rows=max_rows
-            )
+            result = RowsResponse.decode_body(body, schema=header.schema)
             if not result.has_more:
                 self._continuation_expected = False
             return result
         except BaseException as e:
-            # poison() stores Exception | None; wrap non-Exception
-            # BaseException subclasses so the poison cause is still a
-            # real Exception we can inspect.
             self._buffer.poison(
                 e
                 if isinstance(e, Exception)

tests/test_codec.py

@@ -875,7 +875,10 @@ def test_decode_continuation_roundtrip(self) -> None:
         msg1_bytes = header1.encode() + body1
 
         # Build continuation ROWS message with DONE marker
-        body2 = encode_row_header(types)
+        # (C server always includes column_count + column_names)
+        body2 = encode_uint64(2)
+        body2 += encode_text("id") + encode_text("name")
+        body2 += encode_row_header(types)
         body2 += encode_row_values([2, "bob"], types)
         body2 += encode_uint64(ROW_DONE_MARKER)
         header2 = Header(size_words=len(body2) // 8, msg_type=7, schema=0)
@@ -893,20 +896,59 @@ def test_decode_continuation_roundtrip(self) -> None:
         assert initial.rows[0] == [1, "alice"]
 
         # Decode continuation
-        continuation = decoder.decode_continuation(
-            column_names=initial.column_names,
-            column_count=len(initial.column_names),
-        )
+        continuation = decoder.decode_continuation()
         assert isinstance(continuation, RowsResponse)
         assert continuation.has_more is False
         assert len(continuation.rows) == 1
         assert continuation.rows[0] == [2, "bob"]
 
+    def test_decode_continuation_with_column_header(self) -> None:
+        """Continuation frames from the C server include column_count +
+        column_names (same layout as the initial frame). Verify that
+        decode_continuation handles this correctly.
+        """
+        from dqlitewire.constants import ROW_DONE_MARKER, ROW_PART_MARKER, ValueType
+        from dqlitewire.messages.base import Header
+        from dqlitewire.messages.responses import RowsResponse
+        from dqlitewire.tuples import encode_row_header, encode_row_values
+        from dqlitewire.types import encode_text, encode_uint64
+
+        types = [ValueType.INTEGER, ValueType.TEXT]
+
+        # Initial ROWS message (PART marker)
+        body1 = encode_uint64(2)
+        body1 += encode_text("id") + encode_text("name")
+        body1 += encode_row_header(types)
+        body1 += encode_row_values([1, "alice"], types)
+        body1 += encode_uint64(ROW_PART_MARKER)
+        h1 = Header(size_words=len(body1) // 8, msg_type=7, schema=0)
+
+        # Continuation WITH column header (matching C server output)
+        body2 = encode_uint64(2)
+        body2 += encode_text("id") + encode_text("name")
+        body2 += encode_row_header(types)
+        body2 += encode_row_values([2, "bob"], types)
+        body2 += encode_uint64(ROW_DONE_MARKER)
+        h2 = Header(size_words=len(body2) // 8, msg_type=7, schema=0)
+
+        decoder = MessageDecoder(is_request=False)
+        decoder.feed(h1.encode() + body1 + h2.encode() + body2)
+
+        initial = decoder.decode()
+        assert isinstance(initial, RowsResponse)
+        assert initial.has_more is True
+        assert initial.rows[0] == [1, "alice"]
+
+        cont = decoder.decode_continuation()
+        assert isinstance(cont, RowsResponse)
+        assert cont.has_more is False
+        assert cont.rows[0] == [2, "bob"]
+
     def test_decode_continuation_returns_none_when_no_data(self) -> None:
         """decode_continuation should return None when no message is available."""
         decoder = MessageDecoder(is_request=False)
         decoder._continuation_expected = True
-        result = decoder.decode_continuation(column_names=["x"], column_count=1)
+        result = decoder.decode_continuation()
         assert result is None
 
     def test_decode_continuation_raises_on_failure_response(self) -> None:
@@ -922,7 +964,7 @@ def test_decode_continuation_raises_on_failure_response(self) -> None:
         decoder.feed(failure_bytes)
 
         with pytest.raises(ProtocolError, match="disk I/O error") as exc_info:
-            decoder.decode_continuation(column_names=["id"], column_count=1)
+            decoder.decode_continuation()
         # Must be a ProtocolError, NOT a DecodeError (which would mean the
         # failure body was misinterpreted as row data).
         assert type(exc_info.value) is ProtocolError
@@ -939,7 +981,7 @@ def test_decode_continuation_raises_on_unexpected_type(self) -> None:
         decoder.feed(result_bytes)
 
         with pytest.raises(ProtocolError, match="Expected ROWS continuation"):
-            decoder.decode_continuation(column_names=["id"], column_count=1)
+            decoder.decode_continuation()
 
 
 class TestDecoderContinuationExpected:
@@ -1007,8 +1049,10 @@ def test_decode_continuation_clears_flag(self) -> None:
         body1 += encode_uint64(ROW_PART_MARKER)
         h1 = Header(size_words=len(body1) // 8, msg_type=7, schema=0)
 
-        # Continuation frame (has_more=False)
-        body2 = encode_row_header(types)
+        # Continuation frame (has_more=False) — includes column header
+        body2 = encode_uint64(1)
+        body2 += encode_text("id")
+        body2 += encode_row_header(types)
         body2 += encode_row_values([2], types)
         body2 += encode_uint64(ROW_DONE_MARKER)
         h2 = Header(size_words=len(body2) // 8, msg_type=7, schema=0)
@@ -1024,10 +1068,7 @@ def test_decode_continuation_clears_flag(self) -> None:
         assert isinstance(initial, RowsResponse) and initial.has_more
 
         # decode continuation
-        cont = decoder.decode_continuation(
-            column_names=initial.column_names,
-            column_count=len(initial.column_names),
-        )
+        cont = decoder.decode_continuation()
         assert isinstance(cont, RowsResponse) and not cont.has_more
 
         # Now decode() should work again
@@ -1107,7 +1148,7 @@ def test_decode_continuation_raises_when_not_expected(self) -> None:
         decoder.feed(result.encode())
 
         with pytest.raises(ProtocolError, match="no ROWS continuation"):
-            decoder.decode_continuation(column_names=["x"], column_count=1)
+            decoder.decode_continuation()
 
         # The message must NOT have been consumed
         assert decoder.has_message(), (
@@ -1339,7 +1380,7 @@ def test_decode_continuation_poisons_on_wrong_type(self) -> None:
         decoder.feed(empty)
 
         with pytest.raises(ProtocolError, match="Expected ROWS continuation"):
-            decoder.decode_continuation(column_names=["x"], column_count=1)
+            decoder.decode_continuation()
         assert decoder.is_poisoned is True
 
         # Subsequent decode() also fails poisoned.

tests/test_codec_signal_safety.py

@@ -126,8 +126,6 @@ def test_keyboard_interrupt_in_decode_continuation_poisons(self) -> None:
         """Same hazard as decode(): decode_continuation() also needs
         BaseException handling.
         """
-        from dqlitewire.messages.responses import RowsResponse
-
         # Build a minimal ROWS continuation body. We don't actually
         # reach the parser — the tracer injects the interrupt before
         # any real parsing runs — so the frame just needs to carry
@@ -137,6 +135,7 @@ def test_keyboard_interrupt_in_decode_continuation_poisons(self) -> None:
         frame = header + body
 
         dec = MessageDecoder(is_request=False)
+        dec._continuation_expected = True
         dec.feed(frame + frame)
 
         # Inject inside decode_continuation itself (the wrapper
@@ -146,11 +145,7 @@ def test_keyboard_interrupt_in_decode_continuation_poisons(self) -> None:
         sys.settrace(tracer)
         try:
             with contextlib.suppress(KeyboardInterrupt):
-                dec.decode_continuation(
-                    column_names=["a"],
-                    column_count=1,
-                    max_rows=RowsResponse.DEFAULT_MAX_ROWS,
-                )
+                dec.decode_continuation()
         finally:
             sys.settrace(None)
 
@@ -162,10 +157,7 @@ def test_keyboard_interrupt_in_decode_continuation_poisons(self) -> None:
         # still at a valid offset.
         if dec.is_poisoned:
             with pytest.raises(ProtocolError, match="poisoned"):
-                dec.decode_continuation(
-                    column_names=["a"],
-                    column_count=1,
-                )
+                dec.decode_continuation()
         else:
             # Buffer offset must still be sensible.
             assert dec._buffer.available() >= 0