Validate message type in decode_continuation() to surface server errors

decode_continuation() now checks header.msg_type before parsing the body
as row data. If the server sends a FailureResponse mid-stream (e.g.
leadership lost), the error code and message are raised as a ProtocolError
instead of producing a confusing DecodeError from the row-tuple parser.
Unexpected message types also raise ProtocolError.

Fixes #013

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

Author: antoineleclair

src/dqlitewire/codec.py

@@ -204,6 +204,17 @@ def decode_continuation(
 
         header = Header.decode(data[:HEADER_SIZE])
         body = data[HEADER_SIZE : HEADER_SIZE + header.size_words * 8]
+
+        if header.msg_type == ResponseType.FAILURE:
+            failure = FailureResponse.decode_body(body, schema=header.schema)
+            raise ProtocolError(
+                f"Server error during ROWS continuation: [{failure.code}] {failure.message}"
+            )
+        if header.msg_type != ResponseType.ROWS:
+            raise ProtocolError(
+                f"Expected ROWS continuation (type {ResponseType.ROWS}), got type {header.msg_type}"
+            )
+
         return RowsResponse.decode_rows_continuation(
             body, column_names, column_count, max_rows=max_rows
         )

tests/test_codec.py

@@ -804,6 +804,36 @@ def test_decode_continuation_returns_none_when_no_data(self) -> None:
         result = decoder.decode_continuation(column_names=["x"], column_count=1)
         assert result is None
 
+    def test_decode_continuation_raises_on_failure_response(self) -> None:
+        """decode_continuation should surface server error, not a confusing DecodeError."""
+        from dqlitewire.exceptions import ProtocolError
+        from dqlitewire.messages.responses import FailureResponse
+
+        failure = FailureResponse(code=266, message="disk I/O error")
+        failure_bytes = failure.encode()
+
+        decoder = MessageDecoder(is_request=False)
+        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)
+        # 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
+
+    def test_decode_continuation_raises_on_unexpected_type(self) -> None:
+        """decode_continuation should raise ProtocolError for non-ROWS, non-FAILURE type."""
+        from dqlitewire.exceptions import ProtocolError
+
+        result = ResultResponse(last_insert_id=0, rows_affected=0)
+        result_bytes = result.encode()
+
+        decoder = MessageDecoder(is_request=False)
+        decoder.feed(result_bytes)
+
+        with pytest.raises(ProtocolError, match="Expected ROWS continuation"):
+            decoder.decode_continuation(column_names=["id"], column_count=1)
+
 
 class TestDecoderSkipMessage:
     """Test skip_message() and is_skipping on MessageDecoder."""