From d2c0867f76e2007a55eb536722bf780edb8b1bef Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 18:57:17 +0700 Subject: [PATCH] sched: one canonical lock identity per composition; checkpoint coherence and ordering authority (R4 wave 3) --- src/wf_scheduling/ownership.py | 81 ++- src/wf_scheduling/recovery.py | 254 ++++++-- tests/scheduling/test_checkpoint_authority.py | 551 ++++++++++++++++++ tests/scheduling/test_lock_identity.py | 207 +++++++ 4 files changed, 1030 insertions(+), 63 deletions(-) create mode 100644 tests/scheduling/test_checkpoint_authority.py create mode 100644 tests/scheduling/test_lock_identity.py diff --git a/src/wf_scheduling/ownership.py b/src/wf_scheduling/ownership.py index b80d1db8..2addc5d0 100644 --- a/src/wf_scheduling/ownership.py +++ b/src/wf_scheduling/ownership.py @@ -11,11 +11,22 @@ Locking design (Windows-tested, per the store transaction boundary): one new module holding ``msvcrt.locking`` (Windows) / ``fcntl.flock`` (POSIX) on ``/scheduler.lock``. No ad-hoc per-run lock files. +One store composition has exactly one lock identity: the canonical +nearest common ancestor of its store roots (see +:func:`canonical_lock_root`). Any ancestor is NOT good enough — different +ancestors yield different lock files and therefore independent OS locks, +so ancestor containment can never prove exclusivity. Guards require the +held lock's frozen identity to equal the composition identity; overlapping +compositions that share no common ancestor are rejected instead of +claimed safe. + The lock is acquired on the composition root that contains the protected schedule and run stores. Entry-point guards do not trust the caller to have passed the right lock: they validate the held guard against the actual store roots (canonicalized for platform aliases) before any write -or side effect. +or side effect. The acquired identity is frozen at acquisition: later +mutation of the public ``root`` attribute cannot redirect a held handle's +authority. """ from __future__ import annotations @@ -48,12 +59,28 @@ def canonical_store_path(path: Path | str) -> str: return os.path.normcase(os.path.realpath(os.fspath(path))) -def _is_within(candidate: str, base: str) -> bool: - """Whether canonical ``candidate`` lies beneath canonical ``base``.""" - if candidate == base: - return True - prefix = base if base.endswith(os.sep) else base + os.sep - return candidate.startswith(prefix) +def canonical_lock_root(*store_roots: Path | str | None) -> str | None: + """Return the one canonical lock identity for a store composition. + + The identity is the canonical nearest common ancestor of the store + roots; the lock file lives at ``/scheduler.lock``. Both the + held guard's frozen root and the guarded stores map through this + function, so exactly one lock file can authorize one composition. + Returns None when a root is missing or the roots share no common + ancestor (e.g. different drives): such compositions are unsupported + and guards reject them instead of claiming safety. + """ + canonical: list[str] = [] + for store_root in store_roots: + if store_root is None: + return None + canonical.append(canonical_store_path(store_root)) + if not canonical: + return None + try: + return os.path.commonpath(canonical) + except ValueError: + return None class SchedulerOwnership: @@ -64,6 +91,7 @@ class SchedulerOwnership: self.owner = owner self._handle: BinaryIO | None = None self._locked = False + self._acquired_root: str | None = None @property def lock_path(self) -> Path: @@ -80,25 +108,35 @@ class SchedulerOwnership: return self._locked def canonical_root(self) -> str: - """Return the canonical composition root this lock was acquired on.""" + """Return the canonical composition root this lock was acquired on. + + The identity is frozen at acquisition: mutating the public + ``root`` afterwards cannot redirect a held handle's authority. + Before acquisition (or after release) it reports the live root. + """ + if self._acquired_root is not None: + return self._acquired_root return canonical_store_path(self.root) def covers(self, *store_roots: Path | str | None) -> bool: - """Whether this held lock protects the given store roots. + """Whether this held lock is the one lock for the given stores. - Coverage means each root canonicalizes to the lock's composition - root or beneath it. An unrelated (even held) lock, a released lock, - and rootless stores are never covered: guards reject those before - any write or side effect instead of trusting the caller's promise. + Coverage means the frozen acquisition identity equals the + composition's canonical lock identity + (:func:`canonical_lock_root`), not merely that the stores sit + beneath some held ancestor: a lock on a parent directory is an + independent OS lock from the composition's own lock and proves + nothing. An unrelated (even held) lock, a released lock, a + mutated handle, and rootless or ancestor-less stores are never + covered: guards reject those before any write or side effect + instead of trusting the caller's promise. """ if not self._locked or not store_roots: return False - base = self.canonical_root() - return all( - store_root is not None - and _is_within(canonical_store_path(store_root), base) - for store_root in store_roots - ) + if self._acquired_root is None: + return False + expected = canonical_lock_root(*store_roots) + return expected is not None and self._acquired_root == expected def acquire(self) -> SchedulerOwnership: """Acquire the held lock non-blockingly or raise SecondOwnerError.""" @@ -120,6 +158,10 @@ class SchedulerOwnership: raise StartupRejected(f"unsupported scheduler locking: {exc}") from exc self._handle = handle self._locked = True + # Freeze the authority identity now: the public ``root`` may be + # reassigned later, but a held handle must keep proving exactly the + # composition it locked, never the new value. + self._acquired_root = canonical_store_path(self.root) return self def release(self) -> None: @@ -134,6 +176,7 @@ class SchedulerOwnership: finally: self._handle = None self._locked = False + self._acquired_root = None def __enter__(self) -> SchedulerOwnership: return self.acquire() diff --git a/src/wf_scheduling/recovery.py b/src/wf_scheduling/recovery.py index acc7cfed..d0cb39f8 100644 --- a/src/wf_scheduling/recovery.py +++ b/src/wf_scheduling/recovery.py @@ -31,13 +31,25 @@ clear → executor → stopped persist → executing clear → history): reconciliation across the torn ``save_checkpoint``/``save_run`` boundary is owned by the F2 reconcile step, which runs before marker handling. - stopped summary, no marks: existing terminal/attempt reconciliation. -- failed summary: a durable recovery decision is never re-processed. When - the checkpoint pointer still matches the decided state, history is - ensured without touching status, readiness, or diagnostics (repeat - recovery is silent). Only a genuinely newer checkpoint (pointer - mismatch) may reopen the run via the reconcile path; an older - checkpoint can never supersede the abandonment. Failed runs carry - not-applicable resume readiness. +- failed summary: a durable recovery decision is never re-processed. A + matching checkpoint pointer only ensures history; it never revalidates + the decision. A mismatched pointer reopens the run only for a + genuinely newer result: the referenced checkpoint must still exist, the + latest checkpoint must have a strictly greater sequence with attempt + provenance (an attributed result never yields to an older attempt, and + an unattributed result never supersedes an attributed decision), and + the newer checkpoint must be internally coherent (see below). + Anything else — a missing referenced checkpoint, an older stray, a + stale attempt, or an incoherent newer checkpoint — keeps the decision + with history ensured; corruption is additionally noted once as a + durable diagnostic. Failed runs carry not-applicable resume readiness. +- checkpoint authority: before any summary rewrite, the latest checkpoint + must prove coherence — same run identity, checkpoint id matching its + sequence, runtime state decodable through core validation, and the + outer reason agreeing with the decoded stopped state. The outer reason + enum alone is never trusted. Contradictory or corrupt checkpoints fail + closed with durable diagnostics; a successful result is never + fabricated from disagreement. """ from __future__ import annotations @@ -121,19 +133,24 @@ def recover( clear_executing(run_store, run.id) clear_pending(run_store, run.id) diags.append(f"{run.id}:completion-window-cleared") - # Durable-decision stability: a FAILED summary for the current - # checkpoint state is never re-processed. Only a genuinely newer - # checkpoint (pointer mismatch) may reopen it via the reconcile - # path below. History is ensured without mutating the summary. + # Durable-decision stability: a FAILED summary for the decided + # checkpoint state is never re-processed — history is ensured + # without mutating the summary. A mismatched pointer reopens the + # run only for a validated genuinely-newer result (referenced + # checkpoint present, strictly greater sequence, attempt + # provenance, coherent content); anything else keeps the decision. if status == StoredRunStatus.FAILED.value or status == "failed": if _failed_decision_stable(run_store, run, history, now, diags): continue # 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. + # save_checkpoint/save_run boundary: a validated 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). Checkpoints that fail + # authority validation (wrong run, id/sequence disagreement, + # undecodable state, reason disagreeing with the decoded stopped + # state) raise below and fail closed. 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: @@ -284,17 +301,57 @@ def recover( return diags +def _check_checkpoint_coherence(run: Any, checkpoint: Any) -> None: + """Prove a checkpoint may speak for a run summary, or raise. + + Raises KeyError only from the caller's own loads; raises ValueError + (or OSError for unreadable state) when the checkpoint disagrees with + the run: a foreign run identity, a checkpoint id that does not match + its sequence, runtime state that core validation rejects, or an outer + reason that contradicts the decoded stopped state. The outer reason + enum alone is never authority: a COMPLETED label over an interrupted + runtime snapshot must fail closed, never release overlap. + """ + from wf_core import load_run_state_with_upgrade + + if checkpoint.run_id != run.id: + raise ValueError( + f"checkpoint {checkpoint.id!r} names run {checkpoint.run_id!r}, " + f"not {run.id!r}" + ) + if checkpoint.id != f"{run.id}.{checkpoint.sequence:06d}": + raise ValueError( + f"checkpoint {checkpoint.id!r} disagrees with its sequence " + f"{checkpoint.sequence}" + ) + try: + decoded, _ = load_run_state_with_upgrade( + checkpoint.state.model_dump(mode="json") + ) + except ValueError as exc: + raise ValueError( + f"checkpoint {checkpoint.id!r} has undecodable runtime state" + ) from exc + if decoded.status.value != checkpoint.reason.value: + raise ValueError( + f"checkpoint {checkpoint.id!r} reason {checkpoint.reason.value!r} " + f"contradicts its runtime state {decoded.status.value!r}" + ) + + def _reconcile_summary_from_checkpoint( run_store: Any, run: Any, now: datetime ) -> tuple[bool, str]: - """Rewrite a stale summary from its durable stopped checkpoint. + """Rewrite a stale summary from its validated 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). + undispatched-or-unknown, handled by the marker rules). Authority + validation runs first: incoherent checkpoints raise and fail closed + in the caller instead of rewriting the summary. """ from wf_artifacts.runs.models import ResumeReadiness, StoredRunStatus @@ -302,6 +359,7 @@ def _reconcile_summary_from_checkpoint( latest = run_store.get_latest_checkpoint(run.id) except KeyError: return False, getattr(run.status, "value", run.status) + _check_checkpoint_coherence(run, latest) expected = StoredRunStatus(latest.reason.value) readiness = ( ResumeReadiness.READY @@ -342,6 +400,87 @@ def _decision_reason(run: Any) -> str: return "" +KEPT_DECISION_NOTE_CODE = "schedule-recovery-note" + + +def _ensure_failed_history( + run_store: Any, + run: Any, + history: HistoryRecorder, + now: datetime, + diags: list[str], +) -> None: + """Ensure the failed terminal entry without touching the summary.""" + sched_id, intended, revision = _attribution(run_store, run.id) + reason = _decision_reason(run) or "reconciled-on-recovery" + if _reconcile_terminal( + history, + sched_id, + run.id, + "failed", + run.latest_checkpoint_id, + intended, + revision, + reason, + now, + ): + diags.append(f"{run.id}:terminal-reconciled") + + +def _note_decision_kept( + run_store: Any, run: Any, message: str, now: datetime, diags: list[str] +) -> None: + """Durably note once that a decision survived unresolvable state. + + Status and readiness are untouched; the note is appended only when no + identical note exists, so repeated recovery stays silent. + """ + from wf_artifacts import DependencyDiagnostic, DiagnosticSeverity + + if any( + getattr(item, "code", None) == KEPT_DECISION_NOTE_CODE + and str(getattr(item, "message", "")) == message + for item in run.diagnostics + ): + return + run_store.save_run( + run.model_copy( + update={ + "updated_at": now, + "diagnostics": [ + *run.diagnostics, + DependencyDiagnostic( + severity=DiagnosticSeverity.ERROR, + code=KEPT_DECISION_NOTE_CODE, + logical_ref=run.id, + message=message, + repair_hint=None, + ), + ], + } + ) + ) + diags.append(f"{run.id}:decision-kept-noted") + + +def _supersedes_decision(referenced: Any, latest: Any) -> bool: + """Whether a newer checkpoint result may reopen a durable decision. + + Ordering alone is not provenance: the latest checkpoint must carry a + strictly greater sequence AND attempt provenance — an attributed + result never yields to an older attempt, and an unattributed result + never supersedes an attributed decision (unattributed results predate + the resume protocol that produced the decision). Sequence ties and + regressions never supersede: a different checkpoint is not + necessarily a newer result. + """ + if latest.sequence <= referenced.sequence: + return False + if latest.attempt_id is None or referenced.attempt_id is None: + return referenced.attempt_id is None + return latest.attempt_id >= referenced.attempt_id + + def _failed_decision_stable( run_store: Any, run: Any, @@ -353,38 +492,65 @@ def _failed_decision_stable( Returns True when the run needs no further processing this pass: the summary already reflects a recovery decision (or a legacy/external - failure) for the current checkpoint state, so re-running reconcile or - re-failing would only duplicate diagnostics and rewrite readiness. A - genuinely newer checkpoint (pointer mismatch) returns False so the - reconcile path may repair the summary. Unreadable checkpoint state - likewise returns False: the reconcile path fails closed for undecided - runs and keeps decided ones via its own branch. + failure) and no validated newer result supersedes it, so re-running + reconcile or re-failing would only duplicate diagnostics and rewrite + readiness. A pointer match ensures history and stays silent. A + mismatch reopens the run only for a genuinely newer result — + referenced checkpoint present, strictly greater sequence, attempt + provenance, coherent content — via the reconcile path. A missing + referenced checkpoint, an older or unattributable newer checkpoint, + or an incoherent newer checkpoint keeps the decision: missing state + never rolls a run backward and never undoes a durable abandonment. + Corruption is noted once as a durable diagnostic. Unreadable + checkpoint state returns False: the reconcile path fails closed for + undecided runs and keeps decided ones via its own branch. """ try: - latest = run_store.get_latest_checkpoint(run.id) - except KeyError: - latest = None + checkpoints = run_store.list_checkpoints(run.id) except ValueError, OSError: return False pointer = run.latest_checkpoint_id - latest_id = latest.id if latest is not None else None - if latest_id != pointer: - # Genuinely newer (or torn-away) checkpoint state: not stable. + referenced = ( + next((c for c in checkpoints if c.id == pointer), None) + if pointer is not None + else None + ) + latest = max(checkpoints, key=lambda c: c.sequence, default=None) + if pointer is not None and referenced is None: + _ensure_failed_history(run_store, run, history, now, diags) + _note_decision_kept( + run_store, + run, + f"durable recovery decision kept: referenced checkpoint " + f"{pointer!r} is missing", + now, + diags, + ) + return True + if latest is None or latest.id == pointer: + _ensure_failed_history(run_store, run, history, now, diags) + return True + if referenced is None: + # A decision without a checkpoint pointer cannot order stray + # checkpoints: keep the decision with history ensured. + _ensure_failed_history(run_store, run, history, now, diags) + return True + if _supersedes_decision(referenced, latest): + try: + _check_checkpoint_coherence(run, latest) + except (ValueError, OSError) as exc: + _ensure_failed_history(run_store, run, history, now, diags) + _note_decision_kept( + run_store, + run, + f"durable recovery decision kept: newer checkpoint " + f"{latest.id!r} is incoherent: {exc}", + now, + diags, + ) + return True return False - sched_id, intended, revision = _attribution(run_store, run.id) - reason = _decision_reason(run) or "reconciled-on-recovery" - if _reconcile_terminal( - history, - sched_id, - run.id, - "failed", - pointer, - intended, - revision, - reason, - now, - ): - diags.append(f"{run.id}:terminal-reconciled") + _ensure_failed_history(run_store, run, history, now, diags) return True diff --git a/tests/scheduling/test_checkpoint_authority.py b/tests/scheduling/test_checkpoint_authority.py new file mode 100644 index 00000000..0f059a3d --- /dev/null +++ b/tests/scheduling/test_checkpoint_authority.py @@ -0,0 +1,551 @@ +"""Checkpoint authority: coherence and ordering before recovery trusts state. + +A checkpoint becomes recovery authority only when it proves coherence — +same run identity, checkpoint id matching its sequence, runtime state +decodable through core validation, and the outer reason agreeing with the +decoded stopped state. The outer reason enum alone is never trusted. + +A durable recovery decision is superseded only by a validated genuinely +newer result (referenced checkpoint present, strictly greater sequence, +attempt provenance, coherent content). Missing referenced state never +rolls a run backward. +""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any, cast + +import pytest + +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 ( + load_stored_run, + materialize_admitted_view, + persist_admission, + persist_stopped_run, + restore_interrupted_run, +) +from wf_artifacts import CheckpointReason, PinnedRunEnvironment +from wf_artifacts.runs.models import ResumeAttempt, RunCheckpoint +from wf_artifacts.runs.store import FileRunStore +from wf_core import PersistedRunState, RunState, RunStatus, dump_run_state +from wf_scheduling import recovery as sched_recovery +from wf_scheduling.calendar import OneShotSource +from wf_scheduling.models import Schedule +from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError +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) -> Schedule: + now = ts(2026, 9, 8, 12, 0) + return Schedule.model_validate( + { + "id": sid, + "deployment_id": "dep-1", + "trigger": {"kind": "cron", "expression": "0 * * * *", "timezone": "UTC"}, + "input_bindings": [], + "created_at": now.isoformat(), + "updated_at": now.isoformat(), + } + ) + + +def _env() -> PinnedRunEnvironment: + return PinnedRunEnvironment( + deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[] + ) + + +def _recover( + sched_store: FileScheduleStore, run_store: FileRunStore, now: datetime +) -> list[str]: + ownership = SchedulerOwnership(sched_store.root.parent, 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_interrupted( + run_store: FileRunStore, intended: datetime, *, attempt_id: int | None = None +) -> 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="a", + schedule_revision=1, + ) + materialize_admitted_view(store=run_store, admission=admission) + 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=attempt_id, + ) + return run_id + + +def _mark_attempt( + run_store: FileRunStore, run_id: str, attempt_id: int, state: str, now: datetime +) -> None: + run_store.save_resume_attempt( + ResumeAttempt( + run_id=run_id, + attempt_id=attempt_id, + state=cast(Any, state), + 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 _craft_checkpoint( + run_id: str, + sequence: int, + status: RunStatus, + *, + checkpoint_run_id: str | None = None, + checkpoint_id: str | None = None, + attempt_id: int | None = None, + now: datetime | None = None, +) -> RunCheckpoint: + """Build a storable checkpoint with caller-controlled identity fields.""" + return RunCheckpoint( + id=checkpoint_id or f"{run_id}.{sequence:06d}", + run_id=checkpoint_run_id or run_id, + sequence=sequence, + reason=CheckpointReason(status.value), + state=PersistedRunState.model_validate( + dump_run_state( + RunState( + workflow_name="sched", + status=status, + workflow_input={}, + state={}, + ) + ) + ), + attempt_id=attempt_id, + created_at=now or datetime.now(UTC), + ) + + +def _checkpoint_path(root: Path, run_id: str, sequence: int) -> Path: + return root / "runs" / "runs" / run_id / "checkpoints" / f"{sequence:06d}.json" + + +def _tamper_reason(root: Path, run_id: str, sequence: int, reason: str) -> None: + path = _checkpoint_path(root, run_id, sequence) + payload = json.loads(path.read_text(encoding="utf-8")) + payload["reason"] = reason + path.write_text(json.dumps(payload), encoding="utf-8") + + +def _decided_at_000002( + tmp_path: Path, intended: datetime +) -> tuple[FileScheduleStore, FileRunStore, str]: + """Failed decision pointing at .000002 with only .000001 older state. + + Genuine multi-restart sequence: ambiguous fail at .000001, a newer + stopped result under attempt 10, then stale under attempt 11. The + second durable abandonment points at .000002; attempt 11 is DONE so + no attempt matching can mask the ordering decision. + """ + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + run_id = _admit_interrupted(run_store, intended) + _mark_attempt(run_store, run_id, 9, "ACTIVE", intended) + first = _recover(sched_store, run_store, intended) + assert any("failed-closed" in d for d in first) + _mark_attempt(run_store, run_id, 10, "ACTIVE", 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=10, + ) + _mark_attempt(run_store, run_id, 11, "ACTIVE", intended) + second = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=1), + ) + assert any("failed-closed" in d for d in second) + decided = FileRunStore(tmp_path / "runs").get_run(run_id) + assert decided.status.value == "failed" + assert decided.latest_checkpoint_id == f"{run_id}.000002" + assert decided.resume_readiness.value == "not_applicable" + _mark_attempt(FileRunStore(tmp_path / "runs"), run_id, 11, "DONE", intended) + return ( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + run_id, + ) + + +def test_reason_state_disagreement_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_interrupted(run_store, intended) + _tamper_reason(tmp_path, run_id, 1, "completed") + diags = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=1), + ) + assert any("failed-closed" in d for d in diags) + record = FileRunStore(tmp_path / "runs").get_run(run_id) + assert record.status.value == "failed" + assert record.resume_readiness.value == "not_applicable" + assert any("contradicts its runtime state" in d.message for d in record.diagnostics) + # Nothing successful fabricated: no completed/interrupted terminal entries. + fresh_sched = FileScheduleStore(tmp_path / "sched") + assert _entries(fresh_sched, "a", "completed") == [] + assert _entries(fresh_sched, "a", "interrupted") == [] + assert len(_entries(fresh_sched, "a", "failed")) == 1 + # Repeat recovery is stable: the decision stands, nothing new appended. + rerun = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=2), + ) + assert not any(run_id in d for d in rerun) + again = FileRunStore(tmp_path / "runs").get_run(run_id) + assert again.status.value == "failed" + assert len(again.diagnostics) == len(record.diagnostics) + + +def test_run_identity_disagreement_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_interrupted(run_store, intended) + foreign = _craft_checkpoint( + run_id, 2, RunStatus.INTERRUPTED, checkpoint_run_id="run-999999" + ) + _checkpoint_path(tmp_path, run_id, 2).write_text( + json.dumps(foreign.model_dump(mode="json")), encoding="utf-8" + ) + diags = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=1), + ) + assert any("failed-closed" in d for d in diags) + record = FileRunStore(tmp_path / "runs").get_run(run_id) + assert record.status.value == "failed" + assert any("names run" in d.message for d in record.diagnostics) + + +def test_sequence_pointer_incoherence_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_interrupted(run_store, intended) + mismatched = _craft_checkpoint( + run_id, 2, RunStatus.INTERRUPTED, checkpoint_id=f"{run_id}.000001" + ) + _checkpoint_path(tmp_path, run_id, 2).write_text( + json.dumps(mismatched.model_dump(mode="json")), encoding="utf-8" + ) + diags = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=1), + ) + assert any("failed-closed" in d for d in diags) + record = FileRunStore(tmp_path / "runs").get_run(run_id) + assert record.status.value == "failed" + assert any("disagrees with its sequence" in d.message for d in record.diagnostics) + + +def test_undecodable_runtime_state_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_interrupted(run_store, intended) + bogus = _craft_checkpoint(run_id, 2, RunStatus.INTERRUPTED) + payload = bogus.model_dump(mode="json") + payload["state"] = {"version": 2, "state": {"bogus": True}} + _checkpoint_path(tmp_path, run_id, 2).write_text( + json.dumps(payload), encoding="utf-8" + ) + diags = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=1), + ) + assert any("failed-closed" in d for d in diags) + record = FileRunStore(tmp_path / "runs").get_run(run_id) + assert record.status.value == "failed" + assert record.resume_readiness.value == "not_applicable" + assert any("undecodable runtime state" in d.message for d in record.diagnostics) + + +def test_genuine_newer_completed_repairs_torn_summary(tmp_path: Path) -> None: + """Coherent newer checkpoints still repair torn summaries (no over-block).""" + 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_interrupted(run_store, intended) + # Torn boundary: the completed checkpoint persisted, the summary did not. + run_store.save_checkpoint(_craft_checkpoint(run_id, 2, RunStatus.COMPLETED)) + diags = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=1), + ) + assert any("summary-reconciled" in d for d in diags) + record = FileRunStore(tmp_path / "runs").get_run(run_id) + assert record.status.value == "completed" + assert record.latest_checkpoint_id == f"{run_id}.000002" + fresh_sched = FileScheduleStore(tmp_path / "sched") + assert {e["checkpoint_id"] for e in _entries(fresh_sched, "a", "completed")} == { + f"{run_id}.000002" + } + + +def test_missing_referenced_checkpoint_keeps_decision(tmp_path: Path) -> None: + sched_store, run_store, run_id = _decided_at_000002(tmp_path, ts(2026, 9, 8, 12, 0)) + before = FileRunStore(tmp_path / "runs").get_run(run_id) + assert len(before.diagnostics) == 1 + # Referenced durable state goes missing; only older .000001 remains. + _checkpoint_path(tmp_path, run_id, 2).unlink() + diags = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + ts(2026, 9, 8, 12, 2), + ) + assert any("decision-kept-noted" in d for d in diags) + assert not any("summary-reconciled" in d for d in diags) + kept = FileRunStore(tmp_path / "runs").get_run(run_id) + assert kept.status.value == "failed" + assert kept.resume_readiness.value == "not_applicable" + assert len(kept.diagnostics) == 2 + assert any("is missing" in d.message for d in kept.diagnostics) + # Both genuine decisions keep their terminal entries; the keep adds none. + assert len(_entries(FileScheduleStore(tmp_path / "sched"), "a", "failed")) == 2 + # Repeat recovery is fully silent and history stays idempotent. + rerun = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + ts(2026, 9, 8, 12, 3), + ) + assert not any(run_id in d for d in rerun) + again = FileRunStore(tmp_path / "runs").get_run(run_id) + assert again.status.value == "failed" + assert len(again.diagnostics) == 2 + assert _entries(FileScheduleStore(tmp_path / "sched"), "a", "failed") == _entries( + sched_store, "a", "failed" + ) + + +@pytest.mark.parametrize( + ("newer_attempt", "tamper", "expect_repair"), + [ + (11, False, True), # newer sequence + newer attempt: genuine repair + (9, False, False), # newer sequence + older attempt: stale, keep decision + (None, False, False), # newer sequence + unattributed: keep decision + (11, True, False), # newer + provenance but incoherent: keep + note + ], +) +def test_attempt_provenance_combinations( + tmp_path: Path, newer_attempt: int | None, tamper: bool, expect_repair: bool +) -> None: + intended = ts(2026, 9, 8, 12, 0) + _, _, run_id = _decided_at_000002(tmp_path, intended) + run_store = FileRunStore(tmp_path / "runs") + _mark_attempt(run_store, run_id, 11, "ACTIVE", intended) + run_store.save_checkpoint( + _craft_checkpoint(run_id, 3, RunStatus.INTERRUPTED, attempt_id=newer_attempt) + ) + if tamper: + _tamper_reason(tmp_path, run_id, 3, "completed") + diags = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=5), + ) + record = FileRunStore(tmp_path / "runs").get_run(run_id) + if expect_repair: + assert any("fresh-result-resumable" in d for d in diags) + assert record.status.value == "interrupted" + assert record.resume_readiness.value == "ready" + assert run_store.get_resume_attempt(run_id) is not None + assert run_store.get_resume_attempt(run_id).state == "DONE" # type: ignore[union-attr] + else: + assert record.status.value == "failed" + assert record.resume_readiness.value == "not_applicable" + assert not any("summary-reconciled" in d for d in diags) + if tamper: + assert any("decision-kept-noted" in d for d in diags) + assert any("is incoherent" in d.message for d in record.diagnostics) + else: + # Stale provenance is expected debris, not corruption: silent keep. + assert not any("decision-kept-noted" in d for d in diags) + rerun = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=6), + ) + assert not any(run_id in d for d in rerun) + + +def test_pointerless_decision_ignores_stray_checkpoints(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() + 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) + run_store.mark_executing(run_id) + diags = _recover(sched_store, run_store, intended) + assert any("failed-closed" in d for d in diags) + decided = run_store.get_run(run_id) + assert decided.status.value == "failed" + assert decided.latest_checkpoint_id is None + # A stray checkpoint with no ordering basis cannot reopen the decision. + run_store.save_checkpoint(_craft_checkpoint(run_id, 1, RunStatus.INTERRUPTED)) + rerun = _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=1), + ) + assert not any("summary-reconciled" in d for d in rerun) + kept = FileRunStore(tmp_path / "runs").get_run(run_id) + assert kept.status.value == "failed" + assert kept.resume_readiness.value == "not_applicable" + + +def test_inspection_resume_and_poll_agree_after_reconciliation( + tmp_path: Path, +) -> None: + intended = ts(2026, 9, 8, 12, 0) + _, _, run_id = _decided_at_000002(tmp_path, intended) + _checkpoint_path(tmp_path, run_id, 2).unlink() + _recover( + FileScheduleStore(tmp_path / "sched"), + FileRunStore(tmp_path / "runs"), + intended + timedelta(minutes=2), + ) + # Inspection agrees: failed, not applicable. + record, _ = load_stored_run(FileRunStore(tmp_path / "runs"), run_id) + assert record.status.value == "failed" + assert record.resume_readiness.value == "not_applicable" + # Resume refuses. + try: + restore_interrupted_run(FileRunStore(tmp_path / "runs"), run_id) + raise AssertionError("failed run must not restore as interrupted") + except ValueError: + pass + # A poll sweep with nothing due leaves the decided run untouched and + # dispatches nothing from its slot. + sched_store = FileScheduleStore(tmp_path / "sched") + sched_store.save_consumed("a", intended) + ownership = SchedulerOwnership(tmp_path, owner="poll").acquire() + try: + sched = Scheduler( + schedule_store=sched_store, + run_store=FileRunStore(tmp_path / "runs"), + sources={"a": OneShotSource(intended)}, + capacity=4, + preparer=SchedulePreparer( + DictDeployments({"dep-1": {"rev": 1, "required": []}}), + fixture_environment, + ), + dispatcher=ScriptedDispatcher({"*": "hang"}), + ownership=ownership, + ) + sched.poll(intended) + finally: + ownership.release() + after = FileRunStore(tmp_path / "runs").get_run(run_id) + assert after.status.value == "failed" + assert [d.message for d in after.diagnostics] == [ + d.message for d in record.diagnostics + ] + assert [r.id for r in FileRunStore(tmp_path / "runs").list_runs()] == [run_id] + + +def test_wrong_root_rejected_before_writes_or_dispatch(tmp_path: Path) -> None: + comp = tmp_path / "composition" + sched_store = FileScheduleStore(comp / "sched") + run_store = FileRunStore(comp / "runs") + outsider = SchedulerOwnership(tmp_path, owner="outsider").acquire() + try: + assert outsider.held + assert outsider.covers(sched_store.root, run_store.root) is False + with pytest.raises(SecondOwnerError): + sched_recovery.recover( + schedule_store=sched_store, + run_store=run_store, + now=datetime.now(UTC), + ownership=outsider, + ) + assert run_store.list_runs() == [] + assert run_store.list_admissions() == [] + finally: + outsider.release() diff --git a/tests/scheduling/test_lock_identity.py b/tests/scheduling/test_lock_identity.py new file mode 100644 index 00000000..88cc935a --- /dev/null +++ b/tests/scheduling/test_lock_identity.py @@ -0,0 +1,207 @@ +"""One store composition has one lock identity (R4 wave 3, item A). + +Ancestor containment is not authority: a lock on a parent directory is an +independent OS lock from the composition's own lock, so both can be held +simultaneously. Guards require the held lock's frozen identity to equal +the composition's canonical lock identity +(:func:`canonical_lock_root`), and the acquired identity is frozen so a +later root mutation cannot redirect a held handle. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +from datetime import UTC, datetime +from pathlib import Path + +import pytest + +from wf_artifacts.runs.store import FileRunStore +from wf_scheduling import recovery as sched_recovery +from wf_scheduling.ownership import ( + SchedulerOwnership, + SecondOwnerError, + canonical_lock_root, + canonical_store_path, +) +from wf_scheduling.store import FileScheduleStore + + +def _stores(root: Path) -> tuple[FileScheduleStore, FileRunStore]: + return FileScheduleStore(root / "sched"), FileRunStore(root / "runs") + + +def test_canonical_lock_identity_is_deterministic(tmp_path: Path) -> None: + comp = tmp_path / "composition" + sched_store, run_store = _stores(comp) + assert canonical_lock_root( + sched_store.root, run_store.root + ) == canonical_store_path(comp) + # Aliased spellings of the same composition share the identity. + dotted = tmp_path / "sub" / ".." / "composition" + assert canonical_lock_root( + dotted / "sched", dotted / "runs" + ) == canonical_store_path(comp) + # A different composition has a different identity; missing roots have none. + assert canonical_lock_root(tmp_path / "other") != canonical_store_path(comp) + assert canonical_lock_root(None, run_store.root) is None # type: ignore[arg-type] + assert canonical_lock_root() is None + + +def test_parent_and_child_locks_cannot_both_authorize(tmp_path: Path) -> None: + comp = tmp_path / "composition" + sched_store, run_store = _stores(comp) + parent_lock = SchedulerOwnership(tmp_path, owner="parent").acquire() + try: + # A different lock file: simultaneous hold proves independence. + child_lock = SchedulerOwnership(comp, owner="child").acquire() + try: + assert parent_lock.held and child_lock.held + assert child_lock.covers(sched_store.root, run_store.root) is True + assert parent_lock.covers(sched_store.root, run_store.root) is False + now = datetime.now(UTC) + sched_recovery.recover( + schedule_store=sched_store, + run_store=run_store, + now=now, + ownership=child_lock, + ) + before_runs = run_store.list_runs() + before_admissions = run_store.list_admissions() + with pytest.raises(SecondOwnerError): + sched_recovery.recover( + schedule_store=FileScheduleStore(comp / "sched"), + run_store=FileRunStore(comp / "runs"), + now=now, + ownership=parent_lock, + ) + # Wrong-root rejection happens before any write. + assert FileRunStore(comp / "runs").list_runs() == before_runs + assert FileRunStore(comp / "runs").list_admissions() == before_admissions + finally: + child_lock.release() + finally: + parent_lock.release() + + +def test_strict_ancestor_lock_rejected_for_nested_composition(tmp_path: Path) -> None: + comp = tmp_path / "composition" + sched_store, run_store = _stores(comp) + grandparent = SchedulerOwnership(tmp_path.parent, owner="grandparent").acquire() + try: + assert grandparent.covers(sched_store.root, run_store.root) is False + finally: + grandparent.release() + + +def test_competing_processes_share_the_canonical_lock(tmp_path: Path) -> None: + comp = tmp_path / "composition" + comp.mkdir(parents=True) + sched_store, run_store = _stores(comp) + identity = Path(canonical_lock_root(sched_store.root, run_store.root) or "") + holder = SchedulerOwnership(identity, owner="parent").acquire() + try: + script_path = tmp_path / "compete_canonical.py" + script_path.write_text( + "import sys\n" + "sys.path.insert(0, '.')\n" + "from pathlib import Path\n" + "from wf_scheduling.ownership import canonical_lock_root, SchedulerOwnership, SecondOwnerError\n" + f"sched_root = Path({str(sched_store.root)!r})\n" + f"runs_root = Path({str(run_store.root)!r})\n" + "identity = Path(canonical_lock_root(sched_root, runs_root))\n" + "try:\n" + " SchedulerOwnership(identity, owner='child').acquire()\n" + "except SecondOwnerError:\n" + " print('second-owner-rejected')\n" + " raise SystemExit(0)\n" + "print('unexpectedly-acquired')\n" + "raise SystemExit(1)\n", + encoding="utf-8", + ) + proc = subprocess.run( + [sys.executable, str(script_path)], + capture_output=True, + text=True, + timeout=60, + ) + assert proc.returncode == 0, proc.stderr + assert "second-owner-rejected" in proc.stdout + # A lock on the parent is an independent file: it acquires, but it + # cannot authorize the composition while the canonical lock is held. + outsider = SchedulerOwnership(tmp_path, owner="outsider").acquire() + try: + assert outsider.held + assert outsider.covers(sched_store.root, run_store.root) is False + with pytest.raises(SecondOwnerError): + sched_recovery.recover( + schedule_store=FileScheduleStore(comp / "sched"), + run_store=FileRunStore(comp / "runs"), + now=datetime.now(UTC), + ownership=outsider, + ) + finally: + outsider.release() + finally: + holder.release() + + +def test_released_or_missing_ownership_rejected(tmp_path: Path) -> None: + comp = tmp_path / "composition" + sched_store, run_store = _stores(comp) + never = SchedulerOwnership(comp, owner="never-acquired") + assert never.covers(sched_store.root, run_store.root) is False + with pytest.raises(SecondOwnerError): + sched_recovery.recover( + schedule_store=sched_store, + run_store=run_store, + now=datetime.now(UTC), + ownership=never, + ) + held = SchedulerOwnership(comp, owner="test").acquire() + assert held.covers(sched_store.root, run_store.root) is True + held.release() + assert held.covers(sched_store.root, run_store.root) is False + with pytest.raises(SecondOwnerError): + sched_recovery.recover( + schedule_store=sched_store, + run_store=run_store, + now=datetime.now(UTC), + ownership=held, + ) + + +def test_root_mutation_cannot_redirect_held_authority(tmp_path: Path) -> None: + comp = tmp_path / "composition" + sched_store, run_store = _stores(comp) + other_sched, other_runs = _stores(tmp_path / "elsewhere") + ownership = SchedulerOwnership(comp, owner="test").acquire() + try: + assert ownership.covers(sched_store.root, run_store.root) is True + ownership.root = tmp_path / "elsewhere" + # Frozen identity: still proves the locked composition, never the + # mutated one. + assert ownership.canonical_root() == canonical_store_path(comp) + assert ownership.covers(sched_store.root, run_store.root) is True + assert ownership.covers(other_sched.root, other_runs.root) is False + with pytest.raises(SecondOwnerError): + sched_recovery.recover( + schedule_store=other_sched, + run_store=other_runs, + now=datetime.now(UTC), + ownership=ownership, + ) + finally: + ownership.release() + + +def test_canonical_windows_path_handling_retained(tmp_path: Path) -> None: + base = tmp_path / "sched" + dotted = tmp_path / "sub" / ".." / "sched" + assert canonical_store_path(dotted) == canonical_store_path(base) + if os.name == "nt": + assert canonical_store_path("C:\\Temp\\SCHED") == canonical_store_path( + "c:/temp/sched" + )