From 953543c5cbb240f1c74141ec5ac5987554e056d7 Mon Sep 17 00:00:00 2001 From: lda Date: Wed, 9 Sep 2026 19:21:03 +0700 Subject: [PATCH] fix: serialize scheduler transitions and coalesce misfires --- src/wf_api/schedules.py | 24 +++++++++- src/wf_scheduling/poll.py | 60 ++++++++++++++++++++++++ src/wf_scheduling/recovery.py | 24 +++++++--- src/wf_scheduling/store.py | 25 ++++++++++ tests/scheduling/test_poll.py | 76 +++++++++++++++++++++++++++++++ tests/scheduling/test_recovery.py | 41 +++++++++++++++++ 6 files changed, 242 insertions(+), 8 deletions(-) diff --git a/src/wf_api/schedules.py b/src/wf_api/schedules.py index 3048a5ce..919e4783 100644 --- a/src/wf_api/schedules.py +++ b/src/wf_api/schedules.py @@ -14,6 +14,7 @@ from __future__ import annotations from collections.abc import Callable from datetime import UTC, datetime +from functools import wraps from typing import Any from wf_core import RunLimits @@ -21,7 +22,11 @@ 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 ScheduleExistsError, StaleScheduleRevisionError +from wf_scheduling.store import ( + ScheduleExistsError, + StaleScheduleRevisionError, + schedule_store_transaction, +) from .models import ( JsonProjector, @@ -38,6 +43,18 @@ _PROJECT_LIST_SCHEDULES = JsonProjector(ListSchedulesResult) _PROJECT_OCCURRENCE_PAGE = JsonProjector(OccurrencePage) +def _serialize_schedule_write(method: Any) -> Any: + """Serialize one API schedule mutation with the store's local lock.""" + + @wraps(method) + async def wrapped(self: Any, *args: Any, **kwargs: Any) -> Any: + store = self._schedule_store() + with schedule_store_transaction(store): + return await method(self, *args, **kwargs) + + return wrapped + + class _ContextDeploymentDirectory: """Deployment contract source over the API artifact store. @@ -177,6 +194,7 @@ class WorkflowScheduleApi: self._check_trigger(schedule) self._validate_sample(schedule) + @_serialize_schedule_write async def create_schedule( self, *, @@ -261,6 +279,7 @@ class WorkflowScheduleApi: } ) + @_serialize_schedule_write async def update_schedule( self, *, @@ -351,6 +370,7 @@ class WorkflowScheduleApi: stored = store.update_schedule(updated, expected_revision=expected_revision) return _PROJECT_SCHEDULE(stored.model_dump(mode="json")) + @_serialize_schedule_write async def pause_schedule(self, *, schedule_id: str) -> ScheduleResult: """Pause one schedule (mirror the poll-loop paused branch). @@ -376,6 +396,7 @@ class WorkflowScheduleApi: store.save_schedule(schedule) return _PROJECT_SCHEDULE(schedule.model_dump(mode="json")) + @_serialize_schedule_write async def resume_schedule(self, *, schedule_id: str) -> ScheduleResult: """Resume one schedule (mirror ``Scheduler.resume_schedule``). @@ -398,6 +419,7 @@ class WorkflowScheduleApi: store.save_schedule(schedule) return _PROJECT_SCHEDULE(schedule.model_dump(mode="json")) + @_serialize_schedule_write async def delete_schedule(self, *, schedule_id: str) -> ScheduleResult: """Soft-delete one schedule (mirror the poll-loop deleted branch). diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index e56d0b40..d65e1bbd 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -38,6 +38,7 @@ from wf_scheduling.ownership import ( describe_unsupported_layout, ) from wf_scheduling.prepare import InvocationPreparer, PreparationRejected +from wf_scheduling.store import schedule_store_transaction UTC = timezone.utc SCAN_CAP = 100 @@ -303,6 +304,23 @@ class Scheduler: return None def _admit(self, sched: Any, intended: datetime, now: datetime) -> str | None: + """Admit one occurrence under the schedule store's local transaction.""" + with schedule_store_transaction(self.schedule_store): + return self._admit_locked(sched, intended, now) + + def _admit_locked( + self, sched: Any, intended: datetime, now: datetime + ) -> str | None: + # The poller's listing is only a fairness snapshot. Re-read under the + # admission transaction so an edit committed before this point wins; + # an edit after this point waits and affects a later occurrence. + try: + current = self.schedule_store.get_schedule(sched.id) + except KeyError: + return None + if current.revision != sched.revision: + return "schedule-changed" + sched = current if getattr(sched, "blocked_reason", None): raise BlockedSchedule(getattr(sched, "blocked_reason")) if not sched.enabled or sched.deleted or sched.paused: @@ -820,6 +838,48 @@ class Scheduler: ) self.schedule_store.save_consumed(sched.id, now) return "span-skipped" + if ( + sched.misfire == "latest" + and due + and any( + (now - instant).total_seconds() > sched.lateness_allowance_s + for instant in due + ) + ): + # A bounded catch-up can still contain several missed instants. + # Coalesce them before the per-instant loop so ``latest`` never + # turns a short downtime into a replay burst. + latest = due[-1] + old = self.schedule_store.get_candidate(sched.id) + if old is not None and old.intended_at != latest: + self._record( + kind="superseded", + sched_id=sched.id, + intended=old.intended_at, + reason=f"coalesced-into:{latest.isoformat()}", + revision=sched.revision, + now=now, + ) + self.schedule_store.save_candidate( + PendingCandidate( + schedule_id=sched.id, + intended_at=latest, + revision=sched.revision, + ), + schedule_id=sched.id, + ) + self._record( + kind="interval-summary", + sched_id=sched.id, + reason="coalesced-missed-span", + revision=sched.revision, + interval=(consumed, now), + count=-1, + now=now, + ) + self.schedule_store.save_consumed(sched.id, now) + held = self._admit_held_candidate(sched, now) + return "skipped-overlap" if held is None else held last_result = "idle" for instant in due: stored_consumed = self.schedule_store.get_consumed(sched.id) or EPOCH diff --git a/src/wf_scheduling/recovery.py b/src/wf_scheduling/recovery.py index ef901faa..13d6b458 100644 --- a/src/wf_scheduling/recovery.py +++ b/src/wf_scheduling/recovery.py @@ -171,6 +171,23 @@ def recover( clear_executing(run_store, run.id) clear_pending(run_store, run.id) diags.append(f"{run.id}:completion-window-cleared") + if status in ( + StoredRunStatus.ADMITTED.value, + StoredRunStatus.INTERRUPTED.value, + "admitted", + "interrupted", + ): + try: + run_store.get_admission(run.id) + except KeyError: + # Without an admission there is no trusted schedule owner or + # invocation snapshot. Fail the view closed and never let it + # remain active merely because recovery cannot attribute it. + _fail_run(run_store, run, CORRUPT_VIEW_REASON, now, history) + clear_pending(run_store, run.id) + clear_executing(run_store, run.id) + diags.append(f"{run.id}:failed-closed") + continue # 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 @@ -283,13 +300,6 @@ def recover( if _is_pending(run_store, run.id): diags.append(f"{run.id}:pending-dispatch-kept") continue - try: - run_store.get_admission(run.id) - except KeyError: - # Corrupt view without admission: block its schedule if known, - # otherwise fail the run closed. - diags.append(f"{run.id}:corrupt-blocked") - continue if active: _fail_run(run_store, run, AMBIGUOUS_REASON, now, history) else: diff --git a/src/wf_scheduling/store.py b/src/wf_scheduling/store.py index 4cf1ba57..bdb5d900 100644 --- a/src/wf_scheduling/store.py +++ b/src/wf_scheduling/store.py @@ -19,15 +19,34 @@ Scheduler ownership across processes arrives in T11. from __future__ import annotations import json +from collections.abc import Iterator +from contextlib import AbstractContextManager, contextmanager from datetime import datetime, timezone from pathlib import Path from threading import RLock +from typing import Any, cast from .models import OccurrenceRecord, PendingCandidate, Schedule, ensure_schedule_id UTC = timezone.utc +@contextmanager +def schedule_store_transaction(store: object) -> Iterator[None]: + """Serialize a multi-file schedule transition when the store supports it. + + The scheduler also accepts test doubles and future transactional stores. + Those may not expose this optional local transaction seam, in which case + the caller retains its existing behavior. + """ + transaction: Any = getattr(store, "transaction", None) + if callable(transaction): + with cast(AbstractContextManager[None], transaction()): + yield + else: + yield + + def _history_identity(record: OccurrenceRecord) -> tuple[object, ...]: """Return fields that identify one durable occurrence transition.""" return ( @@ -66,6 +85,12 @@ class FileScheduleStore: def schedules_dir(self) -> Path: return self.root / "schedules" + @contextmanager + def transaction(self) -> Iterator[None]: + """Hold the process-local store lock across a compound transition.""" + with self._lock: + yield + # -- schedules ------------------------------------------------------ def create_schedule(self, schedule: Schedule) -> Schedule: """Persist a new schedule; deleted ids are never reusable.""" diff --git a/tests/scheduling/test_poll.py b/tests/scheduling/test_poll.py index b15f8ce1..6870b13b 100644 --- a/tests/scheduling/test_poll.py +++ b/tests/scheduling/test_poll.py @@ -173,6 +173,82 @@ def test_latest_coalesces_to_one_candidate_and_no_double_admit() -> None: sched.ownership.release() +def test_latest_short_downtime_does_not_burst_admit(tmp_path: Path) -> None: + """Latest misfire coalesces several bounded missed starts to one run.""" + sched, store, runs, sources = _harness(tmp_path, script={"*": "complete"}) + start = ts(2026, 9, 8, 9, 0) + now = ts(2026, 9, 8, 12, 0) + _add( + sched, + store, + sources, + "m", + PeriodicSource(timedelta(hours=1), start), + start, + misfire="latest", + ) + + sched.poll(now) + + admitted = [row for row in _history(store, "m") if row["kind"] == "admitted"] + assert len(admitted) == 1 + assert datetime.fromisoformat(admitted[0]["resolved_at"]) == now + sched.ownership.release() + + +def test_schedule_edit_between_poll_snapshot_and_admission_cannot_overwrite_terms( + tmp_path: Path, +) -> None: + """An edit between the poll snapshot and admission wins authoritatively.""" + + class EditOnPollRead(FileScheduleStore): + def __init__(self, root: Path) -> None: + super().__init__(root) + self.armed = False + + def get_schedule(self, schedule_id: str) -> Schedule: + current = super().get_schedule(schedule_id) + if self.armed and schedule_id == "a": + self.armed = False + edited = current.model_copy( + update={ + "revision": current.revision + 1, + "updated_at": current.updated_at + timedelta(seconds=1), + } + ) + super().update_schedule(edited, expected_revision=current.revision) + return current + return current + + sched_store = EditOnPollRead(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sources: dict[str, Any] = {} + delegate = SchedulePreparer( + DictDeployments({"dep-1": {"rev": 1, "required": []}}), + fixture_environment, + ) + + sched = Scheduler( + schedule_store=sched_store, + run_store=run_store, + sources=sources, + capacity=1, + preparer=delegate, + dispatcher=ScriptedDispatcher({"*": "complete"}), + ownership=SchedulerOwnership(tmp_path, owner="test").acquire(), + ) + t0 = ts(2026, 9, 8, 12, 0) + _add(sched, sched_store, sources, "a", OneShotSource(t0), t0 - timedelta(hours=1)) + sched_store.armed = True + try: + result = sched.poll(t0) + assert result["a"] == "admit:schedule-changed" + assert sched_store.get_schedule("a").revision == 2 + assert run_store.list_admissions() == [] + finally: + sched.ownership.release() + + def test_parallel_limits_and_interrupted_slots() -> None: import tempfile diff --git a/tests/scheduling/test_recovery.py b/tests/scheduling/test_recovery.py index f47ca971..8bddf37f 100644 --- a/tests/scheduling/test_recovery.py +++ b/tests/scheduling/test_recovery.py @@ -110,6 +110,47 @@ def test_recovery_fails_abandoned_admitted_without_replay(tmp_path: Path) -> Non assert run_store.get_run(admission.id).status.value == "failed" +def test_recovery_fails_corrupt_view_without_admission(tmp_path: Path) -> None: + """A view with no admission is failed closed instead of left active.""" + from tests.artifacts.test_run_store import artifact as _artifact + from tests.artifacts.test_run_store import deployment as _deployment + from wf_artifacts import PinnedRunEnvironment, ResumeReadiness, WorkflowRunRecord + from wf_artifacts.runs.models import StoredRunStatus + + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sched_store.create_schedule(_sched_model("a")) + now = ts(2026, 9, 8, 12, 0) + run_id = run_store.allocate_run_id() + run_store.save_run( + WorkflowRunRecord( + id=run_id, + status=StoredRunStatus.ADMITTED, + resume_readiness=ResumeReadiness.NOT_APPLICABLE, + environment=PinnedRunEnvironment( + deployment=_deployment(), root_artifact=_artifact(), child_artifacts=[] + ), + latest_checkpoint_id=None, + created_at=now, + updated_at=now, + ) + ) + + ownership = SchedulerOwnership(tmp_path, owner="test").acquire() + try: + diags = sched_recovery.recover( + schedule_store=sched_store, + run_store=run_store, + now=now, + ownership=ownership, + ) + finally: + ownership.release() + + assert any(f"{run_id}:failed-closed" in item for item in diags) + assert run_store.get_run(run_id).status.value == "failed" + + def test_recovery_never_executes_pending_until_poll(tmp_path: Path) -> None: from tests.artifacts.test_run_store import artifact as _artifact