Apply max_continuation_frames cap to the interrupt() drain loop

interrupt() drained in-flight continuation frames after sending
INTERRUPT but only enforced the operation deadline — not the
max_continuation_frames cap that _drain_continuations uses on the
query path. A slow-dripping server answering an INTERRUPT with a
steady stream of small RowsResponse frames could pin the client on
per-frame decode work within a single deadline, reaching the DoS
scenario the cap was introduced to bound.

Count every RowsResponse (including has_more=False "done marker"
frames) and raise ProtocolError when the cap trips, matching the
shape of the sibling check in _drain_continuations. The
operation-deadline guard is preserved; callers who pass
max_continuation_frames=None keep the existing deadline-only
semantics.

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

Author: antoineleclair

src/dqliteclient/protocol.py

@@ -263,8 +263,12 @@ async def interrupt(self, db_id: int) -> None:
 
         # Drain: swallow any trailing continuation frames, break when
         # EmptyResponse arrives. Bound by the single operation deadline
-        # so a non-responsive server cannot stall this forever.
+        # so a non-responsive server cannot stall this forever, and by
+        # the max_continuation_frames cap so a slow-dripping server
+        # cannot pin the client on per-frame decode work inside that
+        # deadline window (same rationale as _drain_continuations).
         deadline = self._operation_deadline()
+        frames = 0
         while True:
             response = await self._read_response(deadline=deadline)
             if isinstance(response, EmptyResponse):
@@ -279,6 +283,13 @@ async def interrupt(self, db_id: int) -> None:
                     f"Expected EmptyResponse after Interrupt, got "
                     f"{type(response).__name__}{self._addr_suffix()}"
                 )
+            frames += 1
+            if self._max_continuation_frames is not None and frames > self._max_continuation_frames:
+                raise ProtocolError(
+                    f"Interrupt drain exceeded max_continuation_frames cap "
+                    f"({self._max_continuation_frames}); server may be "
+                    f"slow-dripping rows{self._addr_suffix()}."
+                )
             # If the RowsResponse signals more frames, keep draining.
             if not response.has_more:
                 # DONE marker arrived: some servers emit the final

tests/test_interrupt.py

@@ -75,6 +75,76 @@ async def test_interrupt_raises_operational_error_on_failure(
             await protocol.interrupt(db_id=1)
         assert exc_info.value.code == 5
 
+    async def test_interrupt_drain_respects_max_continuation_frames(self) -> None:
+        """The drain loop must honour the same max_continuation_frames
+        cap that _drain_continuations uses on the query path.
+        Otherwise a slow-dripping server answering an INTERRUPT with
+        many small RowsResponse frames can pin a client within a
+        single operation deadline.
+        """
+        reader = AsyncMock()
+        writer = MagicMock()
+        writer.drain = AsyncMock()
+        writer.close = MagicMock()
+        writer.wait_closed = AsyncMock()
+        proto = DqliteProtocol(
+            reader,
+            writer,
+            timeout=10.0,
+            address="test:9001",
+            max_continuation_frames=3,
+        )
+        proto._handshake_done = True
+
+        # Four in-flight frames arrive before EmptyResponse; cap = 3
+        # trips on frame 4. RowsResponse with has_more=False is the
+        # canonical "done marker" shape the drain loop already swallows.
+        rows_frame = RowsResponse(
+            column_names=["x"],
+            column_types=[ValueType.INTEGER],
+            rows=[(1,)],
+            row_types=[(ValueType.INTEGER,)],
+            has_more=False,
+        ).encode()
+        empty = EmptyResponse().encode()
+        proto._reader.read = AsyncMock(  # type: ignore[attr-defined]
+            side_effect=[rows_frame * 4 + empty, b""]
+        )
+        with pytest.raises(ProtocolError, match="max_continuation_frames"):
+            await proto.interrupt(db_id=1)
+
+    async def test_interrupt_drain_no_cap_when_governor_unset(self) -> None:
+        """max_continuation_frames=None restores the existing behaviour
+        (bound only by the operation deadline). Regression guard for
+        callers that opt out of the cap."""
+        reader = AsyncMock()
+        writer = MagicMock()
+        writer.drain = AsyncMock()
+        writer.close = MagicMock()
+        writer.wait_closed = AsyncMock()
+        proto = DqliteProtocol(
+            reader,
+            writer,
+            timeout=10.0,
+            address="test:9001",
+            max_continuation_frames=None,
+        )
+        proto._handshake_done = True
+
+        rows_frame = RowsResponse(
+            column_names=["x"],
+            column_types=[ValueType.INTEGER],
+            rows=[(1,)],
+            row_types=[(ValueType.INTEGER,)],
+            has_more=False,
+        ).encode()
+        empty = EmptyResponse().encode()
+        proto._reader.read = AsyncMock(  # type: ignore[attr-defined]
+            side_effect=[rows_frame * 10 + empty, b""]
+        )
+        # No raise: deadline-only behaviour preserved.
+        await proto.interrupt(db_id=1)
+
     async def test_interrupt_raises_protocol_error_on_unexpected_message(
         self, protocol: DqliteProtocol
     ) -> None: