fix: plumb max_rows through MessageDecoder to RowsResponse

RowsResponse.decode_body accepts a max_rows cap, but MessageDecoder
never forwarded it at its two call sites (decode_bytes and
decode_continuation). Callers tightening max_message_size had no
paired knob for in-memory row count. Add a max_rows constructor
parameter, default to RowsResponse.DEFAULT_MAX_ROWS, validate
max_rows >= 1, and forward at both call sites.

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

Author: antoineleclair

src/dqlitewire/codec.py

@@ -181,6 +181,7 @@ def __init__(
         is_request: bool = False,
         version: int = PROTOCOL_VERSION,
         max_message_size: int = ReadBuffer.DEFAULT_MAX_MESSAGE_SIZE,
+        max_rows: int = RowsResponse.DEFAULT_MAX_ROWS,
     ) -> None:
         """Initialize decoder.
 
@@ -194,12 +195,18 @@ def __init__(
             max_message_size: Maximum allowed message size in bytes.
                     Defaults to 64 MiB. Messages exceeding this limit are
                     rejected with DecodeError.
+            max_rows: Maximum number of rows permitted in a single
+                    ``RowsResponse`` frame (including continuation frames).
+                    Defaults to ``RowsResponse.DEFAULT_MAX_ROWS``. Exceeding
+                    this limit raises ``DecodeError``.
         """
         if not is_request and version not in _SUPPORTED_VERSIONS:
             raise ProtocolError(
                 f"Unsupported protocol version: {version:#x}. "
                 f"Supported: {', '.join(f'{v:#x}' for v in sorted(_SUPPORTED_VERSIONS))}"
             )
+        if max_rows < 1:
+            raise ValueError(f"max_rows must be >= 1, got {max_rows}")
         self._buffer = ReadBuffer(max_message_size=max_message_size)
         self._is_request = is_request
         self._type_map = REQUEST_TYPES if is_request else RESPONSE_TYPES
@@ -211,6 +218,7 @@ def __init__(
         # don't receive an inbound handshake, so they skip this check.
         self._handshake_done = not is_request
         self._continuation_expected = False
+        self._max_rows = max_rows
 
     @property
     def version(self) -> int | None:
@@ -334,7 +342,7 @@ def decode_continuation(self) -> RowsResponse | None:
                     f"got type {header.msg_type}"
                 )
 
-            result = RowsResponse.decode_body(body, schema=header.schema)
+            result = RowsResponse.decode_body(body, schema=header.schema, max_rows=self._max_rows)
             if not result.has_more:
                 self._continuation_expected = False
             return result
@@ -449,6 +457,11 @@ def decode_bytes(self, data: bytes) -> Message:
         ):
             return LeaderResponse.decode_body_legacy(body)
 
+        # RowsResponse takes an extra ``max_rows`` cap; other classes
+        # share the generic (body, schema) signature.
+        if msg_class is RowsResponse:
+            return RowsResponse.decode_body(body, schema=header.schema, max_rows=self._max_rows)
+
         return msg_class.decode_body(body, schema=header.schema)
 
     def decode_handshake(self) -> int | None:

tests/test_codec.py

@@ -115,6 +115,68 @@ def test_decoder_default_max_message_size(self) -> None:
         decoder = MessageDecoder()
         assert decoder._buffer._max_message_size == ReadBuffer.DEFAULT_MAX_MESSAGE_SIZE
 
+    def test_decoder_custom_max_rows(self) -> None:
+        """231: max_rows should be forwarded to RowsResponse.decode_body.
+
+        The knob exists on RowsResponse.decode_body, but MessageDecoder
+        previously omitted it at the call site, leaving the 1M default
+        always in force regardless of caller configuration.
+        """
+        from dqlitewire.constants import ValueType
+        from dqlitewire.exceptions import DecodeError
+        from dqlitewire.messages.responses import RowsResponse
+
+        msg = RowsResponse(
+            column_names=["a"],
+            column_types=[ValueType.INTEGER],
+            row_types=[[ValueType.INTEGER]] * 3,
+            rows=[[1], [2], [3]],
+        )
+        encoded = msg.encode()
+
+        decoder = MessageDecoder(max_rows=2)
+        decoder.feed(encoded)
+        with pytest.raises(DecodeError, match="reached maximum 2"):
+            decoder.decode()
+
+    def test_decoder_default_max_rows(self) -> None:
+        """231: default max_rows should match RowsResponse.DEFAULT_MAX_ROWS."""
+        from dqlitewire.messages.responses import RowsResponse
+
+        decoder = MessageDecoder()
+        assert decoder._max_rows == RowsResponse.DEFAULT_MAX_ROWS
+
+    def test_decoder_rejects_zero_max_rows(self) -> None:
+        """231: max_rows < 1 should be rejected at construction time."""
+        with pytest.raises(ValueError, match="max_rows must be >= 1"):
+            MessageDecoder(max_rows=0)
+
+    def test_decoder_continuation_honors_max_rows(self) -> None:
+        """231: decode_continuation should also honor max_rows.
+
+        A multi-frame ROWS stream can exceed max_rows across frames; the
+        per-frame cap still applies to each continuation response.
+        """
+        from dqlitewire.constants import ValueType
+        from dqlitewire.exceptions import DecodeError
+        from dqlitewire.messages.responses import RowsResponse
+
+        cont = RowsResponse(
+            column_names=["a"],
+            column_types=[ValueType.INTEGER],
+            row_types=[[ValueType.INTEGER]] * 3,
+            rows=[[1], [2], [3]],
+            has_more=False,
+        )
+        cont_bytes = cont.encode()
+
+        decoder = MessageDecoder(max_rows=2)
+        # Mark decoder as mid-continuation so decode_continuation() runs.
+        decoder._continuation_expected = True
+        decoder.feed(cont_bytes)
+        with pytest.raises(DecodeError, match="reached maximum 2"):
+            decoder.decode_continuation()
+
     def test_decode_handshake(self) -> None:
         decoder = MessageDecoder(is_request=True)
         decoder.feed(PROTOCOL_VERSION.to_bytes(8, "little"))