diff --git a/src/wf_api/schedules.py b/src/wf_api/schedules.py index 1c6a8596..56fa8e2a 100644 --- a/src/wf_api/schedules.py +++ b/src/wf_api/schedules.py @@ -21,6 +21,7 @@ from wf_scheduling.history import FileScheduleHistoryRecorder, HistoryEntry from wf_scheduling.models import OccurrenceRecord, Schedule from wf_scheduling.occurrences import occurrence_id from wf_scheduling.prepare import PreparationRejected, SchedulePreparer +from wf_scheduling.store import StaleScheduleRevisionError from .models import ( JsonProjector, @@ -207,6 +208,10 @@ class WorkflowScheduleApi: ) self._validate_definition(schedule) stored = store.create_schedule(schedule) + # Creation never backfills time before the revision: the consumed + # watermark starts at creation so catch-up only ever covers + # downtime after this revision, never pre-creation instants. + store.save_consumed(schedule_id, now) return _PROJECT_SCHEDULE(stored.model_dump(mode="json")) async def get_schedule(self, *, schedule_id: str) -> ScheduleResult: @@ -251,14 +256,24 @@ class WorkflowScheduleApi: ``expected_revision + 1`` and ``updated_at`` becomes now, while ``created_at`` and the admitted-run snapshots stay untouched (later edits only affect future admissions). Stale revisions raise - ``StaleScheduleRevisionError``. Side effects mirror - ``Scheduler.edit_schedule``: a held candidate is cleared with a - ``superseded``/``schedule-edit`` history row, and the consumed - watermark advances to at least now (no backfill). + ``StaleScheduleRevisionError`` before any write. Crash-safe + ordering: the candidate is cleared (with a ``superseded`` / + ``schedule-edit`` history row) and the consumed watermark advances + BEFORE the revision bump is persisted, so a crash can only leave + the edit unapplied (safe over-skip under the old revision, freely + retryable) and never a bumped revision that backfills pre-edit + instants on restart. A concurrent edit that commits first still + wins via the store's authoritative revision check; our already + applied watermark advance is a safe over-skip in that case too. """ store = self._schedule_store() now = datetime.now(UTC) current = store.get_schedule(schedule_id) + if current.revision != expected_revision: + raise StaleScheduleRevisionError( + f"stale schedule revision for {schedule_id!r}: " + f"expected {expected_revision}, found {current.revision}" + ) data = current.model_dump(mode="python") if deployment_id is not None: data["deployment_id"] = deployment_id @@ -285,7 +300,11 @@ class WorkflowScheduleApi: self._artifact_store().get_deployment(data["deployment_id"]) updated = Schedule.model_validate(data) self._validate_definition(updated) - stored = store.update_schedule(updated, expected_revision=expected_revision) + # Crash-safe ordering: discard the old revision's unadmitted work + # and advance the watermark BEFORE the revision bump below. A + # crash here leaves the edit unapplied under the old revision + # (retryable); the bumped revision can never observe pre-edit + # instants on restart. old_candidate = store.get_candidate(schedule_id) if old_candidate is not None: FileScheduleHistoryRecorder(store).record( @@ -293,7 +312,7 @@ class WorkflowScheduleApi: schedule_id=schedule_id, kind="superseded", resolved_at=old_candidate.intended_at, - revision=stored.revision, + revision=expected_revision + 1, reason="schedule-edit", created_at=now, ) @@ -303,6 +322,7 @@ class WorkflowScheduleApi: store.save_consumed( schedule_id, max(consumed, now) if consumed is not None else now ) + stored = store.update_schedule(updated, expected_revision=expected_revision) return _PROJECT_SCHEDULE(stored.model_dump(mode="json")) async def pause_schedule(self, *, schedule_id: str) -> ScheduleResult: @@ -310,18 +330,24 @@ class WorkflowScheduleApi: Sets ``paused``, clears any held candidate, and advances the consumed watermark to at least now so the paused span is never - backfilled on resume. No history row is written. + backfilled on resume. No history row is written. Crash-safe + ordering: the candidate is cleared and the watermark advances + BEFORE the flag flip is persisted, so a crash can only leave the + schedule unpaused (retryable) and never a paused flag whose span + backfills on resume. """ store = self._schedule_store() now = datetime.now(UTC) - schedule = store.get_schedule(schedule_id) - schedule.paused = True - store.save_schedule(schedule) + # Existence first: unknown ids raise KeyError before any write. + store.get_schedule(schedule_id) store.save_candidate(None, schedule_id=schedule_id) consumed = store.get_consumed(schedule_id) store.save_consumed( schedule_id, max(consumed, now) if consumed is not None else now ) + schedule = store.get_schedule(schedule_id) + schedule.paused = True + store.save_schedule(schedule) return _PROJECT_SCHEDULE(schedule.model_dump(mode="json")) async def resume_schedule(self, *, schedule_id: str) -> ScheduleResult: @@ -329,28 +355,37 @@ class WorkflowScheduleApi: Clears ``paused`` (resume selects the next future occurrence), clears any held candidate, and advances the consumed watermark to - at least now. No history row is written. + at least now. No history row is written. Crash-safe ordering like + pause: candidate and watermark first, flag flip last. """ store = self._schedule_store() now = datetime.now(UTC) - schedule = store.get_schedule(schedule_id) - schedule.paused = False - store.save_schedule(schedule) + # Existence first: unknown ids raise KeyError before any write. + store.get_schedule(schedule_id) + store.save_candidate(None, schedule_id=schedule_id) consumed = store.get_consumed(schedule_id) store.save_consumed( schedule_id, max(consumed, now) if consumed is not None else now ) - store.save_candidate(None, schedule_id=schedule_id) + schedule = store.get_schedule(schedule_id) + schedule.paused = False + store.save_schedule(schedule) return _PROJECT_SCHEDULE(schedule.model_dump(mode="json")) async def delete_schedule(self, *, schedule_id: str) -> ScheduleResult: """Soft-delete one schedule (mirror the poll-loop deleted branch). - Sets ``deleted`` and clears any held candidate. Runs and history + Clears any held candidate, then sets ``deleted``. Runs and history are untouched, the consumed watermark does not advance, and the id - stays reserved (re-creation is rejected by create). + stays reserved (re-creation is rejected by create). Either crash + half heals: an unclearable candidate on a deleted schedule is + dropped by the poll loop, and a cleared candidate on a live + schedule is rebuilt from the untouched watermark. """ store = self._schedule_store() + # Existence first: unknown ids raise KeyError before any write. + store.get_schedule(schedule_id) + store.save_candidate(None, schedule_id=schedule_id) schedule = store.get_schedule(schedule_id) schedule.deleted = True store.save_schedule(schedule) diff --git a/src/wf_client/schedules.py b/src/wf_client/schedules.py index 47bb2c1c..22b45fee 100644 --- a/src/wf_client/schedules.py +++ b/src/wf_client/schedules.py @@ -32,6 +32,8 @@ class Schedule: enabled: bool paused: bool deleted: bool + exhausted: bool + blocked_reason: str | None overlap: str misfire: str max_active_runs: int @@ -50,6 +52,8 @@ class Schedule: enabled: bool, paused: bool, deleted: bool, + exhausted: bool, + blocked_reason: str | None, overlap: str, misfire: str, max_active_runs: int, @@ -65,6 +69,8 @@ class Schedule: object.__setattr__(self, "enabled", enabled) object.__setattr__(self, "paused", paused) object.__setattr__(self, "deleted", deleted) + object.__setattr__(self, "exhausted", exhausted) + object.__setattr__(self, "blocked_reason", blocked_reason) object.__setattr__(self, "overlap", overlap) object.__setattr__(self, "misfire", misfire) object.__setattr__(self, "max_active_runs", max_active_runs) @@ -135,6 +141,8 @@ class Schedule: enabled=wire["enabled"], paused=wire["paused"], deleted=wire["deleted"], + exhausted=wire["exhausted"], + blocked_reason=wire["blocked_reason"], overlap=wire["overlap"], misfire=wire["misfire"], max_active_runs=wire["max_active_runs"], diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index e93f946c..a2c2ae7f 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -582,6 +582,22 @@ class Scheduler: self._execute_guarded(run.id, now) def _poll_one(self, sched: Any, now: datetime) -> str: + # Fresh read per schedule: the tick lists schedules up front while + # same-process administration may commit an edit, pause, resume, or + # delete mid-tick. All admission decisions below (policies, + # revision, flags) use this fresh copy, never the listing + # snapshot. Two narrow residuals remain: calendar iteration uses + # the tick-start occurrence source (a trigger edit takes effect + # on the next tick), and an admission decision already in flight + # cannot observe a concurrent edit — admission stays idempotent + # per (schedule, instant), so the worst case is one run admitted + # under just-superseded terms, never a duplicate or replay. + try: + sched = self.schedule_store.get_schedule(sched.id) + except KeyError: + # Soft deletes never remove the file; a vanishing schedule is + # unexpectedly gone — treat it as deleted work, never admit. + return "deleted" if sched.deleted: if self.schedule_store.get_candidate(sched.id) is not None: self.schedule_store.save_candidate(None, schedule_id=sched.id) @@ -806,22 +822,33 @@ class Scheduler: # -- administration --------------------------------------------------- def resume_schedule(self, sid: str, now: datetime) -> None: - """Unpause: resume selects the next future occurrence.""" + """Unpause: resume selects the next future occurrence. + + Crash-safe ordering shared with the API surface: the candidate + is cleared and the watermark advances BEFORE the flag flip is + persisted, so a crash can only leave the schedule paused + (retryable) and never an unpaused flag whose span backfills. + """ self._require_ownership() + self.schedule_store.get_schedule(sid) + self.schedule_store.save_candidate(None, schedule_id=sid) + consumed = self.schedule_store.get_consumed(sid) or EPOCH + self.schedule_store.save_consumed(sid, max(consumed, now)) sched = self.schedule_store.get_schedule(sid) sched.paused = False self.schedule_store.save_schedule(sched) - consumed = self.schedule_store.get_consumed(sid) or EPOCH - self.schedule_store.save_consumed(sid, max(consumed, now)) - self.schedule_store.save_candidate(None, schedule_id=sid) def edit_schedule(self, sid: str, now: datetime) -> None: - """Definition edit: new revision, discard old candidates, no backfill.""" + """Definition edit: new revision, discard old candidates, no backfill. + + Crash-safe ordering shared with the API surface: old candidates + are discarded (with a superseded row) and the watermark advances + BEFORE the revision bump is persisted, so a crash can only leave + the edit unapplied under the old revision (retryable) and never + a bumped revision that backfills pre-edit instants on restart. + """ self._require_ownership() sched = self.schedule_store.get_schedule(sid) - sched.revision += 1 - sched.updated_at = now - self.schedule_store.save_schedule(sched) old = self.schedule_store.get_candidate(sid) if old is not None: self._record( @@ -829,11 +856,15 @@ class Scheduler: sched_id=sid, intended=old.intended_at, reason="schedule-edit", - revision=sched.revision, + revision=sched.revision + 1, ) self.schedule_store.save_candidate(None, schedule_id=sid) consumed = self.schedule_store.get_consumed(sid) or EPOCH self.schedule_store.save_consumed(sid, max(consumed, now)) + sched = self.schedule_store.get_schedule(sid) + sched.revision += 1 + sched.updated_at = now + self.schedule_store.save_schedule(sched) def _admission_intended(run_store: Any, run_id: str) -> datetime | None: diff --git a/tests/scheduling/test_poll.py b/tests/scheduling/test_poll.py index c702096d..70073589 100644 --- a/tests/scheduling/test_poll.py +++ b/tests/scheduling/test_poll.py @@ -303,3 +303,53 @@ def test_capacity_wait_then_expire_for_skip(tmp_path: Path) -> None: sched.poll(t0 + timedelta(seconds=30)) assert len([r for r in _history(store, "a") if r["kind"] == "admitted"]) == 1 sched.ownership.release() + + +def test_poll_one_rereads_pause_before_deciding(tmp_path: Path) -> None: + sched, store, runs, sources = _harness(tmp_path, script={"*": "hang"}) + t0 = ts(2026, 9, 8, 12, 0) + _add(sched, store, sources, "a", OneShotSource(t0), t0 - timedelta(hours=1)) + stale = store.get_schedule("a") + # Same-process administration lands after the tick listed schedules: + # the decision must observe the pause, not the stale snapshot. + live = store.get_schedule("a") + live.paused = True + store.save_schedule(live) + assert sched._poll_one(stale, t0 + timedelta(seconds=1)) == "paused" + assert runs.list_runs() == [] + assert runs.list_admissions() == [] + sched.ownership.release() + + +def test_poll_one_uses_fresh_definition_after_edit(tmp_path: Path) -> None: + sched, store, runs, sources = _harness(tmp_path, script={"*": "hang"}) + t0 = ts(2026, 9, 8, 12, 0) + _add( + sched, + store, + sources, + "a", + OneShotSource(t0), + t0 - timedelta(hours=1), + input_bindings=[ + {"target": "team", "expression": {"kind": "literal", "value": "old"}} + ], + ) + stale = store.get_schedule("a") + live = store.get_schedule("a") + live.revision = 2 + from wf_core.models.input_bindings import ScheduleInputBinding + + live.input_bindings = [ + ScheduleInputBinding.model_validate( + {"target": "team", "expression": {"kind": "literal", "value": "new"}} + ) + ] + store.save_schedule(live) + result = sched._poll_one(stale, t0 + timedelta(seconds=1)) + assert result.startswith("admit:run-") + run_id = result.split(":", 1)[1] + admission = runs.get_admission(run_id) + assert admission.resolved_input["team"] == "new" + assert admission.schedule_revision == 2 + sched.ownership.release() diff --git a/tests/wf_api/test_schedules.py b/tests/wf_api/test_schedules.py index 32b8212e..471186d3 100644 --- a/tests/wf_api/test_schedules.py +++ b/tests/wf_api/test_schedules.py @@ -260,6 +260,32 @@ def _history_rows( return rows +class FailConsumedOnce(FileScheduleStore): + """Real schedule store failing exactly one watermark write.""" + + def __init__(self, root: Path) -> None: + super().__init__(root) + self.armed = False + + def save_consumed(self, schedule_id: str, consumed_through: datetime) -> None: + if self.armed: + self.armed = False + raise OSError("injected consumed failure") + super().save_consumed(schedule_id, consumed_through) + + +def _fault_harness( + root: Path, +) -> tuple[WorkflowScheduleApi, FailConsumedOnce, FileRunStore]: + artifact_store = FileWorkflowArtifactStore(root) + artifact_store.save_artifact(_artifact()) + artifact_store.save_deployment(_deployment()) + run_store = FileRunStore(root) + sched_store = FailConsumedOnce(root) + context = _context(artifact_store, run_store, sched_store) + return WorkflowScheduleApi(context), sched_store, run_store + + async def test_create_get_list_round_trip_with_defaults(tmp_path: Path) -> None: api, sched_store, _, _, _ = _harness(tmp_path / "round_trip") @@ -463,13 +489,81 @@ async def test_update_happy_path_only_future_work(tmp_path: Path) -> None: async def test_update_stale_revision_rejected(tmp_path: Path) -> None: - api, _, _, _, _ = _harness(tmp_path / "stale") + api, sched_store, _, _, _ = _harness(tmp_path / "stale") + + created = await api.create_schedule( + schedule_id="s", deployment_id="dep.personal", trigger=_cron() + ) + intended = ts(2026, 9, 8, 12, 0) + sched_store.save_candidate( + PendingCandidate(schedule_id="s", intended_at=intended, revision=1), + schedule_id="s", + ) + consumed_before = sched_store.get_consumed("s") + with pytest.raises(StaleScheduleRevisionError): + await api.update_schedule(schedule_id="s", expected_revision=99) + # Stale edits write nothing: no revision bump, no candidate or + # watermark change, no history row. + assert sched_store.get_schedule("s").revision == 1 + assert sched_store.get_schedule("s").updated_at == datetime.fromisoformat( + created["updated_at"] + ) + candidate = sched_store.get_candidate("s") + assert candidate is not None and candidate.intended_at == intended + assert sched_store.get_consumed("s") == consumed_before + assert _history_rows(sched_store, "s") == [] + + +async def test_create_initializes_consumed_no_backfill(tmp_path: Path) -> None: + api, sched_store, _, _, _ = _harness(tmp_path / "create_consumed") + + created = await api.create_schedule( + schedule_id="s", + deployment_id="dep.personal", + trigger=_cron(), + misfire="latest", + ) + created_at = datetime.fromisoformat(created["created_at"]) + consumed = sched_store.get_consumed("s") + assert consumed is not None and consumed >= created_at + + +async def test_update_watermark_failure_leaves_revision(tmp_path: Path) -> None: + api, sched_store, _ = _fault_harness(tmp_path / "torn_update") await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) - with pytest.raises(StaleScheduleRevisionError): - await api.update_schedule(schedule_id="s", expected_revision=99) + intended = ts(2026, 9, 8, 12, 0) + sched_store.save_candidate( + PendingCandidate(schedule_id="s", intended_at=intended, revision=1), + schedule_id="s", + ) + sched_store.armed = True + with pytest.raises(OSError, match="injected consumed failure"): + await api.update_schedule( + schedule_id="s", + expected_revision=1, + overlap="parallel", + max_active_runs=2, + ) + # Crash-safe ordering: the watermark advance precedes the revision + # bump, so a torn edit leaves the old revision (retryable) and can + # never backfill pre-edit instants under the new revision. + assert sched_store.get_schedule("s").revision == 1 + assert sched_store.get_schedule("s").overlap == "skip" + + +async def test_pause_watermark_failure_leaves_flag(tmp_path: Path) -> None: + api, sched_store, _ = _fault_harness(tmp_path / "torn_pause") + + await api.create_schedule( + schedule_id="s", deployment_id="dep.personal", trigger=_cron() + ) + sched_store.armed = True + with pytest.raises(OSError, match="injected consumed failure"): + await api.pause_schedule(schedule_id="s") + assert sched_store.get_schedule("s").paused is False async def test_unknown_schedule_keyerror(tmp_path: Path) -> None: @@ -491,10 +585,14 @@ async def test_unknown_schedule_keyerror(tmp_path: Path) -> None: async def test_pause_resume_delete_transitions(tmp_path: Path) -> None: api, sched_store, _, _, _ = _harness(tmp_path / "transitions") - await api.create_schedule( + created = await api.create_schedule( schedule_id="s", deployment_id="dep.personal", trigger=_cron() ) - assert sched_store.get_consumed("s") is None + # Creation never backfills time before the revision: the consumed + # watermark starts at creation. + created_consumed = sched_store.get_consumed("s") + assert created_consumed is not None + assert created_consumed >= datetime.fromisoformat(created["created_at"]) intended = ts(2026, 9, 8, 12, 0) sched_store.save_candidate( diff --git a/tests/wf_client/test_schedules.py b/tests/wf_client/test_schedules.py index aba43406..927e94e7 100644 --- a/tests/wf_client/test_schedules.py +++ b/tests/wf_client/test_schedules.py @@ -128,6 +128,8 @@ async def test_schedule_snapshot_exposes_minimal_surface() -> None: assert schedule.enabled is True assert schedule.paused is False assert schedule.deleted is False + assert schedule.exhausted is False + assert schedule.blocked_reason is None assert schedule.overlap == "skip" assert schedule.misfire == "skip" assert schedule.max_active_runs == 1