From 66c0c641bfc2942b33f28c27ae22efaee2707d8f Mon Sep 17 00:00:00 2001 From: Huixin615 Date: Fri, 10 Jul 2026 07:44:45 +0800 Subject: [PATCH] fix(gateway): live-tail malformed redis reconnect ids (#4012) * fix(gateway): live-tail malformed redis reconnect ids * test: avoid redis live-tail heartbeat gaps * fix: handle terminal malformed redis reconnect ids --- README.md | 2 +- backend/AGENTS.md | 2 +- .../deerflow/runtime/stream_bridge/redis.py | 35 ++++-- backend/tests/test_stream_bridge.py | 113 +++++++++++++++--- 4 files changed, 119 insertions(+), 33 deletions(-) diff --git a/README.md b/README.md index f7bf683ce..d13cdea87 100644 --- a/README.md +++ b/README.md @@ -264,7 +264,7 @@ section, when present, overrides the first two for backward compatibility. The unified nginx endpoint is same-origin by default and does not emit browser CORS headers. If you run a split-origin or port-forwarded browser client, set `GATEWAY_CORS_ORIGINS` to comma-separated exact origins such as `http://localhost:3000`; the Gateway then applies the CORS allowlist and matching CSRF origin checks. > [!IMPORTANT] -> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). The Redis stream bridge (`stream_bridge.type: redis`) shares SSE delivery and `Last-Event-ID` replay across workers, with a rolling retained-buffer TTL (`stream_ttl_seconds`) as a cleanup safety net. It does not make run cancellation, request de-duplication, or IM channel state fully cross-worker by itself; use single-worker Gateway or explicit sticky routing/ownership before raising `GATEWAY_WORKERS`. +> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). The Redis stream bridge (`stream_bridge.type: redis`) shares SSE delivery and `Last-Event-ID` replay across workers, with a rolling retained-buffer TTL (`stream_ttl_seconds`) as a cleanup safety net. Malformed reconnect IDs live-tail new events instead of replaying the retained buffer. It does not make run cancellation, request de-duplication, or IM channel state fully cross-worker by itself; use single-worker Gateway or explicit sticky routing/ownership before raising `GATEWAY_WORKERS`. See [CONTRIBUTING.md](CONTRIBUTING.md) for detailed Docker development guide. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 8c7d33d51..e33760434 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -337,7 +337,7 @@ metadata only. - `cancel()` and `create_or_reject(..., multitask_strategy="interrupt"|"rollback")` persist interrupted status through `RunStore.update_status()`, matching normal `set_status()` transitions. - Store-only hydrated runs are readable history. If the current worker has no in-memory task/control state for that run, cancellation APIs can return 409 because this worker cannot stop the task. - `POST /wait` (both thread-scoped and `/api/runs/wait`) drains the stream bridge via `wait_for_run_completion()` instead of bare `await record.task`, so it honours the run's `on_disconnect` setting and cancels the background run on real client disconnect rather than returning a stale checkpoint (issue #3265). -- Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup orphan recovery publishes `END_SENTINEL` and schedules stream cleanup for recovered runs; do not broaden this into a shared-database multi-pod reaper without adding worker ownership/liveness first. +- Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup orphan recovery publishes `END_SENTINEL` and schedules stream cleanup for recovered runs; malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Do not broaden this into a shared-database multi-pod reaper without adding worker ownership/liveness first. - Thread-scoped run creation accepts `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching. - Thread-scoped Gateway runs evaluate an active `ThreadState.goal` after the visible turn completes. `runtime/goal.py` asks a non-thinking evaluator model to judge only visible conversation evidence and return a typed blocker; the evaluator model is created once per run and reused across hidden continuation checks. Satisfied goals are cleared; every non-satisfied evaluation — continuable or stand-down — is persisted with `last_evaluation` (the blocker, reason, and evidence summary; outcomes that stop the loop additionally record a `stand_down_reason` for observability), but only `goal_not_met_yet` evaluations are streamed as hidden `HumanMessage` continuations, and only when a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the thread did not change during evaluation, and the no-progress breaker has not fired. The continuation cap is 8 — a hard maximum in the `0`–`8` range; callers requesting more are clamped (`set_goal`/TUI) or rejected with 422 (`PUT /goal`). The no-progress breaker keys on the latest visible assistant evidence (not the evaluator's free-text reason, which an LLM rewords every turn), so two consecutive continuations that add no new visible assistant output stop the loop after 2 attempts. Model-response cleanup helpers such as think-block stripping and code-fence stripping live in `deerflow.utils.llm_text` so `runtime/goal.py` and Gateway suggestion parsing share the same JSON-prep behavior. diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py b/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py index 8581cefff..c1a488a9d 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py +++ b/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py @@ -6,6 +6,7 @@ import asyncio import inspect import json import logging +import re from collections.abc import AsyncIterator, Mapping from typing import Any @@ -32,6 +33,7 @@ logger = logging.getLogger(__name__) _KIND_EVENT = "event" _KIND_END = "end" +_REDIS_STREAM_ID_RE = re.compile(r"\d+(-\d+)?") # Batch size for ``XREAD``. Reading more than one entry per round-trip collapses # a large ``Last-Event-ID`` replay into far fewer calls; live tailing still @@ -162,6 +164,20 @@ class RedisStreamBridge(StreamBridge): """Return whether Redis still has retained stream data for *run_id*.""" return bool(await self._redis.exists(self._stream_key(run_id))) + async def _resolve_start_stream_id(self, key: str, last_event_id: str | None) -> str: + if last_event_id is None: + return "0-0" + if _REDIS_STREAM_ID_RE.fullmatch(last_event_id): + return last_event_id + entries = await self._redis.xrevrange(key, count=1) + if not entries: + return "0-0" + event_id, fields = entries[0] + payload = self._normalise_fields(fields) + if payload.get("kind") == _KIND_END: + return "0-0" + return self._decode(event_id) + async def subscribe( self, run_id: str, @@ -170,7 +186,7 @@ class RedisStreamBridge(StreamBridge): heartbeat_interval: float = 15.0, ) -> AsyncIterator[StreamEvent]: key = self._stream_key(run_id) - stream_id = last_event_id or "0-0" + stream_id = await self._resolve_start_stream_id(key, last_event_id) block_ms = max(1, int(heartbeat_interval * 1000)) if heartbeat_interval > 0 else 1 consecutive_errors = 0 @@ -178,22 +194,15 @@ class RedisStreamBridge(StreamBridge): try: response = await self._redis.xread({key: stream_id}, count=_XREAD_COUNT, block=block_ms) except ResponseError: - # The only client-controllable stream ID is the Last-Event-ID - # header, so a rejected ID means a malformed reconnect token: - # fall back to replaying from the earliest retained event. We key - # off the control flow rather than the error wording, which is the - # server's text (Redis/Valkey/Dragonfly) and not a stable API. If - # the reset read from "0-0" also fails, the stream/connection is - # genuinely broken, so re-raise. - if stream_id == "0-0": - raise + # Last-Event-ID is client-controlled and validated before XREAD. + # If Redis still rejects the id, fail instead of resetting to + # 0-0, which would replay the whole retained buffer on reconnect. logger.warning( - "Redis rejected Last-Event-ID %r for stream bridge; replaying from earliest retained event", + "Redis rejected stream id %r for stream bridge subscription", stream_id, exc_info=True, ) - stream_id = "0-0" - continue + raise except RedisError: consecutive_errors += 1 if consecutive_errors > _MAX_SUBSCRIBE_RETRIES: diff --git a/backend/tests/test_stream_bridge.py b/backend/tests/test_stream_bridge.py index 781593c51..bf710b436 100644 --- a/backend/tests/test_stream_bridge.py +++ b/backend/tests/test_stream_bridge.py @@ -58,6 +58,10 @@ class _FakeRedis: except TimeoutError: return [] + async def xrevrange(self, name, max="+", min="-", count=None): + entries = list(reversed(self.streams.get(name, []))) + return entries[:count] if count is not None else entries + async def delete(self, name): self.deleted.append(name) self.streams.pop(name, None) @@ -479,6 +483,76 @@ async def test_redis_replays_after_last_event_id(redis_bridge: RedisStreamBridge assert received[-1] is END_SENTINEL +@pytest.mark.anyio +async def test_redis_invalid_last_event_id_tails_live_events(redis_bridge: RedisStreamBridge): + """Malformed reconnect ids should not replay retained Redis events.""" + run_id = "redis-run-invalid-last-event-id" + + await redis_bridge.publish(run_id, "metadata", {"run_id": run_id}) + received = [] + + async def publish_later() -> None: + await anyio.sleep(0.05) + await redis_bridge.publish(run_id, "values", {"step": 1}) + await redis_bridge.publish_end(run_id) + + with anyio.fail_after(2): + async with anyio.create_task_group() as task_group: + task_group.start_soon(publish_later) + async for entry in redis_bridge.subscribe(run_id, last_event_id="-1", heartbeat_interval=0.01): + if entry is HEARTBEAT_SENTINEL: + continue + received.append(entry) + if entry is END_SENTINEL: + break + + assert [entry.event for entry in received[:-1]] == ["values"] + assert received[-1] is END_SENTINEL + + +@pytest.mark.anyio +async def test_redis_invalid_last_event_id_tails_empty_stream(redis_bridge: RedisStreamBridge): + """Malformed reconnect ids should still wait for the first Redis event.""" + run_id = "redis-run-invalid-empty" + received = [] + + async def publish_later() -> None: + await anyio.sleep(0.05) + await redis_bridge.publish(run_id, "metadata", {"run_id": run_id}) + await redis_bridge.publish_end(run_id) + + with anyio.fail_after(2): + async with anyio.create_task_group() as task_group: + task_group.start_soon(publish_later) + async for entry in redis_bridge.subscribe(run_id, last_event_id="-1", heartbeat_interval=0.01): + if entry is HEARTBEAT_SENTINEL: + continue + received.append(entry) + if entry is END_SENTINEL: + break + + assert [entry.event for entry in received[:-1]] == ["metadata"] + assert received[-1] is END_SENTINEL + + +@pytest.mark.anyio +async def test_redis_invalid_last_event_id_on_terminal_run_replays_end(redis_bridge: RedisStreamBridge): + """Malformed reconnect ids on terminal streams should drain END instead of hanging.""" + run_id = "redis-run-invalid-terminal" + + await redis_bridge.publish(run_id, "metadata", {"run_id": run_id}) + await redis_bridge.publish_end(run_id) + + received = [] + async for entry in redis_bridge.subscribe(run_id, last_event_id="-1", heartbeat_interval=1.0): + received.append(entry) + if entry is END_SENTINEL: + break + + assert [entry.event for entry in received[:-1]] == ["metadata"] + assert received[-1] is END_SENTINEL + + @pytest.mark.anyio async def test_redis_heartbeat(redis_bridge: RedisStreamBridge): """Redis bridge should yield heartbeats when XREAD times out on an existing stream.""" @@ -787,9 +861,9 @@ async def test_make_stream_bridge_passes_redis_options(monkeypatch): # `make test` stays green without Redis. Point at a server with # DEER_FLOW_TEST_REDIS_URL (defaults to redis://localhost:6379/15 — DB 15 to # avoid clobbering real data) and select with `pytest -m integration`. They -# cover what _FakeRedis cannot: the real ResponseError fallback for a malformed -# Last-Event-ID, real XADD/XREAD semantics, the server - ID format, -# and MAXLEN trimming. +# cover what _FakeRedis only approximates: real XADD/XREAD semantics, live-tail +# reconnects for malformed Last-Event-ID values, the server - ID +# format, and MAXLEN trimming. REDIS_TEST_URL = os.environ.get("DEER_FLOW_TEST_REDIS_URL", "redis://localhost:6379/15") @@ -878,25 +952,28 @@ async def test_redis_integration_replays_after_last_event_id(real_redis_bridge): @pytest.mark.integration @requires_redis @pytest.mark.anyio -async def test_redis_integration_invalid_last_event_id_falls_back(real_redis_bridge): - """A malformed Last-Event-ID must trigger the ResponseError fallback. - - Real Redis raises ``ResponseError`` for a syntactically-invalid stream ID; - ``_FakeRedis`` cannot reproduce this path, so the ``except ResponseError`` - branch in ``subscribe`` is only exercised here. - """ +async def test_redis_integration_invalid_last_event_id_tails_live_events(real_redis_bridge): + """A malformed Last-Event-ID should wait at the live tail.""" run_id = "integ-bad-leid" await real_redis_bridge.publish(run_id, "metadata", {"run_id": run_id}) - await real_redis_bridge.publish_end(run_id) - received = [] - async for entry in real_redis_bridge.subscribe(run_id, last_event_id="not-a-valid-id", heartbeat_interval=1.0): - received.append(entry) - if entry is END_SENTINEL: - break - # Falls back to replaying from the earliest retained event instead of raising. - assert [e.event for e in received[:-1]] == ["metadata"] + async def publish_later() -> None: + await anyio.sleep(0.05) + await real_redis_bridge.publish(run_id, "values", {"step": 1}) + await real_redis_bridge.publish_end(run_id) + + with anyio.fail_after(2): + async with anyio.create_task_group() as task_group: + task_group.start_soon(publish_later) + async for entry in real_redis_bridge.subscribe(run_id, last_event_id="not-a-valid-id", heartbeat_interval=0.01): + if entry is HEARTBEAT_SENTINEL: + continue + received.append(entry) + if entry is END_SENTINEL: + break + + assert [e.event for e in received[:-1]] == ["values"] assert received[-1] is END_SENTINEL