From be8b40e184debfd1e819cfd2a43221437a9504e7 Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 10:39:45 +0700 Subject: [PATCH] sched: address R3 blockers on pagination, intervals, and fail-closed polling --- src/wf_scheduling/models.py | 26 +++++ src/wf_scheduling/poll.py | 145 +++++++++++++++++++----- src/wf_scheduling/store.py | 104 +++++++++++++---- tests/scheduling/test_schedule_store.py | 10 +- 4 files changed, 233 insertions(+), 52 deletions(-) diff --git a/src/wf_scheduling/models.py b/src/wf_scheduling/models.py index bc5b9629..4a73bca0 100644 --- a/src/wf_scheduling/models.py +++ b/src/wf_scheduling/models.py @@ -58,6 +58,32 @@ class CronTrigger(BaseModel): expression: str timezone: str = "UTC" + @field_validator("expression") + @classmethod + def _require_five_field_cron(cls, value: str) -> str: + # Thin validation: five fields and croniter-constructible. Full DST + # semantics belong to the calendar adapter (croniter owns them). + if len(value.split()) != 5: + raise ValueError(f"cron expression must have exactly 5 fields: {value!r}") + try: + from wf_scheduling.calendar import CronSource + + CronSource(value, "UTC") + except Exception as exc: + raise ValueError(f"invalid cron expression {value!r}: {exc}") from exc + return value + + @field_validator("timezone") + @classmethod + def _require_valid_zone(cls, value: str) -> str: + try: + from zoneinfo import ZoneInfo + + ZoneInfo(value) + except Exception as exc: + raise ValueError(f"invalid time zone {value!r}") from exc + return value + class OneShotTrigger(BaseModel): """Single offset-aware instant trigger.""" diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index 1b34eab9..3bf0e53f 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -22,6 +22,14 @@ SCAN_CAP = 100 EPOCH = datetime(2020, 1, 1, tzinfo=UTC) +class BlockedSchedule(ValueError): + """A schedule is blocked by a corrupt record and fails closed.""" + + +class InvalidScheduleDefinitionError(ValueError): + """A schedule definition or its trigger source is invalid.""" + + class OccurrenceSource(Protocol): """Due-instant source over aware datetimes, UTC at the boundary.""" @@ -53,7 +61,12 @@ def source_for_trigger(trigger: Any) -> OccurrenceSource: def _is_oneshot(source: OccurrenceSource) -> bool: - return hasattr(source, "at") and not hasattr(source, "period") + # Calendar and test-double one-shots share the class name and an ``at`` + # instant; periodic sources carry a ``period`` instead. isinstance covers + # the calendar type; the name check covers duck-typed test doubles. + if isinstance(source, OneShotSource): + return True + return type(source).__name__ == "OneShotSource" and hasattr(source, "at") class Scheduler: @@ -129,12 +142,17 @@ class Scheduler: interval: tuple[datetime, datetime] | None = None, count: int = 0, revision: int | None = None, + now: datetime | None = None, + admitted_at: datetime | None = None, + started_at: datetime | None = None, ) -> None: - oid = ( - occurrence_id(sched_id, intended) - if intended is not None - else f"{sched_id}|no-instant" - ) + created = now if now is not None else datetime.now(UTC) + if intended is not None: + oid = occurrence_id(sched_id, intended) + elif interval is not None: + oid = f"{sched_id}|summary|{interval[0].isoformat()}|{interval[1].isoformat()}" + else: + oid = f"{sched_id}|summary|{created.isoformat()}" self.schedule_store.append_history( OccurrenceRecord( schedule_id=sched_id, @@ -144,18 +162,30 @@ class Scheduler: run_id=run_id, revision=revision, reason=reason, + admitted_at=admitted_at, + started_at=started_at, interval_start=interval[0] if interval else None, interval_end=interval[1] if interval else None, interval_count=count, - created_at=datetime.now(UTC), + created_at=created, ) ) def _admit(self, sched: Any, intended: datetime, now: datetime) -> str | None: if getattr(sched, "blocked_reason", None): - raise ValueError(getattr(sched, "blocked_reason")) + raise BlockedSchedule(getattr(sched, "blocked_reason")) if not sched.enabled or sched.deleted or sched.paused: return None + # Fail closed before any new admission when a corrupt active view + # exists for this schedule. + for run in self._active(sched.id): + try: + self.run_store.get_admission(run.id) + except KeyError: + reason = f"corrupt run view without admission: {run.id}" + sched.blocked_reason = reason + self.schedule_store.save_schedule(sched) + raise BlockedSchedule(reason) from None dep = self.deployments.get(sched.deployment_id) if dep is None: self._record( @@ -164,7 +194,11 @@ class Scheduler: intended=intended, reason="deployment-deleted", revision=sched.revision, + now=now, ) + 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 frozen = { @@ -259,6 +293,8 @@ class Scheduler: run_id=run_id, reason=f"rev={sched.revision}", revision=sched.revision, + now=now, + admitted_at=now, ) if _is_oneshot(self.sources.get(sched.id)): # type: ignore[arg-type] sched.exhausted = True @@ -277,7 +313,15 @@ class Scheduler: 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 not in ("complete", "hang", "interrupt"): + raise ValueError(f"unknown controlled outcome {outcome!r} for {run_id!r}") + try: + record = self.run_store.get_run(run_id) + except KeyError as exc: + raise BlockedSchedule(f"dispatch missing run view: {run_id!r}") from exc + sched_id = self._run_sched(record) + if sched_id is None: + raise BlockedSchedule(f"dispatch missing admission: {run_id!r}") if outcome == "hang": return if outcome == "interrupt": @@ -287,21 +331,24 @@ class Scheduler: self.run_store.save_run(updated) self._record( kind="interrupted", - sched_id=self._run_sched(record) or "", + sched_id=sched_id, intended=_admission_intended(self.run_store, run_id), run_id=run_id, + now=now, + started_at=now, ) 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 "", + sched_id=sched_id, intended=_admission_intended(self.run_store, run_id), run_id=run_id, + now=now, + started_at=now, ) # -- polling -------------------------------------------------------- @@ -311,15 +358,16 @@ class Scheduler: ids = sorted(item.id for item in schedules) if not ids: return {} - start = self._poll_cursor % len(ids) + cursor = self.schedule_store.get_poll_cursor() + start = cursor % len(ids) order = ids[start:] + ids[:start] - self._poll_cursor += 1 + self.schedule_store.save_poll_cursor(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")) + raise BlockedSchedule(getattr(sched, "blocked_reason")) results[sid] = self._poll_one(sched, now) return results @@ -328,10 +376,10 @@ class Scheduler: 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. + pending. F11 is descoped to T10 here: the pending-dispatch marker does + not exist yet, so only runs explicitly flagged ``needs_dispatch`` + dispatch in this sweep and 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): @@ -371,7 +419,23 @@ class Scheduler: 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] + # Fail closed on corrupt active views before calendar work (F5). + for run in self.run_store.list_runs(): + if self._status_value(run) not in ("admitted", "interrupted"): + continue + try: + self.run_store.get_admission(run.id) + except KeyError: + reason = f"corrupt run view without admission: {run.id}" + sched.blocked_reason = reason + self.schedule_store.save_schedule(sched) + raise BlockedSchedule(reason) from None + try: + src = self.sources[sched.id] + except KeyError as exc: + raise InvalidScheduleDefinitionError( + f"no occurrence source for schedule {sched.id!r}" + ) from exc consumed = self.schedule_store.get_consumed(sched.id) or EPOCH if consumed > now: return "clock-rollback-held" @@ -411,8 +475,13 @@ class Scheduler: break if jumped: if sched.misfire == "latest": + from wf_scheduling.calendar import ScheduleExhaustedError + latest = src.prev_before(now) - assert latest is not None + if latest is None: + raise ScheduleExhaustedError( + f"latest-missed lookup exhausted for {sched.id!r}" + ) old = self.schedule_store.get_candidate(sched.id) if old is not None and old.intended_at != latest: self._record( @@ -421,6 +490,7 @@ class Scheduler: intended=old.intended_at, reason=f"coalesced-into:{latest.isoformat()}", revision=sched.revision, + now=now, ) self.schedule_store.save_candidate( PendingCandidate( @@ -435,14 +505,25 @@ class Scheduler: 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) - return self._admit_held_candidate(sched, now) or "candidate-held" + held = self._admit_held_candidate(sched, now) + if held is None: + # Terminal skip inside admission (e.g. overlap) clears the + # candidate; do not misreport it as held. + return "skipped-overlap" + return held self._record( kind="interval-summary", sched_id=sched.id, reason="skipped-missed-span", revision=sched.revision, + interval=(consumed, now), + count=-1, + now=now, ) self.schedule_store.save_consumed(sched.id, now) return "span-skipped" @@ -484,7 +565,13 @@ class Scheduler: schedule_id=sched.id, ) self.schedule_store.save_consumed(sched.id, instant) - last_result = self._admit_held_candidate(sched, now) or "candidate-held" + held = self._admit_held_candidate(sched, now) + if held is None: + last_result = "skipped-overlap" + elif held == "held-undecided": + last_result = "candidate-held" + else: + last_result = held else: if _is_oneshot(src) and not sched.exhausted: self._record( @@ -511,7 +598,13 @@ class Scheduler: 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" + held = self._admit_held_candidate(sched, now) + if held is None: + last_result = "skipped-overlap" + elif held == "held-undecided": + last_result = "candidate-held" + else: + last_result = held return last_result def _admit_held_candidate(self, sched: Any, now: datetime) -> str | None: @@ -524,11 +617,11 @@ class Scheduler: intended=cand.intended_at, reason="schedule-edit", revision=sched.revision, + now=now, ) 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 + return self._admit(sched, cand.intended_at, now) # -- administration --------------------------------------------------- def resume_schedule(self, sid: str, now: datetime) -> None: diff --git a/src/wf_scheduling/store.py b/src/wf_scheduling/store.py index 5a19f267..002c56f9 100644 --- a/src/wf_scheduling/store.py +++ b/src/wf_scheduling/store.py @@ -5,10 +5,11 @@ Layout per schedule ``/schedules//``: - ``schedule.json`` — the schedule definition (deleted schedules stay with ``deleted=True``; ids are never reused to annex old history). - ``candidate.json`` — at most one pending latest candidate. -- ``consumed.json`` — ``{"consumed_through": }`` watermark plus the - schedule revision it belongs to. +- ``consumed.json`` — ``{"consumed_through": }`` watermark. - ``history.json`` — occurrence-history list for cursor pagination over - ``(resolved_at, occurrence_id)``. + ``(resolved_at, occurrence_id)`` (keyset; integer offsets accepted as a + legacy fallback). +- ``_poll_cursor.json`` — round-robin fairness cursor (persisted). Per-file writes are atomic (tmp + rename); the lock is process-local only. Scheduler ownership across processes arrives in T11. @@ -131,12 +132,33 @@ class FileScheduleStore: return datetime.fromisoformat(value) def save_consumed(self, schedule_id: str, consumed_through: datetime) -> None: + if consumed_through.tzinfo is None or consumed_through.utcoffset() is None: + raise ValueError("consumed_through requires an aware datetime") with self._lock: self._write_json( self._consumed_path(schedule_id), {"consumed_through": consumed_through.astimezone(UTC).isoformat()}, ) + def get_poll_cursor(self) -> int: + """Return the persisted round-robin fairness cursor.""" + path = self.schedules_dir / "_poll_cursor.json" + if not path.exists(): + return 0 + try: + raw = json.loads(path.read_text(encoding="utf-8")) + value = raw.get("cursor", 0) if isinstance(raw, dict) else 0 + return value if isinstance(value, int) and value >= 0 else 0 + except ValueError, AttributeError: + return 0 + + def save_poll_cursor(self, cursor: int) -> None: + """Persist the round-robin fairness cursor.""" + with self._lock: + self._write_json( + self.schedules_dir / "_poll_cursor.json", {"cursor": cursor} + ) + # -- history with cursor pagination ----------------------------------- def append_history(self, record: OccurrenceRecord) -> None: """Append one occurrence-history entry for a schedule.""" @@ -146,6 +168,16 @@ class FileScheduleStore: entries.append(record.model_dump(mode="json")) self._write_json(path, entries) + @staticmethod + def _sort_key( + item: OccurrenceRecord, + ) -> tuple[bool, datetime, str]: + return ( + item.resolved_at is None, + item.resolved_at or datetime.max.replace(tzinfo=UTC), + item.occurrence_id, + ) + def list_occurrences( self, schedule_id: str, @@ -153,38 +185,62 @@ class FileScheduleStore: cursor: str | None = None, limit: int = 50, ) -> dict[str, object]: - """Return one history page ordered by ``(resolved_at, occurrence_id)``.""" + """Return one history page ordered by ``(resolved_at, occurrence_id)``. + + Keyset cursors are ``"|"`` (empty iso + for rows without a resolved instant); plain integer offsets remain + accepted as a legacy fallback. + """ if limit < 1 or limit > 100: raise ValueError("limit must be between 1 and 100") records = [ OccurrenceRecord.model_validate(item) for item in self._read_history_locked(schedule_id) ] - records.sort( - key=lambda item: ( - item.resolved_at is None, - item.resolved_at or datetime.max.replace(tzinfo=UTC), - item.occurrence_id, - ) - ) - start = 0 - if cursor is not None: - try: - start = int(cursor) - except ValueError as exc: - raise ValueError( - "cursor must be a non-negative integer offset" - ) from exc - if start < 0: - raise ValueError("cursor must be a non-negative integer offset") + records.sort(key=self._sort_key) total = len(records) - page = records[start : start + limit] - end = start + limit + start_index = 0 + if cursor is not None: + if "|" in cursor: + iso_part, _, oid_part = cursor.partition("|") + key = ( + True, + datetime.max.replace(tzinfo=UTC), + oid_part, + ) + if iso_part: + try: + key = (False, datetime.fromisoformat(iso_part), oid_part) + except ValueError as exc: + raise ValueError("invalid keyset cursor") from exc + start_index = 0 + for index, item in enumerate(records): + if self._sort_key(item) > key: + start_index = index + break + else: + start_index = total + else: + try: + start_index = int(cursor) + except ValueError as exc: + raise ValueError( + "cursor must be a keyset '|' or integer offset" + ) from exc + if start_index < 0: + raise ValueError("cursor must be a non-negative integer offset") + page = records[start_index : start_index + limit] + if start_index + limit < total and page: + last = page[-1] + iso = "" if last.resolved_at is None else last.resolved_at.isoformat() + next_cursor: str | None = f"{iso}|{last.occurrence_id}" + else: + next_cursor = None return { "occurrences": [item.model_dump(mode="json") for item in page], "total": total, "cursor": cursor, - "next_cursor": str(end) if end < total else None, + "next_cursor": next_cursor, "limit": limit, } diff --git a/tests/scheduling/test_schedule_store.py b/tests/scheduling/test_schedule_store.py index 5b752fdc..736aeca0 100644 --- a/tests/scheduling/test_schedule_store.py +++ b/tests/scheduling/test_schedule_store.py @@ -164,11 +164,17 @@ def test_history_pagination_over_resolved_utc(tmp_path: Path) -> None: ) page = store.list_occurrences("a", limit=2) assert page["total"] == 3 - assert page["next_cursor"] == "2" + assert page["next_cursor"] is not None + assert "|" in str(page["next_cursor"]) assert len(page["occurrences"]) == 2 # type: ignore[arg-type] - second = store.list_occurrences("a", cursor="2", limit=2) + second = store.list_occurrences( + "a", cursor=page["next_cursor"], limit=2 # type: ignore[arg-type] + ) assert second["next_cursor"] is None assert len(second["occurrences"]) == 1 # type: ignore[arg-type] + # Legacy integer offsets remain accepted. + legacy = store.list_occurrences("a", cursor="2", limit=2) + assert len(legacy["occurrences"]) == 1 # type: ignore[arg-type] def test_inspection_payload_carries_contract_fields(tmp_path: Path) -> None: