diff --git a/src/wf_api/schedules.py b/src/wf_api/schedules.py index e4c4a9d3..13118d4f 100644 --- a/src/wf_api/schedules.py +++ b/src/wf_api/schedules.py @@ -154,6 +154,25 @@ class WorkflowScheduleApi: raise KeyError("workflow artifact store is not configured") return self.context.artifact_store + def _candidate_is_admitted( + self, *, schedule_id: str, intended_at: datetime + ) -> bool: + """Return whether durable run admission owns this occurrence. + + Candidate and admission files are written at adjacent lifecycle + boundaries. Suppressing the candidate while the admission is already + visible keeps an interleaved read truthful; startup recovery also + removes the stale candidate durably. + """ + run_store = self.context.run_store + if run_store is None: + return False + return any( + admission.schedule_id == schedule_id + and admission.scheduled_at == intended_at + for admission in run_store.list_admissions() + ) + @staticmethod def _check_trigger(schedule: Schedule) -> None: """Reject triggers the poll loop could not build a source for.""" @@ -476,6 +495,11 @@ class WorkflowScheduleApi: raise ValueError("limit must be between 1 and 100") first_page = cursor in (None, "0") candidate = store.get_candidate(schedule_id) + if candidate is not None and self._candidate_is_admitted( + schedule_id=schedule_id, + intended_at=candidate.intended_at, + ): + candidate = None pending_candidate = candidate if first_page else None stored_cursor = None if cursor == _PENDING_PAGE_CURSOR else cursor stored_limit = limit - 1 if pending_candidate is not None else limit diff --git a/src/wf_scheduling/recovery.py b/src/wf_scheduling/recovery.py index 13d6b458..7eb1a489 100644 --- a/src/wf_scheduling/recovery.py +++ b/src/wf_scheduling/recovery.py @@ -88,6 +88,25 @@ CORRUPT_VIEW_REASON = ( ) +def _clear_candidate_for_admission(schedule_store: Any, admission: Any) -> bool: + """Clear a candidate whose occurrence is already durably admitted. + + Admission is the authoritative occurrence identity. Candidate clearing + follows it as a separate file write, so recovery must remove a candidate + left behind by that torn boundary. A newer candidate for another instant + is left untouched. + """ + schedule_id = getattr(admission, "schedule_id", None) + scheduled_at = getattr(admission, "scheduled_at", None) + if schedule_id is None or scheduled_at is None: + return False + candidate = schedule_store.get_candidate(schedule_id) + if candidate is None or candidate.intended_at != scheduled_at: + return False + schedule_store.save_candidate(None, schedule_id=schedule_id) + return True + + def recover( *, schedule_store: Any, @@ -127,12 +146,18 @@ def recover( # Scoped live repair: never touch admissions, views, markers, or # histories belonging to any other run. The live poll sweep owns # pending dispatch; this path only reconciles the one broken run. + try: + admission = run_store.get_admission(only_run_id) + except KeyError: + admission = None + if admission is not None and _clear_candidate_for_admission( + schedule_store, admission + ): + diags.append(f"{only_run_id}:stale-candidate-cleared") try: runs = [run_store.get_run(only_run_id)] except KeyError: - try: - admission = run_store.get_admission(only_run_id) - except KeyError: + if admission is None: return [f"{only_run_id}:run-unknown"] from wf_api.run_lifecycle import materialize_admitted_view @@ -144,6 +169,8 @@ def recover( # materialized views are completed here and flagged pending for the # capacity-checked poll sweep (exactly once, occurrence already consumed). for admission in run_store.list_admissions(): + if _clear_candidate_for_admission(schedule_store, admission): + diags.append(f"{admission.id}:stale-candidate-cleared") try: run_store.get_run(admission.id) except KeyError: diff --git a/tests/wf_api/test_schedules.py b/tests/wf_api/test_schedules.py index 6596e764..dd75e4ed 100644 --- a/tests/wf_api/test_schedules.py +++ b/tests/wf_api/test_schedules.py @@ -38,6 +38,8 @@ from wf_core import InterruptRequest, RunState, RunStatus from wf_platform import CapabilitySource from wf_scheduling.history import FileScheduleHistoryRecorder, HistoryEntry from wf_scheduling.models import PendingCandidate +from wf_scheduling.ownership import SchedulerOwnership +from wf_scheduling.recovery import recover from wf_scheduling.store import ( FileScheduleStore, ScheduleExistsError, @@ -787,6 +789,66 @@ async def test_occurrences_pending_synthesis_first_page_only(tmp_path: Path) -> assert all(row["kind"] != "pending" for row in plain["occurrences"]) +async def test_admitted_occurrence_replaces_retained_candidate_after_recovery( + tmp_path: Path, +) -> None: + """A torn candidate clear cannot expose an admitted occurrence as pending.""" + root = tmp_path / "admitted_candidate" + api, sched_store, run_store, artifact_store, _ = _harness(root) + await api.create_schedule( + schedule_id="s", deployment_id="dep.personal", trigger=_cron() + ) + intended = ts(2026, 9, 8, 13, 0) + admission = persist_admission( + store=run_store, + run_id=run_store.allocate_run_id(), + environment=_env(artifact_store), + resolved_input={}, + max_steps=None, + scheduled_at=intended, + schedule_id="s", + schedule_revision=1, + ) + FileScheduleHistoryRecorder(sched_store).record( + HistoryEntry( + schedule_id="s", + kind="admitted", + resolved_at=intended, + run_id=admission.id, + revision=1, + reason="rev=1", + created_at=intended, + ) + ) + # Fault injection: admission/history are durable, but candidate clearing + # was lost at the following persistence boundary. + sched_store.save_candidate( + PendingCandidate(schedule_id="s", intended_at=intended, revision=1), + schedule_id="s", + ) + + before_recovery = await api.list_schedule_occurrences(schedule_id="s", limit=1) + assert before_recovery["total"] == 1 + assert before_recovery["occurrences"][0]["kind"] == "admitted" + + ownership = SchedulerOwnership(root, owner="test").acquire() + try: + recover( + schedule_store=sched_store, + run_store=run_store, + now=intended, + ownership=ownership, + ) + finally: + ownership.release() + + assert sched_store.get_candidate("s") is None + after_recovery = await api.list_schedule_occurrences(schedule_id="s", limit=1) + assert after_recovery["total"] == 1 + assert after_recovery["occurrences"][0]["kind"] == "admitted" + assert after_recovery["occurrences"][0]["run_id"] == admission.id + + async def test_occurrences_pending_limit_one_traverses_every_row( tmp_path: Path, ) -> None: