From 75de35628d7fd4be9f7c321dbf8956993dc6236b Mon Sep 17 00:00:00 2001 From: yinx65588-boop Date: Sun, 18 Oct 2026 10:46:32 +0800 Subject: [PATCH] fix(scheduler): fence expired launch recovery against fresh claims (#6975) Co-authored-by: yinx65588-boop Co-authored-by: Willem Jiang --- README.md | 1 + .../harness/deerflow/persistence/AGENTS.md | 7 + .../persistence/scheduled_task_runs/sql.py | 7 + .../test_scheduled_launch_recovery_fence.py | 143 ++++++++++++++++++ 4 files changed, 158 insertions(+) create mode 100644 backend/tests/test_scheduled_launch_recovery_fence.py diff --git a/README.md b/README.md index c152f14b0e..03b035c398 100644 --- a/README.md +++ b/README.md @@ -3101,6 +3101,7 @@ Current MVP capabilities: - Editing or duplicating an interval task preserves its saved cadence until the interval is explicitly changed, including sub-minute intervals allowed by the operator's scheduler configuration - Run background scheduled executions as non-interactive DeerFlow runs (`ask_clarification` is not exposed there) - Persist a due execution as `queued` when its reused thread or the global execution budget is busy, then launch it when capacity is available; queued occurrences survive Gateway restarts and fail after `scheduler.queue_timeout_seconds` +- Retrying a busy scheduled task keeps its execution slot while an earlier recovery pass is still in progress. - Share execution slots fairly between task owners: one owner may have at most `scheduler.max_concurrent_runs_per_user` scheduled runs starting or running at a time (default 2, never more than `max_concurrent_runs`; `0` turns the per-owner cap off), and the waiting queue is drained owner by owner, so one owner's backlog never holds back another owner's run. A run that waits longer than `scheduler.queue_timeout_seconds` is skipped, and its history reads "Skipped: it waited too long for a free slot" - Freeze a task's definition while an occurrence is `queued`, `launching`, or `running`, so a durable occurrence cannot silently pick up a different prompt, thread, or schedule; transitioning a task to paused or deleting it cancels an existing waiting occurrence, while `launching`/`running` work must finish before those mutations are retried and an explicit manual trigger may still wait and run without resuming a paused schedule - Pause, resume, trigger, inspect history, and delete tasks diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index f610b3c7e1..6f085db118 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/AGENTS.md @@ -31,6 +31,13 @@ The per-owner run cap `scheduler.max_concurrent_runs_per_user` is effective `min 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. +Expired-launch scans select candidates, not ownership. After taking the parent +and occurrence locks, recheck launch status and lease expiry before recovery. +A busy launch may have requeued and a manual trigger reclaimed the same row +while recovery waited; the new claim can have the same process owner. Preserve +its full row and execution-budget slot. A lease equal to scan time is still live; +missing or strictly older leases remain recoverable. + ## MCP task notifications `mcp_tasks/sql.py::claim_notification_work` never returns a row without a dispatch event. A pending snapshot whose condition resolved before delivery (tracking recovered, or an answered `input_required` prompt returned the task to `working`) is consumed at claim time: `notified_version` advances, the content fingerprint clears so an identical later event is recorded again, stale error/attempt/dispatch state clears, status returns to `none`, and the claim lease is released instead of launching an empty event. A legacy row left without a dispatch event is consumed only when nothing newer exists; a newer recorded event is rebuilt and claimed instead. Consumption is a conditional UPDATE fenced on the event version and phase the claim read, so a concurrent newer event (SQLite has no row locks) is never acknowledged and stays claimable. Coverage: `tests/test_mcp_task_repository.py` (resolved snapshot, legacy retry, expired-claim reclaim, identical-event recurrence, interleaved newer completion, legacy launching/dispatched with and without a newer completion). 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 edce9b61b9..68984d8790 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py @@ -876,6 +876,13 @@ class ScheduledTaskRunRepository: row = await session.get(ScheduledTaskRunRow, row_id, with_for_update=True) if row is None or row.status != "launching": continue + # The scan only identifies candidates. While waiting for the + # parent lock, the old launch can requeue and a manual trigger + # can claim this same occurrence again, even under the same + # lease owner. Revalidate expiry under the write lock before + # clearing any claim or reconciling its underlying run. + if _lease_is_alive(row.lease_expires_at, now=now, grace_seconds=0): + continue candidate = await self._find_underlying_run(session, row, task) row.lease_owner = None row.lease_expires_at = None diff --git a/backend/tests/test_scheduled_launch_recovery_fence.py b/backend/tests/test_scheduled_launch_recovery_fence.py new file mode 100644 index 0000000000..fbb1359715 --- /dev/null +++ b/backend/tests/test_scheduled_launch_recovery_fence.py @@ -0,0 +1,143 @@ +"""Recovery must revalidate launch leases after acquiring the parent lock.""" + +from __future__ import annotations + +import asyncio +import os +import uuid +from datetime import UTC, datetime, timedelta + +import pytest +import pytest_asyncio +from sqlalchemy import text +from support.postgres import asyncpg_test_url + +from deerflow.config.database_config import DatabaseConfig +from deerflow.persistence.engine import close_engine, get_engine, get_session_factory, init_engine_from_config +from deerflow.persistence.scheduled_task_runs import ScheduledTaskRunRepository +from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository + +pytestmark = pytest.mark.asyncio + + +@pytest_asyncio.fixture(params=["sqlite", "postgres"]) +async def repositories(request, tmp_path): + schema = None + if request.param == "postgres": + uri = os.environ.get("TEST_POSTGRES_URI") + if not uri: + pytest.skip("requires TEST_POSTGRES_URI (real PostgreSQL launch recovery)") + schema = f"launch_recovery_{uuid.uuid4().hex}" + config = DatabaseConfig(backend="postgres", postgres_url=asyncpg_test_url(uri), postgres_schema=schema) + else: + config = DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)) + await init_engine_from_config(config) + try: + factory = get_session_factory() + assert factory is not None + yield ScheduledTaskRepository(factory), ScheduledTaskRunRepository(factory), ScheduledTaskRunRepository(factory) + finally: + if schema is not None: + async with get_engine().begin() as connection: + await connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + await close_engine() + + +async def seed_occurrence(tasks, runs, *, now, suffix="1"): + await tasks.create( + task_id=f"task-{suffix}", + user_id="user-1", + thread_id=f"thread-{suffix}", + context_mode="reuse_thread", + assistant_id="lead_agent", + title="Recovery fence", + prompt="Work", + schedule_type="cron", + schedule_spec={"cron": "0 9 * * *"}, + timezone="UTC", + next_run_at=None, + ) + return await runs.create( + run_record_id=f"occurrence-{suffix}", + task_id=f"task-{suffix}", + thread_id=f"thread-{suffix}", + scheduled_for=now, + trigger="manual", + status="queued", + ) + + +@pytest.mark.parametrize("new_owner", ["same-worker", "peer-worker"]) +async def test_expired_scan_cannot_requeue_a_fresh_claim(repositories, monkeypatch, new_owner): + tasks, recovery, launcher = repositories + now = datetime.now(UTC) + await seed_occurrence(tasks, launcher, now=now) + assert await launcher.claim_queued_run("occurrence-1", lease_owner="same-worker", now=now, lease_seconds=5, global_max_concurrent_runs=1) + scan_time = now + timedelta(seconds=6) + selected = asyncio.Event() + release = asyncio.Event() + lock_task = recovery._lock_task + + async def pause_before_parent_lock(session, task_id): + # The scan has selected the expired row, but owns no write lock yet. + selected.set() + await release.wait() + return await lock_task(session, task_id) + + monkeypatch.setattr(recovery, "_lock_task", pause_before_parent_lock) + recovering = asyncio.create_task(recovery.recover_expired_launch_claims(error="expired", now=scan_time)) + try: + await asyncio.wait_for(selected.wait(), timeout=5) + # The original launch conflicts with an interactive run and requeues. + # A later manual trigger claims the same occurrence before the old + # recovery scan acquires its parent lock. No fabricated database rows. + assert await launcher.requeue_claimed_run("occurrence-1", lease_owner="same-worker", error="thread busy") + assert await launcher.claim_queued_run("occurrence-1", lease_owner=new_owner, now=scan_time, lease_seconds=120, global_max_concurrent_runs=1) + before = (await launcher.list_by_task("task-1"))[0] + release.set() + assert await asyncio.wait_for(recovering, timeout=5) == 0 + assert (await launcher.list_by_task("task-1"))[0] == before + # The fresh launch must keep its execution slot, not merely its label. + await seed_occurrence(tasks, launcher, now=now, suffix="2") + assert await launcher.claim_queued_run("occurrence-2", lease_owner="third-worker", now=scan_time, lease_seconds=120, global_max_concurrent_runs=1) is None + finally: + release.set() + await asyncio.gather(recovering, return_exceptions=True) + + +@pytest.mark.parametrize("lease_age", [5, 6]) +async def test_recovery_keeps_boundary_lease_and_requeues_expired_claim(repositories, lease_age): + tasks, recovery, launcher = repositories + now = datetime.now(UTC) + await seed_occurrence(tasks, launcher, now=now) + assert await launcher.claim_queued_run("occurrence-1", lease_owner="worker", now=now, lease_seconds=5, global_max_concurrent_runs=1) + before = (await launcher.list_by_task("task-1"))[0] + recovered = await recovery.recover_expired_launch_claims(error="expired", now=now + timedelta(seconds=lease_age)) + after = (await launcher.list_by_task("task-1"))[0] + if lease_age == 5: + assert recovered == 0 + assert after == before + else: + assert recovered == 1 + assert after["status"] == "queued" + assert after["lease_owner"] is None + assert after["lease_expires_at"] is None + assert after["attempt_count"] == 1 + + +async def test_recovery_requeues_legacy_launch_without_a_lease(repositories): + tasks, recovery, launcher = repositories + now = datetime.now(UTC) + await seed_occurrence(tasks, launcher, now=now) + # The unguarded repository API also supports historical launching rows + # whose lease fields were never populated. They must remain recoverable. + assert await launcher.update_status("occurrence-1", status="launching") + assert (await launcher.list_by_task("task-1"))[0]["lease_expires_at"] is None + + assert await recovery.recover_expired_launch_claims(error="expired", now=now) == 1 + + after = (await launcher.list_by_task("task-1"))[0] + assert after["status"] == "queued" + assert after["lease_owner"] is None + assert after["lease_expires_at"] is None + assert after["attempt_count"] == 0