diff --git a/backend/packages/harness/deerflow/domain/schedule/exceptions.py b/backend/packages/harness/deerflow/domain/schedule/exceptions.py index d07c3eb36..4743cc504 100644 --- a/backend/packages/harness/deerflow/domain/schedule/exceptions.py +++ b/backend/packages/harness/deerflow/domain/schedule/exceptions.py @@ -64,4 +64,22 @@ class ThreadBusyError(ScheduleError): class LaunchFailedError(ScheduleError): - """The run could not be launched for any non-conflict reason.""" + """The launch definitely did not start a run. + + The adapter may raise this only when it is CERTAIN no run exists -- the + service releases the task's active slot on this path, so raising it after + the launch side effect may have happened reopens the #4452 duplicate + execution. When in doubt, raise LaunchIndeterminateError instead. + """ + + +class LaunchIndeterminateError(ScheduleError): + """The launch side effect may have happened, but its identity is unknown. + + Raised by the RunLauncher adapter when the launch call did not fail + cleanly -- the response could not be decoded, the connection dropped after + the request was sent, and so on. The service treats this as launched with + an unknown run id: the execution record stays active so the task's single + active slot remains held (#4452 / #4504), and reconciliation later settles + what actually happened. + """ diff --git a/backend/packages/harness/deerflow/domain/schedule/ports.py b/backend/packages/harness/deerflow/domain/schedule/ports.py index 931485144..0ad531d11 100644 --- a/backend/packages/harness/deerflow/domain/schedule/ports.py +++ b/backend/packages/harness/deerflow/domain/schedule/ports.py @@ -288,12 +288,16 @@ class RunLauncher(Protocol): The contract the adapter MUST honour, because it is what keeps the run runtime and the web framework out of the inner ring: - - the execution thread is already busy -> ThreadBusyError - - anything else goes wrong -> LaunchFailedError + - the execution thread is already busy -> ThreadBusyError + - certainly failed, no run exists -> LaunchFailedError + - a run may exist but its identity is lost -> LaunchIndeterminateError - Nothing else may escape. The domain distinguishes those two because they - lead to different outcomes -- a busy thread on a scheduled dispatch is a - skipped occurrence, while a genuine failure is recorded as one. + Nothing else may escape. The distinction is load-bearing: ThreadBusyError + on a scheduled dispatch is a skipped occurrence; LaunchFailedError + releases the task's single active slot, so the adapter may raise it only + when it is CERTAIN no run started; LaunchIndeterminateError covers every + doubt after the launch request was sent -- an undecodable response, a + dropped connection -- and keeps the slot held (#4452 / #4504). """ async def launch( diff --git a/backend/packages/harness/deerflow/domain/schedule/service.py b/backend/packages/harness/deerflow/domain/schedule/service.py index 7f31d37be..2959a4139 100644 --- a/backend/packages/harness/deerflow/domain/schedule/service.py +++ b/backend/packages/harness/deerflow/domain/schedule/service.py @@ -6,14 +6,16 @@ to the aggregate -- and knows nothing about HTTP, SQL, or the run runtime. `user_id` is always passed in explicitly; resolving the current user is the primary adapter's job. -The dispatch path deliberately mirrors the structure of the pre-migration -`app/scheduler/service.py` (since deleted), including its ordering and its -comments, because its concurrency and idempotency semantics are load-bearing -and are pinned by tests that must keep passing unchanged. +The dispatch path deliberately mirrors the structure of the legacy +`app/scheduler/service.py` (still present until the adapter slice replaces +it), including its ordering, its comments, and its post-launch retention +semantics (#4452 / #4504), because its concurrency and idempotency semantics +are load-bearing and are pinned by tests that must keep passing unchanged. """ from __future__ import annotations +import logging from dataclasses import dataclass, replace from datetime import datetime @@ -29,6 +31,7 @@ from deerflow.domain.schedule.commands import ( from deerflow.domain.schedule.exceptions import ( ActiveRunConflictError, LaunchFailedError, + LaunchIndeterminateError, TaskNotFoundError, ThreadBusyError, ThreadNotFoundError, @@ -51,6 +54,8 @@ from deerflow.domain.schedule.ports import ( ThreadLookup, ) +logger = logging.getLogger(__name__) + # Shared so the has_active fast path and the active-slot race path produce # byte-identical outcomes for the same "task already has an active run" # condition. Two callers must not be able to tell which one rejected them. @@ -260,10 +265,13 @@ class ScheduleService: return self._conflict(execution_thread_id) return await self._record_scheduled_skip(task, thread_id=execution_thread_id, now=now, trigger=trigger) - # Only the launch is guarded. The port contract admits exactly two - # escapes, so a failure of the bookkeeping writes below is a genuine - # fault and propagates instead of being recorded as a failed launch -- - # which would mark an already-running execution as failed. + # Only the launch is guarded, and only its two certain outcomes may + # release the slot. Past this point a live run may exist, so every + # failure -- an indeterminate launch, a bookkeeping write dying -- must + # take the retention path below rather than mark the record failed: + # "failed" is terminal, terminal records do not hold the task's single + # active slot, and a released slot lets the next dispatch launch a + # duplicate of a run that is still alive (#4452, fixed by #4504). try: launched = await self._launcher.launch( thread_id=execution_thread_id, @@ -284,30 +292,16 @@ class ScheduleService: return await self._finalize_skip(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, error=str(exc)) return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome=DispatchOutcome.CONFLICT) except LaunchFailedError as exc: + # The adapter certifies no run started (see the port contract), so + # releasing the slot is safe. return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome=DispatchOutcome.FAILED) + except LaunchIndeterminateError as exc: + # The launch side effect may have happened but its identity could + # not be decoded. Retain the slot with the identity unknown; + # reconciliation settles what actually happened. + return await self._retain_launched(task, record_id=record.record_id, run_id=None, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc)) - await self._runs.update_status( - record.record_id, - status=RunStatus.RUNNING, - run_id=launched.run_id, - started_at=now, - # A fast-failing run can reach handle_run_completion before this - # write lands; never clobber its terminal verdict. - protect_terminal=True, - ) - await self._tasks.record_launch( - task.task_id, - status=task.status_after_launch(trigger=trigger), - next_run_at=task.schedule.next_after(now), - last_run_at=now, - last_run_id=launched.run_id, - last_thread_id=launched.thread_id, - last_error=None, - increment_run_count=True, - # Same race as the record write above. - protect_terminal=True, - ) - return DispatchResult(DispatchOutcome.LAUNCHED, record.record_id, launched.run_id, launched.thread_id, None) + return await self._retain_launched(task, record_id=record.record_id, run_id=launched.run_id, thread_id=launched.thread_id, now=now, trigger=trigger, error=None) # ------------------------------------------------------------------ lifecycle @@ -381,6 +375,64 @@ class ScheduleService: raise TaskNotFoundError("Scheduled task not found") return saved + async def _retain_launched( + self, + task: ScheduledTask, + *, + record_id: str, + run_id: str | None, + thread_id: str, + now: datetime, + trigger: TriggerKind, + error: str | None, + ) -> DispatchResult: + """Record a launch whose run is (or may be) live. Best-effort by design. + + Both writes are attempted independently and their failures logged + rather than raised: the run is already in flight, so there is no + outcome these writes could change -- only bookkeeping they could lose. + A failed record write leaves the row queued, a failed task write + leaves the claim held; both states still hold the task's single + active slot, which is the invariant that matters (#4452). The first + error is surfaced on the result so the caller knows the bookkeeping + is behind, while the outcome stays LAUNCHED. + """ + try: + await self._runs.update_status( + record_id, + status=RunStatus.RUNNING, + run_id=run_id, + started_at=now, + # A fast-failing run can reach handle_run_completion before + # this write lands; never clobber its terminal verdict. + protect_terminal=True, + ) + except Exception: + logger.exception("Scheduled run record %s: post-launch bookkeeping failed; run %s is still live (task %s)", record_id, run_id, task.task_id) + if error is None: + error = f"post-launch bookkeeping failed for record {record_id}" + try: + await self._tasks.record_launch( + task.task_id, + status=task.status_after_launch(trigger=trigger), + next_run_at=task.schedule.next_after(now), + last_run_at=now, + last_run_id=run_id, + last_thread_id=thread_id, + # A bookkeeping transient is not the run's verdict: the launch + # succeeded, so the task list must not show an error while the + # run is in flight. handle_run_completion writes the real one. + last_error=None, + increment_run_count=True, + # Same race as the record write above. + protect_terminal=True, + ) + except Exception: + logger.exception("Scheduled task %s: post-launch update failed; run %s is still live", task.task_id, run_id) + if error is None: + error = f"post-launch bookkeeping failed for task {task.task_id}" + return DispatchResult(DispatchOutcome.LAUNCHED, record_id, run_id, thread_id, error) + def _conflict(self, thread_id: str) -> DispatchResult: """A manual trigger against an active run. diff --git a/backend/tests/test_schedule_service.py b/backend/tests/test_schedule_service.py index 1890b0d97..2cf8fc9d5 100644 --- a/backend/tests/test_schedule_service.py +++ b/backend/tests/test_schedule_service.py @@ -37,6 +37,7 @@ from deerflow.domain.schedule.commands import ( from deerflow.domain.schedule.exceptions import ( InvalidScheduleError, LaunchFailedError, + LaunchIndeterminateError, TaskNotFoundError, TaskNotMutableError, ThreadBusyError, @@ -354,6 +355,147 @@ class TestDispatchOutcomes: assert result.record_id is not None, "the attempt itself is still recorded" +class _RecordWriteFailsOnce(InMemoryScheduledRunRepository): + """`update_status` raises once -- the queued->running write dies mid-flight.""" + + def __init__(self) -> None: + super().__init__() + self.fail_next_update = False + + async def update_status(self, record_id, **kwargs): + if self.fail_next_update: + self.fail_next_update = False + raise RuntimeError("db down: update_status") + return await super().update_status(record_id, **kwargs) + + +class _TaskWriteFailsOnce(InMemoryScheduledTaskRepository): + """`record_launch` raises once -- the parent-task write dies mid-flight.""" + + def __init__(self) -> None: + super().__init__() + self.fail_next_record_launch = False + + async def record_launch(self, task_id, **kwargs): + if self.fail_next_record_launch: + self.fail_next_record_launch = False + raise RuntimeError("db down: record_launch") + return await super().record_launch(task_id, **kwargs) + + +class TestPostLaunchRetention: + """Regression for issue #4452 (fixed on main by #4504), ported to the + domain service. + + Once `launch()` has returned -- or has raised without being able to say + whether a run started -- a live run may exist. A bookkeeping failure after + that point must NOT release the task's single active slot or reclassify + the run: the record stays active, the dispatch still reports LAUNCHED, + and the next dispatch must observe the held slot instead of launching a + duplicate. + """ + + async def test_post_launch_record_write_failure_does_not_release_active_slot(self): + runs = _RecordWriteFailsOnce() + launcher = FakeRunLauncher() + service = make_service(runs=runs, launcher=launcher) + task = await create_cron_task(service) + runs.fail_next_update = True + + first = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + + # The run launched despite the bookkeeping error; the result says so + # and surfaces the error instead of hiding it. + assert first.outcome is DispatchOutcome.LAUNCHED + assert first.run_id == "run-1" + assert first.error is not None + + # The row never reached RUNNING, but queued is still active: the slot + # is held and the next dispatch is an overlap, not a duplicate. + assert runs.all_runs()[0].is_active + second = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + assert len(launcher.calls) == 1, "a duplicate launch is the #4452 bug" + assert second.outcome is DispatchOutcome.SKIPPED + + # The bookkeeping transient is not the run's verdict: record_launch + # still ran, cleared last_error, and counted the execution. + after = await service.get_task(task.task_id, user_id="user-1") + assert after.last_error is None + assert after.run_count == 1 + + async def test_post_launch_task_write_failure_does_not_release_active_slot(self): + tasks = _TaskWriteFailsOnce() + launcher = FakeRunLauncher() + service = make_service(tasks=tasks, launcher=launcher) + task = await create_cron_task(service) + tasks.fail_next_record_launch = True + + first = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + + assert first.outcome is DispatchOutcome.LAUNCHED + assert first.run_id == "run-1" + assert first.error is not None + + # The record write landed before the task write failed, so the row is + # RUNNING with the launched id retained for recovery and cancellation. + stored = (await service.list_task_runs(task.task_id, user_id="user-1"))[0] + assert stored.status is RunStatus.RUNNING + assert stored.run_id == "run-1" + + second = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + assert len(launcher.calls) == 1, "a duplicate launch is the #4452 bug" + assert second.outcome is DispatchOutcome.SKIPPED + + async def test_indeterminate_launch_retains_active_slot(self): + """The port-contract form of main's malformed-launch-result case: the + adapter knows the launch side effect may have happened but cannot + decode the run's identity. The slot must be retained with the identity + unknown, never released via the LaunchFailedError path.""" + runs = InMemoryScheduledRunRepository() + launcher = FakeRunLauncher(fail_with=LaunchIndeterminateError("launch result undecodable")) + service = make_service(runs=runs, launcher=launcher) + task = await create_cron_task(service) + + first = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + + assert first.outcome is DispatchOutcome.LAUNCHED + assert first.run_id is None, "the identity is unknown, not invented" + assert "launch result undecodable" in first.error + + stored = runs.all_runs()[0] + assert stored.status is RunStatus.RUNNING + assert stored.run_id is None + assert stored.is_active + + launcher.fail_with = None + second = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + assert len(launcher.calls) == 1, "a duplicate launch is the #4452 bug" + assert second.outcome is DispatchOutcome.SKIPPED + + # The execution is counted and the task keeps no stale identity. + after = await service.get_task(task.task_id, user_id="user-1") + assert after.last_run_id is None + assert after.run_count == 1 + + async def test_pre_launch_failure_still_releases_active_slot(self): + """Complement pinning the boundary: LaunchFailedError means the + adapter is CERTAIN no run started, so the failure path stays -- the + row goes terminal and the slot is released for the next dispatch.""" + runs = InMemoryScheduledRunRepository() + launcher = FakeRunLauncher(fail_with=LaunchFailedError("provider exploded")) + service = make_service(runs=runs, launcher=launcher) + task = await create_cron_task(service) + + first = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + assert first.outcome is DispatchOutcome.FAILED + assert not runs.all_runs()[0].is_active, "a certain failure must not hold the slot" + + launcher.fail_with = None + second = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) + assert second.outcome is DispatchOutcome.LAUNCHED + assert len(launcher.calls) == 2 + + class TestConflictCollapse: """The fast path and the active-slot rejection must be indistinguishable.