Make trust_server_heartbeat read-timeout widening actually flow through to operation deadline

``DqliteProtocol._operation_deadline()`` keyed off
``self._timeout`` (the un-widened write-path budget) while
``_read_data`` then capped each chunk read at ``min(remaining,
self._read_timeout)``. Since the per-read cap was always
``>= remaining`` after the deadline calculation, the
heartbeat-trust widening of ``_read_timeout`` had NO effect
on the actual deadline budget — the documented
``trust_server_heartbeat=True`` opt-in was dead code on the
read deadline path.

Point ``_operation_deadline()`` at ``self._read_timeout`` so
the widening flows through. Write-path ``_send`` continues
to use ``self._timeout`` directly — only the read side gets
heartbeat extension.

Add a positive test pinning the now-real widening: a read
that takes longer than ``_timeout`` but less than the
widened ``_read_timeout`` completes cleanly. Updates the
sibling deadline test to set both ``_timeout`` and
``_read_timeout`` so the deadline reflects the test's
intended budget.

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

Author: antoineleclair

src/dqliteclient/protocol.py

@@ -739,7 +739,23 @@ def _failure_text(self, response: FailureResponse) -> str:
         return _failure_message(response.message, self._addr_suffix())
 
     def _operation_deadline(self) -> float:
-        """Deadline (monotonic seconds) for a single protocol operation.
+        """Deadline (monotonic seconds) for a single read-side protocol
+        operation.
+
+        Uses ``self._read_timeout`` (NOT ``self._timeout``) so the
+        ``trust_server_heartbeat`` widening — which raises
+        ``_read_timeout`` up to the server-advertised heartbeat
+        capped at ``_HEARTBEAT_READ_TIMEOUT_CAP_SECONDS`` — actually
+        flows through to the per-read deadline budget. Without this,
+        the widening was dead code: ``_read_data``'s
+        ``min(remaining, self._read_timeout)`` always selected
+        ``remaining`` because the deadline came from
+        ``self._timeout`` (un-widened), so the user-configured
+        opt-in had no effect.
+
+        Write-path ``_send`` continues to use ``self._timeout``
+        directly (line ~652) — that's intentional: only the read
+        side gets the heartbeat widening.
 
         Note on multi-phase RPCs: ``timeout`` bounds each phase
         independently — a phase-1 ``_send`` followed by a phase-2
@@ -753,7 +769,7 @@ def _operation_deadline(self) -> float:
         outer call in ``asyncio.timeout`` / ``asyncio.wait_for``.
         This matches go-dqlite's per-phase budgeting.
         """
-        return asyncio.get_running_loop().time() + self._timeout
+        return asyncio.get_running_loop().time() + self._read_timeout
 
     async def _read_continuation(self, deadline: float | None = None) -> RowsResponse:
         """Read and decode a ROWS continuation frame.

tests/test_protocol.py

@@ -139,7 +139,12 @@ async def test_read_response_enforces_operation_deadline(
         import asyncio
         import time
 
+        # Set both the write-path timeout AND the read-path timeout
+        # so the operation deadline (which keys off ``_read_timeout``
+        # so the heartbeat widening flows through) reflects the
+        # tightened budget.
         protocol._timeout = 0.2
+        protocol._read_timeout = 0.2
 
         async def drip_forever(_n: int) -> bytes:
             await asyncio.sleep(0.1)
@@ -155,6 +160,43 @@ async def drip_forever(_n: int) -> bytes:
             f"_read_response must bail at the operation deadline; took {elapsed:.3f}s"
         )
 
+    async def test_read_timeout_widening_actually_extends_operation_deadline(
+        self,
+        protocol: DqliteProtocol,
+        mock_reader: AsyncMock,
+    ) -> None:
+        """Pin: setting ``_read_timeout`` (e.g. via the
+        ``trust_server_heartbeat`` opt-in widening) genuinely extends
+        the per-operation read deadline; the widening is not dead
+        code. A read that takes longer than ``_timeout`` but less
+        than the widened ``_read_timeout`` must complete cleanly.
+        """
+        import asyncio
+
+        from dqlitewire.messages import EmptyResponse
+
+        protocol._timeout = 0.05  # tight write-side
+        protocol._read_timeout = 5.0  # widened read-side
+        body = EmptyResponse().encode()
+
+        sent = [False]
+
+        async def slow_first_read(_n: int) -> bytes:
+            # First call takes longer than _timeout but well within
+            # _read_timeout; deliver the full response so the
+            # decoder completes.
+            if not sent[0]:
+                sent[0] = True
+                await asyncio.sleep(0.2)
+                return body
+            return b""
+
+        mock_reader.read.side_effect = slow_first_read
+
+        # Must NOT raise: the widening flowed through into the
+        # deadline budget. Pre-fix this raised TimeoutError at 0.05s.
+        await protocol._read_response()
+
     async def test_read_data_deadline_already_in_past_raises_immediately(
         self,
         protocol: DqliteProtocol,