From f956682f3145da672e3d5a6ec97427554d70f81c Mon Sep 17 00:00:00 2001 From: ly-wang19 <94427531+ly-wang19@users.noreply.github.com> Date: Tue, 23 Jun 2026 10:25:18 +0800 Subject: [PATCH] perf(runtime): resolve SSE resume offset in O(1) via the event-id seq (#3700) MemoryStreamBridge._resolve_start_offset scanned the retained event buffer (up to queue_maxsize=256 entries) on every subscribe/reconnect carrying a Last-Event-ID. Event ids are "{ts}-{seq}" where seq is a per-run monotonic counter that equals the event's absolute offset, so the offset is computable arithmetically. Parse seq, index into the buffer, and verify the id matches exactly -- a stale/evicted/foreign/malformed id falls back to replay-from-earliest, identical to the previous scan. Co-authored-by: ly-wang19 Co-authored-by: Claude Opus 4.8 (1M context) --- .../deerflow/runtime/stream_bridge/memory.py | 29 +++++++- backend/tests/test_stream_bridge.py | 70 +++++++++++++++++++ 2 files changed, 96 insertions(+), 3 deletions(-) diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py b/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py index cb5b8d1f9..0a03be8af 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py +++ b/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py @@ -48,13 +48,36 @@ class MemoryStreamBridge(StreamBridge): seq = self._counters[run_id] - 1 return f"{ts}-{seq}" + @staticmethod + def _parse_event_seq(event_id: str) -> int | None: + """Extract the per-run sequence number from a ``{ts}-{seq}`` event id. + + ``seq`` (assigned by :meth:`_next_id`) increases by one per published + event, so it equals the event's absolute offset within the run. Returns + ``None`` for ids that do not match the expected format. + """ + _, sep, seq_text = event_id.rpartition("-") + if not sep: + return None + try: + return int(seq_text) + except ValueError: + return None + def _resolve_start_offset(self, stream: _RunStream, last_event_id: str | None) -> int: if last_event_id is None: return stream.start_offset - for index, entry in enumerate(stream.events): - if entry.id == last_event_id: - return stream.start_offset + index + 1 + # Event ids embed a per-run, monotonically increasing ``seq`` that equals + # the event's absolute offset, so locate the event by arithmetic in O(1) + # rather than scanning the retained buffer. The id is verified at the + # computed index, so a stale/evicted/foreign/malformed id still falls back + # to replay-from-earliest — identical to the previous linear scan. + seq = self._parse_event_seq(last_event_id) + if seq is not None: + local_index = seq - stream.start_offset + if 0 <= local_index < len(stream.events) and stream.events[local_index].id == last_event_id: + return stream.start_offset + local_index + 1 if stream.events: logger.warning( diff --git a/backend/tests/test_stream_bridge.py b/backend/tests/test_stream_bridge.py index efd5e7923..6cdf4b67f 100644 --- a/backend/tests/test_stream_bridge.py +++ b/backend/tests/test_stream_bridge.py @@ -334,3 +334,73 @@ async def test_make_stream_bridge_defaults(): """make_stream_bridge() with no config yields a MemoryStreamBridge.""" async with make_stream_bridge() as bridge: assert isinstance(bridge, MemoryStreamBridge) + + +# --------------------------------------------------------------------------- +# _resolve_start_offset: O(1) seq-indexed resolution (parity with linear scan) +# --------------------------------------------------------------------------- + + +def _linear_resolve(stream, last_event_id): + """The original linear-scan resolver, kept as a parity reference.""" + if last_event_id is None: + return stream.start_offset + for index, entry in enumerate(stream.events): + if entry.id == last_event_id: + return stream.start_offset + index + 1 + return stream.start_offset + + +@pytest.mark.parametrize( + "event_id,expected", + [ + ("1718000000000-0", 0), + ("1718000000000-42", 42), + ("garbage", None), # no separator + ("1718000000000-x", None), # non-integer seq + ("", None), + ], +) +def test_parse_event_seq(event_id, expected): + assert MemoryStreamBridge._parse_event_seq(event_id) == expected + + +@pytest.mark.anyio +async def test_resolve_start_offset_matches_linear_scan(): + """The seq-indexed resolver must return exactly what the linear scan returned, + across retained, evicted, foreign (same seq / wrong ts), malformed, and None ids.""" + bridge = MemoryStreamBridge(queue_maxsize=4) + run_id = "run-parity" + ids = [] + for i in range(10): + await bridge.publish(run_id, f"e{i}", {"i": i}) + ids.append(bridge._streams[run_id].events[-1].id) # includes ids that later evict + stream = bridge._streams[run_id] + assert stream.start_offset == 6 # 10 published, buffer of 4 retains seq 6..9 + + # A foreign id: a retained event's seq but a different timestamp -> must NOT match. + ts, _, seq_text = stream.events[0].id.rpartition("-") + foreign_id = f"{int(ts) + 1}-{seq_text}" + + candidates = [None, "garbage", "1718000000000-x", "999999-999999", foreign_id, *ids] + for eid in candidates: + assert bridge._resolve_start_offset(stream, eid) == _linear_resolve(stream, eid), eid + + +@pytest.mark.anyio +async def test_subscribe_with_unknown_last_event_id_replays_from_earliest(): + """A foreign/garbage Last-Event-ID falls back to replaying retained events.""" + bridge = MemoryStreamBridge(queue_maxsize=10) + run_id = "run-unknown-id" + await bridge.publish(run_id, "first", {}) + await bridge.publish(run_id, "second", {}) + await bridge.publish_end(run_id) + + received = [] + async for entry in bridge.subscribe(run_id, last_event_id="not-a-real-id", heartbeat_interval=1.0): + received.append(entry) + if entry is END_SENTINEL: + break + + assert [entry.event for entry in received[:-1]] == ["first", "second"] + assert received[-1] is END_SENTINEL