From f7533c00e5beb2360720254166adcada1d769755 Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 11:43:37 +0700 Subject: [PATCH] sched: checkpoint-first reconcile across torn checkpoint/summary writes (F2) --- src/wf_scheduling/recovery.py | 66 +++- tests/scheduling/controlled.py | 21 ++ tests/scheduling/test_torn_reconcile.py | 454 ++++++++++++++++++++++++ 3 files changed, 538 insertions(+), 3 deletions(-) create mode 100644 tests/scheduling/test_torn_reconcile.py diff --git a/src/wf_scheduling/recovery.py b/src/wf_scheduling/recovery.py index b978837b..ef91d3dd 100644 --- a/src/wf_scheduling/recovery.py +++ b/src/wf_scheduling/recovery.py @@ -20,9 +20,11 @@ clear → executor → stopped persist → executing clear → history): - any admitted run with the executing mark: the executor may already have produced external effects; failed (ACTIVE attempt: ambiguous, else abandoned) with external-effects disclosure, never redispatched. -- admitted + view with neither mark and no ACTIVE attempt: legacy or - manually cleared state whose outcome is unprovable; failed closed - without replay (current code never produces this shape on crash). +- admitted + view with neither mark and no ACTIVE attempt: a durable + stopped checkpoint reconciles the summary first (checkpoint-first rule); + only when no checkpoint exists is the outcome unprovable, and the run is + failed closed without replay (current code never produces this shape on + crash). - stopped summary + executing/pending marks: completion-window leftovers (stopped result persisted, marker clearing lost); markers are cleared, the stopped status stands, nothing is re-executed. Checkpoint-vs-summary @@ -94,6 +96,21 @@ def recover( status = getattr(run.status, "value", run.status) attempt = run_store.get_resume_attempt(run.id) active = attempt is not None and attempt.state == "ACTIVE" + # Checkpoint-first reconcile across the torn + # save_checkpoint/save_run boundary: a durable stopped checkpoint + # wins over a stale summary (status, checkpoint pointer, and + # readiness are rewritten together from the checkpoint's own + # persisted fields; nothing is fabricated). Attempt matching below + # still distinguishes fresh results from stale ones. + try: + changed, status = _reconcile_summary_from_checkpoint(run_store, run, now) + except (ValueError, OSError) as exc: + _fail_run(run_store, run, f"corrupt checkpoint: {exc}", now, history) + diags.append(f"{run.id}:failed-closed") + continue + if changed: + run = run_store.get_run(run.id) + diags.append(f"{run.id}:summary-reconciled") if status in ( StoredRunStatus.INTERRUPTED.value, StoredRunStatus.COMPLETED.value, @@ -217,6 +234,49 @@ def recover( return diags +def _reconcile_summary_from_checkpoint( + run_store: Any, run: Any, now: datetime +) -> tuple[bool, str]: + """Rewrite a stale summary from its durable stopped checkpoint. + + Returns ``(changed, status)``. Only the checkpoint's own persisted + fields (status, checkpoint id, readiness derived exactly as + ``persist_stopped_run`` derives it) are copied; creation time, + environment, and diagnostics are preserved. A run without checkpoints + is untouched (an admitted summary with no checkpoint is genuinely + undispatched-or-unknown, handled by the marker rules). + """ + from wf_artifacts.runs.models import ResumeReadiness, StoredRunStatus + + try: + latest = run_store.get_latest_checkpoint(run.id) + except KeyError: + return False, getattr(run.status, "value", run.status) + expected = StoredRunStatus(latest.reason.value) + readiness = ( + ResumeReadiness.READY + if expected is StoredRunStatus.INTERRUPTED + else ResumeReadiness.NOT_APPLICABLE + ) + if ( + run.status == expected + and run.latest_checkpoint_id == latest.id + and run.resume_readiness == readiness + ): + return False, getattr(run.status, "value", run.status) + run_store.save_run( + run.model_copy( + update={ + "status": expected, + "latest_checkpoint_id": latest.id, + "resume_readiness": readiness, + "updated_at": now, + } + ) + ) + return True, expected.value + + def _attribution( run_store: Any, run_id: str ) -> tuple[str | None, datetime | None, int | None]: diff --git a/tests/scheduling/controlled.py b/tests/scheduling/controlled.py index eee9b4d7..8788d0bf 100644 --- a/tests/scheduling/controlled.py +++ b/tests/scheduling/controlled.py @@ -88,3 +88,24 @@ class ScriptedDispatcher: def finish(self, admission: RunAdmission, outcome: str) -> RunState: """Build the stopped state that settles a hanging run.""" return stopped_state(admission, outcome) + + +class WorkflowDispatcher: + """Execution double driving a real workflow to its first stopped state. + + Dispatch runs the workflow through the genuine ``wf_core`` runtime and + returns the resulting stopped state, so scheduler tests exercise real + execution (including durable interruptions) without enabling any server. + """ + + def __init__(self, workflow: Any, registry: dict[str, Any] | None = None) -> None: + self.workflow = workflow + self.registry = registry or {} + + def dispatch(self, *, admission: RunAdmission, now: datetime) -> DispatchResult: + from wf_core import execute_workflow + + state = execute_workflow( + self.workflow, dict(admission.resolved_input), self.registry + ) + return Stopped(result=state) diff --git a/tests/scheduling/test_torn_reconcile.py b/tests/scheduling/test_torn_reconcile.py new file mode 100644 index 00000000..44a0ed57 --- /dev/null +++ b/tests/scheduling/test_torn_reconcile.py @@ -0,0 +1,454 @@ +"""Torn checkpoint/summary reconciliation: the checkpoint wins (R4/F2). + +``persist_stopped_run`` writes the checkpoint before the summary; a crash +in between leaves a stale summary. Recovery inspects and validates the +durable stopped checkpoint before deciding from the summary, reconciling +status, checkpoint pointer, readiness, and attempt state together — for +first executions and resumes, matching and mismatching attempts, without +fabricating execution output. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any, cast + +from tests.artifacts.test_run_store import artifact as _artifact +from tests.artifacts.test_run_store import deployment as _deployment +from tests.scheduling.controlled import ( + DictDeployments, + WorkflowDispatcher, + fixture_environment, +) +from wf_api.run_lifecycle import ( + load_stored_run, + materialize_admitted_view, + persist_admission, + persist_stopped_run, + restore_interrupted_run, +) +from wf_artifacts import PinnedRunEnvironment +from wf_artifacts.runs.models import ResumeAttempt +from wf_artifacts.runs.store import FileRunStore +from wf_core import ( + END, + Edge, + InterruptNode, + RunState, + RunStatus, + SchemaRef, + StateSchema, + Workflow, + load_run_state, + resume_workflow, +) +from wf_scheduling import recovery as sched_recovery +from wf_scheduling.models import Schedule +from wf_scheduling.ownership import SchedulerOwnership +from wf_scheduling.poll import Scheduler +from wf_scheduling.prepare import SchedulePreparer +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 _env() -> PinnedRunEnvironment: + return PinnedRunEnvironment( + deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[] + ) + + +class FailSaveRunOnce(FileRunStore): + """Real store that drops exactly one summary write (torn boundary).""" + + def __init__(self, root: Path) -> None: + super().__init__(root) + self.armed = False + + def save_run(self, run: Any) -> None: + if self.armed: + self.armed = False + raise OSError("injected save_run failure") + super().save_run(run) + + +def _recover( + sched_store: FileScheduleStore, run_store: FileRunStore, now: datetime +) -> list[str]: + ownership = SchedulerOwnership(sched_store.root, owner="test").acquire() + try: + return sched_recovery.recover( + schedule_store=sched_store, + run_store=run_store, + now=now, + ownership=ownership, + ) + finally: + ownership.release() + + +def _admit_view( + run_store: FileRunStore, + intended: datetime, + *, + schedule_id: str = "a", +) -> str: + run_id = run_store.allocate_run_id() + admission = persist_admission( + store=run_store, + run_id=run_id, + environment=_env(), + resolved_input={}, + max_steps=None, + scheduled_at=intended, + schedule_id=schedule_id, + schedule_revision=1, + ) + materialize_admitted_view(store=run_store, admission=admission) + return run_id + + +def _torn_persist( + run_store: FailSaveRunOnce, + run_id: str, + status: RunStatus, + *, + attempt_id: int | None = None, +) -> None: + record = run_store.get_run(run_id) + run_store.armed = True + try: + persist_stopped_run( + store=run_store, + environment=record.environment, + run=RunState( + workflow_name="sched", status=status, workflow_input={}, state={} + ), + run_id=run_id, + attempt_id=attempt_id, + ) + raise AssertionError("save_run failure must propagate") + except OSError as exc: + assert "injected save_run failure" in str(exc) + + +def _entries(store: FileScheduleStore, sid: str, kind: str) -> list[dict[str, Any]]: + page = store.list_occurrences(sid, limit=100) + return [ + r for r in cast(list[dict[str, Any]], page["occurrences"]) if r["kind"] == kind + ] + + +def _double_interrupt_workflow() -> Workflow: + schema = SchemaRef(type="object", properties={}) + return Workflow( + name="ask_twice", + input_schema=schema, + state_schema=StateSchema.from_field_map({}), + output_schema=schema, + outcomes=["ok"], + start="ask1", + node_defs=[], + nodes=[ + InterruptNode(id="ask1", type="interrupt", kind="approval"), + InterruptNode(id="ask2", type="interrupt", kind="approval"), + ], + edges=[ + Edge.model_validate({"from": "ask1", "outcome": "submitted", "to": "ask2"}), + Edge.model_validate({"from": "ask2", "outcome": "submitted", "to": END}), + ], + ) + + +def test_torn_first_interrupt_reconciles_not_abandons(tmp_path: Path) -> None: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FailSaveRunOnce(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_view(run_store, intended) + _torn_persist(run_store, run_id, RunStatus.INTERRUPTED) + assert run_store.get_run(run_id).status.value == "admitted" + diags = _recover(sched_store, run_store, intended) + assert any("summary-reconciled" in d for d in diags) + assert any("waiting-resumable" in d for d in diags) + assert not any("failed-closed" in d for d in diags) + record = run_store.get_run(run_id) + assert record.status.value == "interrupted" + assert record.latest_checkpoint_id == f"{run_id}.000001" + assert record.resume_readiness.value == "ready" + assert len(_entries(sched_store, "a", "interrupted")) == 1 + # Inspection and resume eligibility agree with the reconciled result. + _, state = load_stored_run(run_store, run_id) + assert state.status is RunStatus.INTERRUPTED + settled, resumed = restore_interrupted_run(run_store, run_id) + assert settled.resume_readiness.value == "ready" + assert resumed.status is RunStatus.INTERRUPTED + + +def test_torn_first_completed_reconciles(tmp_path: Path) -> None: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FailSaveRunOnce(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_view(run_store, intended) + _torn_persist(run_store, run_id, RunStatus.COMPLETED) + diags = _recover(sched_store, run_store, intended) + record = run_store.get_run(run_id) + assert record.status.value == "completed" + assert record.latest_checkpoint_id == f"{run_id}.000001" + assert record.resume_readiness.value == "not_applicable" + assert any("summary-reconciled" in d for d in diags) + assert len(_entries(sched_store, "a", "completed")) == 1 + + +def test_torn_first_failed_reconciles(tmp_path: Path) -> None: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FailSaveRunOnce(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_view(run_store, intended) + _torn_persist(run_store, run_id, RunStatus.FAILED) + diags = _recover(sched_store, run_store, intended) + record = run_store.get_run(run_id) + assert record.status.value == "failed" + assert record.latest_checkpoint_id == f"{run_id}.000001" + assert any("summary-reconciled" in d for d in diags) + assert len(_entries(sched_store, "a", "failed")) == 1 + + +def test_torn_resume_completed_matches_active_attempt(tmp_path: Path) -> None: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FailSaveRunOnce(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_view(run_store, intended) + record = run_store.get_run(run_id) + persist_stopped_run( + store=run_store, + environment=record.environment, + run=RunState( + workflow_name="sched", + status=RunStatus.INTERRUPTED, + workflow_input={}, + state={}, + ), + run_id=run_id, + ) + attempt_id = run_store.allocate_resume_attempt_id() + run_store.save_resume_attempt( + ResumeAttempt( + run_id=run_id, + attempt_id=attempt_id, + state="ACTIVE", + created_at=intended, + updated_at=intended, + ) + ) + _torn_persist(run_store, run_id, RunStatus.COMPLETED, attempt_id=attempt_id) + diags = _recover(sched_store, run_store, intended) + # The fresh COMPLETED checkpoint wins over the stale INTERRUPTED summary: + # reconciled as completed, never as fresh/resumable. + record = run_store.get_run(run_id) + assert record.status.value == "completed" + assert record.latest_checkpoint_id == f"{run_id}.000002" + assert run_store.get_resume_attempt(run_id).state == "DONE" # type: ignore[union-attr] + assert any("summary-reconciled" in d for d in diags) + assert not any("fresh-result-resumable" in d for d in diags) + assert len(_entries(sched_store, "a", "completed")) == 1 + + +def test_torn_resume_interrupted_matches_active_attempt(tmp_path: Path) -> None: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FailSaveRunOnce(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_view(run_store, intended) + record = run_store.get_run(run_id) + persist_stopped_run( + store=run_store, + environment=record.environment, + run=RunState( + workflow_name="sched", + status=RunStatus.INTERRUPTED, + workflow_input={}, + state={}, + ), + run_id=run_id, + ) + attempt_id = run_store.allocate_resume_attempt_id() + run_store.save_resume_attempt( + ResumeAttempt( + run_id=run_id, + attempt_id=attempt_id, + state="ACTIVE", + created_at=intended, + updated_at=intended, + ) + ) + _torn_persist(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=attempt_id) + diags = _recover(sched_store, run_store, intended) + record = run_store.get_run(run_id) + assert record.status.value == "interrupted" + assert record.latest_checkpoint_id == f"{run_id}.000002" + assert record.resume_readiness.value == "ready" + assert run_store.get_resume_attempt(run_id).state == "DONE" # type: ignore[union-attr] + assert any("fresh-result-resumable" in d for d in diags) + assert len(_entries(sched_store, "a", "interrupted")) == 1 + + +def test_stale_checkpoint_never_clears_active_attempt(tmp_path: Path) -> None: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_view(run_store, intended) + record = run_store.get_run(run_id) + persist_stopped_run( + store=run_store, + environment=record.environment, + run=RunState( + workflow_name="sched", + status=RunStatus.INTERRUPTED, + workflow_input={}, + state={}, + ), + run_id=run_id, + attempt_id=3, + ) + run_store.save_resume_attempt( + ResumeAttempt( + run_id=run_id, + attempt_id=5, + state="ACTIVE", + created_at=intended, + updated_at=intended, + ) + ) + diags = _recover(sched_store, run_store, intended) + assert any("failed-closed" in d for d in diags) + assert run_store.get_run(run_id).status.value == "failed" + assert run_store.get_resume_attempt(run_id).state == "ACTIVE" # type: ignore[union-attr] + + +def test_double_recovery_after_reconcile_is_stable(tmp_path: Path) -> None: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FailSaveRunOnce(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_view(run_store, intended) + _torn_persist(run_store, run_id, RunStatus.INTERRUPTED) + first = _recover(sched_store, run_store, intended) + assert any("summary-reconciled" in d for d in first) + before = run_store.get_run(run_id) + second = _recover(sched_store, run_store, intended + timedelta(minutes=1)) + after = run_store.get_run(run_id) + assert not any("summary-reconciled" in d for d in second) + assert not any("terminal-reconciled" in d for d in second) + assert after == before + assert len(_entries(sched_store, "a", "interrupted")) == 1 + + +def test_real_workflow_interrupt_resume_interrupt_with_torn_resume( + tmp_path: Path, +) -> None: + workflow = _double_interrupt_workflow() + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FailSaveRunOnce(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched = Scheduler( + schedule_store=sched_store, + run_store=run_store, + sources={}, + capacity=4, + preparer=SchedulePreparer( + DictDeployments({"dep-1": {"rev": 1, "required": []}}), + fixture_environment, + ), + dispatcher=WorkflowDispatcher(workflow, {}), + ownership=ownership, + ) + sched_store.save_consumed("a", intended - timedelta(hours=1)) + from wf_scheduling.calendar import OneShotSource + + sched.sources["a"] = OneShotSource(intended) + result = sched.poll(intended) + run_id = next(iter(run_store.list_runs())).id + assert result == {"a": f"admit:{run_id}"} + assert run_store.get_run(run_id).status.value == "interrupted" + ownership.release() + # Resume through the real runtime to a second durable interruption, + # then tear the summary write. + fresh = FileRunStore(tmp_path / "runs") + checkpoint = fresh.get_latest_checkpoint(run_id) + st1 = load_run_state(checkpoint.state.model_dump(mode="json")) + st2 = resume_workflow(workflow, st1, {}, resume_payload={}) + assert st2.status is RunStatus.INTERRUPTED + attempt_id = fresh.allocate_resume_attempt_id() + fresh.save_resume_attempt( + ResumeAttempt( + run_id=run_id, + attempt_id=attempt_id, + state="ACTIVE", + created_at=intended, + updated_at=intended, + ) + ) + armed = FailSaveRunOnce(tmp_path / "runs") + armed.armed = True + record = armed.get_run(run_id) + try: + persist_stopped_run( + store=armed, + environment=record.environment, + run=st2, + run_id=run_id, + attempt_id=attempt_id, + ) + raise AssertionError("torn resume persist must raise") + except OSError as exc: + assert "injected save_run failure" in str(exc) + diags = _recover(sched_store, FileRunStore(tmp_path / "runs"), intended) + assert any("summary-reconciled" in d for d in diags) + assert any("fresh-result-resumable" in d for d in diags) + settled = FileRunStore(tmp_path / "runs").get_run(run_id) + assert settled.status.value == "interrupted" + assert settled.latest_checkpoint_id == f"{run_id}.000002" + assert settled.resume_readiness.value == "ready" + _, state = load_stored_run(FileRunStore(tmp_path / "runs"), run_id) + assert state.status is RunStatus.INTERRUPTED + finally: + ownership.release() + + +def test_corrupt_checkpoint_fails_closed(tmp_path: Path) -> None: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_view(run_store, intended) + checkpoint_dir = tmp_path / "runs" / run_id / "checkpoints" + checkpoint_dir.mkdir(parents=True, exist_ok=True) + (checkpoint_dir / "000001.json").write_text("not json", encoding="utf-8") + diags = _recover(sched_store, run_store, intended) + assert any("failed-closed" in d for d in diags) + assert run_store.get_run(run_id).status.value == "failed"