diff --git a/README.md b/README.md index 635e53e15..833e02c38 100644 --- a/README.md +++ b/README.md @@ -262,7 +262,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 holds run state (RunManager and the stream bridge) in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Raising the worker count without a shared cross-worker stream bridge — which is not yet available — breaks run cancellation, SSE reconnects, request de-duplication, and IM channels, because nginx uses no sticky sessions and each worker keeps its own run state. Scale a single worker up with more CPU/RAM (or move the database and sandbox onto dedicated tiers) instead of 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. 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 1dd796801..b5bddca79 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -311,6 +311,7 @@ CORS is same-origin by default when requests enter through nginx on port 2026. S - `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. - 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/app/gateway/deps.py b/backend/app/gateway/deps.py index eefcffc37..28a8a01f0 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -71,6 +71,44 @@ async def _drain_inflight_runs(run_manager: RunManager) -> None: logger.exception("Failed to drain in-flight runs during shutdown") +async def _publish_recovered_run_stream_end( + bridge: StreamBridge, + recovered_runs: list[RunRecord], + *, + cleanup_delay: float = 60.0, +) -> None: + """Terminate retained streams for runs recovered as orphaned at startup.""" + for record in recovered_runs: + stream_exists = getattr(bridge, "stream_exists", None) + if stream_exists is not None: + try: + if not await stream_exists(record.run_id): + logger.debug("Skipping recovered stream end for %s: stream already expired", record.run_id) + continue + except Exception: + logger.debug("Failed to check recovered stream existence for %s", record.run_id, exc_info=True) + try: + await bridge.publish_end(record.run_id) + except Exception: + logger.warning( + "Failed to publish recovered run stream end for %s", + record.run_id, + exc_info=True, + ) + continue + task = asyncio.create_task(bridge.cleanup(record.run_id, delay=cleanup_delay)) + task.add_done_callback(lambda task, run_id=record.run_id: _log_recovered_stream_cleanup_result(task, run_id)) + + +def _log_recovered_stream_cleanup_result(task: asyncio.Task[None], run_id: str) -> None: + if task.cancelled(): + return + try: + task.result() + except Exception: + logger.warning("Failed to clean up recovered run stream for %s", run_id, exc_info=True) + + if TYPE_CHECKING: from app.gateway.auth.local_provider import LocalAuthProvider from app.gateway.auth.repositories.sqlite import SQLiteUserRepository @@ -220,6 +258,9 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen error="Gateway restarted before this run reached a durable final state.", before=now_iso(), ) + sb_config = getattr(config, "stream_bridge", None) + cleanup_delay = getattr(sb_config, "recovered_stream_cleanup_delay_seconds", 60.0) if sb_config else 60.0 + await _publish_recovered_run_stream_end(app.state.stream_bridge, recovered_runs, cleanup_delay=cleanup_delay) await _mark_latest_recovered_threads_error(app.state.run_manager, app.state.thread_store, recovered_runs) try: diff --git a/backend/docs/STREAMING.md b/backend/docs/STREAMING.md index b28e6da70..9c5777941 100644 --- a/backend/docs/STREAMING.md +++ b/backend/docs/STREAMING.md @@ -137,7 +137,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。 +- `runtime/stream_bridge` — 抽象 Queue。`publish/subscribe` 解耦生产者和消费者，支持 `Last-Event-ID` 重连、心跳、多订阅者 fan-out。Redis backend 会在每次 `publish()` / `publish_end()` 刷新 retained stream key TTL；启动恢复将 orphan run 标记为 error 后也会发布 `END_SENTINEL`，避免重连的 SSE 客户端只收到心跳。注意：TTL 是 Redis 内存安全网（防止 key 泄漏），不是 subscriber 终止机制——如果 worker 和 gateway 同时挂掉，已连接的 SSE 客户端在 TTL 过期后仍无法收到 END 信号，需要依赖客户端侧超时。完整的跨 pod subscriber 终止需要 worker 存活检测（liveness），当前版本不包含此功能。 - `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 7d0f05885..6ba9535d6 100644 --- a/backend/packages/harness/deerflow/config/stream_bridge_config.py +++ b/backend/packages/harness/deerflow/config/stream_bridge_config.py @@ -32,6 +32,20 @@ class StreamBridgeConfig(BaseModel): "concurrent SSE clients. Only applies to the redis bridge." ), ) + stream_ttl_seconds: int = Field( + default=86400, + ge=0, + description=( + "Rolling Redis stream key TTL in seconds. The redis bridge refreshes this TTL after " + "each publish and publish_end so retained SSE replay buffers are eventually reclaimed " + "even if cleanup never runs. Set to 0 to disable. Only applies to the redis bridge." + ), + ) + recovered_stream_cleanup_delay_seconds: float = Field( + default=60.0, + ge=0, + description=("Seconds to wait after publishing an END marker for a recovered orphaned run before deleting the stream key. Gives reconnecting SSE clients time to drain the end signal. Only applies to the redis bridge."), + ) # Global configuration instance — None means no stream bridge is configured 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 a70edb9b9..e7538b01d 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/async_provider.py +++ b/backend/packages/harness/deerflow/runtime/stream_bridge/async_provider.py @@ -74,11 +74,13 @@ async def make_stream_bridge(app_config: AppConfig | None = None) -> AsyncIterat redis_url=redis_url, queue_maxsize=config.queue_maxsize, max_connections=config.max_connections, + stream_ttl_seconds=config.stream_ttl_seconds, ) logger.info( - "Stream bridge initialised: redis (queue_maxsize=%d, max_connections=%s)", + "Stream bridge initialised: redis (queue_maxsize=%d, max_connections=%s, stream_ttl_seconds=%d)", config.queue_maxsize, config.max_connections, + config.stream_ttl_seconds, ) try: yield bridge diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py b/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py index de07af599..8581cefff 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py +++ b/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py @@ -63,11 +63,16 @@ class RedisStreamBridge(StreamBridge): queue_maxsize: int = 256, key_prefix: str = "deerflow:stream_bridge", max_connections: int | None = None, + stream_ttl_seconds: int | None = 86400, client: Redis | None = None, ) -> None: self._redis_url = redis_url self._maxsize = max(1, queue_maxsize) 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 + else: + self._stream_ttl_seconds = None # Each live SSE subscriber holds one pooled connection blocked in # ``XREAD ... BLOCK`` for up to ``heartbeat_interval``. ``max_connections`` # caps that pool; ``None`` keeps redis-py's effectively-unbounded default. @@ -77,6 +82,26 @@ class RedisStreamBridge(StreamBridge): def _stream_key(self, run_id: str) -> str: return f"{self._key_prefix}:{run_id}" + async def _xadd_retained(self, key: str, fields: dict[str, str], *, maxlen: int) -> None: + if self._stream_ttl_seconds is None: + await self._redis.xadd( + key, + fields, + maxlen=maxlen, + approximate=False, + ) + return + + async with self._redis.pipeline(transaction=True) as pipe: + pipe.xadd( + key, + fields, + maxlen=maxlen, + approximate=False, + ) + pipe.expire(key, self._stream_ttl_seconds) + await pipe.execute() + @staticmethod def _decode(value: Any) -> str: if isinstance(value, bytes): @@ -113,24 +138,24 @@ class RedisStreamBridge(StreamBridge): ) async def publish(self, run_id: str, event: str, data: Any) -> None: - await self._redis.xadd( - self._stream_key(run_id), + key = self._stream_key(run_id) + await self._xadd_retained( + key, { "kind": _KIND_EVENT, "event": event, "data": self._encode_data(data), }, maxlen=self._maxsize, - approximate=False, ) async def publish_end(self, run_id: str) -> None: # Keep the configured number of data events plus the internal end marker. - await self._redis.xadd( - self._stream_key(run_id), + key = self._stream_key(run_id) + await self._xadd_retained( + key, {"kind": _KIND_END}, maxlen=self._maxsize + 1, - approximate=False, ) async def stream_exists(self, run_id: str) -> bool: diff --git a/backend/tests/test_gateway_run_recovery.py b/backend/tests/test_gateway_run_recovery.py index 69b7c3d64..e5f300726 100644 --- a/backend/tests/test_gateway_run_recovery.py +++ b/backend/tests/test_gateway_run_recovery.py @@ -5,6 +5,7 @@ from __future__ import annotations from contextlib import asynccontextmanager from types import SimpleNamespace +import anyio import pytest from fastapi import FastAPI @@ -57,6 +58,36 @@ class _FakeThreadStore: self.status_updates.append((thread_id, status, user_id)) +class _FakeStreamBridge: + def __init__(self, *, existing_streams: set[str] | None = None) -> None: + self.publish_end_calls: list[str] = [] + self.cleanup_calls: list[tuple[str, float]] = [] + self._existing_streams: set[str] = existing_streams if existing_streams is not None else set() + + async def stream_exists(self, run_id: str) -> bool: + return run_id in self._existing_streams + + async def publish_end(self, run_id: str) -> None: + self.publish_end_calls.append(run_id) + + async def cleanup(self, run_id: str, *, delay: float = 0) -> None: + self.cleanup_calls.append((run_id, delay)) + + +@pytest.mark.anyio +async def test_recovered_run_stream_end_skips_expired_stream(): + """Startup recovery should not recreate an already-expired retained stream.""" + stream_bridge = _FakeStreamBridge(existing_streams=set()) + + await gateway_deps._publish_recovered_run_stream_end( + stream_bridge, + [SimpleNamespace(run_id="expired-run", thread_id="thread-1")], + ) + + assert stream_bridge.publish_end_calls == [] + assert stream_bridge.cleanup_calls == [] + + @pytest.mark.anyio async def test_sqlite_runtime_reconciles_orphaned_runs_on_startup(monkeypatch): """SQLite startup should recover stale active runs before serving requests.""" @@ -64,8 +95,10 @@ async def test_sqlite_runtime_reconciles_orphaned_runs_on_startup(monkeypatch): config = SimpleNamespace( database=SimpleNamespace(backend="sqlite"), run_events=SimpleNamespace(backend="memory"), + stream_bridge=SimpleNamespace(recovered_stream_cleanup_delay_seconds=60.0), ) thread_store = _FakeThreadStore() + stream_bridge = _FakeStreamBridge(existing_streams={"run-1"}) _FakeRunManager.instances.clear() _FakeRunManager.recovered_runs = [SimpleNamespace(run_id="run-1", thread_id="thread-1")] _FakeRunManager.latest_by_thread = {} @@ -79,7 +112,7 @@ async def test_sqlite_runtime_reconciles_orphaned_runs_on_startup(monkeypatch): monkeypatch.setattr(engine_module, "init_engine_from_config", fake_init_engine_from_config) monkeypatch.setattr(engine_module, "get_session_factory", lambda: None) monkeypatch.setattr(engine_module, "close_engine", fake_close_engine) - monkeypatch.setattr(runtime_module, "make_stream_bridge", lambda _config: _fake_context(object())) + monkeypatch.setattr(runtime_module, "make_stream_bridge", lambda _config: _fake_context(stream_bridge)) monkeypatch.setattr(checkpointer_module, "make_checkpointer", lambda _config: _fake_context(object())) monkeypatch.setattr(runtime_module, "make_store", lambda _config: _fake_context(object())) monkeypatch.setattr(thread_meta_module, "make_thread_store", lambda _sf, _store: thread_store) @@ -88,12 +121,15 @@ async def test_sqlite_runtime_reconciles_orphaned_runs_on_startup(monkeypatch): async with gateway_deps.langgraph_runtime(app, config): pass + await anyio.sleep(0) assert len(_FakeRunManager.instances) == 1 assert _FakeRunManager.instances[0].reconcile_calls assert _FakeRunManager.instances[0].reconcile_calls[0]["error"] assert _FakeRunManager.instances[0].list_by_thread_calls == [{"thread_id": "thread-1", "user_id": None, "limit": 1}] assert thread_store.status_updates == [("thread-1", "error", None)] + assert stream_bridge.publish_end_calls == ["run-1"] + assert stream_bridge.cleanup_calls == [("run-1", 60.0)] @pytest.mark.anyio @@ -103,8 +139,10 @@ async def test_sqlite_runtime_does_not_mark_thread_error_when_newer_run_is_succe config = SimpleNamespace( database=SimpleNamespace(backend="sqlite"), run_events=SimpleNamespace(backend="memory"), + stream_bridge=SimpleNamespace(recovered_stream_cleanup_delay_seconds=60.0), ) thread_store = _FakeThreadStore() + stream_bridge = _FakeStreamBridge(existing_streams={"old-running"}) _FakeRunManager.instances.clear() _FakeRunManager.recovered_runs = [SimpleNamespace(run_id="old-running", thread_id="thread-1")] _FakeRunManager.latest_by_thread = {"thread-1": [SimpleNamespace(run_id="newer-success", thread_id="thread-1", status="success")]} @@ -118,7 +156,7 @@ async def test_sqlite_runtime_does_not_mark_thread_error_when_newer_run_is_succe monkeypatch.setattr(engine_module, "init_engine_from_config", fake_init_engine_from_config) monkeypatch.setattr(engine_module, "get_session_factory", lambda: None) monkeypatch.setattr(engine_module, "close_engine", fake_close_engine) - monkeypatch.setattr(runtime_module, "make_stream_bridge", lambda _config: _fake_context(object())) + monkeypatch.setattr(runtime_module, "make_stream_bridge", lambda _config: _fake_context(stream_bridge)) monkeypatch.setattr(checkpointer_module, "make_checkpointer", lambda _config: _fake_context(object())) monkeypatch.setattr(runtime_module, "make_store", lambda _config: _fake_context(object())) monkeypatch.setattr(thread_meta_module, "make_thread_store", lambda _sf, _store: thread_store) @@ -127,7 +165,10 @@ async def test_sqlite_runtime_does_not_mark_thread_error_when_newer_run_is_succe async with gateway_deps.langgraph_runtime(app, config): pass + await anyio.sleep(0) assert len(_FakeRunManager.instances) == 1 assert _FakeRunManager.instances[0].list_by_thread_calls == [{"thread_id": "thread-1", "user_id": None, "limit": 1}] assert thread_store.status_updates == [] + assert stream_bridge.publish_end_calls == ["old-running"] + assert stream_bridge.cleanup_calls == [("old-running", 60.0)] diff --git a/backend/tests/test_stream_bridge.py b/backend/tests/test_stream_bridge.py index b930a5bc4..781593c51 100644 --- a/backend/tests/test_stream_bridge.py +++ b/backend/tests/test_stream_bridge.py @@ -30,6 +30,7 @@ class _FakeRedis: self.conditions = defaultdict(asyncio.Condition) self.counters = defaultdict(int) self.deleted = [] + self.expirations = [] self.closed = False async def xadd(self, name, fields, maxlen=None, approximate=True): @@ -65,10 +66,48 @@ class _FakeRedis: async def exists(self, name): return 1 if name in self.streams else 0 + async def expire(self, name, seconds): + self.expirations.append((name, seconds)) + return True + + def pipeline(self, *, transaction=True): + return _FakeRedisPipeline(self) + async def aclose(self): self.closed = True +class _FakeRedisPipeline: + def __init__(self, redis: _FakeRedis) -> None: + self.redis = redis + self.ops = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return None + + def xadd(self, name, fields, maxlen=None, approximate=True): + self.ops.append(("xadd", name, fields, maxlen, approximate)) + return self + + def expire(self, name, seconds): + self.ops.append(("expire", name, seconds)) + return self + + async def execute(self): + results = [] + for op in self.ops: + if op[0] == "xadd": + _, name, fields, maxlen, approximate = op + results.append(await self.redis.xadd(name, fields, maxlen=maxlen, approximate=approximate)) + elif op[0] == "expire": + _, name, seconds = op + results.append(await self.redis.expire(name, seconds)) + return results + + # --------------------------------------------------------------------------- # Unit tests for MemoryStreamBridge # --------------------------------------------------------------------------- @@ -487,6 +526,43 @@ async def test_redis_cleanup_deletes_stream(redis_bridge: RedisStreamBridge): assert fake.deleted == ["deerflow:stream_bridge:redis-run-cleanup"] +@pytest.mark.anyio +async def test_redis_publish_refreshes_stream_ttl(): + """Redis stream TTL should be rolling on publish and publish_end.""" + fake = _FakeRedis() + bridge = RedisStreamBridge( + redis_url="redis://fake", + queue_maxsize=2, + stream_ttl_seconds=42, + client=fake, + ) + run_id = "redis-run-ttl" + key = "deerflow:stream_bridge:redis-run-ttl" + + await bridge.publish(run_id, "event-1", {"n": 1}) + await bridge.publish(run_id, "event-2", {"n": 2}) + await bridge.publish_end(run_id) + + assert fake.expirations == [(key, 42), (key, 42), (key, 42)] + + +@pytest.mark.anyio +async def test_redis_stream_ttl_can_be_disabled(): + """A zero TTL disables the Redis leak safety net for installations that need it.""" + fake = _FakeRedis() + bridge = RedisStreamBridge( + redis_url="redis://fake", + queue_maxsize=2, + stream_ttl_seconds=0, + client=fake, + ) + + await bridge.publish("redis-run-no-ttl", "event", {}) + await bridge.publish_end("redis-run-no-ttl") + + assert fake.expirations == [] + + @pytest.mark.anyio async def test_redis_subscribe_waits_for_first_publish(redis_bridge: RedisStreamBridge): """A subscriber that starts before the first XADD must not receive END.""" @@ -673,8 +749,8 @@ async def test_make_stream_bridge_uses_docker_redis_env(monkeypatch): @pytest.mark.anyio -async def test_make_stream_bridge_passes_max_connections(monkeypatch): - """max_connections from config should be forwarded to Redis.from_url.""" +async def test_make_stream_bridge_passes_redis_options(monkeypatch): + """Redis options from config should be forwarded to Redis bridge setup.""" import deerflow.runtime.stream_bridge.redis as redis_module captured: dict = {} @@ -685,10 +761,18 @@ async def test_make_stream_bridge_passes_max_connections(monkeypatch): return _FakeRedis() monkeypatch.setattr(redis_module.Redis, "from_url", staticmethod(fake_from_url)) - set_stream_bridge_config(StreamBridgeConfig(type="redis", redis_url="redis://fake:6379/0", max_connections=50)) + set_stream_bridge_config( + StreamBridgeConfig( + type="redis", + redis_url="redis://fake:6379/0", + max_connections=50, + stream_ttl_seconds=42, + ) + ) try: async with make_stream_bridge() as bridge: assert isinstance(bridge, RedisStreamBridge) + assert bridge._stream_ttl_seconds == 42 assert captured["max_connections"] == 50 assert captured["decode_responses"] is True finally: @@ -829,3 +913,34 @@ async def test_redis_integration_maxlen_trims_history(real_redis_bridge): key = real_redis_bridge._stream_key(run_id) length = await real_redis_bridge._redis.xlen(key) assert length == 2 + + +@pytest.mark.integration +@requires_redis +@pytest.mark.anyio +async def test_redis_integration_stream_ttl_reclaims_key(): + """Redis should reclaim retained stream data when cleanup never runs.""" + from redis.asyncio import Redis + + client = Redis.from_url(REDIS_TEST_URL, decode_responses=True) + key_prefix = f"deerflow:test:{uuid.uuid4().hex}" + bridge = RedisStreamBridge( + redis_url=REDIS_TEST_URL, + queue_maxsize=2, + key_prefix=key_prefix, + stream_ttl_seconds=1, + client=client, + ) + run_id = "integ-ttl" + key = bridge._stream_key(run_id) + try: + await bridge.publish(run_id, "metadata", {"run_id": run_id}) + assert await client.exists(key) == 1 + assert await client.ttl(key) >= 0 + await anyio.sleep(2.0) + + assert await client.exists(key) == 0 + finally: + async for existing_key in client.scan_iter(f"{key_prefix}:*"): + await client.delete(existing_key) + await client.aclose() diff --git a/config.example.yaml b/config.example.yaml index 66f9bf064..f8a1fd8e9 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1423,6 +1423,12 @@ run_events: # type: redis # recommended for Docker / multi-worker gateway # redis_url: redis://redis:6379/0 # queue_maxsize: 256 # events retained per run (redis stream MAXLEN) +# 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. +# recovered_stream_cleanup_delay_seconds: 60 # seconds to wait after +# # publishing END for a recovered orphan run +# # before deleting the stream key. # max_connections: 100 # optional pool ceiling. Each live SSE client # # holds one connection blocked in XREAD ... BLOCK # # for up to heartbeat_interval (15s), so hundreds