sched: select latest eligible occurrence <= now; watermarks never regress (F3)

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}"

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()

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()