fix: address cycle-3 review — plumb caps and strengthen tests

Post-review follow-up (correctness + security + test-quality agents
ran on the cycle-3 commits and flagged the items below).

Critical — tests tried real connections to localhost:19001
- test_connect_forwards_max_total_rows refactored to inspect the
  Connection attribute without going through a connect flow. Uses a
  non-existent port (19999) because Connection.__init__ is pure state
  machine; nothing touches the network.
- test_sync_cursor_execute_after_connection_close now uses a
  _ClosedConn fake whose _run_sync raises InterfaceError, mirroring
  the async variant's pattern. No cluster dependency.

Medium — test strength
- rownumber: add increment/fetchmany/fetchall coverage so the
  property's actual claim (advance with each fetched row) is pinned,
  not just the None-without-result-set path.

Critical — plumbing gap
- max_continuation_frames and trust_server_heartbeat now propagate
  through dqlitedbapi.Connection, dqlitedbapi.aio.Connection, and
  the module-level dqlitedbapi.connect(). Callers who go through the
  dbapi (the 99% path) can now tune the DoS governors end-to-end.

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

Author: antoineleclair

src/dqlitedbapi/__init__.py

@@ -95,6 +95,8 @@ def connect(
     database: str = "default",
     timeout: float = 10.0,
     max_total_rows: int | None = 10_000_000,
+    max_continuation_frames: int | None = 100_000,
+    trust_server_heartbeat: bool = False,
 ) -> Connection:
     """Connect to a dqlite database.
 
@@ -108,6 +110,10 @@ def connect(
         max_total_rows: Cumulative row cap across continuation frames
             for a single query. Forwarded to the underlying
             :class:`Connection` (ISSUE-111). ``None`` disables the cap.
+        max_continuation_frames: Per-query continuation-frame cap
+            (ISSUE-98). Forwarded to the underlying :class:`Connection`.
+        trust_server_heartbeat: Let the server-advertised heartbeat
+            widen the per-read deadline (ISSUE-101). Default False.
 
     Returns:
         A Connection object
@@ -121,4 +127,6 @@ def connect(
         database=database,
         timeout=timeout,
         max_total_rows=max_total_rows,
+        max_continuation_frames=max_continuation_frames,
+        trust_server_heartbeat=trust_server_heartbeat,
     )

src/dqlitedbapi/aio/connection.py

@@ -7,7 +7,7 @@
 
 import dqliteclient.exceptions as _client_exc
 from dqliteclient import DqliteConnection
-from dqliteclient.protocol import _validate_max_total_rows
+from dqliteclient.protocol import _validate_max_total_rows, _validate_positive_int_or_none
 from dqlitedbapi.aio.cursor import AsyncCursor
 from dqlitedbapi.connection import _is_no_transaction_error
 from dqlitedbapi.exceptions import InterfaceError, OperationalError, ProgrammingError
@@ -23,6 +23,8 @@ def __init__(
         database: str = "default",
         timeout: float = 10.0,
         max_total_rows: int | None = 10_000_000,
+        max_continuation_frames: int | None = 100_000,
+        trust_server_heartbeat: bool = False,
     ) -> None:
         """Initialize connection (does not connect yet).
 
