diff --git a/src/wf_scheduling/lifecycle.py b/src/wf_scheduling/lifecycle.py index 0965e6cb..2cde31cd 100644 --- a/src/wf_scheduling/lifecycle.py +++ b/src/wf_scheduling/lifecycle.py @@ -143,8 +143,9 @@ class RuntimeDispatcher: persists through the shared lifecycle boundary and records terminal history exactly once). Anything unexpected — an execution failure or a settlement failure such as a torn write — goes through the - service-provided ``abandon`` hook, which reuses startup recovery to - fail the run closed without replay. Cancellation (shutdown drain) is + service-provided ``abandon`` hook, which runs the same per-run + reconciliation as startup recovery but scoped to the broken run id: + sibling executions are never touched. Cancellation (shutdown drain) is re-raised unsettled: the durable executing mark is left for startup recovery, which abandons it truthfully. """ @@ -495,13 +496,19 @@ class SchedulerService: self._settled += 1 def _abandon(self, run_id: str, error: BaseException) -> None: - """Fail a dispatcher-broken run closed through startup recovery. + """Fail one dispatcher-broken run closed through scoped recovery. The run holds a durable executing mark with no genuine stopped - result — exactly the crashed-process shape — so the same recovery - that owns restart abandonment owns it here: no replay, no - fabricated outcome, slot freed. Recovery failures are recorded; - the run then waits for the next restart recovery. + result — exactly the crashed-process shape — so the same per-run + reconciliation that owns restart abandonment owns it here: no + replay, no fabricated outcome, slot freed. The repair is scoped + to this run id only: healthy sibling executions keep their + markers, statuses, and histories, and a late genuine result for + a sibling still settles normally. A torn settlement (the stopped + result was persisted but history/marker writes were lost) is + reconciled as the genuine stopped result, not as a failure. + Recovery failures are recorded; the run then waits for the next + restart recovery. """ with self._lock: self._errors.append(f"{run_id}: {error}") @@ -509,14 +516,33 @@ class SchedulerService: try: from wf_scheduling import recovery as sched_recovery - sched_recovery.recover( + diags = sched_recovery.recover( schedule_store=self.schedule_store, run_store=self.run_store, now=self.clock(), ownership=self.ownership, + only_run_id=run_id, ) except Exception as exc: self._errors.append(f"{run_id}: recovery failed: {exc}") del self._errors[: max(0, len(self._errors) - 20)] return - self._abandoned += 1 + if f"{run_id}:failed-closed" in diags: + self._abandoned += 1 + elif any( + diag.startswith(f"{run_id}:") + and diag.split(":", 1)[1] + in { + "terminal-reconciled", + "completion-window-cleared", + "summary-reconciled", + "attempt-reconciled", + "fresh-result-resumable", + } + for diag in diags + ): + # Scoped repair found a genuine stopped result (torn + # settlement): the run settled, it was not abandoned. + self._settled += 1 + else: + self._abandoned += 1 diff --git a/src/wf_scheduling/recovery.py b/src/wf_scheduling/recovery.py index ed3ccb9e..8a893e93 100644 --- a/src/wf_scheduling/recovery.py +++ b/src/wf_scheduling/recovery.py @@ -91,8 +91,18 @@ def recover( now: datetime, ownership: SchedulerOwnership, history: HistoryRecorder | None = None, + only_run_id: str | None = None, ) -> list[str]: - """Reconcile durable state after a restart without executing work.""" + """Reconcile durable state after a restart without executing work. + + With ``only_run_id``, repair exactly one live run instead of the whole + store: a dispatcher failure (or a torn settlement) for that run is + reconciled with the same checkpoint authority and torn-write rules, + while healthy sibling executions and active resumes are never + touched. Whole-store abandonment stays confined to genuine startup + recovery (``only_run_id=None``), where an executing mark truthfully + means the owner is gone. + """ from wf_artifacts.runs.models import StoredRunStatus sched_root = getattr(schedule_store, "root", None) @@ -109,19 +119,37 @@ def recover( if history is None: history = FileScheduleHistoryRecorder(schedule_store) diags: list[str] = [] - # Admission record is the recovery authority: admitted but never - # materialized views are completed here and flagged pending for the - # capacity-checked poll sweep (exactly once, occurrence already consumed). - for admission in run_store.list_admissions(): + if only_run_id is not None: + # Scoped live repair: never touch admissions, views, markers, or + # histories belonging to any other run. The live poll sweep owns + # pending dispatch; this path only reconciles the one broken run. try: - run_store.get_run(admission.id) + runs = [run_store.get_run(only_run_id)] except KeyError: + try: + admission = run_store.get_admission(only_run_id) + except KeyError: + return [f"{only_run_id}:run-unknown"] from wf_api.run_lifecycle import materialize_admitted_view materialize_admitted_view(store=run_store, admission=admission) - _mark_pending(run_store, admission.id) - diags.append(f"{admission.id}:view-completed-pending-dispatch") - for run in run_store.list_runs(): + _mark_pending(run_store, only_run_id) + return [f"{only_run_id}:view-completed-pending-dispatch"] + else: + # Admission record is the recovery authority: admitted but never + # materialized views are completed here and flagged pending for the + # capacity-checked poll sweep (exactly once, occurrence already consumed). + for admission in run_store.list_admissions(): + try: + run_store.get_run(admission.id) + except KeyError: + from wf_api.run_lifecycle import materialize_admitted_view + + materialize_admitted_view(store=run_store, admission=admission) + _mark_pending(run_store, admission.id) + diags.append(f"{admission.id}:view-completed-pending-dispatch") + runs = run_store.list_runs() + for run in runs: status = getattr(run.status, "value", run.status) attempt = run_store.get_resume_attempt(run.id) active = attempt is not None and attempt.state == "ACTIVE" diff --git a/tests/scheduling/test_lifecycle.py b/tests/scheduling/test_lifecycle.py index f5a05f3a..1ea95076 100644 --- a/tests/scheduling/test_lifecycle.py +++ b/tests/scheduling/test_lifecycle.py @@ -31,6 +31,7 @@ from wf_scheduling.lifecycle import ( ) from wf_scheduling.models import Schedule from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError +from wf_scheduling.poll import Scheduler from wf_scheduling.store import FileScheduleStore @@ -556,3 +557,183 @@ async def test_store_deployment_directory_contract(tmp_path: Path) -> None: directory.deployment_revision("missing.deployment") with pytest.raises(KeyError): directory.required_inputs("missing.deployment") + + +async def _poll_two( + service: SchedulerService, intended: datetime +) -> tuple[str, str]: + service.schedule_store.create_schedule(_sched_model("a", intended)) + service.schedule_store.create_schedule(_sched_model("b", intended)) + result = await service.poll_once(intended + timedelta(seconds=1)) + hang_id = result["a"].removeprefix("admit:") + sib_id = result["b"].removeprefix("admit:") + assert hang_id.startswith("run-") and sib_id.startswith("run-") + return hang_id, sib_id + + +def _assert_sibling_live(service: SchedulerService, hang_id: str) -> None: + """The healthy sibling keeps its markers, status, and clean history.""" + sibling = service.run_store.get_run(hang_id) + assert sibling.status.value == "admitted" + assert service.run_store.is_executing(hang_id) + assert not service.run_store.is_pending_dispatch(hang_id) + assert len(_entries(service.schedule_store, "a", "failed")) == 0 + + +async def test_live_runtime_failure_spares_healthy_sibling( + tmp_path: Path, +) -> None: + """A live execution failure repairs only the broken run (B1). + + Whole-store recovery during live execution used to fail the still + hanging sibling too, whose late success was then rejected as + ``settle non-admitted run``. + """ + intended = ts(2026, 9, 8, 12, 0) + runtime = ScriptedRuntime(lambda n: "hang" if n == 1 else "raise") + service = _service( + tmp_path, + runtime, + SchedulerServiceConfig(poll_interval_s=0.01, capacity=2, auto_tick=False), + ) + try: + await service.start() + hang_id, sib_id = await _poll_two(service, intended) + await _wait_for(lambda: runtime.started.is_set()) + await _wait_for( + lambda: service.run_store.get_run(sib_id).status.value == "failed" + ) + failed = service.run_store.get_run(sib_id) + assert failed.status.value == "failed" + assert not service.run_store.is_executing(sib_id) + assert not service.run_store.is_pending_dispatch(sib_id) + assert len(_entries(service.schedule_store, "b", "failed")) == 1 + _assert_sibling_live(service, hang_id) + runtime.release.set() + await _wait_for(lambda: service.live_executions == 0) + assert service.run_store.get_run(hang_id).status.value == "completed" + assert len(_entries(service.schedule_store, "a", "completed")) == 1 + assert not any("settle non-admitted run" in m for m in service.errors) + finally: + runtime.release.set() + report = await service.stop() + assert report.settled == 1 + assert report.abandoned == 1 + + +async def test_live_settlement_failure_spares_healthy_sibling( + tmp_path: Path, monkeypatch: Any +) -> None: + """A torn settle write repairs only the broken run (B1). + + The settlement seam raises after genuine execution for one run while + its sibling hangs: only that run fails closed, and the sibling still + settles exactly once with its capacity slot freed. + """ + intended = ts(2026, 9, 8, 12, 0) + runtime = ScriptedRuntime(lambda n: "hang" if n == 1 else "complete") + service = _service( + tmp_path, + runtime, + SchedulerServiceConfig(poll_interval_s=0.01, capacity=2, auto_tick=False), + ) + try: + await service.start() + original = Scheduler.record_stopped_execution + fired = {"done": False} + + def _flaky_settle(self: Any, run_id: str, state: Any, now: datetime) -> None: + try: + owned_by_b = ( + self.run_store.get_admission(run_id).schedule_id == "b" + ) + except KeyError: + owned_by_b = False + if owned_by_b and not fired["done"]: + fired["done"] = True + raise RuntimeError("injected settlement failure") + original(self, run_id, state, now) + + monkeypatch.setattr(Scheduler, "record_stopped_execution", _flaky_settle) + hang_id, sib_id = await _poll_two(service, intended) + await _wait_for(lambda: runtime.started.is_set()) + await _wait_for( + lambda: service.run_store.get_run(sib_id).status.value == "failed" + ) + assert fired["done"] + assert not service.run_store.is_executing(sib_id) + assert len(_entries(service.schedule_store, "b", "failed")) == 1 + _assert_sibling_live(service, hang_id) + runtime.release.set() + await _wait_for(lambda: service.live_executions == 0) + assert service.run_store.get_run(hang_id).status.value == "completed" + assert len(_entries(service.schedule_store, "a", "completed")) == 1 + assert not any("settle non-admitted run" in m for m in service.errors) + finally: + runtime.release.set() + report = await service.stop() + assert report.settled == 1 + assert report.abandoned == 1 + + +async def test_live_torn_settlement_reconciles_genuine_result( + tmp_path: Path, monkeypatch: Any +) -> None: + """A torn history write keeps the genuine stopped result (B1). + + The stopped result is durably persisted but the history append is + lost: scoped repair must reconcile the genuine completion (with + checkpoint authority), never fail the run, and never touch the + hanging sibling. + """ + intended = ts(2026, 9, 8, 12, 0) + runtime = ScriptedRuntime(lambda n: "hang" if n == 1 else "complete") + service = _service( + tmp_path, + runtime, + SchedulerServiceConfig(poll_interval_s=0.01, capacity=2, auto_tick=False), + ) + try: + await service.start() + original = Scheduler._record + fired = {"done": False} + + def _flaky_record(self: Any, **kwargs: Any) -> None: + run_id = kwargs.get("run_id") + try: + owned_by_b = ( + run_id is not None + and self.run_store.get_admission(run_id).schedule_id == "b" + ) + except KeyError: + owned_by_b = False + if ( + kwargs.get("kind") == "completed" + and owned_by_b + and not fired["done"] + ): + fired["done"] = True + raise RuntimeError("injected history failure") + original(self, **kwargs) + + monkeypatch.setattr(Scheduler, "_record", _flaky_record) + hang_id, sib_id = await _poll_two(service, intended) + await _wait_for(lambda: runtime.started.is_set()) + await _wait_for( + lambda: len(_entries(service.schedule_store, "b", "completed")) == 1 + ) + assert fired["done"] + repaired = service.run_store.get_run(sib_id) + assert repaired.status.value == "completed" + assert not service.run_store.is_executing(sib_id) + assert len(_entries(service.schedule_store, "b", "failed")) == 0 + _assert_sibling_live(service, hang_id) + runtime.release.set() + await _wait_for(lambda: service.live_executions == 0) + assert service.run_store.get_run(hang_id).status.value == "completed" + assert len(_entries(service.schedule_store, "a", "completed")) == 1 + finally: + runtime.release.set() + report = await service.stop() + assert report.settled == 2 + assert report.abandoned == 0