From 0d86a0d8fef0fa57cdf948b356cae48731c473e1 Mon Sep 17 00:00:00 2001 From: rayhpeng Date: Tue, 28 Jul 2026 19:46:16 +0800 Subject: [PATCH] fix(schedule): stop the completion hook rolling back the launch write A cron task could become permanently unschedulable after a run that failed fast. Reproduced against the real SQL adapters, and against the legacy path for contrast -- the legacy code does not have this bug, so the hexagonal slice introduced it. The window ---------- `dispatch_task` already documents it: "a fast-failing run can reach handle_run_completion before this write lands". The hook therefore reads the task while the dispatch path has not yet written its bookkeeping, and the snapshot it holds still carries the elapsed `next_run_at` the claim was made on. `handle_run_completion` then wrote that whole snapshot back through `save()`, so whichever landed second undid the other. For a cron task the result is terminal in the worst way. `record_launch` writes the fresh fire time, bumps `run_count` and clears the claim; the completion's whole-aggregate write restores the elapsed fire time, rolls `run_count` back and leaves `status='running'` with `lease_expires_at IS NULL`. Neither `claim_due` branch matches that shape (one needs `enabled`, the other needs an expired claim), and `cancel_stuck_once_tasks` only sweeps `once` rows -- so nothing can ever reach the task again. The legacy `app/scheduler/service.py` wrote the same outcome field by field (`update(..., updates={"last_error": ...})`) and never touched scheduling state, which is why it survives the same interleaving. The fix ------- `record_completion` joins `record_launch` as a second deliberately narrow port method, for the same stated reason `record_launch` is not expressed as `save(task)`: the two race, so neither may write through the aggregate. They now own disjoint fields -- the launch owns the schedule, the completion owns the verdict (terminal status plus `last_error`, with `None` meaning "do not move the status", i.e. every cron task). `save()` keeps its documented purpose, the user-initiated whole-aggregate updates (`update_task` / `pause` / `resume`). `handle_run_completion` still reads the task first, but only to ask `status_after_completion`, which reads nothing but the schedule type -- immutable, so that read carries no time-of-check risk. Tests ----- The contract suite missed this because it groups cases by port method: `record_launch` appears only among its own, never interleaved with `save`, and `test_protect_terminal_keeps_a_concurrently_finalized_verdict` pins the mirror-image direction only. The in-memory double replaces the whole row too, so both implementations were faithfully wrong -- the defect was in the contract, not either adapter. Added: `TestRecordCompletion` (5 cases, both implementations), including the interleaving itself; and `test_a_cron_task_survives_a_launch_write_landing_mid_completion`, which drives a real `ScheduleService` over a repo double that commits `record_launch` between the hook's read and its write. Both were watched failing first -- the service case failing with the actual symptom, a rolled back `next_run_at`, not a missing method. Verified end to end against file-backed sqlite with the real service: the task keeps `enabled`, the concurrent fire time and run count survive, the verdict is recorded, and the next poll still claims it. Co-Authored-By: Claude Opus 5 --- backend/AGENTS.md | 1 + .../schedule/scheduled_task_repository.py | 22 ++++++ .../harness/deerflow/domain/schedule/ports.py | 34 ++++++++ .../deerflow/domain/schedule/service.py | 16 ++-- backend/tests/schedule_fakes.py | 16 ++++ backend/tests/test_schedule_fakes.py | 78 +++++++++++++++++++ backend/tests/test_schedule_service.py | 72 +++++++++++++++++ 7 files changed, 234 insertions(+), 5 deletions(-) diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 2f1fd8ec0..da848e0ec 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -18,6 +18,7 @@ DeerFlow is a LangGraph-based AI super agent system with a full-stack architectu - With `stream_subgraphs`, subgraph frames keep their namespace in the SSE event name (`values|`, LangGraph Platform style) instead of impersonating root frames — a delegated subagent inherits the parent checkpoint namespace, so publishing its `values` snapshot as bare `values` replaces the whole thread view in SDK clients (#4399). Root-only consumers (file-tool chunk batcher, subagent event persistence, LLM error-fallback detection) ignore namespaced frames. The web frontend does not request subgraph streaming; subtask progress rides root-namespace `task_*` custom events. - Scheduled-task executions must reuse that same Gateway run lifecycle. The scheduler may decide *when* work runs, but it must dispatch through the existing run path rather than introducing a parallel execution stack. - Scheduled-task dispatch enforces "at most one active run per task when `overlap_policy=skip`" at the DB layer via the partial unique index `uq_scheduled_task_run_active` (`scheduled_task_runs.task_id WHERE status IN ('queued','running')`). `ScheduledTaskService.dispatch_task`'s `has_active_runs` check is a non-atomic fast path (its own session, separated from the `create()` insert by `await` points), so two concurrent dispatches — a manual `POST /scheduled-tasks/{id}/trigger` racing the poller, a double-click, or a client retry — can both pass it; the index is the atomic arbiter, and the losing `create` surfaces as `ActiveScheduledRunConflict` (translated from `IntegrityError` in the repository) and collapses to the same outcome as the fast path (manual → 409 conflict, scheduled → a `"skipped"` tombstone). The scheduled-skip tombstone is created directly as terminal `"skipped"` (not a transient `"queued"`) so it never occupies the active slot the pre-existing run still holds. Sibling of the `runs` table's `uq_runs_thread_active` (PR #4003), which keys on `thread_id` and so does not cover the default `fresh_thread_per_run` context where every dispatch gets a new thread. Index is status-only, not `overlap_policy`-conditional (the policy is fixed to `"skip"` in the MVP). +- The dispatch path and the run-completion hook write to the same `scheduled_tasks` row concurrently, so **the two writes own disjoint field sets and neither may write through the whole aggregate**. A run that fails fast reaches its completion hook before the dispatch path's own bookkeeping write lands, so the hook is holding a snapshot taken *before* it. The launch write (`record_launch`) owns the schedule — `next_run_at`, `last_run_at`, `last_run_id`, `last_thread_id`, `run_count`, the claim — and the completion write (`record_completion`) owns only the verdict: the terminal status and `last_error`. Expressing the completion as a read-modify-write through the aggregate replays the stale snapshot and rolls the launch write back, restoring an elapsed `next_run_at` and leaving the task in `running` with no live claim — a shape neither `claim_due` branch nor `cancel_stuck_once_tasks` can reach, i.e. a cron task that is permanently unschedulable with no recovery path. `record_launch`'s `protect_terminal` flag closes the mirror-image race in the other direction (the hook's terminal status must survive a launch write landing after it). Pinned by the `TestRecordCompletion` contract cases (both the in-memory double and real sqlite) and by `test_a_cron_task_survives_a_launch_write_landing_mid_completion`. **Project Structure**: ``` diff --git a/backend/app/adapters/schedule/scheduled_task_repository.py b/backend/app/adapters/schedule/scheduled_task_repository.py index a7e297387..1d65e89ca 100644 --- a/backend/app/adapters/schedule/scheduled_task_repository.py +++ b/backend/app/adapters/schedule/scheduled_task_repository.py @@ -259,6 +259,28 @@ class SqlScheduledTaskRepository(ScheduledTaskRepository): row.updated_at = datetime.now(UTC) await session.commit() + async def record_completion( + self, + task_id: str, + *, + user_id: str, + status: TaskStatus | None, + error: str | None, + ) -> None: + async with self._sf() as session: + row = await session.get(ScheduledTaskRow, task_id) + if row is None or row.user_id != user_id: + return + # Field-level on purpose: `record_launch` may commit on either side + # of this write, and the two must not undo each other. Assigning + # anything below `last_error` here is what reintroduces the race -- + # see the port docstring. + row.last_error = error + if status is not None: + row.status = str(status) + row.updated_at = datetime.now(UTC) + await session.commit() + async def cancel_stuck_once_tasks(self, *, error: str) -> int: """Reconcile `once` tasks orphaned in `running` by a process crash. diff --git a/backend/packages/harness/deerflow/domain/schedule/ports.py b/backend/packages/harness/deerflow/domain/schedule/ports.py index ae18d2f68..931485144 100644 --- a/backend/packages/harness/deerflow/domain/schedule/ports.py +++ b/backend/packages/harness/deerflow/domain/schedule/ports.py @@ -155,6 +155,40 @@ class ScheduledTaskRepository(Protocol): """ ... + async def record_completion( + self, + task_id: str, + *, + user_id: str, + status: TaskStatus | None, + error: str | None, + ) -> None: + """Write a finished run's verdict onto the task. + + Deliberately NOT expressed as `save(task)`, for the same reason + `record_launch` is not: the two race. A run that fails fast reaches its + completion hook while the dispatch path is still writing its + bookkeeping, so a read-modify-write through the aggregate would replay + a snapshot taken before that write and roll it back -- restoring an + elapsed `next_run_at` and an out-of-date `run_count`, and leaving the + task in `running` with no live claim, which neither `claim_due` branch + nor `cancel_stuck_once_tasks` can reach. + + The two writes therefore own disjoint fields: the launch owns the + schedule, this owns the verdict. Nothing here touches `next_run_at`, + `last_run_at`, `last_run_id`, `last_thread_id`, `run_count` or the + claim. + + `status` is `None` when the verdict must not move the status -- every + cron task, whose schedule outlives any single run. `error` is written + either way, so a cron task still reports what went wrong last time. + + Scoped by `user_id` like every other read: a task belonging to someone + else is left alone rather than reported. An unknown task is ignored -- + one deleted mid-flight simply has nothing to update. + """ + ... + async def cancel_stuck_once_tasks(self, *, error: str) -> int: """Reconcile `once` tasks orphaned mid-flight by a process crash. diff --git a/backend/packages/harness/deerflow/domain/schedule/service.py b/backend/packages/harness/deerflow/domain/schedule/service.py index 3a99a12f6..9a7d78d4d 100644 --- a/backend/packages/harness/deerflow/domain/schedule/service.py +++ b/backend/packages/harness/deerflow/domain/schedule/service.py @@ -350,16 +350,22 @@ class ScheduleService: error=outcome.error, finished_at=now, ) + # Read to ask the aggregate what this outcome means, then write only + # that. `status_after_completion` reads nothing but the schedule type, + # which no concurrent write can change, so this read carries no + # time-of-check risk -- and `record_completion` deliberately does not + # write the fields that a concurrent `record_launch` owns. task = await self._tasks.get(outcome.task_id, user_id=outcome.user_id) if task is None: return # The error is recorded whether or not the status moves: a cron task # keeps its schedule but still reports what went wrong last time. - updated = replace(task, last_error=outcome.error) - new_status = task.status_after_completion(outcome.status) - if new_status is not None: - updated = replace(updated, status=new_status) - await self._tasks.save(updated) + await self._tasks.record_completion( + outcome.task_id, + user_id=outcome.user_id, + status=task.status_after_completion(outcome.status), + error=outcome.error, + ) async def reconcile_on_startup(self, *, error: str) -> tuple[int, int]: """Clean up what a process crash left behind, returning what was fixed. diff --git a/backend/tests/schedule_fakes.py b/backend/tests/schedule_fakes.py index fb2f91542..2e0b84744 100644 --- a/backend/tests/schedule_fakes.py +++ b/backend/tests/schedule_fakes.py @@ -154,6 +154,22 @@ class InMemoryScheduledTaskRepository: row.lease_owner = None row.lease_expires_at = None + async def record_completion( + self, + task_id: str, + *, + user_id: str, + status: TaskStatus | None, + error: str | None, + ) -> None: + row = self._rows.get(task_id) + if row is None or row.task.user_id != user_id: + return + # Only the verdict; every scheduling field belongs to record_launch. + row.task = replace(row.task, last_error=error) + if status is not None: + row.task = replace(row.task, status=status) + async def cancel_stuck_once_tasks(self, *, error: str) -> int: cancelled = 0 for row in self._rows.values(): diff --git a/backend/tests/test_schedule_fakes.py b/backend/tests/test_schedule_fakes.py index 9e5b534a8..3142d560d 100644 --- a/backend/tests/test_schedule_fakes.py +++ b/backend/tests/test_schedule_fakes.py @@ -352,6 +352,84 @@ class TestRecordLaunch: ) +class TestRecordCompletion: + """The completion hook's write. + + Deliberately as narrow as `record_launch` is, and for the same reason: the + two race, so neither may write through the whole aggregate. This one owns + the terminal verdict and nothing else -- every scheduling field belongs to + the launch path, which may commit at any point around it. + """ + + async def test_records_the_terminal_status_and_error(self, tasks): + await tasks.add(make_task(status=TaskStatus.RUNNING, schedule=ONCE)) + + await tasks.record_completion("task-1", user_id="user-1", status=TaskStatus.FAILED, error="boom") + + task = await tasks.get("task-1", user_id="user-1") + assert task.status is TaskStatus.FAILED + assert task.last_error == "boom" + + async def test_a_none_status_records_the_error_and_leaves_the_status(self, tasks): + """A cron task's schedule outlives any single run, so only what went + wrong is recorded.""" + await tasks.add(make_task(status=TaskStatus.ENABLED, schedule=CRON)) + + await tasks.record_completion("task-1", user_id="user-1", status=None, error="boom") + + task = await tasks.get("task-1", user_id="user-1") + assert task.status is TaskStatus.ENABLED + assert task.last_error == "boom" + + async def test_never_rolls_back_a_concurrent_launch_write(self, tasks): + """The regression this method exists to prevent. + + A fast-failing run reaches the completion hook while the dispatch path + is still writing its bookkeeping. Whichever lands second must not undo + the other: the launch owns the schedule, the completion owns the + verdict. + """ + await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1))) + await tasks.claim_due(now=NOW, lease_seconds=120, limit=10) + + next_at = NOW + timedelta(days=1) + await tasks.record_launch( + "task-1", + status=TaskStatus.ENABLED, + next_run_at=next_at, + last_run_at=NOW, + last_run_id="run-1", + last_thread_id="thread-1", + last_error=None, + increment_run_count=True, + protect_terminal=True, + ) + + await tasks.record_completion("task-1", user_id="user-1", status=None, error="boom") + + task = await tasks.get("task-1", user_id="user-1") + assert task.next_run_at == next_at, "the launch path's next fire time must survive" + assert task.run_count == 1, "the launch path's run count must survive" + assert task.last_run_id == "run-1" + assert task.last_thread_id == "thread-1" + assert task.last_error == "boom", "the completion still records its verdict" + # The whole point: the task is still reachable by the next poll. + claimed = await tasks.claim_due(now=NOW + timedelta(days=2), lease_seconds=120, limit=10) + assert [t.task_id for t in claimed] == ["task-1"] + + async def test_another_users_task_is_untouched(self, tasks): + await tasks.add(make_task(status=TaskStatus.ENABLED)) + + await tasks.record_completion("task-1", user_id="someone-else", status=TaskStatus.FAILED, error="boom") + + task = await tasks.get("task-1", user_id="user-1") + assert task.status is TaskStatus.ENABLED + assert task.last_error is None + + async def test_unknown_task_is_ignored(self, tasks): + await tasks.record_completion("nope", user_id="user-1", status=TaskStatus.FAILED, error="boom") + + class TestCancelStuckOnceTasks: async def test_cancels_a_launched_once_task_with_no_claim(self, tasks): """Launched, so the claim was released; the completion hook then died diff --git a/backend/tests/test_schedule_service.py b/backend/tests/test_schedule_service.py index f1dd82e58..84b3d26e0 100644 --- a/backend/tests/test_schedule_service.py +++ b/backend/tests/test_schedule_service.py @@ -64,6 +64,42 @@ class _BlindRunRepo(InMemoryScheduledRunRepository): return False +class _LaunchWritesMidReadTaskRepo(InMemoryScheduledTaskRepository): + """`record_launch` commits between the completion hook's read and its write. + + That ordering is not exotic -- `dispatch_task` performs its two bookkeeping + writes after the launch returns, and a run that fails fast reaches the hook + in between. Modelling it here rather than with real concurrency keeps the + window deterministic. + """ + + def __init__(self, *, launch_next_run_at: datetime) -> None: + super().__init__() + self._launch_next_run_at = launch_next_run_at + self._armed = False + + def arm(self) -> None: + """Fire the interleaving on the next read, once.""" + self._armed = True + + async def get(self, task_id: str, *, user_id: str): + task = await super().get(task_id, user_id=user_id) + if self._armed: + self._armed = False + await self.record_launch( + task_id, + status=TaskStatus.ENABLED, + next_run_at=self._launch_next_run_at, + last_run_at=NOW, + last_run_id="run-1", + last_thread_id="thread-1", + last_error=None, + increment_run_count=True, + protect_terminal=True, + ) + return task + + def make_service( *, tasks: InMemoryScheduledTaskRepository | None = None, @@ -473,6 +509,42 @@ class TestRunCompletion: now=NOW, ) + async def test_a_cron_task_survives_a_launch_write_landing_mid_completion(self): + """The interleaving `dispatch_task` already admits is possible. + + A run that fails fast reaches this hook while the dispatch path has not + yet written its bookkeeping, so the hook reads a task still carrying the + elapsed `next_run_at`. If the write-back replays that whole snapshot, + the launch path's fresh fire time is rolled back and the task lands in + `running` with no live claim -- a shape neither `claim_due` branch nor + `cancel_stuck_once_tasks` can reach, i.e. permanently unschedulable. + """ + tasks = _LaunchWritesMidReadTaskRepo(launch_next_run_at=NOW + timedelta(days=1)) + runs = InMemoryScheduledRunRepository() + service = make_service(tasks=tasks, runs=runs) + task = await create_cron_task(service) + await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + record = runs.all_runs()[0] + tasks.arm() + + await service.handle_run_completion( + RunOutcome( + task_id=task.task_id, + record_id=record.record_id, + run_id="run-1", + user_id="user-1", + status=RunStatus.FAILED, + error="boom", + ), + now=NOW, + ) + + after = await service.get_task(task.task_id, user_id="user-1") + assert after.last_error == "boom", "the completion still records its verdict" + assert after.next_run_at == NOW + timedelta(days=1), "the launch path's fire time must survive" + claimed = await tasks.claim_due(now=NOW + timedelta(days=2), lease_seconds=120, limit=10) + assert [t.task_id for t in claimed] == [task.task_id], "the task must remain schedulable" + class TestReconcileOnStartup: async def test_sweeps_orphaned_runs_and_stuck_once_tasks(self):