diff --git a/src/wf_scheduling/ownership.py b/src/wf_scheduling/ownership.py index 6b4ca0c6..b80d1db8 100644 --- a/src/wf_scheduling/ownership.py +++ b/src/wf_scheduling/ownership.py @@ -9,11 +9,18 @@ startup is rejected instead of running unprotected. 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. +on ``/scheduler.lock``. No ad-hoc per-run lock files. + +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. """ from __future__ import annotations +import os from pathlib import Path from typing import BinaryIO @@ -26,8 +33,31 @@ class StartupRejected(Exception): """Scheduler startup rejected where locking is unsupported.""" +def canonical_store_path(path: Path | str) -> str: + """Canonicalize a store path for ownership-coverage comparison. + + ``realpath`` resolves symlinks, junctions, ``..`` segments, and + relative paths; ``normcase`` folds case and separators on Windows + (identity on POSIX, which stays case-sensitive). Both sides of every + comparison pass through this function, so platform aliases of one + directory compare equal. Exotic aliases (subst drives, UNC-vs-mapped + paths, 8.3 short names where the OS preserves them) are out of scope: + operators should acquire the lock on the same canonical composition + root the stores live under. + """ + 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) + + class SchedulerOwnership: - """Held exclusive ownership of a schedule store root.""" + """Held exclusive ownership of a store composition root.""" def __init__(self, root: Path, *, owner: str) -> None: self.root = root @@ -49,6 +79,27 @@ class SchedulerOwnership: """ return self._locked + def canonical_root(self) -> str: + """Return the canonical composition root this lock was acquired on.""" + return canonical_store_path(self.root) + + def covers(self, *store_roots: Path | str | None) -> bool: + """Whether this held lock protects the given store roots. + + 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. + """ + 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 + ) + def acquire(self) -> SchedulerOwnership: """Acquire the held lock non-blockingly or raise SecondOwnerError.""" self.lock_path.parent.mkdir(parents=True, exist_ok=True) diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index 9fb22cad..7bd79d6a 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -139,13 +139,20 @@ class Scheduler: def _require_ownership(self) -> None: """Reject schedule mutation/dispatch without proven live ownership. - Runs before any store write or dispatcher side effect: without a - held lock this process cannot prove exclusive ownership, so polling + Runs before any store write or dispatcher side effect: the held + lock must cover the actual schedule and run store composition, not + merely be held on some unrelated directory. Without covering + ownership this process cannot prove exclusive ownership, so polling or administering schedules would risk double admission. """ - if self.ownership is None or not self.ownership.held: + ownership = self.ownership + if ownership is None or not ownership.covers( + getattr(self.schedule_store, "root", None), + getattr(self.run_store, "root", None), + ): raise SecondOwnerError( - "scheduler ownership is required before polling or mutating schedules" + "scheduler ownership must cover the schedule and run stores " + "before polling or mutating schedules" ) # -- helpers ------------------------------------------------------ @@ -431,6 +438,12 @@ class Scheduler: :class:`wf_core.RunState`. Persists through the shared lifecycle boundary, clears the executing mark, and records terminal history. Never re-invokes the dispatcher. + + An admitted status alone is not enough: pending, never-dispatched + work is also admitted. Settlement requires the durable executing + transition and refuses contradictory pending/executing state (a + crash between the transition writes owns that run now, not this + caller); it also refuses runs that already stopped. """ self._require_ownership() from wf_api.run_lifecycle import persist_stopped_run @@ -448,6 +461,12 @@ class Scheduler: raise BlockedSchedule(f"settle missing run view: {run_id!r}") from exc if self._status_value(record) != "admitted": raise BlockedSchedule(f"settle non-admitted run: {run_id!r}") + if self.run_store.is_pending_dispatch(run_id): + raise BlockedSchedule( + f"settle contradictory pending/executing run: {run_id!r}" + ) + if not self.run_store.is_executing(run_id): + raise BlockedSchedule(f"settle without executing transition: {run_id!r}") stopped = persist_stopped_run( store=self.run_store, environment=admission.environment, diff --git a/src/wf_scheduling/prepare.py b/src/wf_scheduling/prepare.py index 03fb8595..2496235c 100644 --- a/src/wf_scheduling/prepare.py +++ b/src/wf_scheduling/prepare.py @@ -17,7 +17,9 @@ from datetime import datetime from typing import Any, Protocol from wf_artifacts.runs.models import PinnedRunEnvironment +from wf_core.errors import WorkflowExecutionError from wf_core.runtime.input_sources import resolve_schedule_input_bindings +from wf_core.runtime.ops.schemas import validate_payload_against_schema from wf_scheduling.occurrences import occurrence_id @@ -63,7 +65,14 @@ class SchedulePreparer: Occurrence references resolve through the shared schedule-expression contract (never graph paths); target conflicts, over-budget trees, and invalid resolved input fail closed as preflight rejections before any - run identity is allocated. + run identity is allocated. The resolved object is additionally + validated against the pinned root artifact's input schema with the same + validator the runtime uses for fresh runs, so a schedule cannot admit + work the deployment contract would refuse. + + The environment is built once per preparation: the deployment revision, + required inputs, and input schema all describe that single captured + snapshot, never independently resolved versions. """ def __init__( @@ -77,9 +86,12 @@ class SchedulePreparer: def prepare( self, *, sched: Any, intended: datetime, now: datetime ) -> PreparedInvocation | PreparationRejected: + environment = self._build_environment(sched) try: - revision = self._deployments.deployment_revision(sched.deployment_id) - required = list(self._deployments.required_inputs(sched.deployment_id)) + revision = self._deployments.deployment_revision(environment.deployment.id) + required = list( + self._deployments.required_inputs(environment.deployment.id) + ) except KeyError: return PreparationRejected(reason="deployment-deleted") occurrence = { @@ -98,8 +110,16 @@ class SchedulePreparer: missing = [key for key in required if key not in resolved] if missing: return PreparationRejected(reason=f"missing-input:{missing}") + try: + validate_payload_against_schema( + environment.root_artifact.input_schema, + dict(resolved), + f"schedule {sched.id} input", + ) + except WorkflowExecutionError as exc: + return PreparationRejected(reason=f"invalid-input:schema:{exc}") return PreparedInvocation( - environment=self._build_environment(sched), + environment=environment, resolved_input=dict(resolved), max_steps=getattr(sched, "max_steps", None), deployment_revision=revision, diff --git a/src/wf_scheduling/recovery.py b/src/wf_scheduling/recovery.py index ef91d3dd..acc7cfed 100644 --- a/src/wf_scheduling/recovery.py +++ b/src/wf_scheduling/recovery.py @@ -31,6 +31,13 @@ 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. """ from __future__ import annotations @@ -72,10 +79,14 @@ def recover( """Reconcile durable state after a restart without executing work.""" from wf_artifacts.runs.models import StoredRunStatus - if ownership is None or not ownership.held: + if ownership is None or not ownership.covers( + getattr(schedule_store, "root", None), + getattr(run_store, "root", None), + ): raise SecondOwnerError( - "scheduler ownership is required before recovery: an unowned " - "recovery could abandon or redispatch another owner's work" + "scheduler ownership must cover the schedule and run stores " + "before recovery: an unowned recovery could abandon or " + "redispatch another owner's work" ) if history is None: history = FileScheduleHistoryRecorder(schedule_store) @@ -96,6 +107,27 @@ 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" + if status in ( + StoredRunStatus.INTERRUPTED.value, + StoredRunStatus.COMPLETED.value, + StoredRunStatus.FAILED.value, + "interrupted", + "completed", + "failed", + ) and (is_executing(run_store, run.id) or _is_pending(run_store, run.id)): + # Completion-window leftovers: the stopped result was persisted + # but marker clearing was lost. The stopped status stands; + # nothing is re-executed. + 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. + 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 @@ -105,6 +137,25 @@ def recover( try: changed, status = _reconcile_summary_from_checkpoint(run_store, run, now) except (ValueError, OSError) as exc: + if _has_recovery_decision(run): + # A decided run stays decided even when checkpoint state + # corrupts afterwards: ensure history from the summary + # pointer, note the corruption, never re-fail. + sched_id, intended, revision = _attribution(run_store, run.id) + if _reconcile_terminal( + history, + sched_id, + run.id, + "failed", + run.latest_checkpoint_id, + intended, + revision, + _decision_reason(run) or "reconciled-on-recovery", + now, + ): + diags.append(f"{run.id}:terminal-reconciled") + diags.append(f"{run.id}:corrupt-checkpoint-noted") + continue _fail_run(run_store, run, f"corrupt checkpoint: {exc}", now, history) diags.append(f"{run.id}:failed-closed") continue @@ -119,9 +170,8 @@ def recover( "completed", "failed", ) and (is_executing(run_store, run.id) or _is_pending(run_store, run.id)): - # Completion-window leftovers: the stopped result was persisted - # but marker clearing was lost. The stopped status stands; - # nothing is re-executed. + # A repaired summary may carry completion-window leftovers (the + # stopped result was persisted but marker clearing was lost). clear_executing(run_store, run.id) clear_pending(run_store, run.id) diags.append(f"{run.id}:completion-window-cleared") @@ -277,6 +327,67 @@ def _reconcile_summary_from_checkpoint( return True, expected.value +def _has_recovery_decision(run: Any) -> bool: + """Whether the summary carries a durable recovery-failure decision.""" + return any( + getattr(item, "code", None) == "schedule-recovery" for item in run.diagnostics + ) + + +def _decision_reason(run: Any) -> str: + """Return the first durable recovery-failure reason, if any.""" + for item in run.diagnostics: + if getattr(item, "code", None) == "schedule-recovery": + return str(getattr(item, "message", "")) + return "" + + +def _failed_decision_stable( + run_store: Any, + run: Any, + history: HistoryRecorder, + now: datetime, + diags: list[str], +) -> bool: + """Leave a decided FAILED summary untouched, ensuring history only. + + 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. + """ + try: + latest = run_store.get_latest_checkpoint(run.id) + except KeyError: + latest = None + 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. + 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") + return True + + def _attribution( run_store: Any, run_id: str ) -> tuple[str | None, datetime | None, int | None]: @@ -366,12 +477,13 @@ def _fail_run( ) -> 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. + The reason is always written into the run summary diagnostics, and a + failed run is never resumable. 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 + from wf_artifacts.runs.models import ResumeReadiness, StoredRunStatus diagnostic = DependencyDiagnostic( severity=DiagnosticSeverity.ERROR, @@ -383,6 +495,7 @@ def _fail_run( updated = run.model_copy( update={ "status": StoredRunStatus.FAILED, + "resume_readiness": ResumeReadiness.NOT_APPLICABLE, "updated_at": now, "diagnostics": [*run.diagnostics, diagnostic], } diff --git a/tests/scheduling/test_dispatch_transition.py b/tests/scheduling/test_dispatch_transition.py index 20f5f6cf..a6e759d8 100644 --- a/tests/scheduling/test_dispatch_transition.py +++ b/tests/scheduling/test_dispatch_transition.py @@ -83,7 +83,7 @@ def _due(sched: Scheduler, store: FileScheduleStore, intended: datetime) -> None def test_executing_is_set_before_executor_runs(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: observed: dict[str, bool] = {} @@ -128,7 +128,7 @@ def test_failure_before_transition_never_dispatches(tmp_path: Path) -> None: sched_store = FileScheduleStore(tmp_path / "sched") run_store = FailMarkOnce(tmp_path / "runs") - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched = Scheduler( schedule_store=sched_store, @@ -160,7 +160,7 @@ def test_failure_before_transition_never_dispatches(tmp_path: Path) -> None: def test_capacity_shortage_keeps_pending_undispatched(tmp_path: Path) -> None: calls: list[str] = [] - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs = _harness(tmp_path, ownership, capacity=0) intended = ts(2026, 9, 8, 12, 0) @@ -185,7 +185,7 @@ def test_capacity_shortage_keeps_pending_undispatched(tmp_path: Path) -> None: def test_pending_on_terminal_run_clears_without_redispatch(tmp_path: Path) -> None: calls: list[str] = [] - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs = _harness(tmp_path, ownership) @@ -230,7 +230,7 @@ def test_pending_without_admission_fails_closed(tmp_path: Path) -> None: WorkflowRunRecord, ) - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"}) intended = ts(2026, 9, 8, 12, 0) @@ -260,7 +260,7 @@ def test_pending_without_admission_fails_closed(tmp_path: Path) -> None: def test_executing_admitted_is_abandoned_by_recovery(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"}) intended = ts(2026, 9, 8, 12, 0) @@ -289,7 +289,7 @@ def test_executing_admitted_is_abandoned_by_recovery(tmp_path: Path) -> None: def test_settle_hanging_run_persists_and_clears(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"}) intended = ts(2026, 9, 8, 12, 0) @@ -350,7 +350,7 @@ def _child_poll_and_die( capacity=4, preparer=_Prep(_Dict({"dep-1": {"rev": 1, "required": []}}), _env), dispatcher=_SD({"*": killer}), - ownership=_Own(base / "sched", owner="child").acquire(), + ownership=_Own(base, owner="child").acquire(), ) sched.poll(intended) @@ -376,7 +376,7 @@ def test_subprocess_death_after_side_effect_dispatches_once(tmp_path: Path) -> N # Restart with fresh objects: the side effect count stays one and the run # is abandoned, never redispatched. - ownership = SchedulerOwnership(root / "sched", owner="parent").acquire() + ownership = SchedulerOwnership(root, owner="parent").acquire() try: sched, store, runs = _harness(root, ownership, script={"*": "hang"}) sched.sources["a"] = OneShotSource(intended) diff --git a/tests/scheduling/test_history_reconcile.py b/tests/scheduling/test_history_reconcile.py index 38bddc7e..5c3ed764 100644 --- a/tests/scheduling/test_history_reconcile.py +++ b/tests/scheduling/test_history_reconcile.py @@ -124,7 +124,7 @@ def _entries(store: FileScheduleStore, sid: str, kind: str) -> list[dict[str, An def _recover( sched_store: FileScheduleStore, run_store: FileRunStore, now: datetime ) -> list[str]: - ownership = SchedulerOwnership(sched_store.root, owner="test").acquire() + ownership = SchedulerOwnership(sched_store.root.parent, owner="test").acquire() try: return sched_recovery.recover( schedule_store=sched_store, @@ -223,7 +223,7 @@ def test_history_write_failure_reconciles_once_on_recovery(tmp_path: Path) -> No sched_store = FailCompletedHistoryOnce(tmp_path / "sched") run_store = FileRunStore(tmp_path / "runs") - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched = Scheduler( schedule_store=sched_store, @@ -271,7 +271,7 @@ def test_history_write_failure_reconciles_once_on_recovery(tmp_path: Path) -> No def test_dispatch_and_recovery_share_entry_identity(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched_store = FileScheduleStore(tmp_path / "sched") run_store = FileRunStore(tmp_path / "runs") @@ -383,7 +383,7 @@ def test_admission_tear_returns_existing_run_without_dup(tmp_path: Path) -> None ) 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() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched = _owned_scheduler( sched_store, run_store, ownership, script={"*": "hang"} @@ -426,7 +426,7 @@ def test_parallel_tear_never_double_admits(tmp_path: Path) -> None: ) 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() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched = _owned_scheduler( sched_store, run_store, ownership, script={"*": "hang"} @@ -460,7 +460,7 @@ def test_consumed_write_failure_recovers_without_dup(tmp_path: Path) -> None: 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() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched = _owned_scheduler( sched_store, run_store, ownership, script={"*": "hang"} diff --git a/tests/scheduling/test_latest_boundary.py b/tests/scheduling/test_latest_boundary.py index 088c9a97..51070ae4 100644 --- a/tests/scheduling/test_latest_boundary.py +++ b/tests/scheduling/test_latest_boundary.py @@ -98,7 +98,7 @@ def _admitted(store: FileScheduleStore, sid: str) -> list[dict[str, Any]]: def test_exact_boundary_admits_latest_once(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs, sources = _harness( tmp_path, ownership, script={"*": "hang"} @@ -120,7 +120,7 @@ def test_exact_boundary_admits_latest_once(tmp_path: Path) -> None: def test_second_poll_at_same_instant_admits_nothing(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs, sources = _harness( tmp_path, ownership, script={"*": "hang"} @@ -140,7 +140,7 @@ def test_second_poll_at_same_instant_admits_nothing(tmp_path: Path) -> None: def test_between_ticks_admits_latest(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs, sources = _harness( tmp_path, ownership, script={"*": "hang"} @@ -163,7 +163,7 @@ def test_between_ticks_admits_latest(tmp_path: Path) -> None: def test_restart_after_catchup_admits_nothing(tmp_path: Path) -> None: now = ts(2026, 9, 8, 12, 0) - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, _, sources = _harness(tmp_path, ownership, script={"*": "hang"}) store.create_schedule( @@ -174,7 +174,7 @@ def test_restart_after_catchup_admits_nothing(tmp_path: Path) -> None: sched.poll(now) finally: ownership.release() - ownership2 = SchedulerOwnership(tmp_path / "sched", owner="test2").acquire() + ownership2 = SchedulerOwnership(tmp_path, owner="test2").acquire() try: sched2, store2, runs2, sources2 = _harness( tmp_path, ownership2, script={"*": "hang"} @@ -188,7 +188,7 @@ def test_restart_after_catchup_admits_nothing(tmp_path: Path) -> None: def test_long_catchup_stays_bounded(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs, sources = _harness( tmp_path, ownership, script={"*": "complete"} @@ -208,7 +208,7 @@ def test_long_catchup_stays_bounded(tmp_path: Path) -> None: def test_skip_policy_exact_boundary_spans_without_admission( tmp_path: Path, ) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs, sources = _harness( tmp_path, ownership, script={"*": "hang"} @@ -229,7 +229,7 @@ def test_skip_policy_exact_boundary_spans_without_admission( def test_overlap_skip_exact_boundary_skips_once(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs, sources = _harness( tmp_path, ownership, script={"*": "hang"} diff --git a/tests/scheduling/test_ownership_binding.py b/tests/scheduling/test_ownership_binding.py new file mode 100644 index 00000000..d80a6f8b --- /dev/null +++ b/tests/scheduling/test_ownership_binding.py @@ -0,0 +1,179 @@ +"""Ownership is bound to the protected store composition (R4 binding). + +A held lock on an unrelated directory must not authorize mutation of a +store composition it does not cover. The guard validates the held guard +against the actual schedule/run store roots (canonicalized for platform +aliases) instead of trusting the caller to have passed the right lock. +""" + +from __future__ import annotations + +import os +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + +import pytest + +from tests.scheduling.controlled import ( + DictDeployments, + ScriptedDispatcher, + fixture_environment, +) +from wf_artifacts.runs.store import FileRunStore +from wf_scheduling.calendar import OneShotSource +from wf_scheduling.dispatch import StillRunning +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, **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 _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 _due(sched: Scheduler, store: FileScheduleStore, intended: datetime) -> None: + store.create_schedule(_sched_model("a")) + store.save_consumed("a", intended - timedelta(hours=1)) + sched.sources["a"] = OneShotSource(intended) + + +def test_unrelated_held_lock_rejected_before_writes(tmp_path: Path) -> None: + calls: list[str] = [] + + def spy(admission: Any, now: datetime) -> Any: + calls.append(admission.id) + return StillRunning() + + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + ownership = SchedulerOwnership(tmp_path / "other", owner="unrelated").acquire() + try: + sched = _scheduler(sched_store, run_store, ownership) + intended = ts(2026, 9, 8, 12, 0) + _due(sched, sched_store, intended) + with pytest.raises(SecondOwnerError): + sched.poll(intended) + assert calls == [] + assert run_store.list_runs() == [] + assert run_store.list_admissions() == [] + assert sched_store.list_occurrences("a", limit=100)["total"] == 0 + finally: + ownership.release() + + +def test_two_lock_dirs_cannot_operate_same_stores(tmp_path: Path) -> None: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + intended = ts(2026, 9, 8, 12, 0) + covering = SchedulerOwnership(tmp_path, owner="covering").acquire() + other = SchedulerOwnership(tmp_path / "other", owner="other").acquire() + try: + first = _scheduler(sched_store, run_store, covering, script={"*": "hang"}) + _due(first, sched_store, intended) + assert first.poll(intended)["a"].startswith("admit:run-") + # The other directory's lock acquires fine (different lock file) but + # must not authorize the same stores. + second = _scheduler(sched_store, run_store, other, script={"*": "hang"}) + second.sources["a"] = OneShotSource(intended) + with pytest.raises(SecondOwnerError): + second.poll(intended + timedelta(minutes=1)) + assert len(run_store.list_runs()) == 1 + finally: + covering.release() + other.release() + + +def test_recover_with_unrelated_lock_rejected(tmp_path: Path) -> None: + from tests.artifacts.test_run_store import artifact as _artifact + from tests.artifacts.test_run_store import deployment as _deployment + from wf_api.run_lifecycle import materialize_admitted_view, persist_admission + from wf_artifacts import PinnedRunEnvironment + from wf_scheduling import recovery as sched_recovery + + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + admission = persist_admission( + store=run_store, + run_id=run_store.allocate_run_id(), + environment=PinnedRunEnvironment( + deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[] + ), + resolved_input={}, + max_steps=None, + ) + materialize_admitted_view(store=run_store, admission=admission) + ownership = SchedulerOwnership(tmp_path / "other", owner="unrelated").acquire() + try: + with pytest.raises(SecondOwnerError): + sched_recovery.recover( + schedule_store=sched_store, + run_store=run_store, + now=ts(2026, 9, 8, 12, 0), + ownership=ownership, + ) + assert run_store.get_run(admission.id).status.value == "admitted" + finally: + ownership.release() + + +def test_canonicalizer_handles_aliases(tmp_path: Path) -> None: + from wf_scheduling.ownership import canonical_store_path + + 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" + ) + else: + assert canonical_store_path(dotted) != canonical_store_path(tmp_path / "other") + + +def test_covers_rejects_empty_root_set(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + assert ownership.held + assert ownership.covers() is False + assert ownership.covers(tmp_path) is True + finally: + ownership.release() diff --git a/tests/scheduling/test_ownership_guard.py b/tests/scheduling/test_ownership_guard.py index 05076272..bdb8d0a6 100644 --- a/tests/scheduling/test_ownership_guard.py +++ b/tests/scheduling/test_ownership_guard.py @@ -83,7 +83,7 @@ def test_poll_without_ownership_rejects_before_writes(tmp_path: Path) -> None: calls.append(admission.id) raise AssertionError("dispatcher must not run without ownership") - ownership = SchedulerOwnership(tmp_path / "sched", owner="never-acquired") + ownership = SchedulerOwnership(tmp_path, owner="never-acquired") sched, store, runs = _scheduler(tmp_path, ownership) sched.dispatcher = ScriptedDispatcher({"*": spy}) intended = ts(2026, 9, 8, 12, 0) @@ -98,7 +98,7 @@ def test_poll_without_ownership_rejects_before_writes(tmp_path: Path) -> None: def test_poll_with_released_ownership_rejects(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() sched, store, runs = _scheduler(tmp_path, ownership) intended = ts(2026, 9, 8, 12, 0) _due_setup(sched, store, intended) @@ -133,7 +133,7 @@ def test_recover_without_ownership_rejects_before_writes(tmp_path: Path) -> None schedule_revision=1, ) materialize_admitted_view(store=run_store, admission=admission) - ownership = SchedulerOwnership(tmp_path / "sched", owner="never-acquired") + ownership = SchedulerOwnership(tmp_path, owner="never-acquired") with pytest.raises(SecondOwnerError): sched_recovery.recover( schedule_store=sched_store, @@ -146,7 +146,7 @@ def test_recover_without_ownership_rejects_before_writes(tmp_path: Path) -> None def test_held_ownership_allows_poll_and_recovery(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs = _scheduler(tmp_path, ownership, script={"*": "hang"}) intended = ts(2026, 9, 8, 12, 0) @@ -164,9 +164,9 @@ def test_held_ownership_allows_poll_and_recovery(tmp_path: Path) -> None: def test_second_owner_cannot_acquire_for_poll(tmp_path: Path) -> None: - first = SchedulerOwnership(tmp_path / "sched", owner="first").acquire() + first = SchedulerOwnership(tmp_path, owner="first").acquire() try: - second = SchedulerOwnership(tmp_path / "sched", owner="second") + second = SchedulerOwnership(tmp_path, owner="second") with pytest.raises(SecondOwnerError): second.acquire() assert not second.held @@ -181,7 +181,7 @@ def test_second_owner_cannot_acquire_for_poll(tmp_path: Path) -> None: def test_admin_mutations_require_ownership(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test") + ownership = SchedulerOwnership(tmp_path, owner="test") sched, store, _ = _scheduler(tmp_path, ownership) store.create_schedule(_sched_model("a", paused=True)) with pytest.raises(SecondOwnerError): @@ -193,7 +193,7 @@ def test_admin_mutations_require_ownership(tmp_path: Path) -> None: def test_settle_requires_ownership(tmp_path: Path) -> None: - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched, store, runs = _scheduler(tmp_path, ownership, script={"*": "hang"}) intended = ts(2026, 9, 8, 12, 0) diff --git a/tests/scheduling/test_poll.py b/tests/scheduling/test_poll.py index a9a239e0..c702096d 100644 --- a/tests/scheduling/test_poll.py +++ b/tests/scheduling/test_poll.py @@ -88,7 +88,7 @@ def _harness( capacity=capacity, preparer=preparer, dispatcher=ScriptedDispatcher(script), - ownership=SchedulerOwnership(tmp_path / "sched", owner="test").acquire(), + ownership=SchedulerOwnership(tmp_path, owner="test").acquire(), ) return sched, sched_store, run_store, sources diff --git a/tests/scheduling/test_preparation.py b/tests/scheduling/test_preparation.py index af3a7d4c..2620d656 100644 --- a/tests/scheduling/test_preparation.py +++ b/tests/scheduling/test_preparation.py @@ -67,7 +67,7 @@ def _scheduler( capacity=4, preparer=preparer, dispatcher=ScriptedDispatcher(script), - ownership=SchedulerOwnership(tmp_path / "sched", owner="test").acquire(), + ownership=SchedulerOwnership(tmp_path, owner="test").acquire(), ) return sched, sched_store, run_store @@ -173,7 +173,7 @@ def test_missing_required_input_rejects_without_a_run(tmp_path: Path) -> None: capacity=4, preparer=preparer, dispatcher=ScriptedDispatcher({"*": "hang"}), - ownership=SchedulerOwnership(tmp_path / "sched", owner="test").acquire(), + ownership=SchedulerOwnership(tmp_path, owner="test").acquire(), ) intended = ts(2026, 9, 8, 12, 0) sched_store.create_schedule(_sched_model("need")) diff --git a/tests/scheduling/test_preparation_schema.py b/tests/scheduling/test_preparation_schema.py new file mode 100644 index 00000000..2dee03c5 --- /dev/null +++ b/tests/scheduling/test_preparation_schema.py @@ -0,0 +1,298 @@ +"""Resolved input validates against the pinned workflow schema (R4 item 3). + +Preparation checks more than required key names: the resolved object must +satisfy the pinned root artifact's input schema (same snapshot that is +captured in the admission) before any run identity is allocated. Invalid +input rejects the occurrence without admission, run, or dispatch. +""" + +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 +from wf_artifacts.runs.models import PinnedRunEnvironment +from wf_artifacts.runs.store import FileRunStore +from wf_scheduling.calendar import OneShotSource +from wf_scheduling.dispatch import StillRunning +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) + + +STRICT_SCHEMA: dict[str, Any] = { + "type": "object", + "properties": { + "count": {"type": "integer"}, + "nested": { + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + "additionalProperties": False, + }, + }, + "required": ["count"], + "additionalProperties": False, +} + + +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 _strict_env(sched: Any) -> PinnedRunEnvironment: + artifact = _artifact().model_copy( + update={"input_schema": dict(STRICT_SCHEMA)}, + ) + deployment = _deployment().model_copy(update={"id": sched.deployment_id}) + return PinnedRunEnvironment( + deployment=deployment, root_artifact=artifact, child_artifacts=[] + ) + + +def _scheduler( + tmp_path: Path, + ownership: SchedulerOwnership, + bindings: list[dict[str, Any]], + *, + script: dict | None = None, +) -> tuple[Scheduler, FileScheduleStore, FileRunStore, list[str]]: + calls: list[str] = [] + 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": []}}), + _strict_env, + ), + dispatcher=ScriptedDispatcher(script), + ownership=ownership, + ) + intended = ts(2026, 9, 8, 12, 0) + sched_store.create_schedule(_sched_model("s", input_bindings=bindings)) + sched_store.save_consumed("s", intended - timedelta(hours=1)) + sched.sources["s"] = OneShotSource(intended) + dispatches = cast(ScriptedDispatcher, sched.dispatcher) + outcome = (script or {}).get("*", "complete") + assert isinstance(outcome, str) + + def spy(admission: Any, now: datetime) -> Any: + calls.append(admission.id) + if outcome == "hang": + return StillRunning() + from wf_scheduling.dispatch import Stopped + + return Stopped(result=dispatches.finish(admission, outcome)) + + dispatches.script = {"*": spy} + return sched, sched_store, run_store, calls + + +def _rejected_entry(store: FileScheduleStore) -> dict[str, Any]: + page = store.list_occurrences("s", limit=100) + rows = cast(list[dict[str, Any]], page["occurrences"]) + assert len(rows) == 1 + assert rows[0]["kind"] == "preflight-rejected" + return rows[0] + + +def test_wrong_scalar_type_rejects_without_admission(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + sched, store, runs, calls = _scheduler( + tmp_path, + ownership, + [ + { + "target": "count", + "expression": { + "kind": "literal", + "value": "definitely not integer", + }, + } + ], + ) + intended = ts(2026, 9, 8, 12, 0) + assert sched.poll(intended) == {"s": "admit:None"} + entry = _rejected_entry(store) + assert "invalid-input" in entry["reason"] + assert runs.list_runs() == [] + assert runs.list_admissions() == [] + assert calls == [] + finally: + ownership.release() + + +def test_nested_constraint_violation_rejects(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + sched, store, runs, calls = _scheduler( + tmp_path, + ownership, + [ + {"target": "count", "expression": {"kind": "literal", "value": 3}}, + { + "target": "nested", + "expression": { + "kind": "object", + "fields": {"name": {"kind": "literal", "value": 7}}, + }, + }, + ], + ) + assert sched.poll(ts(2026, 9, 8, 12, 0)) == {"s": "admit:None"} + _rejected_entry(store) + assert runs.list_runs() == [] + assert calls == [] + finally: + ownership.release() + + +def test_disallowed_extra_property_rejects(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + sched, store, runs, calls = _scheduler( + tmp_path, + ownership, + [ + {"target": "count", "expression": {"kind": "literal", "value": 3}}, + { + "target": "surprise", + "expression": {"kind": "literal", "value": "x"}, + }, + ], + ) + assert sched.poll(ts(2026, 9, 8, 12, 0)) == {"s": "admit:None"} + _rejected_entry(store) + assert runs.list_runs() == [] + assert calls == [] + finally: + ownership.release() + + +def test_valid_structured_input_admits(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + sched, store, runs, calls = _scheduler( + tmp_path, + ownership, + [ + {"target": "count", "expression": {"kind": "literal", "value": 3}}, + { + "target": "nested", + "expression": { + "kind": "object", + "fields": {"name": {"kind": "literal", "value": "ok"}}, + }, + }, + ], + script={"*": "hang"}, + ) + intended = ts(2026, 9, 8, 12, 0) + result = sched.poll(intended) + assert result["s"].startswith("admit:run-") + run_id = result["s"].split(":", 1)[1] + assert runs.get_admission(run_id).resolved_input == { + "count": 3, + "nested": {"name": "ok"}, + } + assert len(calls) == 1 + finally: + ownership.release() + + +def test_changed_contract_rejects_before_admission(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + sched, store, runs, calls = _scheduler( + tmp_path, + ownership, + [ + {"target": "count", "expression": {"kind": "literal", "value": 3}}, + ], + ) + # The pinned contract now additionally requires a token the + # schedule does not provide. + strict = dict(STRICT_SCHEMA) + strict["required"] = ["count", "token"] + artifact = _artifact().model_copy(update={"input_schema": strict}) + + def changed_env(sched: Any) -> PinnedRunEnvironment: + deployment = _deployment().model_copy(update={"id": sched.deployment_id}) + return PinnedRunEnvironment( + deployment=deployment, root_artifact=artifact, child_artifacts=[] + ) + + sched.preparer = SchedulePreparer( + DictDeployments({"dep-1": {"rev": 2, "required": []}}), changed_env + ) + assert sched.poll(ts(2026, 9, 8, 12, 0)) == {"s": "admit:None"} + _rejected_entry(store) + assert runs.list_runs() == [] + assert runs.list_admissions() == [] + assert calls == [] + finally: + ownership.release() + + +def test_environment_built_once_per_preparation(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + builds: list[str] = [] + + def counting_env(sched: Any) -> PinnedRunEnvironment: + builds.append(sched.id) + return _strict_env(sched) + + 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": []}}), counting_env + ), + dispatcher=ScriptedDispatcher({"*": "hang"}), + ownership=ownership, + ) + intended = ts(2026, 9, 8, 12, 0) + sched_store.create_schedule( + _sched_model( + "s", + input_bindings=[ + {"target": "count", "expression": {"kind": "literal", "value": 1}} + ], + ) + ) + sched_store.save_consumed("s", intended - timedelta(hours=1)) + sched.sources["s"] = OneShotSource(intended) + sched.poll(intended) + assert builds == ["s"] + finally: + ownership.release() diff --git a/tests/scheduling/test_recovery.py b/tests/scheduling/test_recovery.py index 8f4ee123..f47ca971 100644 --- a/tests/scheduling/test_recovery.py +++ b/tests/scheduling/test_recovery.py @@ -58,7 +58,7 @@ def test_recovery_materializes_missing_view_as_pending(tmp_path: Path) -> None: resolved_input={}, max_steps=None, ) - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: diags = sched_recovery.recover( schedule_store=sched_store, @@ -96,7 +96,7 @@ def test_recovery_fails_abandoned_admitted_without_replay(tmp_path: Path) -> Non schedule_revision=1, ) materialize_admitted_view(store=run_store, admission=admission) - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: diags = sched_recovery.recover( schedule_store=sched_store, @@ -133,7 +133,7 @@ def test_recovery_never_executes_pending_until_poll(tmp_path: Path) -> None: schedule_id="a", schedule_revision=1, ) - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: diags = sched_recovery.recover( schedule_store=sched_store, @@ -147,7 +147,7 @@ def test_recovery_never_executes_pending_until_poll(tmp_path: Path) -> None: # Recovery itself produced no terminal history; the poll sweep dispatches. assert sched_store.list_occurrences("a", limit=100)["total"] == 0 sources = {"a": OneShotSource(ts(2026, 9, 8, 12, 0))} - ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched = Scheduler( schedule_store=sched_store, diff --git a/tests/scheduling/test_recovery_stability.py b/tests/scheduling/test_recovery_stability.py new file mode 100644 index 00000000..bde04fd9 --- /dev/null +++ b/tests/scheduling/test_recovery_stability.py @@ -0,0 +1,307 @@ +"""Durable recovery failure is stable across repeated recovery (R4 item 2). + +An older checkpoint cannot supersede a durable abandonment decision: once +recovery fails a run, repeating recovery must leave status, readiness, +diagnostics, and history stable. Only a genuinely newer durable stopped +result may repair a torn summary. Failed recovery carries not-applicable +resume readiness. +""" + +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 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 RunState, RunStatus +from wf_scheduling import recovery as sched_recovery +from wf_scheduling.models import Schedule +from wf_scheduling.ownership import SchedulerOwnership +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 _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 _stopped( + 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 _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) + _stopped(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=attempt_id) + return run_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 test_abandonment_decision_is_stable_across_recovery() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + sched_store = FileScheduleStore(root / "sched") + run_store = FileRunStore(root / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_interrupted(run_store, intended) + _mark_active(run_store, run_id, 9, intended) + first = _recover(sched_store, run_store, intended) + assert any("failed-closed" in d for d in first) + record = run_store.get_run(run_id) + assert record.status.value == "failed" + assert record.resume_readiness.value == "not_applicable" + assert len(record.diagnostics) == 1 + # Fresh store objects across the restart boundary: everything stable. + sched_store2 = FileScheduleStore(root / "sched") + run_store2 = FileRunStore(root / "runs") + second = _recover(sched_store2, run_store2, intended + timedelta(minutes=1)) + assert not any(run_id in d for d in second) + again = run_store2.get_run(run_id) + assert again.status.value == "failed" + assert again.resume_readiness.value == "not_applicable" + assert len(again.diagnostics) == 1 + assert _entries(sched_store2, "a", "failed") == _entries( + sched_store, "a", "failed" + ) + assert len(_entries(sched_store2, "a", "failed")) == 1 + + +def test_failed_readiness_and_inspection_agree() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + sched_store = FileScheduleStore(root / "sched") + run_store = FileRunStore(root / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_interrupted(run_store, intended) + _mark_active(run_store, run_id, 9, intended) + _recover(sched_store, run_store, intended) + record, _ = load_stored_run(run_store, run_id) + assert record.resume_readiness.value == "not_applicable" + try: + restore_interrupted_run(run_store, run_id) + raise AssertionError("failed run must not restore as interrupted") + except ValueError: + pass + + +def test_genuinely_newer_result_repairs_after_decision() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + sched_store = FileScheduleStore(root / "sched") + run_store = FileRunStore(root / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_interrupted(run_store, intended) + _mark_active(run_store, run_id, 9, intended) + _recover(sched_store, run_store, intended) + assert run_store.get_run(run_id).status.value == "failed" + # A genuinely newer stopped result under a new matching attempt: the + # newer checkpoint repairs the summary and completes the attempt. + _mark_active(run_store, run_id, 10, intended) + _stopped(run_store, run_id, RunStatus.INTERRUPTED, attempt_id=10) + diags = _recover(sched_store, run_store, intended + timedelta(minutes=1)) + assert any("fresh-result-resumable" in d for d in diags) + record = run_store.get_run(run_id) + assert record.status.value == "interrupted" + assert record.resume_readiness.value == "ready" + assert run_store.get_resume_attempt(run_id).state == "DONE" # type: ignore[union-attr] + interrupted = _entries(sched_store, "a", "interrupted") + assert {e["checkpoint_id"] for e in interrupted} == {f"{run_id}.000002"} + + +def test_completed_mismatch_decision_is_stable() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + sched_store = FileScheduleStore(root / "sched") + run_store = FileRunStore(root / "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) + _stopped(run_store, run_id, RunStatus.COMPLETED, attempt_id=3) + _mark_active(run_store, run_id, 5, intended) + _recover(sched_store, run_store, intended) + assert run_store.get_run(run_id).status.value == "failed" + second = _recover( + FileScheduleStore(root / "sched"), + FileRunStore(root / "runs"), + intended + timedelta(minutes=1), + ) + assert not any(run_id in d for d in second) + assert len(FileRunStore(root / "runs").get_run(run_id).diagnostics) == 1 + + +def test_legacy_failed_run_gains_history_without_refail() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + sched_store = FileScheduleStore(root / "sched") + run_store = FileRunStore(root / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_interrupted(run_store, intended) + record = run_store.get_run(run_id) + from wf_artifacts.runs.models import StoredRunStatus + + run_store.save_run(record.model_copy(update={"status": StoredRunStatus.FAILED})) + assert _entries(sched_store, "a", "failed") == [] + diags = _recover(sched_store, run_store, intended) + assert any("terminal-reconciled" in d for d in diags) + assert not any("failed-closed" in d for d in diags) + assert run_store.get_run(run_id).diagnostics == [] + assert len(_entries(sched_store, "a", "failed")) == 1 + + +def test_crash_between_decision_writes_recovers_history_once() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + + class FailFailedHistoryOnce(FileScheduleStore): + def __init__(self, root: Path) -> None: + super().__init__(root) + self.armed = False + + def append_history(self, record: Any) -> None: + if self.armed and getattr(record, "kind", None) == "failed": + self.armed = False + raise OSError("injected failed-history failure") + super().append_history(record) + + sched_store = FailFailedHistoryOnce(root / "sched") + run_store = FileRunStore(root / "runs") + sched_store.create_schedule(_sched_model("a")) + intended = ts(2026, 9, 8, 12, 0) + run_id = _admit_interrupted(run_store, intended) + _mark_active(run_store, run_id, 9, intended) + sched_store.armed = True + import pytest + + with pytest.raises(OSError, match="injected failed-history failure"): + _recover(sched_store, run_store, intended) + record = run_store.get_run(run_id) + assert record.status.value == "failed" + assert len(record.diagnostics) == 1 + assert _entries(sched_store, "a", "failed") == [] + # The decision (status + reason) survived; only history is missing. + second = _recover( + FileScheduleStore(root / "sched"), + FileRunStore(root / "runs"), + intended + timedelta(minutes=1), + ) + assert not any("failed-closed" in d for d in second) + assert len(FileRunStore(root / "runs").get_run(run_id).diagnostics) == 1 + assert len(_entries(FileScheduleStore(root / "sched"), "a", "failed")) == 1 diff --git a/tests/scheduling/test_settle_guards.py b/tests/scheduling/test_settle_guards.py new file mode 100644 index 00000000..845b5f99 --- /dev/null +++ b/tests/scheduling/test_settle_guards.py @@ -0,0 +1,252 @@ +"""Late settlement requires a real executing transition (R4 item 4). + +An admitted status also describes pending, never-dispatched work, so the +async-completion seam must demand the executing marker and refuse +contradictory pending/executing state before persisting a late result. +Settling a stopped run twice is rejected; ownership stays enforced. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any, cast + +import pytest + +from tests.scheduling.controlled import ( + DictDeployments, + ScriptedDispatcher, + fixture_environment, +) +from tests.scheduling.controlled import ScriptedDispatcher as SD +from wf_artifacts.runs.store import FileRunStore +from wf_scheduling.calendar import OneShotSource +from wf_scheduling.models import Schedule +from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError +from wf_scheduling.poll import BlockedSchedule, 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 _harness( + root: Path, + ownership: SchedulerOwnership, + *, + script: dict | None = None, +) -> tuple[Scheduler, FileScheduleStore, FileRunStore]: + sched_store = FileScheduleStore(root / "sched") + run_store = FileRunStore(root / "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(script), + ownership=ownership, + ) + return sched, sched_store, run_store + + +def _due(sched: Scheduler, store: FileScheduleStore, intended: datetime) -> None: + store.create_schedule(_sched_model("a")) + store.save_consumed("a", intended - timedelta(hours=1)) + sched.sources["a"] = OneShotSource(intended) + + +def test_pending_undispatched_run_rejects_settlement(tmp_path: Path) -> None: + from tests.artifacts.test_run_store import artifact as _artifact + from tests.artifacts.test_run_store import deployment as _deployment + from wf_api.run_lifecycle import materialize_admitted_view, persist_admission + from wf_artifacts import PinnedRunEnvironment + from wf_core import RunState, RunStatus + + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + sched, _, runs = _harness(tmp_path, ownership, script={"*": "hang"}) + admission = persist_admission( + store=runs, + run_id=runs.allocate_run_id(), + environment=PinnedRunEnvironment( + deployment=_deployment(), + root_artifact=_artifact(), + child_artifacts=[], + ), + resolved_input={}, + max_steps=None, + scheduled_at=ts(2026, 9, 8, 12, 0), + schedule_id="a", + schedule_revision=1, + ) + materialize_admitted_view(store=runs, admission=admission) + runs.mark_pending_dispatch(admission.id) + state = RunState( + workflow_name="sched", + status=RunStatus.COMPLETED, + workflow_input={}, + state={}, + ) + with pytest.raises(BlockedSchedule): + sched.record_stopped_execution(admission.id, state, ts(2026, 9, 8, 12, 0)) + assert runs.get_run(admission.id).status.value == "admitted" + assert runs.is_pending_dispatch(admission.id) + assert not runs.is_executing(admission.id) + with pytest.raises(KeyError): + runs.get_latest_checkpoint(admission.id) + finally: + ownership.release() + + +def test_missing_executing_marker_rejects_settlement(tmp_path: Path) -> None: + from tests.artifacts.test_run_store import artifact as _artifact + from tests.artifacts.test_run_store import deployment as _deployment + from wf_api.run_lifecycle import materialize_admitted_view, persist_admission + from wf_artifacts import PinnedRunEnvironment + from wf_core import RunState, RunStatus + + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + sched, _, runs = _harness(tmp_path, ownership, script={"*": "hang"}) + admission = persist_admission( + store=runs, + run_id=runs.allocate_run_id(), + environment=PinnedRunEnvironment( + deployment=_deployment(), + root_artifact=_artifact(), + child_artifacts=[], + ), + resolved_input={}, + max_steps=None, + scheduled_at=ts(2026, 9, 8, 12, 0), + schedule_id="a", + schedule_revision=1, + ) + materialize_admitted_view(store=runs, admission=admission) + assert not runs.is_pending_dispatch(admission.id) + state = RunState( + workflow_name="sched", + status=RunStatus.COMPLETED, + workflow_input={}, + state={}, + ) + with pytest.raises(BlockedSchedule): + sched.record_stopped_execution(admission.id, state, ts(2026, 9, 8, 12, 0)) + assert runs.get_run(admission.id).status.value == "admitted" + with pytest.raises(KeyError): + runs.get_latest_checkpoint(admission.id) + finally: + ownership.release() + + +def test_contradictory_pending_and_executing_rejects_settlement( + tmp_path: Path, +) -> None: + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"}) + intended = ts(2026, 9, 8, 12, 0) + _due(sched, store, intended) + sched.poll(intended) + run_id = runs.list_runs()[0].id + assert runs.is_executing(run_id) + # Crash shape between mark_executing and pending-clear: both markers. + runs.mark_pending_dispatch(run_id) + dispatcher = cast(SD, sched.dispatcher) + admission = runs.get_admission(run_id) + with pytest.raises(BlockedSchedule): + sched.record_stopped_execution( + run_id, dispatcher.finish(admission, "complete"), intended + ) + assert runs.get_run(run_id).status.value == "admitted" + assert runs.is_executing(run_id) + assert runs.is_pending_dispatch(run_id) + with pytest.raises(KeyError): + runs.get_latest_checkpoint(run_id) + finally: + ownership.release() + + +def test_genuine_execution_settles_successfully(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"}) + intended = ts(2026, 9, 8, 12, 0) + _due(sched, store, intended) + sched.poll(intended) + run_id = runs.list_runs()[0].id + dispatcher = cast(SD, sched.dispatcher) + admission = runs.get_admission(run_id) + sched.record_stopped_execution( + run_id, dispatcher.finish(admission, "complete"), intended + ) + assert runs.get_run(run_id).status.value == "completed" + assert not runs.is_executing(run_id) + finally: + ownership.release() + + +def test_duplicate_settlement_cannot_overwrite_stopped_result( + tmp_path: Path, +) -> None: + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"}) + intended = ts(2026, 9, 8, 12, 0) + _due(sched, store, intended) + sched.poll(intended) + run_id = runs.list_runs()[0].id + dispatcher = cast(SD, sched.dispatcher) + admission = runs.get_admission(run_id) + sched.record_stopped_execution( + run_id, dispatcher.finish(admission, "complete"), intended + ) + with pytest.raises(BlockedSchedule): + sched.record_stopped_execution( + run_id, dispatcher.finish(admission, "interrupt"), intended + ) + assert runs.get_run(run_id).status.value == "completed" + assert len(runs.list_checkpoints(run_id)) == 1 + finally: + ownership.release() + + +def test_settle_without_ownership_rejected(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + sched, store, runs = _harness(tmp_path, ownership, script={"*": "hang"}) + intended = ts(2026, 9, 8, 12, 0) + _due(sched, store, intended) + sched.poll(intended) + run_id = runs.list_runs()[0].id + dispatcher = cast(SD, sched.dispatcher) + admission = runs.get_admission(run_id) + ownership.release() + with pytest.raises(SecondOwnerError): + sched.record_stopped_execution( + run_id, dispatcher.finish(admission, "complete"), intended + ) + assert runs.get_run(run_id).status.value == "admitted" + finally: + ownership.release() diff --git a/tests/scheduling/test_torn_reconcile.py b/tests/scheduling/test_torn_reconcile.py index 44a0ed57..30c4d2bd 100644 --- a/tests/scheduling/test_torn_reconcile.py +++ b/tests/scheduling/test_torn_reconcile.py @@ -92,7 +92,7 @@ class FailSaveRunOnce(FileRunStore): def _recover( sched_store: FileScheduleStore, run_store: FileRunStore, now: datetime ) -> list[str]: - ownership = SchedulerOwnership(sched_store.root, owner="test").acquire() + ownership = SchedulerOwnership(sched_store.root.parent, owner="test").acquire() try: return sched_recovery.recover( schedule_store=sched_store, @@ -373,7 +373,7 @@ def test_real_workflow_interrupt_resume_interrupt_with_torn_resume( 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() + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() try: sched = Scheduler( schedule_store=sched_store,