@@ -33,13 +35,21 @@ def __init__(
             max_total_rows: Cumulative row cap across continuation
                 frames. Forwarded to the underlying DqliteConnection;
                 ``None`` disables the cap.
+            max_continuation_frames: Per-query continuation-frame cap
+                (ISSUE-98). Forwarded to the underlying DqliteConnection.
+            trust_server_heartbeat: When True, let the server-advertised
+                heartbeat widen the per-read deadline (ISSUE-101).
         """
         if not math.isfinite(timeout) or timeout <= 0:
             raise ProgrammingError(f"timeout must be a positive finite number, got {timeout}")
         self._address = address
         self._database = database
         self._timeout = timeout
         self._max_total_rows = _validate_max_total_rows(max_total_rows)
+        self._max_continuation_frames = _validate_positive_int_or_none(
+            max_continuation_frames, "max_continuation_frames"
+        )
+        self._trust_server_heartbeat = trust_server_heartbeat
         self._async_conn: DqliteConnection | None = None
         self._closed = False
         # asyncio primitives MUST be created inside the loop they will
@@ -79,6 +89,8 @@ async def _ensure_connection(self) -> DqliteConnection:
                 database=self._database,
                 timeout=self._timeout,
                 max_total_rows=self._max_total_rows,
+                max_continuation_frames=self._max_continuation_frames,
+                trust_server_heartbeat=self._trust_server_heartbeat,
             )
             try:
                 await conn.connect()

src/dqlitedbapi/connection.py

@@ -10,7 +10,7 @@
 
 import dqliteclient.exceptions as _client_exc
 from dqliteclient import DqliteConnection
-from dqliteclient.protocol import _validate_max_total_rows
+from dqliteclient.protocol import _validate_max_total_rows, _validate_positive_int_or_none
 from dqlitedbapi.cursor import Cursor
 from dqlitedbapi.exceptions import InterfaceError, OperationalError, ProgrammingError
 
@@ -94,6 +94,8 @@ def __init__(
         database: str = "default",
         timeout: float = 10.0,
         max_total_rows: int | None = 10_000_000,
+        max_continuation_frames: int | None = 100_000,
+        trust_server_heartbeat: bool = False,
     ) -> None:
         """Initialize connection (does not connect yet).
 
@@ -107,13 +109,25 @@ def __init__(
             max_total_rows: Cumulative row cap across continuation
                 frames for a single query. Forwarded to the underlying
                 :class:`DqliteConnection`. ``None`` disables the cap.
+            max_continuation_frames: Per-query continuation-frame cap
+                (ISSUE-98). Bounds Python-side decode work a hostile
+                server can inflict by drip-feeding 1-row frames.
+                Forwarded to the underlying :class:`DqliteConnection`.
+            trust_server_heartbeat: When True, widen the per-read
+                deadline to the server-advertised heartbeat (subject to
+                a 300 s hard cap). Default False so the configured
+                ``timeout`` is authoritative (ISSUE-101).
         """
         if not math.isfinite(timeout) or timeout <= 0:
             raise ProgrammingError(f"timeout must be a positive finite number, got {timeout}")
         self._address = address
         self._database = database
         self._timeout = timeout
         self._max_total_rows = _validate_max_total_rows(max_total_rows)
+        self._max_continuation_frames = _validate_positive_int_or_none(
+            max_continuation_frames, "max_continuation_frames"
+        )
+        self._trust_server_heartbeat = trust_server_heartbeat
         self._async_conn: DqliteConnection | None = None
         self._closed = False
         self._loop: asyncio.AbstractEventLoop | None = None
@@ -228,6 +242,8 @@ async def _get_async_connection(self) -> DqliteConnection:
                 database=self._database,
                 timeout=self._timeout,
                 max_total_rows=self._max_total_rows,
+                max_continuation_frames=self._max_continuation_frames,
+                trust_server_heartbeat=self._trust_server_heartbeat,
             )
             try:
                 await conn.connect()

tests/test_cycle3_hardening.py

@@ -43,6 +43,65 @@ class _FakeAsyncConn:
         cursor = AsyncCursor(_FakeAsyncConn())  # type: ignore[arg-type]
         assert cursor.rownumber is None
 
+    def test_rownumber_increments_with_fetchone(self) -> None:
+        """After each fetchone(), rownumber points at the next row."""
+        from dqlitedbapi.cursor import Cursor
+
+        class _FakeConn:
+            def _check_thread(self) -> None: ...
+
+        cursor = Cursor(_FakeConn())  # type: ignore[arg-type]
+        cursor._description = [("x", None, None, None, None, None, None)]
+        cursor._rows = [(1,), (2,), (3,)]
+
+        assert cursor.rownumber == 0  # before any fetch, cursor points at row 0
+
+        assert cursor.fetchone() == (1,)
+        assert cursor.rownumber == 1
+
+        assert cursor.fetchone() == (2,)
+        assert cursor.rownumber == 2
+
+        assert cursor.fetchone() == (3,)
+        assert cursor.rownumber == 3  # past the end
+
+        # Further fetches return None and do not advance rownumber past len
+        assert cursor.fetchone() is None
+        assert cursor.rownumber == 3
+
+    def test_rownumber_after_fetchall(self) -> None:
+        """fetchall advances rownumber to end of result set."""
+        from dqlitedbapi.cursor import Cursor
+
+        class _FakeConn:
+            def _check_thread(self) -> None: ...
+
+        cursor = Cursor(_FakeConn())  # type: ignore[arg-type]
+        cursor._description = [("x", None, None, None, None, None, None)]
+        cursor._rows = [(i,) for i in range(5)]
+        assert cursor.rownumber == 0
+
+        rows = cursor.fetchall()
+        assert len(rows) == 5
+        assert cursor.rownumber == 5
+
+    def test_rownumber_after_fetchmany(self) -> None:
+        """fetchmany advances rownumber by the number of rows fetched."""
+        from dqlitedbapi.cursor import Cursor
+
+        class _FakeConn:
+            def _check_thread(self) -> None: ...
+
+        cursor = Cursor(_FakeConn())  # type: ignore[arg-type]
+        cursor._description = [("x", None, None, None, None, None, None)]
+        cursor._rows = [(i,) for i in range(10)]
+
+        cursor.fetchmany(3)
+        assert cursor.rownumber == 3
+
+        cursor.fetchmany(4)
+        assert cursor.rownumber == 7
+
 
 class TestFetchmanyNegativeSize:
     def test_sync_fetchmany_rejects_negative(self) -> None:
@@ -221,27 +280,36 @@ def test_constraint_violation_is_not_silenced(self) -> None:
 
 
 class TestConnectForwardsMaxTotalRows:
-    """ISSUE-111: module-level connect() forwards max_total_rows."""
-
-    def test_connect_forwards_max_total_rows(self) -> None:
-        from dqlitedbapi import connect
-        from dqlitedbapi.connection import Connection
+    """ISSUE-111: module-level connect() forwards max_total_rows.
 
-        conn = connect("localhost:19001", max_total_rows=500)
-        try:
-            assert isinstance(conn, Connection)
-            assert conn._max_total_rows == 500
-        finally:
-            conn.close()
+    These tests only verify parameter plumbing — they do not open a
+    socket. Connection.__init__ is pure state machine; no cluster
+    needed. The previous implementation called conn.close() which
+    required a running cluster for the event-loop thread to wind down
+    cleanly.
+    """
 
-    def test_connect_forwards_none_for_max_total_rows(self) -> None:
+    @pytest.mark.parametrize(
+        "max_total_rows,expected",
+        [(500, 500), (None, None), (10_000, 10_000)],
+    )
+    def test_connect_forwards_max_total_rows(
+        self, max_total_rows: int | None, expected: int | None
+    ) -> None:
         from dqlitedbapi import connect
+        from dqlitedbapi.connection import Connection
 
-        conn = connect("localhost:19001", max_total_rows=None)
-        try:
-            assert conn._max_total_rows is None
-        finally:
-            conn.close()
+        # connect() does NOT actually connect to the server — it
+        # instantiates a Connection with the given address and defers
+        # the real TCP until first use. Inspect the attribute and then
+        # skip close() because close() on a never-connected connection
+        # is a silent no-op (no loop thread was started).
+        conn = connect("localhost:19999", max_total_rows=max_total_rows)
+        assert isinstance(conn, Connection)
+        assert conn._max_total_rows == expected
+        # conn.close() on an unused connection is a no-op; no cluster
+        # contact happens.
+        conn.close()
 
 
 class TestIsRowReturning:

tests/test_cycle3_tests_pinning.py

@@ -21,11 +21,23 @@ class TestCursorAfterExternalConnectionClose:
     """
 
     def test_sync_cursor_execute_after_connection_close(self) -> None:
-        import dqlitedbapi
+        # Simulate a closed connection without touching any cluster.
+        # The contract we're pinning: when the underlying connection's
+        # _run_sync raises InterfaceError (its documented response to a
+        # closed connection), the cursor surface propagates it.
+        from dqlitedbapi.cursor import Cursor
 
-        conn = dqlitedbapi.connect("localhost:19001")
-        cursor = conn.cursor()
-        conn.close()
+        class _ClosedConn:
+            _closed = True
+
+            def _check_thread(self) -> None:
+                return None
+
+            def _run_sync(self, coro) -> None:  # noqa: ANN001
+                coro.close()
+                raise InterfaceError("Connection is closed")
+
+        cursor = Cursor(_ClosedConn())  # type: ignore[arg-type]
 
         with pytest.raises(InterfaceError):
             cursor.execute("SELECT 1")