fix: enforce per-operation deadline on protocol reads

self._timeout applied to each individual read. A slow server trickling
bytes just under the timeout on every chunk could keep a call alive
indefinitely. query_sql with N continuation frames was also unbounded:
N × self._timeout.

Establish a per-operation deadline at the start of _read_response and
_read_continuation; _read_data caps each chunk's timeout by the
remaining budget. query_sql now threads one deadline through the
initial response and every continuation, so the wall-time budget
spans the whole call. Also reject non-finite (inf/nan) timeouts at
DqliteConnection construction.

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

Author: antoineleclair

src/dqliteclient/connection.py

@@ -80,8 +80,10 @@ def __init__(
             database: Database name to open
             timeout: Connection timeout in seconds
         """
-        if timeout <= 0:
-            raise ValueError(f"timeout must be positive, got {timeout}")
+        import math
+
+        if not math.isfinite(timeout) or timeout <= 0:
+            raise ValueError(f"timeout must be a positive finite number, got {timeout}")
         self._address = address
         self._database = database
         self._timeout = timeout

src/dqliteclient/protocol.py

@@ -184,7 +184,11 @@ async def query_sql(
         self._writer.write(request.encode())
         await self._send()
 
-        response = await self._read_response()
+        # Single deadline spans the initial response plus every continuation
+        # frame; otherwise a server that split a reply into N frames could
+        # legitimately take N * self._timeout to complete.
+        deadline = self._operation_deadline()
+        response = await self._read_response(deadline=deadline)
 
         if isinstance(response, FailureResponse):
             raise OperationalError(response.code, response.message)
@@ -200,7 +204,7 @@ async def query_sql(
         # marker), matching the C dqlite server's wire format.
         all_rows = list(response.rows)
         while response.has_more:
-            next_response = await self._read_continuation()
+            next_response = await self._read_continuation(deadline=deadline)
             if not next_response.rows and next_response.has_more:
                 # Server claimed "more coming" but delivered zero rows in a
                 # continuation frame. That would spin forever (known
@@ -221,36 +225,67 @@ async def _send(self) -> None:
         except (ConnectionError, OSError, RuntimeError) as e:
             raise DqliteConnectionError(f"Write failed: {e}") from e
 
-    async def _read_data(self) -> bytes:
-        """Read data from the stream with timeout.
+    async def _read_data(self, deadline: float | None = None) -> bytes:
+        """Read a chunk from the stream, bounded by a per-operation deadline.
+
+        If ``deadline`` is set (monotonic time), the per-chunk timeout is
+        capped by the remaining budget — a slow-drip server that returned
+        just under the per-read timeout on every chunk used to be able to
+        keep a call alive indefinitely.
 
         Transport errors (ConnectionResetError, BrokenPipeError, OSError,
         RuntimeError("Transport is closed")) are wrapped in
         DqliteConnectionError to match the write-path behaviour.
         """
+        if deadline is not None:
+            remaining = deadline - asyncio.get_running_loop().time()
+            if remaining <= 0:
+                raise DqliteConnectionError(f"Operation exceeded {self._timeout}s deadline")
+            timeout = min(remaining, self._timeout)
+        else:
+            timeout = self._timeout
         try:
-            data = await asyncio.wait_for(self._reader.read(4096), timeout=self._timeout)
+            data = await asyncio.wait_for(self._reader.read(4096), timeout=timeout)
         except TimeoutError:
-            raise DqliteConnectionError(f"Server read timed out after {self._timeout}s") from None
+            raise DqliteConnectionError(f"Server read timed out after {timeout:.1f}s") from None
         except (ConnectionError, OSError, RuntimeError) as e:
             raise DqliteConnectionError(f"Read failed: {e}") from e
         if not data:
             raise DqliteConnectionError("Connection closed by server")
         return data
 
-    async def _read_continuation(self) -> RowsResponse:
-        """Read and decode a ROWS continuation frame."""
+    def _operation_deadline(self) -> float:
+        """Deadline (monotonic seconds) for a single protocol operation."""
+        return asyncio.get_running_loop().time() + self._timeout
+
+    async def _read_continuation(self, deadline: float | None = None) -> RowsResponse:
+        """Read and decode a ROWS continuation frame.
+
+        If ``deadline`` is None, a fresh per-operation deadline is set;
+        query_sql passes its own deadline so the budget spans every
+        continuation frame, not each one individually.
+        """
+        if deadline is None:
+            deadline = self._operation_deadline()
         while True:
             result = self._decoder.decode_continuation()
             if result is not None:
                 return result
-            data = await self._read_data()
+            data = await self._read_data(deadline=deadline)
             self._decoder.feed(data)
 
-    async def _read_response(self) -> Message:
-        """Read and decode the next response message."""
+    async def _read_response(self, deadline: float | None = None) -> Message:
+        """Read and decode the next response message.
+
+        If ``deadline`` is None, a fresh per-operation deadline is set for
+        this one response; callers that span multiple reads (e.g. query_sql
+        across continuation frames) pass an externally-held deadline so
+        the cumulative wall time is bounded.
+        """
+        if deadline is None:
+            deadline = self._operation_deadline()
         while not self._decoder.has_message():
-            data = await self._read_data()
+            data = await self._read_data(deadline=deadline)
             self._decoder.feed(data)
 
         message = self._decoder.decode()

tests/test_connection.py

@@ -90,13 +90,21 @@ def test_min_valid_port(self) -> None:
 
 class TestDqliteConnection:
     def test_zero_timeout_raises(self) -> None:
-        with pytest.raises(ValueError, match="timeout must be positive"):
+        with pytest.raises(ValueError, match="timeout must be"):
             DqliteConnection("localhost:9001", timeout=0)
 
     def test_negative_timeout_raises(self) -> None:
-        with pytest.raises(ValueError, match="timeout must be positive"):
+        with pytest.raises(ValueError, match="timeout must be"):
             DqliteConnection("localhost:9001", timeout=-1)
 
+    def test_infinite_timeout_raises(self) -> None:
+        with pytest.raises(ValueError, match="finite"):
+            DqliteConnection("localhost:9001", timeout=float("inf"))
+
+    def test_nan_timeout_raises(self) -> None:
+        with pytest.raises(ValueError, match="finite"):
+            DqliteConnection("localhost:9001", timeout=float("nan"))
+
     def test_init(self) -> None:
         conn = DqliteConnection("localhost:9001", database="test", timeout=5.0)
         assert conn.address == "localhost:9001"

tests/test_protocol.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from dqliteclient.exceptions import OperationalError, ProtocolError
+from dqliteclient.exceptions import DqliteConnectionError, OperationalError, ProtocolError
 from dqliteclient.protocol import DqliteProtocol
 from dqlitewire.messages import (
     FailureResponse,
@@ -69,6 +69,33 @@ async def test_reader_errors_are_wrapped(
             await protocol._read_data()
         assert exc_info.value.__cause__ is err
 
+    async def test_read_response_enforces_operation_deadline(
+        self,
+        protocol: DqliteProtocol,
+        mock_reader: AsyncMock,
+    ) -> None:
+        """Even if each individual read returns just under the per-read
+        timeout, the cumulative operation deadline must fire.
+        """
+        import asyncio
+        import time
+
+        protocol._timeout = 0.2
+
+        async def drip_forever(_n: int) -> bytes:
+            await asyncio.sleep(0.1)
+            return b"\x00"  # 1 byte, never completes a message
+
+        mock_reader.read.side_effect = drip_forever
+
+        t0 = time.monotonic()
+        with pytest.raises(DqliteConnectionError, match="deadline|timed out"):
+            await protocol._read_response()
+        elapsed = time.monotonic() - t0
+        assert elapsed < 1.0, (
+            f"_read_response must bail at the operation deadline; took {elapsed:.3f}s"
+        )
+
     async def test_handshake_success(
         self,
         protocol: DqliteProtocol,