Cumulative row cap + atomicity docstring on query paths

ISSUE-05 — 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.

ISSUE-03 — 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