From b0cb3a3a8cc15b2723581c11855b06560da17dda Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Wed, 16 Sep 2026 20:56:01 +0900 Subject: [PATCH] fix(scheduler): serialize the SQLite launch-budget claim (#5469) * fix(scheduler): serialize the SQLite launch-budget claim claim_queued_run counts the executing occurrences and then promotes one row to launching. Postgres serializes that pair with a transaction advisory lock; SQLite had no counterpart. pysqlite does not begin a transaction for a SELECT, so the budget count ran in autocommit and the deferred transaction reserved the writer only at the promoting UPDATE. Claimers racing on distinct rows therefore read the same stale count, each passed its own status == 'queued' CAS, and max_concurrent_runs was exceeded. A manual trigger overlapping the poller reaches this concurrently within one process, and scheduler.multi_instance over a shared database file reaches it across processes. Two claims of the same row were already safe, which is why the existing coverage did not catch it. Take the writer before the count with BEGIN IMMEDIATE, the idiom ThreadMetaRepository already uses for its read-modify-write paths and the same reservation _lock_task makes for a parent row. The claim targets one row but the budget is global, so this has to be the database-wide writer rather than a row lock. * docs(changelog): reference #5469 in the SQLite launch-budget entry * test(scheduler): pin the launch-budget test's connection reuse The warm-up gather is what makes the claimers actually overlap, but it silently depended on the SQLite engine keeping pooled connections. If that engine ever moved to a non-pooling class, every claimer would open its own connection, the per-connection PRAGMA setup would stagger them, and this test would pass against an unserialized claim instead of failing -- the cold-pool case it exists to avoid. Assert that the warm-up left connections checked in. A non-pooling class does not implement checkedin() at all, so a missing counter reads as zero reuse and reports the same explanation rather than an AttributeError. Verified against NullPool: the guard fails with "NullPool left 0 connections pooled after the warm-up". Only pool_size connections survive the gather (the overflow is discarded), which is why the pre-fix failure is exactly five claimants over a cap of one rather than eight. --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 9 +++ CHANGELOG_zh.md | 6 ++ backend/AGENTS.md | 2 +- .../persistence/scheduled_task_runs/sql.py | 14 +++- backend/tests/test_scheduled_task_queue.py | 69 +++++++++++++++++++ 5 files changed, 98 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index da8cf5b8b..d9d59b996 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -582,6 +582,14 @@ This section accumulates work toward the **2.1.0** milestone ### Fixed +- **scheduler:** Enforce the global `max_concurrent_runs` budget on SQLite, + which previously only held on Postgres. Claiming a queued occurrence counts + the executing rows and then promotes one row to `launching`, and Postgres + serializes that pair with an advisory lock. SQLite's deferred transaction + reserved the writer only at the promoting UPDATE, so claimants racing on + distinct rows — a manual trigger overlapping the poller, or a second Gateway + process sharing the database file — all read the same stale count, all passed + the budget check, and the configured cap was exceeded. ([#5469]) - **sandbox:** Stop AIO's `glob` from reporting an exactly-full result as truncated. Its `include_dirs` branch returned as soon as it had collected `max_results` matches, so a listing that held exactly that many — and no more @@ -2878,4 +2886,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#5427]: https://github.com/bytedance/deer-flow/pull/5427 [#5431]: https://github.com/bytedance/deer-flow/pull/5431 [#5447]: https://github.com/bytedance/deer-flow/pull/5447 +[#5469]: https://github.com/bytedance/deer-flow/pull/5469 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 32de22166..cfad49cb4 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -397,6 +397,11 @@ ### 修复 +- **调度器：** 在 SQLite 上同样强制执行全局 `max_concurrent_runs`，此前该上限只在 Postgres 上成立。 + 认领排队中的 occurrence 时，会先统计正在执行的行，再把其中一行提升为 `launching`，Postgres 用 + advisory lock 将这两步串行化。而 SQLite 的 deferred 事务直到那条提升用的 UPDATE 才占用 writer， + 因此在不同行上并发认领的调用方——手动触发与轮询重叠，或第二个 Gateway 进程共用同一个数据库 + 文件——会读到同一个过期计数并全部通过预算检查，导致实际运行数超过配置的上限。([#5469]) - **沙箱：** AIO 的 `glob` 不再把"恰好填满"的结果报告为截断。其 `include_dirs` 分支在收集到 `max_results` 个匹配时就立即返回，因此一个只有这么多匹配、后面再无匹配的目录列表也会被标记为 被截断，工具据此告诉模型结果不完整。该分支本就持有整份目录列表，现在改为多看一个匹配再判断， @@ -2197,3 +2202,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#5427]: https://github.com/bytedance/deer-flow/pull/5427 [#5431]: https://github.com/bytedance/deer-flow/pull/5431 [#5447]: https://github.com/bytedance/deer-flow/pull/5447 +[#5469]: https://github.com/bytedance/deer-flow/pull/5469 diff --git a/backend/AGENTS.md b/backend/AGENTS.md index a546fff19..cf3c38536 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -20,7 +20,7 @@ The backend runs a LangGraph-based super agent with sandbox execution, persisten - The background scheduler is single-instance by default. `scheduler.multi_instance=true` opts into lease-aware recovery across Gateway instances and requires shared Postgres, `run_ownership.heartbeat_enabled=true`, and `run_events.backend=db`; otherwise startup rejects the configuration. Live scheduled runs are preserved when a peer starts; expired launch claims return to the durable queue, expired run leases are atomically taken over, stale launch writes are fenced by lease ownership, and the Postgres advisory-locked budget makes `max_concurrent_runs` a shared global cap for `launching`/`running` rows. - Long-running MCP work uses a separate durable task runtime (`McpTaskService` + `mcp_tasks`, lease-based recovery) rather than keeping remote task IDs or status polling inside the Agent loop; only submit remains Agent-visible, the database is the source of truth, and `ThreadState` receives only a bounded current-thread projection. Full contract (leases, cancellation fencing, delivery idempotency, management-tool exposure): [packages/harness/deerflow/mcp/AGENTS.md](packages/harness/deerflow/mcp/AGENTS.md). - MCP task notification retries, dead-lettering, and the cancel endpoint's worker-stopped 503 are part of that same contract — see [packages/harness/deerflow/mcp/AGENTS.md](packages/harness/deerflow/mcp/AGENTS.md). -- Scheduled-task dispatch permits one active occurrence per task via `uq_scheduled_task_run_active` (`task_id WHERE status IN ('queued','launching','running')`). Durable `queued` rows survive restarts; only lease-fenced `launching` may call Gateway launch; `running` references the durable run. Stable admission idempotency keys reuse that run after recovery. Reused-thread `ConflictError` returns `launching` to `queued`; other launch errors become `failed`. Atomic queue claims enforce `max_concurrent_runs`, excluding waiting rows. Repeated triggers coalesce; same-thread FIFO blocks behind older active rows. Queue admission, PATCH/resume, pause and delete lock the parent before the occurrence, freezing active task definitions. Pause/delete atomically cancel `queued` work but reject `launching`/`running`; PATCH/resume reject all active states. Only queued conflicts offer pause cancellation. Manual triggers may queue/run while paused. Recovery locks task/run pairs in task-id/run-id order and restores `run_id`, `started_at` and live errors before releasing launch claims. Launch/failure/timeout updates use one parent-first transaction to prevent interleaved claims. Queue timeout fails the occurrence and advances scheduled work to prevent immediate requeue. Repository boundaries coerce serialized timestamps before SQL `DateTime` binding. +- Scheduled-task dispatch permits one active occurrence per task via `uq_scheduled_task_run_active` (`task_id WHERE status IN ('queued','launching','running')`). Durable `queued` rows survive restarts; only lease-fenced `launching` may call Gateway launch; `running` references the durable run. Stable admission idempotency keys reuse that run after recovery. Reused-thread `ConflictError` returns `launching` to `queued`; other launch errors become `failed`. Atomic queue claims enforce `max_concurrent_runs`, excluding waiting rows; the budget count and its UPDATE are separate statements, so writers must serialize before the count (Postgres advisory lock; SQLite `BEGIN IMMEDIATE`, whose deferred transaction otherwise reserves the writer only at the UPDATE) or claims on distinct rows overshoot the cap. Repeated triggers coalesce; same-thread FIFO blocks behind older active rows. Queue admission, PATCH/resume, pause and delete lock the parent before the occurrence, freezing active task definitions. Pause/delete atomically cancel `queued` work but reject `launching`/`running`; PATCH/resume reject all active states. Only queued conflicts offer pause cancellation. Manual triggers may queue/run while paused. Recovery locks task/run pairs in task-id/run-id order and restores `run_id`, `started_at` and live errors before releasing launch claims. Launch/failure/timeout updates use one parent-first transaction to prevent interleaved claims. Queue timeout fails the occurrence and advances scheduled work to prevent immediate requeue. Repository boundaries coerce serialized timestamps before SQL `DateTime` binding. - `POST /api/scheduled-tasks/preview-cron` requires authenticated `threads:read`. Bounded cron previews call the shared scheduler calculator in `asyncio.to_thread`, preserving its DST semantics. Capture the optional aware reference once; return UTC and offset-bearing local occurrences without acquiring task/thread/run stores or dispatching work. This advisory API does not reserve execution. - `extensions_config.json` is written at runtime by the Gateway (`PUT`/`PATCH /api/mcp/config`, the MCP enable switch, skill updates), so the production compose mounts it read-write while `config.yaml` stays `:ro`; Helm copies its ConfigMap seed into a writable home-volume directory before Gateway starts. Every read-modify-write holds both `extensions_config_write_lock` and the sidecar advisory `extensions_config_file_lock`, because the process-local lock alone loses updates across workers. Docker mounts the compose file as its own mount point, and Linux refuses `rename()` over a mount point with `EBUSY` even when the mount is writable — so `atomic_write_extensions_config` keeps the temp-file-plus-rename path and falls back to an in-place overwrite only on `EBUSY`. That fallback is deliberately non-atomic (a crash mid-write truncates the file); it exists because the alternative is a write that can never succeed, and only its first occurrence per target is logged at warning level. Any other `errno` still propagates. Pinned by `tests/test_compose_extensions_config_writable.py`, `tests/test_extensions_config_atomic_write.py`, and `tests/test_helm_extensions_config_writable.py`. diff --git a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py index f9220bdb3..215abeb9b 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py @@ -312,11 +312,23 @@ class ScheduledTaskRunRepository: ) -> dict[str, Any] | None: """Atomically move one waiting row into the lease-fenced launch phase.""" async with self._sf() as session: - if session.get_bind().dialect.name == "postgresql": + dialect = session.get_bind().dialect.name + if dialect == "postgresql": await session.execute( text("SELECT pg_advisory_xact_lock(:lock_key)"), {"lock_key": _SCHEDULER_BUDGET_LOCK_KEY}, ) + elif dialect == "sqlite": + # The budget count below is only meaningful if no peer can + # claim between it and the UPDATE. A deferred SQLite + # transaction does not reserve the writer until that UPDATE, + # which is too late: every claimer would read the same stale + # count and pass. BEGIN IMMEDIATE takes the writer first, the + # same reservation _lock_task makes for a parent row, and it + # serializes claimers in other processes sharing the file too. + # The claim targets one row, but the budget is global, so this + # has to be the database-wide writer rather than a row lock. + await session.execute(text("BEGIN IMMEDIATE")) executing = await session.scalar(select(func.count()).select_from(ScheduledTaskRunRow).where(ScheduledTaskRunRow.status.in_(EXECUTING_RUN_STATUSES))) if int(executing or 0) >= global_max_concurrent_runs: await session.rollback() diff --git a/backend/tests/test_scheduled_task_queue.py b/backend/tests/test_scheduled_task_queue.py index 7be1f58b4..61bdaebf2 100644 --- a/backend/tests/test_scheduled_task_queue.py +++ b/backend/tests/test_scheduled_task_queue.py @@ -285,6 +285,75 @@ async def test_only_one_worker_can_claim_a_queued_run(tmp_path): await close_engine() +async def test_global_launch_budget_holds_when_distinct_rows_are_claimed_concurrently(tmp_path): + """The global launch budget must survive claims racing on *distinct* rows. + + ``claim_queued_run`` counts executing rows and then promotes one row to + ``launching``. Postgres serializes that pair with an advisory lock. SQLite + needs ``BEGIN IMMEDIATE`` for the same reason ``ThreadMetaRepository`` + does: a deferred transaction does not reserve the writer until the UPDATE, + so every claimer reads the same stale count and overshoots + ``max_concurrent_runs``. + + Distinct rows are the load-bearing part. Two claims of the *same* row are + already safe via the ``status == "queued"`` CAS, which is what + ``test_only_one_worker_can_claim_a_queued_run`` covers. + """ + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + try: + sf = get_session_factory() + assert sf is not None + run_repo = ScheduledTaskRunRepository(sf) + now = datetime(2026, 8, 20, 9, 0, tzinfo=UTC) + claimants = 8 + for index in range(claimants): + await run_repo.create( + run_record_id=f"task-run-budget-{index}", + task_id=f"task-budget-{index}", + thread_id=f"thread-budget-{index}", + scheduled_for=now, + trigger="scheduled", + status="queued", + ) + + # Open every connection the claimers need up front. On a cold pool the + # per-connection PRAGMA setup staggers them enough to hide the race. + await asyncio.gather(*(run_repo.count_active_runs() for _ in range(claimants))) + + # That warm-up is load-bearing, and it only works because the SQLite + # engine keeps pooled connections (init_engine_from_config builds it on + # SQLAlchemy's default AsyncAdaptedQueuePool, so `pool_size` of them + # survive this gather while the overflow is discarded). Under a + # non-pooling class such as NullPool every claimer would open its own + # connection, the PRAGMA setup would serialize them, and this test + # would pass against an unserialized claim instead of failing. Assert + # the reuse so that a pool change breaks this test loudly rather than + # quietly draining it of guard strength. + pool = sf.kw["bind"].sync_engine.pool + # A non-pooling class does not implement checkedin() at all, so treat a + # missing counter as "nothing was reused" and report it the same way. + pooled = pool.checkedin() if hasattr(pool, "checkedin") else 0 + assert pooled >= 2, f"{type(pool).__name__} left {pooled} connections pooled after the warm-up; the claimers cannot overlap, so this test would pass against an unserialized claim" + + claims = await asyncio.gather( + *( + run_repo.claim_queued_run( + f"task-run-budget-{index}", + lease_owner=f"worker-{index}", + now=now, + lease_seconds=120, + global_max_concurrent_runs=1, + ) + for index in range(claimants) + ) + ) + + assert sum(claim is not None for claim in claims) == 1 + assert await run_repo.count_active_runs() == 1 + finally: + await close_engine() + + async def test_same_thread_queue_is_claimed_in_fifo_order(tmp_path): await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) try: