From 052ad76f044991325dcaa0eb5cbbdf37abe4b56e Mon Sep 17 00:00:00 2001 From: yetuge <182123929+yetuge@users.noreply.github.com> Date: Sun, 27 Sep 2026 21:00:09 +0800 Subject: [PATCH] fix(scheduler): re-read the run row under the writer before guarding update_status (#5860) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit update_status guarded expected_lease_owner (and the protect_terminal backfill) on a plain session.get. On SQLite a concurrent requeue plus re-claim can commit between that read and this session's write, so the stale guard passed and the late write clobbered the fresh claim — clearing the new owner's lease fencing and overwriting its status. The guard now re-reads under the parent's writer lock (the same task -> run lock order as fail_launching_run / expire_queued_runs), with populate_existing so the identity map does not serve the pre-lock row back untouched. This is the run-row counterpart of the staleness #5777 fixed for the task-level lease release. Red/green: tests/test_scheduled_task_dispatch_race.py gains a parking session proxy that commits a requeue + re-claim by another scheduler between update_status's read and its write — on main the late write returned True and clobbered the claim; with the change it is fenced (False) and the re-claim survives. Signed-off-by: yetuge <2219677952@qq.com> --- .../persistence/scheduled_task_runs/sql.py | 12 +++ .../test_scheduled_task_dispatch_race.py | 98 +++++++++++++++++++ 2 files changed, 110 insertions(+) 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 215abeb9b2..d4b1c39518 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py @@ -636,6 +636,18 @@ class ScheduledTaskRunRepository: ) -> bool: async with self._sf() as session: row = await session.get(ScheduledTaskRunRow, run_record_id) + if row is None: + return False + # The lease-owner / terminal guards are only as fresh as the row + # they read: on SQLite a plain SELECT sees a snapshot that a + # concurrent requeue + re-claim can invalidate before this session + # commits (the same staleness #5777 fixed for the parent task's + # lease release). Take the parent's writer first — the same + # task -> run lock order every other mutating path uses — and + # re-read the row under it (populate_existing: the identity map + # would otherwise serve the pre-lock row back untouched). + await self._lock_task(session, row.task_id) + row = await session.get(ScheduledTaskRunRow, run_record_id, with_for_update=True, populate_existing=True) if row is None: return False if protect_terminal and row.status in TERMINAL_RUN_STATUSES: diff --git a/backend/tests/test_scheduled_task_dispatch_race.py b/backend/tests/test_scheduled_task_dispatch_race.py index 80b18f61f5..9c4cc22a71 100644 --- a/backend/tests/test_scheduled_task_dispatch_race.py +++ b/backend/tests/test_scheduled_task_dispatch_race.py @@ -26,6 +26,7 @@ from app.scheduler.service import ScheduledTaskService from deerflow.config.database_config import DatabaseConfig from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config from deerflow.persistence.scheduled_task_runs import ActiveScheduledRunConflict, ScheduledTaskRunRepository +from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository pytestmark = pytest.mark.asyncio @@ -278,3 +279,100 @@ async def test_release_dispatch_lease_does_not_revive_a_paused_task(tmp_path): assert task["lease_owner"] is None finally: await close_engine() + + +class _ParkingRunRepoSession: + """AsyncSession proxy that parks the first load of a given run row, so a + second writer can commit inside the reader's WAL snapshot window — the + deterministic stand-in for a lease-expiry requeue + re-claim racing a late + status write.""" + + def __init__(self, real, state) -> None: + self._real = real + self._state = state + + async def __aenter__(self): + await self._real.__aenter__() + return self + + async def __aexit__(self, *exc): + return await self._real.__aexit__(*exc) + + def __getattr__(self, name): + return getattr(self._real, name) + + async def get(self, *args, **kwargs): + row = await self._real.get(*args, **kwargs) + state = self._state + if state["armed"] and not state["done"] and args and args[0] is ScheduledTaskRunRow and args[1] == state["run_record_id"]: + state["done"] = True + state["holding"].set() + await state["resume"].wait() + return row + + +async def test_update_status_lease_guard_does_not_read_a_stale_snapshot(tmp_path): + """A late ``update_status`` must fence against a re-claim that landed + after its read. + + ``update_status`` guarded ``expected_lease_owner`` on a plain SELECT. On + SQLite (WAL) that guard read a pre-requeue snapshot: a lease-expiry + requeue plus a re-claim by another scheduler that committed while the late + write was in flight passed the stale guard and clobbered the fresh claim. + The guard now re-reads under the parent's writer lock — the same + staleness #5777 fixed for the task-level lease release. + """ + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + try: + sf = get_session_factory() + assert sf is not None + now = datetime.now(UTC) + task_repo = ScheduledTaskRepository(sf) + await _seed_task(task_repo, "task-stale-guard", next_run_at=now) + + state = { + "armed": False, + "done": False, + "holding": asyncio.Event(), + "resume": asyncio.Event(), + "run_record_id": "run-stale-guard", + } + + def hooked_factory(): + return _ParkingRunRepoSession(sf(), state) + + run_repo = ScheduledTaskRunRepository(sf) + parked_repo = ScheduledTaskRunRepository(hooked_factory) + plain_repo = ScheduledTaskRunRepository(sf) + + created = await run_repo.create( + run_record_id="run-stale-guard", + task_id="task-stale-guard", + thread_id="th-1", + scheduled_for=now, + trigger="cron", + status="queued", + ) + assert created["status"] == "queued" + claimed = await run_repo.claim_queued_run("run-stale-guard", lease_owner="scheduler-A", now=now, lease_seconds=120, global_max_concurrent_runs=10) + assert claimed is not None and claimed["lease_owner"] == "scheduler-A" + + state["armed"] = True + late = asyncio.create_task(parked_repo.update_status("run-stale-guard", status="failed", error="late completion", expected_lease_owner="scheduler-A")) + await asyncio.wait_for(state["holding"].wait(), timeout=5) + + # A lease-expiry requeue plus a fresh claim by another scheduler land + # while the late write is parked between its read and its write. + assert await plain_repo.requeue_claimed_run("run-stale-guard", lease_owner="scheduler-A") is True + reclaimed = await plain_repo.claim_queued_run("run-stale-guard", lease_owner="scheduler-B", now=now, lease_seconds=120, global_max_concurrent_runs=10) + assert reclaimed is not None and reclaimed["lease_owner"] == "scheduler-B" + state["resume"].set() + + assert await asyncio.wait_for(late, timeout=10) is False, "the stale guard must fence the late write" + + row = await plain_repo.get_active_run("task-stale-guard") + assert row is not None + assert row["status"] == "launching", "the re-claim must survive the late write" + assert row["lease_owner"] == "scheduler-B" + finally: + await close_engine()