diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index f3245a83..890588dd 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -66,6 +66,36 @@ def _is_oneshot(source: OccurrenceSource) -> bool: return isinstance(source, OneShotSource) +def _save_consumed_max(store: Any, sched_id: str, instant: datetime) -> None: + """Advance the consumed watermark without ever moving it backwards. + + Admission decides instants out of order across retries (a held + candidate admitted after the watermark already advanced past it); a + backwards write would resurrect the intervening instants on restart. + """ + existing = store.get_consumed(sched_id) + if existing is None or instant > existing: + store.save_consumed(sched_id, instant) + + +def _latest_eligible(source: OccurrenceSource, now: datetime) -> datetime | None: + """Select the latest eligible occurrence ``<= now`` with bounded queries. + + ``prev_before`` is exclusive, so a poll exactly at a due instant would + miss it. ``prev_before(now)`` is the greatest occurrence strictly before + ``now``; at most one occurrence (``now`` itself) can lie in between, so + a single bounded ``next_after`` probe closes the gap without any custom + calendar math. + """ + latest = source.prev_before(now) + if latest is None: + return None + forward = source.next_after(latest) + if forward is not None and forward <= now: + latest = forward + return latest + + class Scheduler: """File-store scheduler core with injected clock and collaborators. @@ -215,7 +245,7 @@ class Scheduler: cand = self.schedule_store.get_candidate(sched.id) if cand is not None and cand.intended_at == intended: self.schedule_store.save_candidate(None, schedule_id=sched.id) - self.schedule_store.save_consumed(sched.id, intended) + _save_consumed_max(self.schedule_store, sched.id, intended) return None active = self._active(sched.id) if sched.overlap == "skip" and active: @@ -229,7 +259,7 @@ class Scheduler: cand = self.schedule_store.get_candidate(sched.id) if cand is not None and cand.intended_at == intended: self.schedule_store.save_candidate(None, schedule_id=sched.id) - self.schedule_store.save_consumed(sched.id, intended) + _save_consumed_max(self.schedule_store, sched.id, intended) return None if sched.overlap == "parallel" and len(active) >= sched.max_active_runs: self._record( @@ -242,7 +272,7 @@ class Scheduler: cand = self.schedule_store.get_candidate(sched.id) if cand is not None and cand.intended_at == intended: self.schedule_store.save_candidate(None, schedule_id=sched.id) - self.schedule_store.save_consumed(sched.id, intended) + _save_consumed_max(self.schedule_store, sched.id, intended) return None if self._task_load() >= self.capacity: if sched.misfire == "latest": @@ -254,7 +284,7 @@ class Scheduler: ), schedule_id=sched.id, ) - self.schedule_store.save_consumed(sched.id, intended) + _save_consumed_max(self.schedule_store, sched.id, intended) return "held" if (now - intended).total_seconds() > sched.lateness_allowance_s: self._record( @@ -264,7 +294,7 @@ class Scheduler: reason="capacity-deadline", revision=sched.revision, ) - self.schedule_store.save_consumed(sched.id, intended) + _save_consumed_max(self.schedule_store, sched.id, intended) return None return "held-undecided" run_id = self.run_store.allocate_run_id() @@ -297,7 +327,7 @@ class Scheduler: cand = self.schedule_store.get_candidate(sched.id) if cand is not None and cand.intended_at == intended: self.schedule_store.save_candidate(None, schedule_id=sched.id) - self.schedule_store.save_consumed(sched.id, intended) + _save_consumed_max(self.schedule_store, sched.id, intended) from wf_api.run_lifecycle import materialize_admitted_view self.run_store.mark_pending_dispatch(run_id) @@ -557,7 +587,7 @@ class Scheduler: if sched.misfire == "latest": from wf_scheduling.calendar import ScheduleExhaustedError - latest = src.prev_before(now) + latest = _latest_eligible(src, now) if latest is None: raise ScheduleExhaustedError( f"latest-missed lookup exhausted for {sched.id!r}" diff --git a/tests/scheduling/test_latest_boundary.py b/tests/scheduling/test_latest_boundary.py new file mode 100644 index 00000000..088c9a97 --- /dev/null +++ b/tests/scheduling/test_latest_boundary.py @@ -0,0 +1,256 @@ +"""Exact-boundary latest catch-up admits one occurrence, not two (R4/F3). + +The scan-cap/latest branch used exclusive ``prev_before(now)`` for an +inclusive latest-due query and let admission move the consumed watermark +backwards, so a poll exactly at a due instant admitted the previous tick +and the next poll admitted the due tick. The fix selects the latest +eligible occurrence <= now (one bounded forward probe reusing the adapter) +and never moves the consumed watermark backwards. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any, cast + +from tests.scheduling.controlled import ( + DictDeployments, + ScriptedDispatcher, + fixture_environment, +) +from wf_artifacts.runs.store import FileRunStore +from wf_scheduling.calendar import CronSource +from wf_scheduling.models import Schedule +from wf_scheduling.ownership import SchedulerOwnership +from wf_scheduling.poll import SCAN_CAP, 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, s: int = 0) -> datetime: + return datetime(y, mo, d, h, mi, s, tzinfo=UTC) + + +class CountingCron: + """CronSource wrapper counting bounded-lookup calls.""" + + def __init__(self, expression: str, zone: str) -> None: + self._source = CronSource(expression, zone) + self.next_calls = 0 + self.prev_calls = 0 + + def next_after(self, instant: datetime) -> datetime | None: + self.next_calls += 1 + return self._source.next_after(instant) + + def prev_before(self, instant: datetime) -> datetime | None: + self.prev_calls += 1 + return self._source.prev_before(instant) + + +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": "* * * * *", "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, + *, + capacity: int = 4, + script: dict | None = None, +) -> tuple[Scheduler, FileScheduleStore, FileRunStore, dict]: + sched_store = FileScheduleStore(root / "sched") + run_store = FileRunStore(root / "runs") + sources: dict = {} + sched = Scheduler( + schedule_store=sched_store, + run_store=run_store, + sources=sources, + capacity=capacity, + preparer=SchedulePreparer( + DictDeployments({"dep-1": {"rev": 1, "required": []}}), + fixture_environment, + ), + dispatcher=ScriptedDispatcher(script), + ownership=ownership, + ) + return sched, sched_store, run_store, sources + + +def _admitted(store: FileScheduleStore, sid: 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"] == "admitted" + ] + + +def test_exact_boundary_admits_latest_once(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched, store, runs, sources = _harness( + tmp_path, ownership, script={"*": "hang"} + ) + now = ts(2026, 9, 8, 12, 0) + store.create_schedule( + _sched_model("m", misfire="latest", overlap="parallel", max_active_runs=4) + ) + store.save_consumed("m", now - timedelta(days=3)) + sources["m"] = CronSource("* * * * *", "UTC") + sched.poll(now) + assert len(runs.list_runs()) == 1 + admitted = _admitted(store, "m") + assert len(admitted) == 1 + assert datetime.fromisoformat(admitted[0]["resolved_at"]) == now + assert store.get_consumed("m") == now + finally: + ownership.release() + + +def test_second_poll_at_same_instant_admits_nothing(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched, store, runs, sources = _harness( + tmp_path, ownership, script={"*": "hang"} + ) + now = ts(2026, 9, 8, 12, 0) + store.create_schedule( + _sched_model("m", misfire="latest", overlap="parallel", max_active_runs=4) + ) + store.save_consumed("m", now - timedelta(days=3)) + sources["m"] = CronSource("* * * * *", "UTC") + sched.poll(now) + sched.poll(now) + assert len(runs.list_runs()) == 1 + assert len(_admitted(store, "m")) == 1 + finally: + ownership.release() + + +def test_between_ticks_admits_latest(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched, store, runs, sources = _harness( + tmp_path, ownership, script={"*": "hang"} + ) + now = ts(2026, 9, 8, 12, 0, 30) + store.create_schedule( + _sched_model("m", misfire="latest", overlap="parallel", max_active_runs=4) + ) + store.save_consumed("m", now - timedelta(days=3)) + sources["m"] = CronSource("* * * * *", "UTC") + sched.poll(now) + admitted = _admitted(store, "m") + assert len(admitted) == 1 + assert datetime.fromisoformat(admitted[0]["resolved_at"]) == ts( + 2026, 9, 8, 12, 0 + ) + finally: + ownership.release() + + +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() + try: + sched, store, _, sources = _harness(tmp_path, ownership, script={"*": "hang"}) + store.create_schedule( + _sched_model("m", misfire="latest", overlap="parallel", max_active_runs=4) + ) + store.save_consumed("m", now - timedelta(days=3)) + sources["m"] = CronSource("* * * * *", "UTC") + sched.poll(now) + finally: + ownership.release() + ownership2 = SchedulerOwnership(tmp_path / "sched", owner="test2").acquire() + try: + sched2, store2, runs2, sources2 = _harness( + tmp_path, ownership2, script={"*": "hang"} + ) + sources2["m"] = CronSource("* * * * *", "UTC") + sched2.poll(now) + assert len(runs2.list_runs()) == 1 + assert len(_admitted(store2, "m")) == 1 + finally: + ownership2.release() + + +def test_long_catchup_stays_bounded(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched, store, runs, sources = _harness( + tmp_path, ownership, script={"*": "complete"} + ) + now = ts(2026, 9, 8, 12, 0) + store.create_schedule(_sched_model("m", misfire="latest")) + store.save_consumed("m", now - timedelta(days=3)) + src = CountingCron("* * * * *", "UTC") + sources["m"] = src + sched.poll(now) + assert src.next_calls + src.prev_calls <= SCAN_CAP + 2 + assert len(_admitted(store, "m")) == 1 + finally: + ownership.release() + + +def test_skip_policy_exact_boundary_spans_without_admission( + tmp_path: Path, +) -> None: + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched, store, runs, sources = _harness( + tmp_path, ownership, script={"*": "hang"} + ) + now = ts(2026, 9, 8, 12, 0) + store.create_schedule(_sched_model("s", misfire="skip")) + store.save_consumed("s", now - timedelta(days=3)) + sources["s"] = CronSource("* * * * *", "UTC") + sched.poll(now) + assert runs.list_runs() == [] + assert _admitted(store, "s") == [] + assert store.get_consumed("s") == now + sched.poll(now) + assert runs.list_runs() == [] + assert _admitted(store, "s") == [] + finally: + ownership.release() + + +def test_overlap_skip_exact_boundary_skips_once(tmp_path: Path) -> None: + ownership = SchedulerOwnership(tmp_path / "sched", owner="test").acquire() + try: + sched, store, runs, sources = _harness( + tmp_path, ownership, script={"*": "hang"} + ) + now = ts(2026, 9, 8, 12, 0) + store.create_schedule(_sched_model("o", misfire="latest", overlap="skip")) + store.save_consumed("o", now - timedelta(minutes=1)) + sources["o"] = CronSource("* * * * *", "UTC") + # Poll exactly at the due instant: timely admission of 12:00. + sched.poll(now) + assert len(runs.list_runs()) == 1 + # Exactly at the next due instant with the first run still active: + # one terminal skip, no candidate resurrection afterwards. + sched.poll(now + timedelta(minutes=1)) + page = store.list_occurrences("o", limit=100) + kinds = [r["kind"] for r in cast(list[dict[str, Any]], page["occurrences"])] + assert kinds.count("skipped-overlap") == 1 + assert store.get_candidate("o") is None + sched.poll(now + timedelta(minutes=1)) + page = store.list_occurrences("o", limit=100) + kinds = [r["kind"] for r in cast(list[dict[str, Any]], page["occurrences"])] + assert kinds.count("skipped-overlap") == 1 + finally: + ownership.release() diff --git a/tests/scheduling/test_poll.py b/tests/scheduling/test_poll.py index 4d63ada8..a9a239e0 100644 --- a/tests/scheduling/test_poll.py +++ b/tests/scheduling/test_poll.py @@ -256,10 +256,15 @@ def test_long_downtime_is_bounded(tmp_path: Path) -> None: assert src.next_calls + src.prev_calls <= SCAN_CAP + 2 admitted = [r for r in _history(store, "m") if r["kind"] == "admitted"] assert len(admitted) == 1 - assert datetime.fromisoformat(admitted[0]["resolved_at"]) == ts(2026, 9, 8, 11, 59) + # Latest-eligible <= now: a poll exactly at a due instant selects that + # instant (F3), not its exclusive predecessor. + assert datetime.fromisoformat(admitted[0]["resolved_at"]) == ts(2026, 9, 8, 12, 0) assert ( len([r for r in _history(store, "m") if r["kind"] == "interval-summary"]) == 1 ) + # A second poll at the same instant admits nothing more. + sched.poll(ts(2026, 9, 8, 12, 0, 0)) + assert len([r for r in _history(store, "m") if r["kind"] == "admitted"]) == 1 sched.ownership.release()