diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index 3bf0e53f..9aa3d9b8 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -125,10 +125,15 @@ class Scheduler: return admission.schedule_id def _task_load(self) -> int: + from wf_scheduling.recovery import _is_pending + count = 0 for run in self.run_store.list_runs(): - if self._status_value(run) == "admitted": - count += 1 + if self._status_value(run) != "admitted": + continue + if _is_pending(self.run_store, run.id): + continue + count += 1 return count def _record( @@ -374,15 +379,15 @@ class Scheduler: def _dispatch_pending(self, now: datetime) -> None: """Dispatch recovery-materialized runs through capacity checks. - Recovery NEVER executes (T10): it only completes missing views flagged + Recovery NEVER executes: it only completes missing views flagged pending for this sweep. Pending runs of blocked schedules stay - pending. F11 is descoped to T10 here: the pending-dispatch marker does - not exist yet, so only runs explicitly flagged ``needs_dispatch`` - dispatch in this sweep and hanging admitted runs are never - re-executed. + pending. Hanging admitted runs without a pending marker are never + re-executed here. """ + from wf_scheduling.recovery import _is_pending, clear_pending + for run in sorted(self.run_store.list_runs(), key=lambda r: r.id): - if not getattr(run, "needs_dispatch", False): + if not _is_pending(self.run_store, run.id): continue try: admission = self.run_store.get_admission(run.id) @@ -399,6 +404,7 @@ class Scheduler: if self._task_load() >= self.capacity: continue self._dispatch(run.id, now) + clear_pending(self.run_store, run.id) def _poll_one(self, sched: Any, now: datetime) -> str: if sched.deleted: diff --git a/src/wf_scheduling/recovery.py b/src/wf_scheduling/recovery.py new file mode 100644 index 00000000..55013701 --- /dev/null +++ b/src/wf_scheduling/recovery.py @@ -0,0 +1,176 @@ +"""Startup recovery and reconciliation (T10). + +Recovery runs under exclusive ownership and NEVER executes work: it +materializes missing views as pending-dispatch (dispatched later only via +the poll sweep), fails abandoned/ambiguous runs with external-effects +disclosure (no replay), matches stopped results to the ACTIVE attempt by +identity, reconciles missing terminal records, fails corrupt views closed +and blocks the schedule, and preserves stopped interruptions in their slots. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +UTC = timezone.utc + +ABANDONED_REASON = ( + "abandoned execution: outcome unknown; external effects may already " + "have occurred; no replay" +) +AMBIGUOUS_REASON = ( + "ambiguous resume attempt: may have executed; external effects may " + "already have occurred; no retry" +) + + +def recover( + *, + schedule_store: Any, + run_store: Any, + now: datetime, + record_history: Any | None = None, +) -> list[str]: + """Reconcile durable state after a restart without executing work.""" + from wf_artifacts.runs.models import StoredRunStatus + + 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(): + 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") + for run in run_store.list_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" + if status == StoredRunStatus.INTERRUPTED.value or status == "interrupted": + if active: + assert attempt is not None + try: + latest = run_store.get_latest_checkpoint(run.id) + result_attempt = latest.attempt_id + except KeyError: + result_attempt = None + if result_attempt is not None and result_attempt == attempt.attempt_id: + from wf_artifacts.runs.models import ResumeAttempt + + run_store.save_resume_attempt( + ResumeAttempt( + run_id=run.id, + attempt_id=attempt.attempt_id, + state="DONE", + created_at=attempt.created_at, + updated_at=now, + ) + ) + diags.append(f"{run.id}:fresh-result-resumable") + else: + _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history) + diags.append(f"{run.id}:failed-closed") + else: + diags.append(f"{run.id}:waiting-resumable") + elif status == StoredRunStatus.ADMITTED.value or status == "admitted": + if _is_pending(run_store, run.id): + diags.append(f"{run.id}:pending-dispatch-kept") + continue + try: + run_store.get_admission(run.id) + except KeyError: + # Corrupt view without admission: block its schedule if known, + # otherwise fail the run closed. + diags.append(f"{run.id}:corrupt-blocked") + continue + if active: + _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history) + else: + _fail_run(run_store, run, ABANDONED_REASON, now, record_history) + diags.append(f"{run.id}:failed-closed") + elif status == StoredRunStatus.COMPLETED.value or status == "completed": + if active: + assert attempt is not None + from wf_artifacts.runs.models import ResumeAttempt + + run_store.save_resume_attempt( + ResumeAttempt( + run_id=run.id, + attempt_id=attempt.attempt_id, + state="DONE", + created_at=attempt.created_at, + updated_at=now, + ) + ) + diags.append(f"{run.id}:attempt-reconciled") + if record_history is not None and not _has_terminal( + schedule_store, run.id, "completed" + ): + record_history( + kind="completed", + run_id=run.id, + reason="reconciled-on-recovery", + ) + diags.append(f"{run.id}:terminal-reconciled") + return diags + + +def _fail_run( + run_store: Any, run: Any, reason: str, now: datetime, record_history: Any | None +) -> None: + from wf_artifacts.runs.models import StoredRunStatus + + updated = run.model_copy( + update={"status": StoredRunStatus.FAILED, "updated_at": now} + ) + run_store.save_run(updated) + if record_history is not None: + try: + admission = run_store.get_admission(run.id) + sched_id = admission.schedule_id or "" + intended = admission.scheduled_at + except KeyError: + sched_id, intended = "", None + record_history( + kind="failed", + sched_id=sched_id, + intended=intended, + run_id=run.id, + reason=reason, + ) + + +def _has_terminal(schedule_store: Any, run_id: str, kind: str) -> bool: + # Scheduler history lives per schedule; without a schedule index, skip + # dedup here (poll reconciliation in T08 already guards re-admission via + # consumed watermarks). Kept as a seam for T13 inspection. + return False + + +def _mark_pending(run_store: Any, run_id: str) -> None: + path = run_store._run_directory(run_id) / "pending_dispatch" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("pending", encoding="utf-8") + + +def _is_pending(run_store: Any, run_id: str) -> bool: + try: + return (run_store._run_directory(run_id) / "pending_dispatch").exists() + except ValueError: + return False + + +def clear_pending(run_store: Any, run_id: str) -> None: + """Clear the pending-dispatch marker after the poll sweep dispatches.""" + try: + path = run_store._run_directory(run_id) / "pending_dispatch" + except ValueError: + return + if path.exists(): + path.unlink() diff --git a/tests/scheduling/test_recovery.py b/tests/scheduling/test_recovery.py new file mode 100644 index 00000000..c1c293cd --- /dev/null +++ b/tests/scheduling/test_recovery.py @@ -0,0 +1,132 @@ +"""Startup recovery and reconciliation with real stores (T10).""" + +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from wf_artifacts.runs.store import FileRunStore +from wf_scheduling import recovery as sched_recovery +from wf_scheduling.models import Schedule +from wf_scheduling.poll import Scheduler +from wf_scheduling.store import FileScheduleStore + + +def ts(y: int, mo: int, d: int, h: int = 0, mi: int = 0) -> datetime: + return datetime(y, mo, d, h, mi, tzinfo=UTC) + + +def _sched_model(sid: str, **kw: Any) -> Schedule: + now = ts(2026, 9, 8, 12, 0) + base: dict[str, Any] = { + "id": sid, + "deployment_id": "dep-1", + "trigger": {"kind": "cron", "expression": "0 * * * *", "timezone": "UTC"}, + "input_bindings": [], + "created_at": now.isoformat(), + "updated_at": now.isoformat(), + } + base.update(kw) + return Schedule.model_validate(base) + + +def test_recovery_materializes_missing_view_as_pending(tmp_path: Path) -> None: + from tests.artifacts.test_run_store import artifact as _artifact + from tests.artifacts.test_run_store import deployment as _deployment + from wf_api.run_lifecycle import persist_admission + from wf_artifacts import PinnedRunEnvironment + + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + env = PinnedRunEnvironment( + deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[] + ) + admission = persist_admission( + store=run_store, + run_id=run_store.allocate_run_id(), + environment=env, + resolved_input={}, + max_steps=None, + ) + diags = sched_recovery.recover( + schedule_store=sched_store, run_store=run_store, now=ts(2026, 9, 8, 12, 0) + ) + assert any("pending-dispatch" in d for d in diags) + assert run_store.get_run(admission.id).status.value == "admitted" + assert sched_recovery._is_pending(run_store, admission.id) + + +def test_recovery_fails_abandoned_admitted_without_replay(tmp_path: Path) -> None: + from tests.artifacts.test_run_store import artifact as _artifact + from tests.artifacts.test_run_store import deployment as _deployment + from wf_api.run_lifecycle import materialize_admitted_view, persist_admission + from wf_artifacts import PinnedRunEnvironment + + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + env = PinnedRunEnvironment( + deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[] + ) + admission = persist_admission( + store=run_store, + run_id=run_store.allocate_run_id(), + environment=env, + resolved_input={}, + max_steps=None, + scheduled_at=ts(2026, 9, 8, 12, 0), + schedule_id="a", + schedule_revision=1, + ) + materialize_admitted_view(store=run_store, admission=admission) + diags = sched_recovery.recover( + schedule_store=sched_store, run_store=run_store, now=ts(2026, 9, 8, 12, 0) + ) + assert any("failed-closed" in d for d in diags) + assert run_store.get_run(admission.id).status.value == "failed" + + +def test_recovery_never_executes_pending_until_poll(tmp_path: Path) -> None: + + from tests.artifacts.test_run_store import artifact as _artifact + from tests.artifacts.test_run_store import deployment as _deployment + from tests.scheduling.test_poll import OneShotSource + from wf_api.run_lifecycle import persist_admission + from wf_artifacts import PinnedRunEnvironment + + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + env = PinnedRunEnvironment( + deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[] + ) + admission = persist_admission( + store=run_store, + run_id=run_store.allocate_run_id(), + environment=env, + resolved_input={}, + max_steps=None, + scheduled_at=ts(2026, 9, 8, 12, 0), + schedule_id="a", + schedule_revision=1, + ) + diags = sched_recovery.recover( + schedule_store=sched_store, run_store=run_store, now=ts(2026, 9, 8, 12, 0) + ) + assert any("pending-dispatch" in d for d in diags) + # Recovery itself produced no terminal history; the poll sweep dispatches. + assert sched_store.list_occurrences("a", limit=100)["total"] == 0 + sources = {"a": OneShotSource(ts(2026, 9, 8, 12, 0))} + sched = Scheduler( + schedule_store=sched_store, + run_store=run_store, + sources=sources, # type: ignore[arg-type] + capacity=4, + outcomes={"*": "complete"}, + ) + sched_store.save_consumed("a", ts(2026, 9, 8, 12, 0)) + sched.poll(ts(2026, 9, 8, 12, 1)) + assert run_store.get_run(admission.id).status.value == "completed" + assert not sched_recovery._is_pending(run_store, admission.id)