From 5a6056b144b4bb6c8635ce5c97a848f420418c3d Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 11:39:13 +0700 Subject: [PATCH] sched: typed history recorder with exactly-once terminal reconcile (F4) --- src/wf_scheduling/history.py | 115 +++++ src/wf_scheduling/models.py | 8 + src/wf_scheduling/poll.py | 71 ++- src/wf_scheduling/recovery.py | 253 ++++++++--- src/wf_scheduling/store.py | 23 + tests/scheduling/test_history_reconcile.py | 490 +++++++++++++++++++++ tests/scheduling/test_preparation.py | 1 + 7 files changed, 879 insertions(+), 82 deletions(-) create mode 100644 src/wf_scheduling/history.py create mode 100644 tests/scheduling/test_history_reconcile.py diff --git a/src/wf_scheduling/history.py b/src/wf_scheduling/history.py new file mode 100644 index 00000000..9b374662 --- /dev/null +++ b/src/wf_scheduling/history.py @@ -0,0 +1,115 @@ +"""Typed occurrence-history interface with idempotent reconciliation (R4/F4). + +Both the poll loop and startup recovery record through :class:`HistoryRecorder`, +so completed/failed/interrupted entries share one shape and one idempotency +identity: ``(run_id, kind, checkpoint_id)``. A resumed run that interrupts +again produces a new checkpoint id and therefore a new entry; repeating +recovery never duplicates an entry. The file-backed recorder derives entry +identity the same way the scheduler always has (occurrence instants hash to +``occurrence_id``; interval summaries use their span key). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Protocol + +from wf_scheduling.models import OccurrenceKind, OccurrenceRecord +from wf_scheduling.occurrences import occurrence_id + +UTC = timezone.utc + +TERMINAL_KINDS: tuple[str, str, str] = ("completed", "interrupted", "failed") + + +@dataclass(frozen=True) +class HistoryEntry: + """One occurrence-history entry with a stable idempotency identity.""" + + schedule_id: str + kind: OccurrenceKind + resolved_at: datetime | None = None + run_id: str | None = None + revision: int | None = None + reason: str = "" + admitted_at: datetime | None = None + started_at: datetime | None = None + checkpoint_id: str | None = None + interval_start: datetime | None = None + interval_end: datetime | None = None + interval_count: int = 0 + created_at: datetime | None = None + + +class HistoryRecorder(Protocol): + """Occurrence-history sink shared by polling and recovery.""" + + def record(self, entry: HistoryEntry) -> None: + """Append one entry (callers dedup stopped results first).""" + ... + + def has_terminal( + self, + schedule_id: str, + run_id: str, + kind: str, + checkpoint_id: str | None = None, + ) -> bool: + """Whether this exact stopped result was already reconciled.""" + ... + + +def entry_occurrence_id(entry: HistoryEntry, created: datetime) -> str: + """Derive the history occurrence id with the scheduler's standing rules.""" + if entry.resolved_at is not None: + return occurrence_id(entry.schedule_id, entry.resolved_at) + if entry.interval_start is not None and entry.interval_end is not None: + start = entry.interval_start.isoformat() + end = entry.interval_end.isoformat() + return f"{entry.schedule_id}|summary|{start}|{end}" + return f"{entry.schedule_id}|summary|{created.isoformat()}" + + +class FileScheduleHistoryRecorder: + """History recorder backed by a file schedule store.""" + + def __init__(self, schedule_store: Any) -> None: + self.schedule_store = schedule_store + + def record(self, entry: HistoryEntry) -> None: + created = ( + entry.created_at if entry.created_at is not None else datetime.now(UTC) + ) + self.schedule_store.append_history( + OccurrenceRecord( + schedule_id=entry.schedule_id, + occurrence_id=entry_occurrence_id(entry, created), + kind=entry.kind, + resolved_at=entry.resolved_at, + run_id=entry.run_id, + revision=entry.revision, + reason=entry.reason, + admitted_at=entry.admitted_at, + started_at=entry.started_at, + checkpoint_id=entry.checkpoint_id, + interval_start=entry.interval_start, + interval_end=entry.interval_end, + interval_count=entry.interval_count, + created_at=created, + ) + ) + + def has_terminal( + self, + schedule_id: str, + run_id: str, + kind: str, + checkpoint_id: str | None = None, + ) -> bool: + return self.schedule_store.has_history_entry( + schedule_id, + run_id=run_id, + kind=kind, + checkpoint_id=checkpoint_id, + ) diff --git a/src/wf_scheduling/models.py b/src/wf_scheduling/models.py index 4a73bca0..69835d80 100644 --- a/src/wf_scheduling/models.py +++ b/src/wf_scheduling/models.py @@ -156,6 +156,14 @@ class OccurrenceRecord(BaseModel): reason: str = "" admitted_at: datetime | None = None started_at: datetime | None = None + checkpoint_id: str | None = Field( + default=None, + description=( + "Stopped checkpoint this entry reconciles, for exactly-once " + "recovery dedup: a resumed run that stops again carries a new " + "checkpoint id and therefore a new entry." + ), + ) interval_start: datetime | None = None interval_end: datetime | None = None interval_count: int = 0 diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index 890588dd..f0d2fc72 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -25,8 +25,12 @@ from wf_scheduling.calendar import ( OneShotSource, ) from wf_scheduling.dispatch import RunDispatcher, StillRunning -from wf_scheduling.models import OccurrenceRecord, PendingCandidate -from wf_scheduling.occurrences import occurrence_id +from wf_scheduling.history import ( + FileScheduleHistoryRecorder, + HistoryEntry, + HistoryRecorder, +) +from wf_scheduling.models import PendingCandidate from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError from wf_scheduling.prepare import InvocationPreparer, PreparationRejected @@ -116,6 +120,7 @@ class Scheduler: preparer: InvocationPreparer, dispatcher: RunDispatcher, ownership: SchedulerOwnership, + history: HistoryRecorder | None = None, ) -> None: self.schedule_store = schedule_store self.run_store = run_store @@ -124,6 +129,11 @@ class Scheduler: self.preparer = preparer self.dispatcher = dispatcher self.ownership = ownership + self.history: HistoryRecorder = ( + history + if history is not None + else FileScheduleHistoryRecorder(schedule_store) + ) self._poll_cursor = 0 def _require_ownership(self) -> None: @@ -191,18 +201,12 @@ class Scheduler: now: datetime | None = None, admitted_at: datetime | None = None, started_at: datetime | None = None, + checkpoint_id: str | None = None, ) -> None: created = now if now is not None else datetime.now(UTC) - if intended is not None: - oid = occurrence_id(sched_id, intended) - elif interval is not None: - oid = f"{sched_id}|summary|{interval[0].isoformat()}|{interval[1].isoformat()}" - else: - oid = f"{sched_id}|summary|{created.isoformat()}" - self.schedule_store.append_history( - OccurrenceRecord( + self.history.record( + HistoryEntry( schedule_id=sched_id, - occurrence_id=oid, kind=kind, # type: ignore[arg-type] resolved_at=intended, run_id=run_id, @@ -210,6 +214,7 @@ class Scheduler: reason=reason, admitted_at=admitted_at, started_at=started_at, + checkpoint_id=checkpoint_id, interval_start=interval[0] if interval else None, interval_end=interval[1] if interval else None, interval_count=count, @@ -217,6 +222,20 @@ class Scheduler: ) ) + def _existing_occurrence_run(self, sched_id: str, intended: datetime) -> str | None: + """Return the run already owning this occurrence, if any. + + Identity is ``(schedule_id, resolved UTC instant)`` from the durable + admission record — the admission persist is the decision point, so + admissions are scanned rather than views (a crashed admission may + not have a view yet). Manual runs carry no scheduled instant and + never match. + """ + for admission in self.run_store.list_admissions(): + if admission.schedule_id == sched_id and admission.scheduled_at == intended: + return admission.id + return None + def _admit(self, sched: Any, intended: datetime, now: datetime) -> str | None: if getattr(sched, "blocked_reason", None): raise BlockedSchedule(getattr(sched, "blocked_reason")) @@ -232,6 +251,30 @@ class Scheduler: sched.blocked_reason = reason self.schedule_store.save_schedule(sched) raise BlockedSchedule(reason) from None + existing = self._existing_occurrence_run(sched.id, intended) + if existing is not None: + # The occurrence already owns a run (crash between the admission + # persist and the watermark/history writes, or a lost watermark + # write): an occurrence is immutable and never replayed. Advance + # the watermark, reconcile a missing admitted entry exactly once, + # and return the owner without dispatching (the pending sweep + # owns dispatch). + cand = self.schedule_store.get_candidate(sched.id) + if cand is not None and cand.intended_at == intended: + self.schedule_store.save_candidate(None, schedule_id=sched.id) + _save_consumed_max(self.schedule_store, sched.id, intended) + if not self.history.has_terminal(sched.id, existing, "admitted", None): + self._record( + kind="admitted", + sched_id=sched.id, + intended=intended, + run_id=existing, + reason=f"rev={sched.revision}", + revision=sched.revision, + now=now, + admitted_at=now, + ) + return existing prepared = self.preparer.prepare(sched=sched, intended=intended, now=now) if isinstance(prepared, PreparationRejected): self._record( @@ -377,6 +420,7 @@ class Scheduler: run_id=run_id, now=now, started_at=now, + checkpoint_id=stopped.latest_checkpoint_id, ) def record_stopped_execution(self, run_id: str, result: Any, now: datetime) -> None: @@ -422,6 +466,7 @@ class Scheduler: run_id=run_id, now=now, started_at=now, + checkpoint_id=stopped.latest_checkpoint_id, ) # -- polling -------------------------------------------------------- @@ -476,7 +521,7 @@ class Scheduler: run, CORRUPT_PENDING_REASON, now, - None, + self.history, ) clear_pending(self.run_store, run.id) clear_executing(self.run_store, run.id) @@ -487,7 +532,7 @@ class Scheduler: run, CORRUPT_PENDING_REASON, now, - None, + self.history, ) clear_pending(self.run_store, run.id) clear_executing(self.run_store, run.id) diff --git a/src/wf_scheduling/recovery.py b/src/wf_scheduling/recovery.py index 8f34c5e8..b978837b 100644 --- a/src/wf_scheduling/recovery.py +++ b/src/wf_scheduling/recovery.py @@ -36,6 +36,11 @@ from __future__ import annotations from datetime import datetime, timezone from typing import Any +from wf_scheduling.history import ( + FileScheduleHistoryRecorder, + HistoryEntry, + HistoryRecorder, +) from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError UTC = timezone.utc @@ -60,7 +65,7 @@ def recover( run_store: Any, now: datetime, ownership: SchedulerOwnership, - record_history: Any | None = None, + history: HistoryRecorder | None = None, ) -> list[str]: """Reconcile durable state after a restart without executing work.""" from wf_artifacts.runs.models import StoredRunStatus @@ -70,6 +75,8 @@ def recover( "scheduler ownership is required before recovery: an unowned " "recovery could abandon or redispatch another owner's work" ) + 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 @@ -102,40 +109,51 @@ def recover( clear_pending(run_store, run.id) diags.append(f"{run.id}:completion-window-cleared") if status == StoredRunStatus.INTERRUPTED.value or status == "interrupted": + checkpoint_id, result_attempt = _latest_result(run_store, run.id) + sched_id, intended, revision = _attribution(run_store, run.id) 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, - ) - ) + _complete_attempt(run_store, run.id, attempt, now) diags.append(f"{run.id}:fresh-result-resumable") + if _reconcile_terminal( + history, + sched_id, + run.id, + "interrupted", + checkpoint_id, + intended, + revision, + "fresh-result", + now, + ): + diags.append(f"{run.id}:terminal-reconciled") else: - _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history) + _fail_run(run_store, run, AMBIGUOUS_REASON, now, history) diags.append(f"{run.id}:failed-closed") else: diags.append(f"{run.id}:waiting-resumable") + if _reconcile_terminal( + history, + sched_id, + run.id, + "interrupted", + checkpoint_id, + intended, + revision, + "reconciled-on-recovery", + now, + ): + diags.append(f"{run.id}:terminal-reconciled") elif status == StoredRunStatus.ADMITTED.value or status == "admitted": if is_executing(run_store, run.id): # The executor may already have produced external effects: # abandon, never retry. Both markers are cleared so a later # recovery does not re-fail the now-terminal run. if active: - _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history) + _fail_run(run_store, run, AMBIGUOUS_REASON, now, history) else: - _fail_run(run_store, run, ABANDONED_REASON, now, record_history) + _fail_run(run_store, run, ABANDONED_REASON, now, history) clear_pending(run_store, run.id) clear_executing(run_store, run.id) diags.append(f"{run.id}:failed-closed") @@ -151,80 +169,177 @@ def recover( diags.append(f"{run.id}:corrupt-blocked") continue if active: - _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history) + _fail_run(run_store, run, AMBIGUOUS_REASON, now, history) else: - _fail_run(run_store, run, ABANDONED_REASON, now, record_history) + _fail_run(run_store, run, ABANDONED_REASON, now, history) diags.append(f"{run.id}:failed-closed") elif status == StoredRunStatus.COMPLETED.value or status == "completed": + checkpoint_id, completed_attempt = _latest_result(run_store, run.id) + sched_id, intended, revision = _attribution(run_store, run.id) if active: assert attempt is not None - try: - latest = run_store.get_latest_checkpoint(run.id) - completed_attempt = latest.attempt_id - except KeyError: - completed_attempt = None if ( completed_attempt is not None and completed_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, - ) - ) + _complete_attempt(run_store, run.id, attempt, now) diags.append(f"{run.id}:attempt-reconciled") else: - _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history) + _fail_run(run_store, run, AMBIGUOUS_REASON, now, history) diags.append(f"{run.id}:failed-closed") continue - if record_history is not None and not _has_terminal( - schedule_store, run.id, "completed" + if _reconcile_terminal( + history, + sched_id, + run.id, + "completed", + checkpoint_id, + intended, + revision, + "reconciled-on-recovery", + now, + ): + diags.append(f"{run.id}:terminal-reconciled") + elif status == StoredRunStatus.FAILED.value or status == "failed": + sched_id, intended, revision = _attribution(run_store, run.id) + if _reconcile_terminal( + history, + sched_id, + run.id, + "failed", + run.latest_checkpoint_id, + intended, + revision, + "reconciled-on-recovery", + now, ): - record_history( - kind="completed", - run_id=run.id, - reason="reconciled-on-recovery", - ) diags.append(f"{run.id}:terminal-reconciled") return diags +def _attribution( + run_store: Any, run_id: str +) -> tuple[str | None, datetime | None, int | None]: + """Return ``(schedule_id, scheduled_at, schedule_revision)`` for a run. + + ``(None, None, None)`` when no admission owns the run: schedule history + cannot attribute it, so callers persist the reason on the run itself. + """ + try: + admission = run_store.get_admission(run_id) + except KeyError: + return None, None, None + return ( + admission.schedule_id, + admission.scheduled_at, + admission.schedule_revision, + ) + + +def _latest_result(run_store: Any, run_id: str) -> tuple[str | None, Any]: + """Return ``(checkpoint_id, attempt_id)`` of the latest stopped checkpoint.""" + try: + latest = run_store.get_latest_checkpoint(run_id) + except KeyError: + return None, None + return latest.id, latest.attempt_id + + +def _complete_attempt(run_store: Any, run_id: str, attempt: Any, now: datetime) -> None: + """Mark the ACTIVE attempt DONE after its fresh result was recognized.""" + 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, + ) + ) + + +def _reconcile_terminal( + history: HistoryRecorder, + sched_id: str | None, + run_id: str, + kind: str, + checkpoint_id: str | None, + intended: datetime | None, + revision: int | None, + reason: str, + now: datetime, +) -> bool: + """Record one stopped-result entry unless this exact result is known. + + Returns whether an entry was appended. Idempotency identity is + ``(run_id, kind, checkpoint_id)``: repeating recovery never duplicates + an entry, while a resumed run that stops again (new checkpoint id) + records a new one. + """ + if sched_id is None: + return False + if history.has_terminal(sched_id, run_id, kind, checkpoint_id): + return False + history.record( + HistoryEntry( + schedule_id=sched_id, + kind=kind, # type: ignore[arg-type] + resolved_at=intended, + run_id=run_id, + revision=revision, + reason=reason, + checkpoint_id=checkpoint_id, + created_at=now, + ) + ) + return True + + def _fail_run( - run_store: Any, run: Any, reason: str, now: datetime, record_history: Any | None + run_store: Any, + run: Any, + reason: str, + now: datetime, + history: HistoryRecorder, ) -> None: + """Fail a run closed with its truthful reason durably preserved. + + The reason is always written into the run summary diagnostics; when an + admission attributes the run to a schedule, a failed occurrence entry is + reconciled exactly once through the history recorder. + """ + from wf_artifacts import DependencyDiagnostic, DiagnosticSeverity from wf_artifacts.runs.models import StoredRunStatus + diagnostic = DependencyDiagnostic( + severity=DiagnosticSeverity.ERROR, + code="schedule-recovery", + logical_ref=run.id, + message=reason, + repair_hint=None, + ) updated = run.model_copy( - update={"status": StoredRunStatus.FAILED, "updated_at": now} + update={ + "status": StoredRunStatus.FAILED, + "updated_at": now, + "diagnostics": [*run.diagnostics, diagnostic], + } ) 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 + sched_id, intended, revision = _attribution(run_store, run.id) + _reconcile_terminal( + history, + sched_id, + run.id, + "failed", + run.latest_checkpoint_id, + intended, + revision, + reason, + now, + ) def _mark_pending(run_store: Any, run_id: str) -> None: diff --git a/src/wf_scheduling/store.py b/src/wf_scheduling/store.py index 002c56f9..893b4fb4 100644 --- a/src/wf_scheduling/store.py +++ b/src/wf_scheduling/store.py @@ -168,6 +168,29 @@ class FileScheduleStore: entries.append(record.model_dump(mode="json")) self._write_json(path, entries) + def has_history_entry( + self, + schedule_id: str, + *, + run_id: str, + kind: str, + checkpoint_id: str | None = None, + ) -> bool: + """Whether this exact stopped result already has a history entry. + + Idempotency identity is ``(run_id, kind, checkpoint_id)``: repeats + of one recovery pass dedup, while a resumed run that stops again + (new checkpoint id) records a new entry. + """ + for item in self._read_history_locked(schedule_id): + if ( + item.get("run_id") == run_id + and item.get("kind") == kind + and item.get("checkpoint_id") == checkpoint_id + ): + return True + return False + @staticmethod def _sort_key( item: OccurrenceRecord, diff --git a/tests/scheduling/test_history_reconcile.py b/tests/scheduling/test_history_reconcile.py new file mode 100644 index 00000000..38bddc7e --- /dev/null +++ b/tests/scheduling/test_history_reconcile.py @@ -0,0 +1,490 @@ +"""Occurrence-history reconciliation through one typed interface (R4/F4). + +Polling and recovery share :class:`HistoryRecorder`: stopped results carry +``(run_id, kind, checkpoint_id)`` identity, failure reasons persist on the +run summary even when no schedule attribution exists, and repeating +recovery never duplicates an entry while a resumed re-interruption records +anew. +""" + +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, + ScriptedDispatcher, + fixture_environment, +) +from wf_api.run_lifecycle import ( + materialize_admitted_view, + persist_admission, + persist_stopped_run, +) +from wf_artifacts import PinnedRunEnvironment +from wf_artifacts.runs.models import ResumeAttempt +from wf_artifacts.runs.store import FileRunStore +from wf_core import RunState, RunStatus +from wf_scheduling import recovery as sched_recovery +from wf_scheduling.history import FileScheduleHistoryRecorder +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=[] + ) + + +def _admit_owned( + run_store: FileRunStore, + run_id: str, + intended: datetime, + *, + revision: int = 1, +) -> Any: + admission = persist_admission( + store=run_store, + run_id=run_id, + environment=_env(), + resolved_input={}, + max_steps=None, + scheduled_at=intended, + schedule_id="a", + schedule_revision=revision, + ) + materialize_admitted_view(store=run_store, admission=admission) + return admission + + +def _stop( + run_store: FileRunStore, + run_id: str, + status: RunStatus, + *, + attempt_id: int | None = None, +) -> Any: + record = run_store.get_run(run_id) + return 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, + ) + + +def _mark_active( + run_store: FileRunStore, run_id: str, attempt_id: int, now: datetime +) -> None: + run_store.save_resume_attempt( + ResumeAttempt( + run_id=run_id, + attempt_id=attempt_id, + state="ACTIVE", + created_at=now, + updated_at=now, + ) + ) + + +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 _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 test_failed_run_persists_reason_without_callback(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 = run_store.allocate_run_id() + _admit_owned(run_store, run_id, intended) + run_store.mark_executing(run_id) + diags = _recover(sched_store, run_store, intended) + assert any("failed-closed" in d for d in diags) + record = run_store.get_run(run_id) + assert record.status.value == "failed" + assert any(sched_recovery.ABANDONED_REASON in d.message for d in record.diagnostics) + failed = _entries(sched_store, "a", "failed") + assert len(failed) == 1 + assert failed[0]["run_id"] == run_id + assert sched_recovery.ABANDONED_REASON in failed[0]["reason"] + + +def test_waiting_interrupted_reconciles_exactly_once(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 = run_store.allocate_run_id() + _admit_owned(run_store, run_id, intended) + stopped = _stop(run_store, run_id, RunStatus.INTERRUPTED) + diags = _recover(sched_store, run_store, intended) + assert any("waiting-resumable" in d for d in diags) + assert any("terminal-reconciled" in d for d in diags) + first = _entries(sched_store, "a", "interrupted") + assert len(first) == 1 + assert first[0]["checkpoint_id"] == stopped.latest_checkpoint_id + diags = _recover(sched_store, run_store, intended + timedelta(minutes=1)) + assert _entries(sched_store, "a", "interrupted") == first + + +def test_fresh_interrupted_completes_attempt_and_reconciles(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 = run_store.allocate_run_id() + _admit_owned(run_store, run_id, intended) + attempt_id = run_store.allocate_resume_attempt_id() + _mark_active(run_store, run_id, attempt_id, intended) + _stop(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=attempt_id) + diags = _recover(sched_store, run_store, intended) + assert any("fresh-result-resumable" in d for d in diags) + assert run_store.get_resume_attempt(run_id).state == "DONE" # type: ignore[union-attr] + assert len(_entries(sched_store, "a", "interrupted")) == 1 + _recover(sched_store, run_store, intended + timedelta(minutes=1)) + assert len(_entries(sched_store, "a", "interrupted")) == 1 + + +def test_completed_reconciles_exactly_once(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 = run_store.allocate_run_id() + _admit_owned(run_store, run_id, intended) + stopped = _stop(run_store, run_id, RunStatus.COMPLETED) + diags = _recover(sched_store, run_store, intended) + assert any("terminal-reconciled" in d for d in diags) + first = _entries(sched_store, "a", "completed") + assert len(first) == 1 + assert first[0]["checkpoint_id"] == stopped.latest_checkpoint_id + assert first[0]["run_id"] == run_id + _recover(sched_store, run_store, intended + timedelta(minutes=1)) + assert _entries(sched_store, "a", "completed") == first + + +def test_history_write_failure_reconciles_once_on_recovery(tmp_path: Path) -> None: + class FailCompletedHistoryOnce(FileScheduleStore): + def __init__(self, root: Path) -> None: + super().__init__(root) + self.armed = True + + def append_history(self, record: Any) -> None: + if self.armed and getattr(record, "kind", None) == "completed": + self.armed = False + raise OSError("injected history failure") + super().append_history(record) + + sched_store = FailCompletedHistoryOnce(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + 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=ScriptedDispatcher({"*": "complete"}), + ownership=ownership, + ) + intended = ts(2026, 9, 8, 12, 0) + sched_store.create_schedule(_sched_model("a")) + sched_store.save_consumed("a", intended - timedelta(hours=1)) + from wf_scheduling.calendar import OneShotSource + + sched.sources["a"] = OneShotSource(intended) + try: + sched.poll(intended) + raise AssertionError("history failure must propagate") + except OSError as exc: + assert "injected history failure" in str(exc) + run_id = run_store.list_runs()[0].id + assert run_store.get_run(run_id).status.value == "completed" + assert _entries(sched_store, "a", "completed") == [] + diags = sched_recovery.recover( + schedule_store=sched_store, + run_store=run_store, + now=intended, + ownership=ownership, + ) + assert any("terminal-reconciled" in d for d in diags) + assert len(_entries(sched_store, "a", "completed")) == 1 + sched_recovery.recover( + schedule_store=sched_store, + run_store=run_store, + now=intended + timedelta(minutes=1), + ownership=ownership, + ) + assert len(_entries(sched_store, "a", "completed")) == 1 + finally: + ownership.release() + + +def test_dispatch_and_recovery_share_entry_identity(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sched = Scheduler( + schedule_store=sched_store, + run_store=run_store, + sources={}, + capacity=4, + preparer=SchedulePreparer( + DictDeployments({"dep-1": {"rev": 1, "required": []}}), + fixture_environment, + ), + dispatcher=ScriptedDispatcher({"*": "complete"}), + ownership=ownership, + ) + intended = ts(2026, 9, 8, 12, 0) + sched_store.create_schedule(_sched_model("a")) + sched_store.save_consumed("a", intended - timedelta(hours=1)) + from wf_scheduling.calendar import OneShotSource + + sched.sources["a"] = OneShotSource(intended) + sched.poll(intended) + run_id = run_store.list_runs()[0].id + dispatched = _entries(sched_store, "a", "completed") + assert len(dispatched) == 1 + checkpoint_id = run_store.get_run(run_id).latest_checkpoint_id + assert dispatched[0]["checkpoint_id"] == checkpoint_id + recorder = FileScheduleHistoryRecorder(sched_store) + assert recorder.has_terminal("a", run_id, "completed", checkpoint_id) + # Recovery on fresh objects recognizes the entry: no duplicate. + diags = sched_recovery.recover( + schedule_store=FileScheduleStore(tmp_path / "sched"), + run_store=FileRunStore(tmp_path / "runs"), + now=intended, + ownership=ownership, + ) + assert not any("terminal-reconciled" in d for d in diags) + assert len(_entries(sched_store, "a", "completed")) == 1 + finally: + ownership.release() + + +def test_resumed_reinterruption_records_anew(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 = run_store.allocate_run_id() + _admit_owned(run_store, run_id, intended) + first = run_store.allocate_resume_attempt_id() + _mark_active(run_store, run_id, first, intended) + _stop(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=first) + _recover(sched_store, run_store, intended) + assert run_store.get_resume_attempt(run_id).state == "DONE" # type: ignore[union-attr] + # The resumed run interrupts again under a new attempt: a new checkpoint + # id means a new history entry, not a dedup hit. + second = run_store.allocate_resume_attempt_id() + assert second != first + _mark_active(run_store, run_id, second, intended) + _stop(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=second) + _recover(sched_store, run_store, intended + timedelta(minutes=5)) + entries = _entries(sched_store, "a", "interrupted") + assert len(entries) == 2 + assert {e["checkpoint_id"] for e in entries} == { + f"{run_id}.000001", + f"{run_id}.000002", + } + + +def _owned_scheduler( + sched_store: FileScheduleStore, + run_store: FileRunStore, + ownership: SchedulerOwnership, + *, + script: dict | None = None, +) -> Scheduler: + return Scheduler( + schedule_store=sched_store, + run_store=run_store, + sources={}, + capacity=4, + preparer=SchedulePreparer( + DictDeployments({"dep-1": {"rev": 1, "required": []}}), + fixture_environment, + ), + dispatcher=ScriptedDispatcher(script), + ownership=ownership, + ) + + +def test_admission_tear_returns_existing_run_without_dup(tmp_path: Path) -> None: + from wf_scheduling.calendar import OneShotSource + + 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) + # Crash shape: admission + view persisted, history/consumed writes lost. + 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="a", + schedule_revision=1, + ) + materialize_admitted_view(store=run_store, admission=admission) + sched_store.save_consumed("a", intended - timedelta(hours=1)) + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched = _owned_scheduler( + sched_store, run_store, ownership, script={"*": "hang"} + ) + sched.sources["a"] = OneShotSource(intended) + assert sched.poll(intended) == {"a": f"admit:{run_id}"} + assert [r.id for r in run_store.list_runs()] == [run_id] + assert [a.id for a in run_store.list_admissions()] == [run_id] + assert sched_store.get_consumed("a") == intended + admitted = _entries(sched_store, "a", "admitted") + assert len(admitted) == 1 + assert admitted[0]["run_id"] == run_id + # Re-polling never duplicates the occurrence or its history entry. + sched.poll(intended) + assert [r.id for r in run_store.list_runs()] == [run_id] + assert len(_entries(sched_store, "a", "admitted")) == 1 + finally: + ownership.release() + + +def test_parallel_tear_never_double_admits(tmp_path: Path) -> None: + from wf_scheduling.calendar import OneShotSource + + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sched_store.create_schedule( + _sched_model("a", overlap="parallel", max_active_runs=4) + ) + intended = ts(2026, 9, 8, 12, 0) + 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="a", + schedule_revision=1, + ) + materialize_admitted_view(store=run_store, admission=admission) + sched_store.save_consumed("a", intended - timedelta(hours=1)) + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched = _owned_scheduler( + sched_store, run_store, ownership, script={"*": "hang"} + ) + sched.sources["a"] = OneShotSource(intended) + sched.poll(intended) + sched.poll(intended) + assert [r.id for r in run_store.list_runs()] == [run_id] + assert [a.id for a in run_store.list_admissions()] == [run_id] + finally: + ownership.release() + + +def test_consumed_write_failure_recovers_without_dup(tmp_path: Path) -> None: + from wf_scheduling.calendar import OneShotSource + + class FailConsumedOnce(FileScheduleStore): + def __init__(self, root: Path) -> None: + super().__init__(root) + self.armed = False + + def save_consumed(self, schedule_id: str, consumed_through: Any) -> None: + if self.armed: + self.armed = False + raise OSError("injected consumed failure") + super().save_consumed(schedule_id, consumed_through) + + sched_store = FailConsumedOnce(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + sched_store.save_consumed("a", intended - timedelta(hours=1)) + sched_store.armed = True # arm only the poll's watermark write + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched = _owned_scheduler( + sched_store, run_store, ownership, script={"*": "hang"} + ) + sched.sources["a"] = OneShotSource(intended) + with __import__("pytest").raises(OSError, match="injected consumed"): + sched.poll(intended) + assert [a.id for a in run_store.list_admissions()] != [] + first = run_store.list_admissions()[0].id + # Retry: the one-shot is exhausted so no new admission is possible; + # the failed watermark write cannot duplicate the occurrence. + assert sched.poll(intended) == {"a": "exhausted"} + assert [a.id for a in run_store.list_admissions()] == [first] + # Recovery completes the orphaned admission; the sweep dispatches it + # exactly once with no duplicate run. + diags = sched_recovery.recover( + schedule_store=sched_store, + run_store=run_store, + now=intended, + ownership=ownership, + ) + assert any("pending-dispatch" in d for d in diags) + sched.poll(intended) + assert [r.id for r in run_store.list_runs()] == [first] + assert run_store.is_executing(first) + finally: + ownership.release() diff --git a/tests/scheduling/test_preparation.py b/tests/scheduling/test_preparation.py index 54735da9..af3a7d4c 100644 --- a/tests/scheduling/test_preparation.py +++ b/tests/scheduling/test_preparation.py @@ -83,6 +83,7 @@ def test_scheduler_requires_typed_collaborators() -> None: "preparer", "dispatcher", "ownership", + "history", }