From 73cb08e5ca53aed9f4b0848c98171bd7c10fad7f Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 10:32:30 +0700 Subject: [PATCH] sched: add poll loop with overlap, misfire, fairness, and capacity (T08) --- src/wf_scheduling/poll.py | 590 ++++++++++++++++++++++++++++++++++ tests/scheduling/test_poll.py | 293 +++++++++++++++++ 2 files changed, 883 insertions(+) create mode 100644 src/wf_scheduling/poll.py create mode 100644 tests/scheduling/test_poll.py diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py new file mode 100644 index 00000000..1b34eab9 --- /dev/null +++ b/src/wf_scheduling/poll.py @@ -0,0 +1,590 @@ +"""Poll loop: overlap, misfire, candidates, fairness, capacity (T08). + +Mirrors the reference state model (probes/deployment_scheduling_verify/ +test_schedule_state_model.py) against real file stores. Calendar iteration +uses ``OccurrenceSource`` (``next_after``/``prev_before`` only, never +enumeration); latest-missed catch-up is one bounded ``prev_before`` query +(F1). Overlap decisions precede capacity checks; terminal skips never +reappear; ``latest`` retains at most one candidate. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any, Protocol + +from wf_scheduling.calendar import CronSource, OneShotSource +from wf_scheduling.models import OccurrenceRecord, PendingCandidate +from wf_scheduling.occurrences import occurrence_id + +UTC = timezone.utc +SCAN_CAP = 100 +EPOCH = datetime(2020, 1, 1, tzinfo=UTC) + + +class OccurrenceSource(Protocol): + """Due-instant source over aware datetimes, UTC at the boundary.""" + + def next_after(self, instant: datetime) -> datetime | None: ... + def prev_before(self, instant: datetime) -> datetime | None: ... + + +def source_for_trigger(trigger: Any) -> OccurrenceSource: + """Build a calendar source from a schedule trigger model.""" + kind = trigger.kind if hasattr(trigger, "kind") else trigger.get("kind") + if kind == "cron": + expression = ( + trigger.expression + if hasattr(trigger, "expression") + else trigger["expression"] + ) + zone = ( + trigger.timezone + if hasattr(trigger, "timezone") + else trigger.get("timezone", "UTC") + ) + return CronSource(expression, zone) + if kind == "oneshot": + at = trigger.at if hasattr(trigger, "at") else trigger["at"] + if isinstance(at, str): + at = datetime.fromisoformat(at) + return OneShotSource(at) + raise ValueError(f"unknown trigger kind {kind!r}") + + +def _is_oneshot(source: OccurrenceSource) -> bool: + return hasattr(source, "at") and not hasattr(source, "period") + + +class Scheduler: + """File-store scheduler core with injected clock and controlled dispatch. + + ``outcomes`` maps run ids (or ``"*"``) to ``complete`` | ``hang`` | + ``interrupt`` for deterministic tests. Production dispatch (real runtime) + is wired in T12; here ``hang`` keeps an admitted run occupying its + schedule slot, ``interrupt`` marks it interrupted (schedule slot only), + and ``complete`` marks it completed (releases overlap). + """ + + def __init__( + self, + *, + schedule_store: Any, + run_store: Any, + sources: dict[str, OccurrenceSource], + capacity: int, + outcomes: dict[str, str] | None = None, + deployments: dict[str, dict] | None = None, + ) -> None: + self.schedule_store = schedule_store + self.run_store = run_store + self.sources = sources + self.capacity = capacity + self.outcomes = outcomes or {} + self.deployments = deployments or {"dep-1": {"rev": 1, "required": []}} + self._poll_cursor = 0 + + # -- helpers ------------------------------------------------------ + @staticmethod + def _status_value(run: Any) -> Any: + status = getattr(run, "status", None) + return getattr(status, "value", status) + + def _active(self, sched_id: str) -> list[Any]: + active: list[Any] = [] + for run in self.run_store.list_runs(): + if self._status_value(run) not in ("admitted", "interrupted"): + continue + direct = getattr(run, "sched_id", None) + if direct is not None: + if direct == sched_id: + active.append(run) + continue + if self._run_sched(run) == sched_id: + active.append(run) + return active + + def _run_sched(self, run: Any) -> str | None: + try: + admission = self.run_store.get_admission(run.id) + except KeyError: + return None + return admission.schedule_id + + def _task_load(self) -> int: + count = 0 + for run in self.run_store.list_runs(): + if self._status_value(run) == "admitted": + count += 1 + return count + + def _record( + self, + *, + kind: str, + sched_id: str, + intended: datetime | None = None, + run_id: str | None = None, + reason: str = "", + interval: tuple[datetime, datetime] | None = None, + count: int = 0, + revision: int | None = None, + ) -> None: + oid = ( + occurrence_id(sched_id, intended) + if intended is not None + else f"{sched_id}|no-instant" + ) + self.schedule_store.append_history( + OccurrenceRecord( + schedule_id=sched_id, + occurrence_id=oid, + kind=kind, # type: ignore[arg-type] + resolved_at=intended, + run_id=run_id, + revision=revision, + reason=reason, + interval_start=interval[0] if interval else None, + interval_end=interval[1] if interval else None, + interval_count=count, + created_at=datetime.now(UTC), + ) + ) + + def _admit(self, sched: Any, intended: datetime, now: datetime) -> str | None: + if getattr(sched, "blocked_reason", None): + raise ValueError(getattr(sched, "blocked_reason")) + if not sched.enabled or sched.deleted or sched.paused: + return None + dep = self.deployments.get(sched.deployment_id) + if dep is None: + self._record( + kind="preflight-rejected", + sched_id=sched.id, + intended=intended, + reason="deployment-deleted", + revision=sched.revision, + ) + self.schedule_store.save_consumed(sched.id, intended) + return None + frozen = { + "team": "eng", + "report_time": intended.isoformat(), + "sched": sched.id, + "dep_rev": dep["rev"], + } + missing = [k for k in dep.get("required", []) if k not in frozen] + if missing: + self._record( + kind="preflight-rejected", + sched_id=sched.id, + intended=intended, + reason=f"missing-input:{missing}", + revision=sched.revision, + ) + 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) + return None + active = self._active(sched.id) + if sched.overlap == "skip" and active: + self._record( + kind="skipped-overlap", + sched_id=sched.id, + intended=intended, + reason=f"active={[getattr(r, 'id', None) for r in active]}", + revision=sched.revision, + ) + 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) + return None + if sched.overlap == "parallel" and len(active) >= sched.max_active_runs: + self._record( + kind="skipped-overlap", + sched_id=sched.id, + intended=intended, + reason=f"max_active={sched.max_active_runs}", + revision=sched.revision, + ) + 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) + return None + if self._task_load() >= self.capacity: + if sched.misfire == "latest": + self.schedule_store.save_candidate( + PendingCandidate( + schedule_id=sched.id, + intended_at=intended, + revision=sched.revision, + ), + schedule_id=sched.id, + ) + self.schedule_store.save_consumed(sched.id, intended) + return "held" + if (now - intended).total_seconds() > sched.lateness_allowance_s: + self._record( + kind="skipped-misfire", + sched_id=sched.id, + intended=intended, + reason="capacity-deadline", + revision=sched.revision, + ) + self.schedule_store.save_consumed(sched.id, intended) + return None + return "held-undecided" + run_id = self.run_store.allocate_run_id() + from wf_artifacts.runs.models import RunAdmission + + admission = RunAdmission( + id=run_id, + environment=_test_environment(sched), + resolved_input=dict(frozen), + max_steps=getattr(sched, "max_steps", None), + scheduled_at=intended, + schedule_id=sched.id, + schedule_revision=sched.revision, + deployment_revision=dep["rev"], + created_at=now, + ) + self.run_store.save_admission(admission) + self._record( + kind="admitted", + sched_id=sched.id, + intended=intended, + run_id=run_id, + reason=f"rev={sched.revision}", + revision=sched.revision, + ) + if _is_oneshot(self.sources.get(sched.id)): # type: ignore[arg-type] + sched.exhausted = True + self.schedule_store.save_schedule(sched) + 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) + from wf_api.run_lifecycle import materialize_admitted_view + + materialize_admitted_view(store=self.run_store, admission=admission) + self._dispatch(run_id, now) + return run_id + + def _dispatch(self, run_id: str, now: datetime) -> None: + from wf_artifacts.runs.models import StoredRunStatus + + outcome = self.outcomes.get(run_id, self.outcomes.get("*", "complete")) + record = self.run_store.get_run(run_id) + if outcome == "hang": + return + if outcome == "interrupt": + updated = record.model_copy( + update={"status": StoredRunStatus.INTERRUPTED, "updated_at": now} + ) + self.run_store.save_run(updated) + self._record( + kind="interrupted", + sched_id=self._run_sched(record) or "", + intended=_admission_intended(self.run_store, run_id), + run_id=run_id, + ) + return + assert outcome == "complete" + updated = record.model_copy( + update={"status": StoredRunStatus.COMPLETED, "updated_at": now} + ) + self.run_store.save_run(updated) + self._record( + kind="completed", + sched_id=self._run_sched(record) or "", + intended=_admission_intended(self.run_store, run_id), + run_id=run_id, + ) + + # -- polling -------------------------------------------------------- + def poll(self, now: datetime) -> dict[str, str]: + self._dispatch_pending(now) + schedules = self.schedule_store.list_schedules(include_deleted=True) + ids = sorted(item.id for item in schedules) + if not ids: + return {} + start = self._poll_cursor % len(ids) + order = ids[start:] + ids[:start] + self._poll_cursor += 1 + results: dict[str, str] = {} + by_id = {item.id: item for item in schedules} + for sid in order: + sched = by_id[sid] + if getattr(sched, "blocked_reason", None): + raise ValueError(getattr(sched, "blocked_reason")) + results[sid] = self._poll_one(sched, now) + return results + + def _dispatch_pending(self, now: datetime) -> None: + """Dispatch recovery-materialized runs through capacity checks. + + Recovery NEVER executes (T10): it only completes missing views flagged + pending for this sweep. Pending runs of blocked schedules stay + pending. TODO(T10): introduce the pending-dispatch marker (F11) once + dispatch marks distinguish crash-after-dispatch from never-dispatched; + until then only runs explicitly flagged ``needs_dispatch`` dispatch + here so hanging admitted runs are never re-executed. + """ + for run in sorted(self.run_store.list_runs(), key=lambda r: r.id): + if not getattr(run, "needs_dispatch", False): + continue + try: + admission = self.run_store.get_admission(run.id) + except KeyError: + continue + if admission.schedule_id is None: + continue + try: + sched = self.schedule_store.get_schedule(admission.schedule_id) + except KeyError: + continue + if getattr(sched, "blocked_reason", None): + continue + if self._task_load() >= self.capacity: + continue + self._dispatch(run.id, now) + + def _poll_one(self, sched: Any, now: datetime) -> str: + if sched.deleted: + if self.schedule_store.get_candidate(sched.id) is not None: + self.schedule_store.save_candidate(None, schedule_id=sched.id) + return "deleted" + if not sched.enabled: + if self.schedule_store.get_candidate(sched.id) is not None: + self.schedule_store.save_candidate(None, schedule_id=sched.id) + consumed = self.schedule_store.get_consumed(sched.id) or EPOCH + self.schedule_store.save_consumed(sched.id, max(consumed, now)) + return "disabled" + if sched.exhausted: + return "exhausted" + if sched.paused: + if self.schedule_store.get_candidate(sched.id) is not None: + self.schedule_store.save_candidate(None, schedule_id=sched.id) + consumed = self.schedule_store.get_consumed(sched.id) or EPOCH + self.schedule_store.save_consumed(sched.id, max(consumed, now)) + return "paused" + src = self.sources[sched.id] + consumed = self.schedule_store.get_consumed(sched.id) or EPOCH + if consumed > now: + return "clock-rollback-held" + due: list[datetime] = [] + cursor = consumed + jumped = False + while True: + nxt = src.next_after(cursor) + if nxt is None or nxt > now: + if ( + not due + and _is_oneshot(src) + and nxt is None + and not sched.exhausted + and sched.misfire == "skip" + and getattr(src, "at", None) is not None + and getattr(src, "at") <= now + ): + at = getattr(src, "at") + self._record( + kind="exhausted", + sched_id=sched.id, + intended=at, + reason="oneshot-expired-skip", + revision=sched.revision, + ) + sched.exhausted = True + self.schedule_store.save_schedule(sched) + self.schedule_store.save_consumed(sched.id, now) + self.schedule_store.save_candidate(None, schedule_id=sched.id) + return "exhausted" + break + due.append(nxt) + cursor = nxt + if len(due) >= SCAN_CAP: + jumped = True + break + if jumped: + if sched.misfire == "latest": + latest = src.prev_before(now) + assert latest is not None + 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, + ) + 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, + ) + self.schedule_store.save_consumed(sched.id, now) + return self._admit_held_candidate(sched, now) or "candidate-held" + self._record( + kind="interval-summary", + sched_id=sched.id, + reason="skipped-missed-span", + revision=sched.revision, + ) + self.schedule_store.save_consumed(sched.id, now) + return "span-skipped" + last_result = "idle" + for instant in due: + stored_consumed = self.schedule_store.get_consumed(sched.id) or EPOCH + if instant <= stored_consumed: + continue + age = (now - instant).total_seconds() + if age <= sched.lateness_allowance_s: + old = self.schedule_store.get_candidate(sched.id) + if old is not None and old.intended_at != instant: + self._record( + kind="superseded", + sched_id=sched.id, + intended=old.intended_at, + reason=f"admitted-newer:{instant.isoformat()}", + revision=sched.revision, + ) + self.schedule_store.save_candidate(None, schedule_id=sched.id) + result = self._admit(sched, instant, now) + last_result = f"admit:{result}" + elif sched.misfire == "latest": + old = self.schedule_store.get_candidate(sched.id) + if old is not None and old.intended_at != instant: + self._record( + kind="superseded", + sched_id=sched.id, + intended=old.intended_at, + reason=f"coalesced-into:{instant.isoformat()}", + revision=sched.revision, + ) + self.schedule_store.save_candidate( + PendingCandidate( + schedule_id=sched.id, + intended_at=instant, + revision=sched.revision, + ), + schedule_id=sched.id, + ) + self.schedule_store.save_consumed(sched.id, instant) + last_result = self._admit_held_candidate(sched, now) or "candidate-held" + else: + if _is_oneshot(src) and not sched.exhausted: + self._record( + kind="exhausted", + sched_id=sched.id, + intended=instant, + reason="oneshot-expired-skip", + revision=sched.revision, + ) + sched.exhausted = True + self.schedule_store.save_schedule(sched) + self.schedule_store.save_consumed(sched.id, instant) + last_result = "exhausted" + continue + self._record( + kind="skipped-misfire", + sched_id=sched.id, + intended=instant, + reason=f"age={age:.0f}s", + revision=sched.revision, + ) + self.schedule_store.save_consumed(sched.id, instant) + last_result = "skipped-misfire" + if last_result in ("idle",): + cand = self.schedule_store.get_candidate(sched.id) + if cand is not None: + last_result = self._admit_held_candidate(sched, now) or "candidate-held" + return last_result + + def _admit_held_candidate(self, sched: Any, now: datetime) -> str | None: + cand = self.schedule_store.get_candidate(sched.id) + if cand is None or cand.revision != sched.revision: + if cand is not None and cand.revision != sched.revision: + self._record( + kind="superseded", + sched_id=sched.id, + intended=cand.intended_at, + reason="schedule-edit", + revision=sched.revision, + ) + self.schedule_store.save_candidate(None, schedule_id=sched.id) + return None + result = self._admit(sched, cand.intended_at, now) + return result if result != "held-undecided" else None + + # -- administration --------------------------------------------------- + def resume_schedule(self, sid: str, now: datetime) -> None: + """Unpause: resume selects the next future occurrence.""" + 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.""" + 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( + kind="superseded", + sched_id=sid, + intended=old.intended_at, + reason="schedule-edit", + revision=sched.revision, + ) + 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)) + + +def _test_environment(sched: Any) -> Any: + from wf_artifacts import PinnedRunEnvironment, WorkflowArtifact, WorkflowDeployment + + deployment = WorkflowDeployment( + id=getattr(sched, "deployment_id", "dep-1"), + artifact_id="wf-1", + artifact_version=1, + bindings=[], + ) + artifact = WorkflowArtifact( + id="wf-1", + version=1, + title="Wf-1", + input_schema={"type": "object", "properties": {}}, + output_schema={"type": "object", "properties": {}}, + outcomes=("ok",), + plan={"name": "wf-1", "nodes": [], "edges": []}, + ) + return PinnedRunEnvironment( + deployment=deployment, root_artifact=artifact, child_artifacts=[] + ) + + +def _admission_intended(run_store: Any, run_id: str) -> datetime | None: + try: + return run_store.get_admission(run_id).scheduled_at + except KeyError: + return None diff --git a/tests/scheduling/test_poll.py b/tests/scheduling/test_poll.py new file mode 100644 index 00000000..aa49b7f5 --- /dev/null +++ b/tests/scheduling/test_poll.py @@ -0,0 +1,293 @@ +"""Scheduler poll loop against real file stores (T08).""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + +from wf_artifacts.runs.models import StoredRunStatus +from wf_artifacts.runs.store import FileRunStore +from wf_scheduling.models import Schedule +from wf_scheduling.poll import SCAN_CAP, Scheduler +from wf_scheduling.store import FileScheduleStore + +UTC_TZ = UTC + + +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_TZ) + + +class PeriodicSource: + def __init__(self, period: timedelta, start: datetime) -> None: + self.period = period + self.start = start + self.next_calls = 0 + self.prev_calls = 0 + + def next_after(self, instant: datetime) -> datetime | None: + self.next_calls += 1 + if instant < self.start: + return self.start + n = (instant - self.start) // self.period + 1 + return self.start + n * self.period + + def prev_before(self, instant: datetime) -> datetime | None: + self.prev_calls += 1 + if instant <= self.start: + return None + n = (instant - self.start - timedelta(microseconds=1)) // self.period + return self.start + n * self.period + + +class OneShotSource: + def __init__(self, at: datetime) -> None: + self.at = at + self.next_calls = 0 + self.prev_calls = 0 + + def next_after(self, instant: datetime) -> datetime | None: + self.next_calls += 1 + return self.at if instant < self.at else None + + def prev_before(self, instant: datetime) -> datetime | None: + self.prev_calls += 1 + return self.at if instant > self.at else None + + +def _sched_model(sid: str, start_hint: str = "cron", **kw: Any) -> Schedule: + now = ts(2026, 9, 8, 12, 0) + base: dict[str, Any] = { + "id": sid, + "deployment_id": "dep-1", + "trigger": {"kind": "cron", "expression": "0 * * * *", "timezone": "UTC"}, + "input_bindings": [], + "created_at": now.isoformat(), + "updated_at": now.isoformat(), + } + base.update(kw) + return Schedule.model_validate(base) + + +def _harness( + tmp_path: Path, + *, + capacity: int = 4, + outcomes: dict[str, str] | None = None, + deployments: dict[str, dict] | None = None, +) -> tuple[Scheduler, FileScheduleStore, FileRunStore, dict]: + sched_store = FileScheduleStore(tmp_path / "sched") + run_store = FileRunStore(tmp_path / "runs") + sources: dict = {} + sched = Scheduler( + schedule_store=sched_store, + run_store=run_store, + sources=sources, + capacity=capacity, + outcomes=outcomes, + deployments=deployments + if deployments is not None + else {"dep-1": {"rev": 1, "required": []}}, + ) + return sched, sched_store, run_store, sources + + +def _add( + sched: Scheduler, + store: FileScheduleStore, + sources: dict, + sid: str, + src: Any, + consumed: datetime, + **kw: Any, +) -> Any: + model = _sched_model(sid, **kw) + store.create_schedule(model) + store.save_consumed(sid, consumed) + sources[sid] = src + return store.get_schedule(sid) + + +def _history(store: FileScheduleStore, sid: str) -> list[dict]: + page = store.list_occurrences(sid, limit=100) + return page["occurrences"] # type: ignore[return-value] + + +def test_overlap_skip_blocks_and_late_drops() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + sched, store, runs, sources = _harness(root, capacity=4, outcomes={"*": "hang"}) + t0 = ts(2026, 9, 8, 12, 0) + _add( + sched, + store, + sources, + "a", + PeriodicSource(timedelta(minutes=10), t0), + t0 - timedelta(minutes=10), + ) + sched.poll(t0) + assert len(runs.list_runs()) == 1 + sched.poll(t0 + timedelta(minutes=10)) + kinds = [(r["kind"], r["resolved_at"]) for r in _history(store, "a")] + assert any(k == "skipped-overlap" and "12:10" in str(v) for k, v in kinds) + sched.poll(t0 + timedelta(minutes=30)) + kinds = [(r["kind"], r["resolved_at"]) for r in _history(store, "a")] + assert any(k == "skipped-misfire" for k, _ in kinds) + + +def test_latest_coalesces_to_one_candidate_and_no_double_admit() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + sched, store, runs, sources = _harness(root, capacity=0) + _add( + sched, + store, + sources, + "h", + PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)), + ts(2026, 9, 8, 9, 0), + misfire="latest", + ) + sched.poll(ts(2026, 9, 8, 12, 20)) + cand = store.get_candidate("h") + assert cand is not None and cand.intended_at == ts(2026, 9, 8, 12, 0) + assert [r for r in _history(store, "h") if r["kind"] == "admitted"] == [] + # Capacity returns at 13:00 while 13:00 is also due: exactly one + # admission for 13:00, 12:00 superseded, never both. + sched.capacity = 4 + sched.outcomes = {"*": "complete"} + sched.poll(ts(2026, 9, 8, 13, 0)) + admitted = [r for r in _history(store, "h") if r["kind"] == "admitted"] + assert len(admitted) == 1 + assert datetime.fromisoformat(admitted[0]["resolved_at"]) == ts( + 2026, 9, 8, 13, 0 + ) + assert store.get_candidate("h") is None + + +def test_parallel_limits_and_interrupted_slots() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + sched, store, runs, sources = _harness(root, capacity=4, outcomes={"*": "hang"}) + t0 = ts(2026, 9, 8, 12, 0) + _add( + sched, + store, + sources, + "p", + PeriodicSource(timedelta(minutes=5), t0), + t0 - timedelta(minutes=5), + overlap="parallel", + max_active_runs=2, + ) + sched.poll(t0) + assert sched._task_load() == 1 + first = runs.list_runs()[0] + runs.save_run(first.model_copy(update={"status": StoredRunStatus.INTERRUPTED})) + assert sched._task_load() == 0 + sched.poll(t0 + timedelta(minutes=5)) + assert len(runs.list_runs()) == 2 + sched.poll(t0 + timedelta(minutes=10)) + assert any(r["kind"] == "skipped-overlap" for r in _history(store, "p")) + # Lowering the limit never cancels; blocks new admission instead. + sched_model = store.get_schedule("p") + sched_model.max_active_runs = 1 + store.save_schedule(sched_model) + second = [r for r in runs.list_runs() if r.id != first.id][0] + runs.save_run(second.model_copy(update={"status": StoredRunStatus.INTERRUPTED})) + sched.poll(t0 + timedelta(minutes=15)) + assert any( + r["kind"] == "skipped-overlap" and "12:15" in str(r["resolved_at"]) + for r in _history(store, "p") + ) + + +def test_pause_is_not_downtime() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + sched, store, runs, sources = _harness( + root, capacity=4, outcomes={"*": "complete"} + ) + _add( + sched, + store, + sources, + "a", + PeriodicSource(timedelta(hours=1), ts(2026, 9, 8, 9, 0)), + ts(2026, 9, 8, 9, 0), + misfire="latest", + ) + sched.poll(ts(2026, 9, 8, 9, 0)) + model = store.get_schedule("a") + model.paused = True + store.save_schedule(model) + sched.poll(ts(2026, 9, 8, 10, 30)) + assert store.get_candidate("a") is None + sched.resume_schedule("a", ts(2026, 9, 8, 12, 30)) + sched.poll(ts(2026, 9, 8, 12, 30)) + admitted = [ + r["resolved_at"] for r in _history(store, "a") if r["kind"] == "admitted" + ] + assert "2026-09-08T10:00:00+00:00" not in admitted + assert "2026-09-08T11:00:00+00:00" not in admitted + + +def test_long_downtime_is_bounded(tmp_path: Path) -> None: + sched, store, runs, sources = _harness( + tmp_path, capacity=4, outcomes={"*": "complete"} + ) + src = PeriodicSource(timedelta(minutes=1), ts(2023, 9, 8, 12, 0)) + _add(sched, store, sources, "m", src, ts(2023, 9, 8, 12, 0), misfire="latest") + sched.poll(ts(2026, 9, 8, 12, 0, 0)) + 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) + assert ( + len([r for r in _history(store, "m") if r["kind"] == "interval-summary"]) == 1 + ) + + +def test_fairness_slow_schedule_not_starved(tmp_path: Path) -> None: + sched, store, runs, sources = _harness(tmp_path, capacity=1, outcomes={"*": "hang"}) + t0 = ts(2026, 9, 8, 12, 0) + _add( + sched, + store, + sources, + "fast", + PeriodicSource(timedelta(minutes=1), t0), + t0 - timedelta(minutes=1), + ) + _add(sched, store, sources, "slow", OneShotSource(t0), t0 - timedelta(hours=1)) + sched.poll(t0) + assert any( + r["kind"] == "admitted" and r["schedule_id"] == "fast" + for r in _history(store, "fast") + ) + for run in runs.list_runs(): + runs.save_run(run.model_copy(update={"status": StoredRunStatus.COMPLETED})) + sched.poll(t0 + timedelta(seconds=30)) + assert any(r["kind"] == "admitted" for r in _history(store, "slow")) + + +def test_capacity_wait_then_expire_for_skip(tmp_path: Path) -> None: + sched, store, runs, sources = _harness(tmp_path, capacity=0) + t0 = ts(2026, 9, 8, 12, 0) + _add(sched, store, sources, "a", OneShotSource(t0), t0 - timedelta(hours=1)) + assert sched.poll(t0) == {"a": "admit:held-undecided"} + assert store.get_candidate("a") is None + sched.capacity = 1 + sched.outcomes = {"*": "complete"} + sched.poll(t0 + timedelta(seconds=30)) + assert len([r for r in _history(store, "a") if r["kind"] == "admitted"]) == 1