Cumulative row cap + atomicity docstring on query paths

- DqliteProtocol now enforces a max_total_rows cap (default
  10_000_000) across continuation frames. A server that drip-feeds
  1 row per frame inside the per-operation deadline used to be able
  to have the client accumulate arbitrary rows; the cap bounds that
  in addition to the existing wall-clock deadline. Set
  max_total_rows=None to disable.

- Document the atomicity post-condition of query_sql_typed: a
  mid-stream failure raises before any rows are returned; the local
  row list is discarded; the connection is invalidated by the
  enclosing _run_protocol path. No code change — the invariant
  already held, we're making it explicit.

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

Author: antoineleclair

src/dqliteclient/protocol.py

@@ -36,13 +36,20 @@ def __init__(
         reader: asyncio.StreamReader,
         writer: asyncio.StreamWriter,
         timeout: float = 15.0,
+        max_total_rows: int | None = 10_000_000,
     ) -> None:
         self._reader = reader
         self._writer = writer
         self._decoder = MessageDecoder(is_request=False)
         self._client_id = 0
         self._heartbeat_timeout = 0
         self._timeout = timeout
+        # Cumulative cap across continuation frames for a single query.
+        # A hostile or buggy server can drip-feed 1-row-per-frame inside
+        # the per-operation deadline; without a cumulative cap, clients
+        # could legitimately allocate hundreds of millions of rows over
+        # the full deadline. None disables the cap.
+        self._max_total_rows = max_total_rows
 
     async def handshake(self, client_id: int | None = None) -> int:
         """Perform protocol handshake.
@@ -180,6 +187,11 @@ async def query_sql_typed(
         column_types are the wire-level ``ValueType`` integer tags from the
         first response frame — what DBAPI cursor.description maps into
         ``type_code``.
+
+        Atomicity: a mid-stream server failure or an unexpected message
+        type raises before any rows are returned to the caller; the
+        local row list is discarded. The connection is invalidated so
+        callers don't accidentally reuse it with torn protocol state.
         """
         request = QuerySqlRequest(db_id=db_id, sql=sql, params=params if params is not None else [])
         self._writer.write(request.encode())
@@ -197,6 +209,17 @@ async def query_sql_typed(
         all_rows = list(response.rows)
         while response.has_more:
             next_response = await self._read_continuation(deadline=deadline)
+            if not next_response.rows and next_response.has_more:
+                raise ProtocolError(
+                    "ROWS continuation made no progress: frame had 0 rows and has_more=True"
+                )
+            if self._max_total_rows is not None and (
+                len(all_rows) + len(next_response.rows) > self._max_total_rows
+            ):
+                raise ProtocolError(
+                    f"Query exceeded max_total_rows cap ({self._max_total_rows}); "
+                    f"reduce result size or raise the cap on the connection/pool."
+                )
             all_rows.extend(next_response.rows)
             response = next_response
         return column_names, column_types, all_rows
@@ -245,6 +268,13 @@ async def query_sql(
                 raise ProtocolError(
                     "ROWS continuation made no progress: frame had 0 rows and has_more=True"
                 )
+            if self._max_total_rows is not None and (
+                len(all_rows) + len(next_response.rows) > self._max_total_rows
+            ):
+                raise ProtocolError(
+                    f"Query exceeded max_total_rows cap ({self._max_total_rows}); "
+                    f"reduce result size or raise the cap on the connection/pool."
+                )
             all_rows.extend(next_response.rows)
             response = next_response