From aabfdeebc9880e2b7fa3b8bd443d48cc72dca8fd Mon Sep 17 00:00:00 2001 From: spud <92900806+jamespud@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:16:53 +0800 Subject: [PATCH] fix(streaming): make Redis stream END publication idempotent (#7572) * fix(streaming): make Redis stream END publication idempotent * fix(streaming): align Redis test fake with Lua END checks Remove the unreachable owned missing-stream branch, mirror tail-only data checks in both fake script paths, and pin legacy END and owned missing-buffer behavior against real Redis. * fix(streaming): gate legacy END scans after writer rollout * fix(helm): synchronize embedded config version with schema * test(runtime): use real stream bridge config in close coverage Keep new configuration defaults in lifecycle fixtures and surface owner-task failures instead of masking them as event timeouts. Drain every helper task and cover context entry/close failures. --------- Co-authored-by: Willem Jiang --- README.md | 2 +- backend/app/gateway/AGENTS.md | 2 +- backend/docs/CONFIGURATION.md | 1 + backend/docs/STREAMING.md | 2 +- .../deerflow/config/stream_bridge_config.py | 8 + .../harness/deerflow/runtime/AGENTS.md | 15 + .../runtime/stream_bridge/async_provider.py | 1 + .../deerflow/runtime/stream_bridge/redis.py | 92 +++- backend/tests/AGENTS.md | 4 + ...est_runtime_provider_close_cancellation.py | 47 +- backend/tests/test_stream_bridge.py | 484 +++++++++++++++++- config.example.yaml | 9 +- deploy/helm/deer-flow/README.md | 2 +- deploy/helm/deer-flow/values.yaml | 2 +- 14 files changed, 641 insertions(+), 30 deletions(-) diff --git a/README.md b/README.md index 768f55c681..ca1aade939 100644 --- a/README.md +++ b/README.md @@ -658,7 +658,7 @@ Administrators can suspend and restore accounts from **Settings → Users** (adm DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser-facing scheme and origin behind a proxy. The bundled nginx sets `X-Forwarded-Proto`, but preserves an upstream HTTPS value and does not overwrite every forwarded header. Configure the outer trusted proxy to replace or strip client-supplied forwarding headers before traffic reaches DeerFlow. > [!IMPORTANT] -> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. Kubernetes replicas run one worker per Pod, which the worker count cannot see: declare them with `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`, which deploy tooling such as a Helm chart can set from its replica count) so the startup gate enforces the same prerequisites instead of staying inert. Declared instances that store credentials (`channel_connections`, or MCP servers using the OAuth `refresh_token` grant, whose rotating tokens they share through the database) must also share one at-rest `DEER_FLOW_CREDENTIALS_KEY` (the Helm chart and `make up` generate it; back it up, see [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)). The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a trimmed reconnect cursor cannot be verified for complete replay, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. The Memory bridge can replay without recovery when the cursor is its exact last evicted event: every successor is still retained. Older or unverifiable evicted cursors still trigger recovery. An already-observed Redis stream removed during replay also signals a gap, so slow subscribers reload durable state instead of heartbeating forever after cleanup. Redis also tracks terminal stream producers with a short lease: if a worker dies after committing a terminal run status but before publishing END, SSE and `/wait` finish after the lease expires instead of waiting for the retained stream's TTL. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); IM bot credentials entered in the web UI live in the shared `channel_runtime_configs` table, encrypted with `DEER_FLOW_CREDENTIALS_KEY` (an existing `channels/runtime-config.json` is imported once and then deleted), so a save on any instance reaches the provider's connection within one lease renewal; each provider's platform connection is held by one instance through the shared `channel_leases` table — `channels.leader_lease` (default `auto`, on wherever the multi-process signal is) hands it to another instance when the holder dies, and the startup gate refuses `enabled: false` while an IM channel other than GitHub is enabled (see [IM channel leader lease](backend/docs/CONFIGURATION.md#im-channel-leader-lease-channelsleader_lease)); follow-up buffers still need their own multi-worker coordination. To try this topology on one machine, `make dev-multi` starts two Gateways on shared Postgres, Redis and `DEER_FLOW_HOME`, and `make dev-multi-check` runs the cross-instance checks; see the local two-Gateway harness in [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#local-two-gateway-harness). +> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. Kubernetes replicas run one worker per Pod, which the worker count cannot see: declare them with `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`, which deploy tooling such as a Helm chart can set from its replica count) so the startup gate enforces the same prerequisites instead of staying inert. Declared instances that store credentials (`channel_connections`, or MCP servers using the OAuth `refresh_token` grant, whose rotating tokens they share through the database) must also share one at-rest `DEER_FLOW_CREDENTIALS_KEY` (the Helm chart and `make up` generate it; back it up, see [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)). The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a trimmed reconnect cursor cannot be verified for complete replay, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. The Memory bridge can replay without recovery when the cursor is its exact last evicted event: every successor is still retained. Older or unverifiable evicted cursors still trigger recovery. An already-observed Redis stream removed during replay also signals a gap, so slow subscribers reload durable state instead of heartbeating forever after cleanup. Redis also tracks terminal stream producers with a short lease: if a worker dies after committing a terminal run status but before publishing END, SSE and `/wait` finish after the lease expires instead of waiting for the retained stream's TTL. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. Redis END publication is atomic and idempotent: owner and recovery retries confirm an existing marker without appending another, events cannot be appended once END is the stream tail, and an unowned recovery retry after cleanup does not recreate the stream. Legacy streams with an earlier END followed by late data can still accept data. The Redis compatibility scan (`stream_bridge.legacy_end_scan_enabled: true`) also runs on first END publication; disable it only after old writers stop and all legacy replay buffers expire or are removed, then restart every Gateway to use tail-only END detection. SSE, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); IM bot credentials entered in the web UI live in the shared `channel_runtime_configs` table, encrypted with `DEER_FLOW_CREDENTIALS_KEY` (an existing `channels/runtime-config.json` is imported once and then deleted), so a save on any instance reaches the provider's connection within one lease renewal; each provider's platform connection is held by one instance through the shared `channel_leases` table — `channels.leader_lease` (default `auto`, on wherever the multi-process signal is) hands it to another instance when the holder dies, and the startup gate refuses `enabled: false` while an IM channel other than GitHub is enabled (see [IM channel leader lease](backend/docs/CONFIGURATION.md#im-channel-leader-lease-channelsleader_lease)); follow-up buffers still need their own multi-worker coordination. To try this topology on one machine, `make dev-multi` starts two Gateways on shared Postgres, Redis and `DEER_FLOW_HOME`, and `make dev-multi-check` runs the cross-instance checks; see the local two-Gateway harness in [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#local-two-gateway-harness). > > The producer-lease recovery behavior above is gated by `stream_bridge.producer_lease_recovery_enabled`. Keep it disabled while old Gateway binaries or their runs remain; enable it on every instance only after the writer rollout and drain complete. > diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 0e90cd47fc..ff70bcf115 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -211,7 +211,7 @@ or `deployment.multi_instance: true`. - `DELETE /api/threads/{id}` holds a durable `delete` reservation for the whole cleanup and removes thread filesystem data (off-loop, drained on cancel), checkpoints, owner-scoped historical runs, run events, feedback and thread metadata best-effort. Historical-run cleanup removes only `operation_kind="run"` rows, so internal thread-operation rows — including the reservation protecting that request — stay durable until `RunManager.reserve_thread_operation()` exits. This cleans persisted rows; preventing an already-admitted write from re-creating state for a deleted thread is a separate thread-incarnation contract, not part of this cleanup. - Creator SSE and `/wait` apply `on_disconnect` only to confirmed request disconnect or cancellation; bridge failures propagate without cancelling the run. `/wait` consumes bridge END; errors return SDK `__error__`, never stale state. - Memory and Redis `StreamBridge` implementations retain only `stream_bridge.queue_maxsize` data events. A syntactically valid `Last-Event-ID` older than the retained watermark, or a live subscriber that falls behind it, yields `StreamGap` before any partial replay. `sse_consumer` maps that control item to an id-less SSE `gap` payload (`stream_replay_gap`) and intentionally leaves the run active; internal `/wait` consumers resume from its latest retained ID because they only need terminal completion. Redis checks bounds plus the non-blocking read in one transaction, using blocking `XREAD` only as a wake-up before repeating the atomic snapshot. For a no-cursor subscriber that established a wait on an empty stream, the first wake response remains provisional until that next snapshot verifies its tail is still retained; this closes the pre-first-delivery trimming window without changing malformed-cursor live tailing. The correctness tradeoff is one three-command snapshot pipeline per poll plus the blocking wake round trip while idle. Malformed cursor behavior remains backend-specific. Memory treats a syntactically numeric cursor below its watermark conservatively as a gap even when the evicted timestamp can no longer be verified; unknown ids at or above the watermark retain the legacy replay-from-earliest policy. -- 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 and lease-driven periodic orphan recovery share one Gateway stream-terminalization path: after `RunManager` durably marks a run `error` with `stop_reason=orphan_recovered`, Gateway publishes `END_SENTINEL` and schedules stream cleanup. The periodic store scan, per-row status writes, and Gateway callback run as one supervised single-flight task, so a slow pass is skipped at the next interval instead of piling up or pausing the sole lease-renewal loop. Store retries have bounded attempts/backoff; an individual operation still relies on the database driver/pool timeout. `RunManager.shutdown()` gives active user runs priority within its shared deadline, then drains or cancels orphan recovery. Gateway tracks delayed recovered-stream cleanups and converts unfinished delays to immediate deletes before closing the bridge; the Redis TTL remains the outage safety net. Only startup recovery, before the runtime yields to requests, projects the latest affected thread to `error`; periodic recovery deliberately avoids that non-atomic projection because `ThreadMetaStore` has no `latest_run_id` conditional-update contract. Store-only SSE and `/wait` consumers wait for the bridge's real END marker after an ordinary durable terminal status, because status persistence can precede tail events. Terminal `orphan_recovered` and `scheduled_task_orphan_recovered` rows permit heartbeat fallback because their producer is gone; ordinary terminal rows still await END. Malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Keep cross-component recovery orchestration in Gateway through the generic `RunManager.on_orphans_recovered` callback; do not introduce a harness-to-app dependency. Callback failure warnings include every recovered `run_id` so operators can identify rows whose Gateway-side terminalization needs inspection. +- Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on data and first-END writes) as a leak safety net, not as a run timeout. Startup and lease-driven periodic orphan recovery share one Gateway stream-terminalization path: after `RunManager` durably marks a run `error` with `stop_reason=orphan_recovered`, Gateway publishes `END_SENTINEL` and schedules stream cleanup. The periodic store scan, per-row status writes, and Gateway callback run as one supervised single-flight task, so a slow pass is skipped at the next interval instead of piling up or pausing the sole lease-renewal loop. Store retries have bounded attempts/backoff; an individual operation still relies on the database driver/pool timeout. `RunManager.shutdown()` gives active user runs priority within its shared deadline, then drains or cancels orphan recovery. Gateway tracks delayed recovered-stream cleanups and converts unfinished delays to immediate deletes before closing the bridge; the Redis TTL remains the outage safety net. Only startup recovery, before the runtime yields to requests, projects the latest affected thread to `error`; periodic recovery deliberately avoids that non-atomic projection because `ThreadMetaStore` has no `latest_run_id` conditional-update contract. Store-only SSE and `/wait` consumers wait for the bridge's real END marker after an ordinary durable terminal status, because status persistence can precede tail events. Terminal `orphan_recovered` and `scheduled_task_orphan_recovered` rows permit heartbeat fallback because their producer is gone; ordinary terminal rows still await END. Malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Keep cross-component recovery orchestration in Gateway through the generic `RunManager.on_orphans_recovered` callback; do not introduce a harness-to-app dependency. Callback failure warnings include every recovered `run_id` so operators can identify rows whose Gateway-side terminalization needs inspection. - Thread-scoped create, stream and wait share an optional owner/thread-scoped `Idempotency-Key` (internal callers: owner-only, cross-thread reuse 409); never pass the raw external key to `RunManager`'s process-wide index. Stores resolve an existing key before `interrupt`/`rollback`, so a retry reuses rather than interrupts its run. Reuse requires matching input, assistant and conversation references; keyed resume stores a versioned strict typed-JSON SHA-256 (`true`, `1` and `1.0` differ) in the private `runs.idempotency_request_json` column, never public `kwargs`; ordinary input needs no extra identity. New readers accept the exact old-writer resume identity shape from `kwargs`, every response strips it; identity-less legacy resume fails closed. Validate strict resume JSON with or without a key; persist the identity only for keyed resume admission. Missing headers stay non-idempotent; stateless `/api/runs/*` is outside this contract (it creates a thread before admission). - For a rolling upgrade, route every keyed resume submission/retry on create/stream/wait only to upgraded workers; otherwise pause that traffic or replace old workers first. Old workers ignore the private identity and can reuse `approve` for `deny` with null input. Old-reader schema compatibility does not protect admission. Upgraded workers reject identity-less legacy resumes with 409. Rollout details: `README.md`, `backend/docs/API.md`. - Reused `/wait` returns durable status/error, never the latest thread checkpoint; `task is None` is not completion, and store-backed fields must refresh after waiting. Capture reuse before awaiting because the cached record's `idempotency_reused` flag is sticky. A creating `/runs/stream` retry whose terminal stream is gone opts into `gap`/`stream_replay_gap` via `emit_gap_on_missing_stream`, while observer joins and default/stateless consumers emit `end`; never infer this from disconnect policy or the sticky flag. A reused running `store_only` record on a process-local bridge returns 409 without `Retry-After`. diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index f7f06ea223..975491abfd 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -763,6 +763,7 @@ Notes: - Set `multi_instance: true` (or export `DEER_FLOW_MULTI_INSTANCE=1`, which lets deploy tooling such as a Helm chart set it from its replica count) on every instance that shares one database. Startup then enforces the same prerequisites as `GATEWAY_WORKERS > 1`: `database.backend: postgres`, `run_events.backend: db`, `run_ownership.heartbeat_enabled: true`, and a Redis stream bridge (`stream_bridge.type: redis` or `DEER_FLOW_STREAM_BRIDGE_REDIS_URL`). It refuses an explicit `sandbox.ownership.type: memory`, process-local browser tools, and `scheduler.enabled: true` without `scheduler.multi_instance: true`. - `DEER_FLOW_MULTI_INSTANCE` treats blank, `0`, `false`, `no` and `off` as "not declared"; any other value declares a multi-instance deployment, so a typo fails closed. - `GET /health/ready` pings the Redis stream bridge on every probe and answers 503 (`stream_bridge: unreachable`) while Redis is down, so the orchestrator drains that instance instead of routing it runs it cannot publish or stream; the memory bridge reports `not_configured`. When `sandbox.provisioner_url` is set, the body also carries the provisioner's `/health` verdict (`provisioner: ok|unreachable`), which never changes the status code because every instance shares one provisioner. That probe follows the sandbox clients' proxy policy: loopback, private, link-local and cluster-local provisioner addresses bypass `HTTP_PROXY`, external hosts keep the environment's proxy settings. +- `stream_bridge.legacy_end_scan_enabled` defaults to `true` for Redis: when the tail is not END, END publication scans all retained entries, including a normal run's first END. Cost grows with the retained entry count and total payload bytes inside atomic Lua, blocking other Redis clients; `queue_maxsize` bounds entries, not bytes. Stop every old writer and expire or remove every legacy replay buffer before setting this to `false` on all Gateways and restarting them. With `stream_ttl_seconds: 0`, remove those buffers explicitly. Disabled mode detects END only at the tail, so both data and END checks have constant cost relative to retained history; it cannot deduplicate an old mid-stream END. No performance ceiling is implied by the default entry cap. - `stream_bridge.producer_lease_recovery_enabled` is a restart-required activation gate for recovering a terminal Redis stream whose producer died before publishing END. Keep it `false` during the binary rollout: upgraded writers already publish producer markers, while old writers do not. After every old Gateway has stopped and its runs have drained, set it to `true` on every instance and restart them together. Enabling it while an old worker may still finalize a run can make a new consumer mistake that untracked legacy producer for completed history. - The declaration also drives the `agent_storage.backend: file` divergence warning, the inbound webhook dedupe warning, the `auth.local.throttle_storage: memory` warning, and the WeChat QR-login guard, which otherwise only look at the worker count. - Login lockouts are shared through the database: `auth.local.throttle_storage` (startup-only) defaults to `auto`, which keeps the per-IP failed-login counters for `POST /api/v1/auth/login/local` in the `login_throttle` table whenever `database.backend` is `sqlite` or `postgres`, so every instance enforces one `max_login_attempts` limit per client IP and a lockout on one instance holds on all of them. `memory` keeps the historical per-process counter (N instances give an attacker N × `max_login_attempts` guesses; a declared multi-instance deployment logs a warning), and `db` forces the table (it falls back to memory with a warning when the database backend is `memory`, and refuses to start when the configured database's engine is unavailable; `auto` falls back with a warning in that case). `max_login_attempts` and `lockout_seconds` stay live-read. diff --git a/backend/docs/STREAMING.md b/backend/docs/STREAMING.md index 5606e60785..9355d6686c 100644 --- a/backend/docs/STREAMING.md +++ b/backend/docs/STREAMING.md @@ -143,7 +143,7 @@ sequenceDiagram 关键组件： - `runtime/runs/worker.py::run_agent` — 在 `asyncio.Task` 里跑 `agent.astream()`，把每个 chunk 通过 `serialize(chunk, mode=mode)` 转成 JSON，再 `bridge.publish()`。 -- `runtime/stream_bridge` — 抽象 Queue。`publish/subscribe` 解耦生产者和消费者，支持 `Last-Event-ID` 重连、心跳、多订阅者 fan-out。`stream_bridge.heartbeat_interval_seconds`（默认 15 秒）是 bridge 实例的默认心跳周期，统一作用于 Gateway SSE、`/wait` 和内部 channel watcher；显式传给 `subscribe()` 的值仍可覆盖单次订阅。Memory 和 Redis 都只保留 `queue_maxsize` 条数据事件；游标早于保留水位线时返回 `StreamGap`，不会从当前最早事件静默部分重放。Redis backend 会在每次 `publish()` / `publish_end()` 刷新 retained stream key TTL；启动恢复与基于 worker lease 的周期恢复共用 Gateway stream terminalization 路径：`RunManager` 先将 orphan run 持久化为 `error` 并写入显式的 `stop_reason=orphan_recovered`，随后 Gateway 发布 `END_SENTINEL` 并安排 stream cleanup。周期扫描、逐行状态写入和 Gateway callback 作为一个受监督的 single-flight 后台 task 执行；慢任务不会堆积，也不会阻塞唯一的 lease heartbeat。shutdown 优先收敛活跃 run，再处理恢复 task；尚未执行的延迟 stream cleanup 会改为立即删除。只有 runtime `yield` 前、无并发请求的启动恢复会把最新受影响 thread 标记为 error；周期恢复不做非原子的 thread 投影。普通 durable terminal status 本身仍不足以证明尾帧已发布；Redis producer 在 durable admission 前取得短租约，经 request task、worker、finalizer 转移并保留到 END。只有显式 `orphan_recovered` 或 durable terminal 加已失效的 tracked producer lease 才能在 heartbeat 上触发 END fallback；lease loss 会阻止后续 stream 写入，active 或 untracked producer 继续等待，避免截断尾帧。TTL 仍是 Redis 内存和故障安全网，不是正常的 subscriber 终止机制。 +- `runtime/stream_bridge` — 抽象 Queue。`publish/subscribe` 解耦生产者和消费者，支持 `Last-Event-ID` 重连、心跳、多订阅者 fan-out。`stream_bridge.heartbeat_interval_seconds`（默认 15 秒）是 bridge 实例的默认心跳周期，统一作用于 Gateway SSE、`/wait` 和内部 channel watcher；显式传给 `subscribe()` 的值仍可覆盖单次订阅。Memory 和 Redis 都只保留 `queue_maxsize` 条数据事件；游标早于保留水位线时返回 `StreamGap`，不会从当前最早事件静默部分重放。Redis backend 会在写入数据或首个 END 时刷新 retained stream key TTL；启动恢复与基于 worker lease 的周期恢复共用 Gateway stream terminalization 路径：`RunManager` 先将 orphan run 持久化为 `error` 并写入显式的 `stop_reason=orphan_recovered`，随后 Gateway 发布 `END_SENTINEL` 并安排 stream cleanup。周期扫描、逐行状态写入和 Gateway callback 作为一个受监督的 single-flight 后台 task 执行；慢任务不会堆积，也不会阻塞唯一的 lease heartbeat。shutdown 优先收敛活跃 run，再处理恢复 task；尚未执行的延迟 stream cleanup 会改为立即删除。只有 runtime `yield` 前、无并发请求的启动恢复会把最新受影响 thread 标记为 error；周期恢复不做非原子的 thread 投影。普通 durable terminal status 本身仍不足以证明尾帧已发布；Redis producer 在 durable admission 前取得短租约，经 request task、worker、finalizer 转移并保留到 END。只有显式 `orphan_recovered` 或 durable terminal 加已失效的 tracked producer lease 才能在 heartbeat 上触发 END fallback；lease loss 会阻止后续 stream 写入，active 或 untracked producer 继续等待，避免截断尾帧。Redis END 通过同一个 Lua 原子完成 lease 校验、既有 END 检查和必要写入，因此重复调用、owner/recovery 竞争和 ACK 丢失重试至多追加一个 END；当 END 已位于 stream 尾部时，后续事件写入会被拒绝；旧 stream 中 END 后已有迟到数据时，仍允许继续写入数据。cleanup 后无 owner 的迟到 END 重试是成功 no-op，不会复活已删除的 stream。普通事件只做 O(1) 尾部 END 检查。`stream_bridge.legacy_end_scan_enabled` 默认 `true`：尾部不是 END 时扫描整个保留窗口，包括正常 run 的首次 END；原子 Lua 的耗时随保留条目数量和 payload 总字节数增长，期间会阻塞其他 Redis 客户端。只有全部旧 writer 停止且旧 replay buffer 已过期或清除后，才可设为 `false` 并重启所有 Gateway，使 END 检查也只看尾部。TTL 为 0 时必须显式清除旧 buffer。TTL 仍是 Redis 内存和故障安全网，不是正常的 subscriber 终止机制。 - `stream_bridge.producer_lease_recovery_enabled` 是上述 producer-lease fallback 的 restart-required activate gate。升级时先保持 `false` 完成所有 writer 的滚动发布并排空旧 run，再在所有 Gateway 上同时设为 `true` 并重启；mixed-version 阶段不得启用。 - `app/gateway/services.py::sse_consumer` — 从 bridge 订阅，格式化为 SSE wire 帧。 - `runtime/serialization.py::serialize` — mode-aware 序列化；`messages` mode 下 `serialize_messages_tuple` 把 `(chunk, metadata)` 转成 `[chunk.model_dump(), metadata]`。 diff --git a/backend/packages/harness/deerflow/config/stream_bridge_config.py b/backend/packages/harness/deerflow/config/stream_bridge_config.py index f72f17df34..ae7b5274a8 100644 --- a/backend/packages/harness/deerflow/config/stream_bridge_config.py +++ b/backend/packages/harness/deerflow/config/stream_bridge_config.py @@ -54,6 +54,14 @@ class StreamBridgeConfig(BaseModel): "even if cleanup never runs. Set to 0 to disable. Only applies to the redis bridge." ), ) + legacy_end_scan_enabled: bool = Field( + default=True, + description=( + "Scan retained Redis history when END is not the tail, including first END publication. " + "Disable only after all old writers stop and all legacy replay buffers expire or are removed. " + "Disabling makes END detection tail-only. Restart every Gateway after changing; Redis only." + ), + ) producer_lease_recovery_enabled: bool = Field( default=False, description=("Allow terminal Redis consumers to recover a missing END from an expired producer lease. Enable only after every Gateway writer has been upgraded and pre-upgrade runs have drained."), diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 69e3234971..05d1beb1e1 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -42,6 +42,21 @@ Memory/Redis bridges keep startup-only `stream_bridge.heartbeat_interval_seconds Reconnect cursor parsing requires ASCII digits and must treat integer-conversion failures as malformed input. Redis cursors additionally require unsigned 64-bit components; use the same parser before sending a cursor to `XREAD` and when detecting replay gaps. Preserve malformed-cursor policy: memory replays retained events, Redis tails active streams and replays ended streams through `end`. `tests/test_stream_bridge.py` directly tests both parsers offline, including oversized sequences, Redis component overflow, non-ASCII digits, and valid uint64 boundaries; integration tests also exercise malformed cursors against real Redis. +### Redis Stream END Publication + +Keep the producer-token check first, then END detection, then `XADD` in one Lua +script. Ordinary event publication checks only the stream tail (`XREVRANGE ... +COUNT 1`), so token streaming remains O(1) as retained history grows. END +publication checks the tail first. With `legacy_end_scan_enabled` (default true), +a non-END tail triggers a full retained-window scan even on first END; its +payload-byte cost blocks Redis. Disable only after old writers stop and legacy +buffers expire or are removed; restart all Gateways together. Disabled mode is +tail-only and cannot deduplicate an old mid-stream END. Do not add an unbounded boolean state key for +this fast path; cleanup, expiry, and a new producer generation would then need a +separate lifecycle contract. An unowned recovery END for a missing stream is a +successful no-op; a live owner's write retains its existing missing-buffer +`XADD` behavior. Test fakes must mirror the scan switch and both tail fast paths. Coverage: `tests/test_stream_bridge.py`. + ### Memory Stream Replay Memory keeps the exact last evicted event ID as bounded per-run metadata. A diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/async_provider.py b/backend/packages/harness/deerflow/runtime/stream_bridge/async_provider.py index 098ca7fc46..0d8153d5d4 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/async_provider.py +++ b/backend/packages/harness/deerflow/runtime/stream_bridge/async_provider.py @@ -85,6 +85,7 @@ async def make_stream_bridge(app_config: AppConfig | None = None) -> AsyncIterat heartbeat_interval=config.heartbeat_interval_seconds, max_connections=config.max_connections, stream_ttl_seconds=config.stream_ttl_seconds, + legacy_end_scan_enabled=config.legacy_end_scan_enabled, producer_lease_recovery_enabled=getattr( config, "producer_lease_recovery_enabled", diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py b/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py index a024d6ccce..6ac5bc6e55 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py +++ b/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py @@ -40,6 +40,10 @@ _KIND_EVENT = "event" _KIND_END = "end" _REDIS_STREAM_ID_RE = re.compile(r"[0-9]+(-[0-9]+)?") _MAX_STREAM_ID_COMPONENT = 2**64 - 1 +# Lua XADD scripts return one of these integers before any event ID. +_XADD_BLOCKED = 0 +_XADD_ALREADY_ENDED = 1 +_XADD_STREAM_MISSING = 2 # 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 @@ -71,9 +75,44 @@ redis.call("expire", KEYS[1], ARGV[2]) if tonumber(ARGV[3]) > 0 then redis.call("expire", KEYS[2], ARGV[3]) end return 1 """ -_COMPARE_XADD_SCRIPT = """ -if redis.call("get", KEYS[1]) ~= ARGV[1] then return false end -if redis.call("get", KEYS[2]) ~= ARGV[1] then return false end +_END_MARKER_CHECK_LUA = """ +local function entry_is_end(fields) + for index = 1, #fields, 2 do + if fields[index] == "kind" and fields[index + 1] == "end" then + return true + end + end + return false +end + +local function tail_is_end(stream_key) + local entries = redis.call("xrevrange", stream_key, "+", "-", "COUNT", 1) + return #entries > 0 and entry_is_end(entries[1][2]) +end + +local function has_end_marker(stream_key, legacy_scan_enabled) + if tail_is_end(stream_key) then return true end + if legacy_scan_enabled ~= "1" then return false end + -- Compatibility path for retained streams written by the old buggy writer: + -- an END may precede later entries. The first END on a normal stream also + -- reaches this scan; operators may disable it only after legacy buffers drain. + local entries = redis.call("xrange", stream_key, "-", "+") + for _, entry in ipairs(entries) do + if entry_is_end(entry[2]) then return true end + end + return false +end +""" +_COMPARE_XADD_SCRIPT = ( + _END_MARKER_CHECK_LUA + + """ +if redis.call("get", KEYS[1]) ~= ARGV[1] then return 0 end +if redis.call("get", KEYS[2]) ~= ARGV[1] then return 0 end +if ARGV[4] == "end" then + if has_end_marker(KEYS[3], ARGV[7]) then return 1 end +elseif tail_is_end(KEYS[3]) then + return 1 +end local event_id = redis.call( "xadd", KEYS[3], "MAXLEN", "=", ARGV[2], "*", "kind", ARGV[4], "event", ARGV[5], "data", ARGV[6] @@ -81,8 +120,17 @@ local event_id = redis.call( if tonumber(ARGV[3]) > 0 then redis.call("expire", KEYS[3], ARGV[3]) end return event_id """ -_XADD_IF_NO_ACTIVE_PRODUCER_SCRIPT = """ -if redis.call("exists", KEYS[1]) == 1 then return false end +) +_XADD_IF_NO_ACTIVE_PRODUCER_SCRIPT = ( + _END_MARKER_CHECK_LUA + + """ +if redis.call("exists", KEYS[1]) == 1 then return 0 end +if ARGV[3] == "end" and redis.call("exists", KEYS[2]) == 0 then return 2 end +if ARGV[3] == "end" then + if has_end_marker(KEYS[2], ARGV[6]) then return 1 end +elseif tail_is_end(KEYS[2]) then + return 1 +end local event_id = redis.call( "xadd", KEYS[2], "MAXLEN", "=", ARGV[1], "*", "kind", ARGV[3], "event", ARGV[4], "data", ARGV[5] @@ -90,6 +138,7 @@ local event_id = redis.call( if tonumber(ARGV[2]) > 0 then redis.call("expire", KEYS[2], ARGV[2]) end return event_id """ +) _CLEANUP_IF_TOKEN_SCRIPT = """ local tracked = redis.call("get", KEYS[2]) local active = redis.call("get", KEYS[3]) @@ -139,11 +188,13 @@ class RedisStreamBridge(StreamBridge): max_connections: int | None = None, stream_ttl_seconds: int | None = 86400, producer_lease_recovery_enabled: bool = False, + legacy_end_scan_enabled: bool = True, client: Redis | None = None, ) -> None: super().__init__(heartbeat_interval=heartbeat_interval) self._redis_url = redis_url self._maxsize = max(1, queue_maxsize) + self._legacy_end_scan_enabled = legacy_end_scan_enabled self._key_prefix = key_prefix.rstrip(":") if stream_ttl_seconds is not None and stream_ttl_seconds > 0: self._stream_ttl_seconds = stream_ttl_seconds @@ -442,11 +493,17 @@ class RedisStreamBridge(StreamBridge): fields.get("kind", _KIND_EVENT), fields.get("event", ""), fields.get("data", ""), + int(self._legacy_end_scan_enabled), ) - if result: - return - self._cancel_lost_producer(run_id, lease) - raise RuntimeError(f"Redis stream producer lease was lost for run {run_id}") + if result == _XADD_BLOCKED: + self._cancel_lost_producer(run_id, lease) + raise RuntimeError(f"Redis stream producer lease was lost for run {run_id}") + if result == _XADD_ALREADY_ENDED: + if fields.get("kind") == _KIND_END: + return + raise RuntimeError(f"Redis stream for run {run_id} has already ended") + # Every remaining successful Lua reply is an XADD stream ID. + assert isinstance(result, (str, bytes)), f"Unexpected Redis stream append result: {result!r}" async def _xadd_without_active_producer( self, @@ -465,10 +522,21 @@ class RedisStreamBridge(StreamBridge): fields.get("kind", _KIND_EVENT), fields.get("event", ""), fields.get("data", ""), + int(self._legacy_end_scan_enabled), ) - if result: - return - raise RuntimeError(f"Cannot publish stream data for run {run_id} because another active producer exists") + if result == _XADD_BLOCKED: + raise RuntimeError(f"Cannot publish stream data for run {run_id} because another active producer exists") + if result == _XADD_ALREADY_ENDED: + if fields.get("kind") == _KIND_END: + return + raise RuntimeError(f"Redis stream for run {run_id} has already ended") + if result == _XADD_STREAM_MISSING: + if fields.get("kind") == _KIND_END: + logger.debug("Skipping stream END for %s: retained stream no longer exists", run_id) + return + raise RuntimeError(f"Redis stream for run {run_id} no longer exists") + # Every remaining successful Lua reply is an XADD stream ID. + assert isinstance(result, (str, bytes)), f"Unexpected Redis stream append result: {result!r}" async def _stop_producer_lease( self, diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md index ae611e4129..2213a68b0d 100644 --- a/backend/tests/AGENTS.md +++ b/backend/tests/AGENTS.md @@ -18,6 +18,10 @@ ownership and capture cleanup. Retain the real XLSX conversion coverage. Backend tests must preserve the runtime invariants they exercise without changing production execution topology. +Provider cleanup tests use concrete configuration models and event gates. +Surface owner-task failures while awaiting lifecycle events; join owners and +event waiters during teardown. + `test_bench_sandbox_prewarm.py` controls observation timestamps through the benchmark module's clock binding. Keep asyncio's clock real, initialize `Trace.started` from the controlled clock, and advance ticks at event gates. diff --git a/backend/tests/test_runtime_provider_close_cancellation.py b/backend/tests/test_runtime_provider_close_cancellation.py index c93b10ad65..cad6d707ef 100644 --- a/backend/tests/test_runtime_provider_close_cancellation.py +++ b/backend/tests/test_runtime_provider_close_cancellation.py @@ -1,8 +1,10 @@ import asyncio +from contextlib import asynccontextmanager from types import SimpleNamespace import pytest +from deerflow.config.stream_bridge_config import StreamBridgeConfig from deerflow.runtime.checkpoint_cache import provider as checkpoint_provider from deerflow.runtime.checkpoint_cache import redis as checkpoint_redis from deerflow.runtime.stream_bridge import async_provider as stream_provider @@ -30,6 +32,20 @@ class _BlockingStreamBridge: await self.allow_close.wait() +async def _wait_for_owner_event(task: asyncio.Task[None], event: asyncio.Event) -> None: + waiter = asyncio.create_task(event.wait()) + try: + done, _ = await asyncio.wait((task, waiter), timeout=1, return_when=asyncio.FIRST_COMPLETED) + if task in done: + await task + raise AssertionError("provider owner finished before its lifecycle event") + if waiter not in done: + raise TimeoutError("provider did not reach its lifecycle event") + finally: + waiter.cancel() + await asyncio.gather(waiter, return_exceptions=True) + + async def _assert_close_is_drained(cm, close_started: asyncio.Event, allow_close: asyncio.Event) -> None: entered = asyncio.Event() leave = asyncio.Event() @@ -42,9 +58,9 @@ async def _assert_close_is_drained(cm, close_started: asyncio.Event, allow_close task: asyncio.Task[None] | None = None try: task = asyncio.create_task(owner()) - await asyncio.wait_for(entered.wait(), timeout=1) + await _wait_for_owner_event(task, entered) leave.set() - await asyncio.wait_for(close_started.wait(), timeout=1) + await _wait_for_owner_event(task, close_started) task.cancel() for _ in range(5): @@ -61,8 +77,9 @@ async def _assert_close_is_drained(cm, close_started: asyncio.Event, allow_close await task finally: allow_close.set() - if task is not None and not task.done(): - task.cancel() + if task is not None: + if not task.done(): + task.cancel() await asyncio.gather(task, return_exceptions=True) @@ -107,7 +124,7 @@ async def test_checkpoint_cache_redis_context_drains_close_across_repeated_cance async def test_stream_bridge_context_drains_close_across_repeated_cancellation(monkeypatch: pytest.MonkeyPatch) -> None: bridge = _BlockingStreamBridge() monkeypatch.setattr(stream_memory, "MemoryStreamBridge", lambda **_kwargs: bridge) - config = SimpleNamespace(stream_bridge=SimpleNamespace(type="memory", queue_maxsize=8, heartbeat_interval_seconds=1.0)) + config = SimpleNamespace(stream_bridge=StreamBridgeConfig(type="memory", queue_maxsize=8, heartbeat_interval_seconds=1.0)) await _assert_close_is_drained( stream_provider.make_stream_bridge(config), @@ -123,7 +140,7 @@ async def test_stream_bridge_redis_context_drains_close_across_repeated_cancella bridge = _BlockingStreamBridge() monkeypatch.setattr(stream_redis, "RedisStreamBridge", lambda *_args, **_kwargs: bridge) config = SimpleNamespace( - stream_bridge=SimpleNamespace( + stream_bridge=StreamBridgeConfig( type="redis", redis_url="redis://localhost:6379/0", queue_maxsize=8, @@ -138,3 +155,21 @@ async def test_stream_bridge_redis_context_drains_close_across_repeated_cancella bridge.close_started, bridge.allow_close, ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure_stage", ["enter", "close"]) +async def test_close_drain_helper_propagates_owner_failure(failure_stage: str) -> None: + error = RuntimeError(f"provider {failure_stage} failed") + + @asynccontextmanager + async def failing_context(): + if failure_stage == "enter": + raise error + yield + raise error + + with pytest.raises(RuntimeError, match=f"provider {failure_stage} failed") as raised: + await _assert_close_is_drained(failing_context(), asyncio.Event(), asyncio.Event()) + + assert raised.value is error diff --git a/backend/tests/test_stream_bridge.py b/backend/tests/test_stream_bridge.py index 2b00f9686a..ce7e1dab4d 100644 --- a/backend/tests/test_stream_bridge.py +++ b/backend/tests/test_stream_bridge.py @@ -5,7 +5,7 @@ import os import re import uuid from collections import defaultdict -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, Mock import anyio import pytest @@ -50,6 +50,7 @@ class _FakeRedis: self.closed = False self.ping_calls = 0 self.ping_error: Exception | None = None + self._eval_lock = asyncio.Lock() async def ping(self): self.ping_calls += 1 @@ -107,7 +108,20 @@ class _FakeRedis: async def mget(self, *names): return [self.values.get(name) for name in names] + def _tail_is_end(self, stream_key): + entries = self.streams.get(stream_key, []) + return bool(entries and entries[-1][1].get("kind") == "end") + + def _has_end_marker(self, stream_key): + return any(fields.get("kind") == "end" for _event_id, fields in self.streams.get(stream_key, [])) + async def eval(self, script, numkeys, *args): + # Redis executes a Lua script atomically. Keep the fake's assertion + # surface faithful even though individual emulated commands yield. + async with self._eval_lock: + return await self._eval(script, numkeys, *args) + + async def _eval(self, script, numkeys, *args): keys = args[:numkeys] values = args[numkeys:] if numkeys == 2 and not values: @@ -133,11 +147,18 @@ class _FakeRedis: return 0 await self.delete(stream_key, tracking_key, active_key) return 1 - if numkeys == 2 and len(values) == 5: + if numkeys == 2 and len(values) == 6: active_key, stream_key = keys - maxlen, stream_ttl, kind, event, data = values + maxlen, stream_ttl, kind, event, data, legacy_scan_enabled = values if active_key in self.values: - return None + return 0 + if kind == "end" and stream_key not in self.streams: + return 2 + if kind == "end": + if self._tail_is_end(stream_key) or (int(legacy_scan_enabled) and self._has_end_marker(stream_key)): + return 1 + elif self._tail_is_end(stream_key): + return 1 event_id = await self.xadd( stream_key, {"kind": kind, "event": event, "data": data}, @@ -166,9 +187,14 @@ class _FakeRedis: return 1 if numkeys == 3: active_key, tracking_key, stream_key = keys - token, maxlen, stream_ttl, kind, event, data = values + token, maxlen, stream_ttl, kind, event, data, legacy_scan_enabled = values if self.values.get(active_key) != token or self.values.get(tracking_key) != token: - return None + return 0 + if kind == "end": + if self._tail_is_end(stream_key) or (int(legacy_scan_enabled) and self._has_end_marker(stream_key)): + return 1 + elif self._tail_is_end(stream_key): + return 1 event_id = await self.xadd( stream_key, {"kind": kind, "event": event, "data": data}, @@ -227,6 +253,24 @@ class _DelayedBlockingReadRedis: return response +class _LostAckRedis: + """Delegate to Redis but drop one successful EVAL response.""" + + def __init__(self, redis) -> None: + self._delegate = redis + self.drop_next_successful_eval = False + + def __getattr__(self, name): + return getattr(self._delegate, name) + + async def eval(self, script, numkeys, *args): + result = await self._delegate.eval(script, numkeys, *args) + if self.drop_next_successful_eval and result: + self.drop_next_successful_eval = False + raise TimeoutError("simulated lost Redis EVAL acknowledgement") + return result + + class _FakeRedisPipeline: def __init__(self, redis: _FakeRedis) -> None: self.redis = redis @@ -704,6 +748,148 @@ async def test_redis_publish_subscribe(redis_bridge: RedisStreamBridge): assert received[-1] is END_SENTINEL +@pytest.mark.anyio +async def test_redis_publish_end_is_idempotent(redis_bridge: RedisStreamBridge): + """Concurrent and repeated END publication must append at most one marker.""" + fake = redis_bridge._redis + run_id = "redis-run-idempotent-end" + key = redis_bridge._stream_key(run_id) + + await redis_bridge.start_producer_lease(run_id) + try: + await redis_bridge.publish(run_id, "metadata", {"run_id": run_id}) + await asyncio.gather(*(redis_bridge.publish_end(run_id) for _ in range(8))) + await redis_bridge.publish_end(run_id) + + entries = fake.streams[key] + assert [fields["kind"] for _event_id, fields in entries] == ["event", "end"] + + received = [entry async for entry in redis_bridge.subscribe(run_id, heartbeat_interval=1.0)] + assert [entry.event for entry in received[:-1]] == ["metadata"] + assert received[-1] is END_SENTINEL + finally: + await redis_bridge.finish_producer_lease(run_id) + + +@pytest.mark.anyio +async def test_redis_publish_end_recognizes_legacy_end_before_late_events(redis_bridge: RedisStreamBridge): + """An END written by the old buggy writer still prevents a duplicate END.""" + fake = redis_bridge._redis + run_id = "redis-run-legacy-end-not-tail" + key = redis_bridge._stream_key(run_id) + fake.streams[key] = [ + ("1-0", {"kind": "event", "event": "metadata", "data": "{}"}), + ("2-0", {"kind": "end"}), + ("3-0", {"kind": "event", "event": "late", "data": "{}"}), + ] + + await redis_bridge.publish_end(run_id) + + assert [fields["kind"] for _event_id, fields in fake.streams[key]].count("end") == 1 + + +@pytest.mark.anyio +@pytest.mark.parametrize("owned", [False, True], ids=["recovery", "owner"]) +async def test_redis_legacy_midstream_end_allows_data_but_deduplicates_end(owned: bool): + """Data checks only the tail; END retries still inspect the retained window.""" + fake = _FakeRedis() + bridge = RedisStreamBridge(redis_url="redis://fake", queue_maxsize=8, client=fake) + run_id = "redis-run-legacy-data-after-end" + key = bridge._stream_key(run_id) + if owned: + await bridge.start_producer_lease(run_id) + try: + await fake.xadd(key, {"kind": "end"}) + await fake.xadd(key, {"kind": "event", "event": "legacy-late", "data": "{}"}) + before = list(fake.streams[key]) + + await bridge.publish(run_id, "current-data", {"step": 2}) + + entries = list(fake.streams[key]) + assert entries[:2] == before + assert len(entries) == 3 + assert entries[-1][1]["event"] == "current-data" + await bridge.publish_end(run_id) + assert fake.streams[key] == entries + finally: + if owned: + await bridge.finish_producer_lease(run_id) + + +@pytest.mark.anyio +async def test_redis_owned_end_recreates_missing_buffer_with_live_lease(): + """Only unowned recovery END is a no-op on a missing buffer.""" + fake = _FakeRedis() + bridge = RedisStreamBridge(redis_url="redis://fake", client=fake) + run_id = "redis-run-owned-missing-buffer" + key = bridge._stream_key(run_id) + await bridge.start_producer_lease(run_id) + try: + await bridge.publish(run_id, "metadata", {}) + await fake.delete(key) + + await bridge.publish_end(run_id) + + assert [fields["kind"] for _event_id, fields in fake.streams[key]] == ["end"] + finally: + await bridge.finish_producer_lease(run_id) + + +@pytest.mark.anyio +async def test_redis_publish_after_end_is_rejected_without_appending(redis_bridge: RedisStreamBridge): + """Once END is retained, later event publication cannot corrupt terminal order.""" + fake = redis_bridge._redis + run_id = "redis-run-publish-after-end" + key = redis_bridge._stream_key(run_id) + + await redis_bridge.publish(run_id, "metadata", {"run_id": run_id}) + await redis_bridge.publish_end(run_id) + before = list(fake.streams[key]) + + with pytest.raises(RuntimeError, match="already ended"): + await redis_bridge.publish(run_id, "late", {"step": 2}) + + assert fake.streams[key] == before + + +@pytest.mark.anyio +async def test_redis_event_end_race_never_appends_after_end(redis_bridge: RedisStreamBridge): + """The atomic terminal check keeps END last even when an event races it.""" + fake = redis_bridge._redis + run_id = "redis-run-event-end-race" + key = redis_bridge._stream_key(run_id) + + await redis_bridge.publish(run_id, "metadata", {"run_id": run_id}) + event_result, end_result = await asyncio.gather( + redis_bridge.publish(run_id, "racing-event", {"step": 2}), + redis_bridge.publish_end(run_id), + return_exceptions=True, + ) + + assert end_result is None + assert event_result is None or "already ended" in str(event_result) + kinds = [fields["kind"] for _event_id, fields in fake.streams[key]] + assert kinds.count("end") == 1 + assert kinds[-1] == "end" + + +@pytest.mark.anyio +async def test_redis_recovery_end_after_cleanup_does_not_resurrect_stream(redis_bridge: RedisStreamBridge): + """A delayed recovery END is a no-op when cleanup already removed the stream.""" + fake = redis_bridge._redis + run_id = "redis-run-cleaned-end-retry" + key = redis_bridge._stream_key(run_id) + + await redis_bridge.publish(run_id, "metadata", {"run_id": run_id}) + await redis_bridge.publish_end(run_id) + await redis_bridge.cleanup(run_id) + assert key not in fake.streams + + await redis_bridge.publish_end(run_id) + + assert key not in fake.streams + + @pytest.mark.anyio async def test_redis_replays_after_last_event_id(redis_bridge: RedisStreamBridge): """Redis XREAD should resume after Last-Event-ID.""" @@ -1098,6 +1284,30 @@ async def test_redis_finish_does_not_delete_a_successor_token(): assert fake.values[active_key] == "successor-token" +@pytest.mark.anyio +async def test_redis_lost_owner_does_not_get_idempotent_end_success(): + """Lease validation precedes retained-END detection for a stale owner.""" + fake = _FakeRedis() + bridge = RedisStreamBridge(redis_url="redis://fake", client=fake) + run_id = "redis-run-lost-owner-existing-end" + active_key = bridge._producer_active_key(run_id) + key = bridge._stream_key(run_id) + replacement_owner = asyncio.create_task(asyncio.Event().wait()) + + await bridge.start_producer_lease(run_id) + bridge.transfer_producer_lease(run_id, replacement_owner) + await bridge.publish_end(run_id) + fake.values[active_key] = "successor-token" + try: + with pytest.raises(RuntimeError, match="producer lease was lost"): + await bridge.publish_end(run_id) + assert [fields["kind"] for _event_id, fields in fake.streams[key]].count("end") == 1 + finally: + await bridge.finish_producer_lease(run_id) + replacement_owner.cancel() + await asyncio.gather(replacement_owner, return_exceptions=True) + + @pytest.mark.anyio async def test_redis_publish_is_atomically_fenced_by_producer_token(): fake = _FakeRedis() @@ -2103,6 +2313,7 @@ async def test_make_stream_bridge_passes_redis_options(monkeypatch): max_connections=50, stream_ttl_seconds=42, producer_lease_recovery_enabled=True, + legacy_end_scan_enabled=False, ) ) try: @@ -2111,6 +2322,7 @@ async def test_make_stream_bridge_passes_redis_options(monkeypatch): assert bridge.heartbeat_interval == 2.5 assert bridge._stream_ttl_seconds == 42 assert bridge.producer_lease_recovery_enabled is True + assert bridge._legacy_end_scan_enabled is False assert captured["max_connections"] == 50 assert captured["decode_responses"] is True finally: @@ -2188,6 +2400,189 @@ async def test_redis_integration_publish_subscribe_and_id_format(real_redis_brid assert re.match(r"^\d+-\d+$", received[0].id), received[0].id +@pytest.mark.integration +@requires_redis +@pytest.mark.anyio +async def test_redis_integration_end_is_idempotent_across_independent_clients(): + """Concurrent END writers must serialize through Redis, not local process state.""" + from redis.asyncio import Redis + + clients = [ + Redis.from_url(REDIS_TEST_URL, decode_responses=True), + Redis.from_url(REDIS_TEST_URL, decode_responses=True), + ] + key_prefix = f"deerflow:test:{uuid.uuid4().hex}" + bridges = [ + RedisStreamBridge( + redis_url=REDIS_TEST_URL, + queue_maxsize=2, + key_prefix=key_prefix, + client=client, + ) + for client in clients + ] + run_id = "integ-end-idempotent" + key = bridges[0]._stream_key(run_id) + try: + await bridges[0].publish(run_id, "metadata", {"run_id": run_id}) + await asyncio.gather(*(bridge.publish_end(run_id) for bridge in bridges)) + await asyncio.gather(*(bridge.publish_end(run_id) for bridge in bridges)) + + entries = await clients[0].xrange(key) + assert [fields["kind"] for _event_id, fields in entries] == ["event", "end"] + + first_event_id = entries[0][0] + replayed = [ + entry + async for entry in bridges[1].subscribe( + run_id, + last_event_id=first_event_id, + heartbeat_interval=1.0, + ) + ] + assert replayed == [END_SENTINEL] + finally: + async for existing_key in clients[0].scan_iter(f"{key_prefix}:*"): + await clients[0].delete(existing_key) + await asyncio.gather(*(client.aclose() for client in clients)) + + +@pytest.mark.integration +@requires_redis +@pytest.mark.anyio +async def test_redis_integration_recovery_end_cannot_race_active_owner(real_redis_bridge): + """Recovery may append END only after the valid producer's lease is released.""" + from redis.asyncio import Redis + + recovery_client = Redis.from_url(REDIS_TEST_URL, decode_responses=True) + recovery = RedisStreamBridge( + redis_url=REDIS_TEST_URL, + queue_maxsize=2, + key_prefix=real_redis_bridge._key_prefix, + client=recovery_client, + ) + run_id = "integ-end-owner-recovery-race" + key = real_redis_bridge._stream_key(run_id) + try: + await real_redis_bridge.publish(run_id, "metadata", {"run_id": run_id}) + await real_redis_bridge.start_producer_lease(run_id) + + owner_result, recovery_result = await asyncio.gather( + real_redis_bridge.publish_end(run_id), + recovery.publish_end(run_id), + return_exceptions=True, + ) + assert owner_result is None + assert isinstance(recovery_result, RuntimeError) + assert "active producer" in str(recovery_result) + + entries = await recovery_client.xrange(key) + assert [fields["kind"] for _event_id, fields in entries] == ["event", "end"] + + await real_redis_bridge.finish_producer_lease(run_id) + await recovery.publish_end(run_id) + entries = await recovery_client.xrange(key) + assert [fields["kind"] for _event_id, fields in entries] == ["event", "end"] + finally: + await real_redis_bridge.finish_producer_lease(run_id) + await recovery_client.aclose() + + +@pytest.mark.integration +@requires_redis +@pytest.mark.anyio +async def test_redis_integration_end_retry_after_lost_ack_is_idempotent(real_redis_bridge): + """A retry confirms the retained END without appending a duplicate.""" + run_id = "integ-end-lost-ack" + key = real_redis_bridge._stream_key(run_id) + + await real_redis_bridge.start_producer_lease(run_id) + try: + await real_redis_bridge.publish(run_id, "metadata", {"run_id": run_id}) + lossy = _LostAckRedis(real_redis_bridge._redis) + real_redis_bridge._redis = lossy + lossy.drop_next_successful_eval = True + + with pytest.raises(TimeoutError, match="lost Redis EVAL acknowledgement"): + await real_redis_bridge.publish_end(run_id) + + await real_redis_bridge.publish_end(run_id) + + entries = await lossy.xrange(key) + assert [fields["kind"] for _event_id, fields in entries] == ["event", "end"] + finally: + await real_redis_bridge.finish_producer_lease(run_id) + + +@pytest.mark.integration +@requires_redis +@pytest.mark.anyio +async def test_redis_integration_cleanup_prevents_late_end_from_recreating_stream(real_redis_bridge): + """An END retry after cleanup is acknowledged as a no-op, not a resurrection.""" + run_id = "integ-end-after-cleanup" + key = real_redis_bridge._stream_key(run_id) + + await real_redis_bridge.publish(run_id, "metadata", {"run_id": run_id}) + await real_redis_bridge.publish_end(run_id) + await real_redis_bridge.cleanup(run_id) + assert await real_redis_bridge._redis.exists(key) == 0 + + await real_redis_bridge.publish_end(run_id) + + assert await real_redis_bridge._redis.exists(key) == 0 + + +@pytest.mark.integration +@requires_redis +@pytest.mark.anyio +@pytest.mark.parametrize("owned", [False, True], ids=["recovery", "owner"]) +async def test_redis_integration_legacy_midstream_end_allows_data_but_deduplicates_end(real_redis_bridge, owned: bool): + """Pin the fake's legacy-stream decisions against the real Lua scripts.""" + client = real_redis_bridge._redis + bridge = RedisStreamBridge(redis_url=REDIS_TEST_URL, queue_maxsize=8, key_prefix=real_redis_bridge._key_prefix, client=client) + run_id = "integ-legacy-data-after-end" + key = bridge._stream_key(run_id) + if owned: + await bridge.start_producer_lease(run_id) + try: + await client.xadd(key, {"kind": "end"}) + await client.xadd(key, {"kind": "event", "event": "legacy-late", "data": "{}"}) + before = await client.xrange(key) + + await bridge.publish(run_id, "current-data", {"step": 2}) + + entries = await client.xrange(key) + assert entries[:2] == before + assert len(entries) == 3 + assert entries[-1][1]["event"] == "current-data" + await bridge.publish_end(run_id) + assert await client.xrange(key) == entries + finally: + if owned: + await bridge.finish_producer_lease(run_id) + + +@pytest.mark.integration +@requires_redis +@pytest.mark.anyio +async def test_redis_integration_owned_end_recreates_missing_buffer_with_live_lease(real_redis_bridge): + """A live owner retains the existing XADD behavior after buffer removal.""" + bridge = real_redis_bridge + client = bridge._redis + run_id = "integ-owned-missing-buffer" + key = bridge._stream_key(run_id) + await bridge.start_producer_lease(run_id) + try: + await bridge.publish(run_id, "metadata", {}) + await client.delete(key) + + await bridge.publish_end(run_id) + + assert [fields["kind"] for _event_id, fields in await client.xrange(key)] == ["end"] + finally: + await bridge.finish_producer_lease(run_id) + + @pytest.mark.integration @requires_redis @pytest.mark.anyio @@ -2385,3 +2780,80 @@ async def test_redis_integration_stream_ttl_reclaims_key(): async for existing_key in client.scan_iter(f"{key_prefix}:*"): await client.delete(existing_key) await client.aclose() + + +@pytest.mark.parametrize("enabled", [True, False]) +def test_stream_bridge_config_legacy_end_scan_switch(enabled): + assert StreamBridgeConfig().legacy_end_scan_enabled is True + assert StreamBridgeConfig(legacy_end_scan_enabled=enabled).legacy_end_scan_enabled is enabled + + +@pytest.mark.anyio +@pytest.mark.parametrize("owned", [False, True], ids=["recovery", "owner"]) +@pytest.mark.parametrize("enabled", [True, False], ids=["compatibility-scan", "tail-only"]) +async def test_redis_end_scan_switch_and_tail_fast_path(owned, enabled): + fake = _FakeRedis() + fake._has_end_marker = Mock(wraps=fake._has_end_marker) + bridge = RedisStreamBridge(redis_url="redis://fake", queue_maxsize=8, legacy_end_scan_enabled=enabled, client=fake) + run_id = "scan-switch" + if owned: + await bridge.start_producer_lease(run_id) + try: + await bridge.publish(run_id, "values", {"content": "x" * 4096}) + assert fake._has_end_marker.call_count == 0 + await bridge.publish_end(run_id) + assert fake._has_end_marker.call_count == int(enabled) + await bridge.publish_end(run_id) + with pytest.raises(RuntimeError, match="already ended"): + await bridge.publish(run_id, "late", {}) + assert fake._has_end_marker.call_count == int(enabled) + assert [fields["kind"] for _, fields in fake.streams[bridge._stream_key(run_id)]] == ["event", "end"] + finally: + if owned: + await bridge.finish_producer_lease(run_id) + await bridge.close() + + +@pytest.mark.integration +@requires_redis +@pytest.mark.anyio +@pytest.mark.parametrize("owned", [False, True], ids=["recovery", "owner"]) +@pytest.mark.parametrize("enabled", [True, False], ids=["compatibility-scan", "tail-only"]) +async def test_redis_integration_end_scan_switch_with_xrange_denied(real_redis_bridge, owned, enabled): + """Deny XRANGE so real Lua proves that tail-only END never scans history.""" + from redis.asyncio import ConnectionPool, Redis + from redis.exceptions import ResponseError + + admin = real_redis_bridge._redis + server = await admin.info("server") + if int(server["redis_version"].split(".", 1)[0]) < 6: + pytest.skip("This scan-path probe uses Redis 6+ ACLs") + username = "end-scan-" + uuid.uuid4().hex + prefix = real_redis_bridge._key_prefix + await admin.acl_setuser(username, enabled=True, reset=True, passwords=["+fixture-password"], keys=[f"{prefix}:*"], categories=["+all"], commands=["-xrange"]) + pool = ConnectionPool.from_url(REDIS_TEST_URL, decode_responses=True) + pool.connection_kwargs.update(username=username, password="fixture-password") + client = Redis(connection_pool=pool) + bridge = RedisStreamBridge(redis_url=REDIS_TEST_URL, queue_maxsize=8, key_prefix=prefix, legacy_end_scan_enabled=enabled, client=client) + run_id = "scan-switch-acl" + try: + if owned: + await bridge.start_producer_lease(run_id) + await bridge.publish(run_id, "values", {"content": "x" * 4096}) + if enabled: + with pytest.raises(ResponseError, match="(?i)acl|permission|user executing|allowed"): + await bridge.publish_end(run_id) + assert await admin.xlen(bridge._stream_key(run_id)) == 1 + else: + await bridge.publish_end(run_id) + await bridge.publish_end(run_id) + with pytest.raises(RuntimeError, match="already ended"): + await bridge.publish(run_id, "late", {}) + assert [fields["kind"] for _, fields in await admin.xrange(bridge._stream_key(run_id))] == ["event", "end"] + finally: + try: + await bridge.close() + finally: + await client.aclose() + await pool.disconnect() + await admin.acl_deluser(username) diff --git a/config.example.yaml b/config.example.yaml index 88b3b324c0..86e882d6fe 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -20,7 +20,7 @@ # ============================================================================ # Bump this number when the config schema changes. # Run `make config-upgrade` to merge new fields into your local config.yaml. -config_version: 62 +config_version: 63 # Optional operator-owned literal prompt extensions (see CONFIGURATION.md): # lead_prompt_overlay: @@ -2990,6 +2990,13 @@ run_ownership: # stream_ttl_seconds: 86400 # rolling TTL for retained stream buffers. # # Refreshed on each publish/publish_end; set 0 # # to disable. This is not a run timeout. +# legacy_end_scan_enabled: true # restart-required Redis compatibility scan. +# # First END scans all retained payloads while true. +# # Set false only after all old writers stop and +# # every legacy buffer expires or is removed. +# # With TTL disabled, remove old buffers explicitly. +# # Restart all Gateways together; false uses only +# # the tail for O(1) END detection. # producer_lease_recovery_enabled: false # restart-required activation gate. # # First roll this binary to every Gateway and # # drain pre-upgrade runs; then enable on every diff --git a/deploy/helm/deer-flow/README.md b/deploy/helm/deer-flow/README.md index 93bd7a49fc..840cf9488e 100644 --- a/deploy/helm/deer-flow/README.md +++ b/deploy/helm/deer-flow/README.md @@ -135,7 +135,7 @@ they resolve from the `secrets` map): ```yaml config: | - config_version: 62 + config_version: 63 models: - name: gpt-4 use: langchain_openai:ChatOpenAI diff --git a/deploy/helm/deer-flow/values.yaml b/deploy/helm/deer-flow/values.yaml index c4e0451f6d..73c80a13c3 100644 --- a/deploy/helm/deer-flow/values.yaml +++ b/deploy/helm/deer-flow/values.yaml @@ -367,7 +367,7 @@ ingress: # -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never # inline literal secret values here. The default enables provisioner sandbox. config: | - config_version: 62 + config_version: 63 log_level: info recursion_limit: 100 max_recursion_limit: 1000