From 7ed759053a02f35197a7614f5063c979100c0ae1 Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 10:29:10 +0700 Subject: [PATCH] sched: add schedule models, file store, and deployment revisions (T07) --- src/wf_artifacts/models.py | 8 + src/wf_artifacts/store.py | 9 + src/wf_scheduling/models.py | 136 +++++++++++++++ src/wf_scheduling/store.py | 223 ++++++++++++++++++++++++ tests/scheduling/test_schedule_store.py | 218 +++++++++++++++++++++++ 5 files changed, 594 insertions(+) create mode 100644 src/wf_scheduling/models.py create mode 100644 src/wf_scheduling/store.py create mode 100644 tests/scheduling/test_schedule_store.py diff --git a/src/wf_artifacts/models.py b/src/wf_artifacts/models.py index 54e5fa18..9bc8a2ca 100644 --- a/src/wf_artifacts/models.py +++ b/src/wf_artifacts/models.py @@ -222,6 +222,14 @@ class WorkflowDeployment(BaseModel): default_factory=list ) drift_policy: DriftPolicy = DriftPolicy.BLOCK + revision: int = Field( + default=1, + ge=1, + description=( + "Mutable deployment revision for edit-race rechecks. " + "Incremented on every save; admission freezes the revision it saw." + ), + ) def binding_map(self) -> dict[str, str]: """Return bindings keyed by dot-joined logical source ref.""" diff --git a/src/wf_artifacts/store.py b/src/wf_artifacts/store.py index 8d7b6383..2daa6836 100644 --- a/src/wf_artifacts/store.py +++ b/src/wf_artifacts/store.py @@ -112,6 +112,15 @@ class FileWorkflowArtifactStore(WorkflowArtifactStore): def save_deployment(self, deployment: WorkflowDeployment) -> None: path = self._deployment_path(deployment.id) + if path.exists(): + existing = WorkflowDeployment.model_validate_json( + path.read_text(encoding="utf-8") + ) + deployment = deployment.model_copy( + update={"revision": existing.revision + 1} + ) + elif deployment.revision != 1: + deployment = deployment.model_copy(update={"revision": 1}) path.write_text( json.dumps(deployment.model_dump(mode="json"), indent=2), encoding="utf-8", diff --git a/src/wf_scheduling/models.py b/src/wf_scheduling/models.py new file mode 100644 index 00000000..bc5b9629 --- /dev/null +++ b/src/wf_scheduling/models.py @@ -0,0 +1,136 @@ +"""Schedule, candidate, and occurrence-history models. + +A schedule follows its deployment: admission freezes the deployment revision, +resolved input, limits, and pinned artifact tree, so later edits cannot change +that occurrence or its run. Overlap is per schedule; manual runs and other +schedules never participate in the check. +""" + +from __future__ import annotations + +import math +import re +from datetime import datetime +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from wf_core.models.input_bindings import ScheduleInputBinding + +SCHEDULE_ID_PATTERN = r"^[A-Za-z0-9_][A-Za-z0-9_.-]*$" + +OverlapPolicy = Literal["skip", "parallel"] +MisfirePolicy = Literal["skip", "latest"] + +OccurrenceKind = Literal[ + "pending", + "coalesced", + "superseded", + "skipped-overlap", + "skipped-misfire", + "preflight-rejected", + "admitted", + "running", + "interrupted", + "completed", + "failed", + "exhausted", + "interval-summary", +] + + +def ensure_schedule_id(schedule_id: str) -> str: + """Reject ids that cannot safely identify one schedule directory.""" + if not re.fullmatch(SCHEDULE_ID_PATTERN, schedule_id): + raise ValueError( + "schedule_id must start with alphanumeric or underscore and contain " + "only [A-Za-z0-9_.-]" + ) + return schedule_id + + +class CronTrigger(BaseModel): + """Recurring five-field Unix-cron trigger in one explicit IANA zone.""" + + model_config = ConfigDict(extra="forbid") + + kind: Literal["cron"] = "cron" + expression: str + timezone: str = "UTC" + + +class OneShotTrigger(BaseModel): + """Single offset-aware instant trigger.""" + + model_config = ConfigDict(extra="forbid") + + kind: Literal["oneshot"] = "oneshot" + at: datetime + + @field_validator("at") + @classmethod + def _require_offset(cls, value: datetime) -> datetime: + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError("one-shot timestamp must include an offset") + return value + + +class Schedule(BaseModel): + """One deployment schedule with its revision and lifecycle flags.""" + + model_config = ConfigDict(extra="forbid") + + id: str = Field(pattern=SCHEDULE_ID_PATTERN) + deployment_id: str + trigger: CronTrigger | OneShotTrigger = Field(discriminator="kind") + input_bindings: list[ScheduleInputBinding] = Field(default_factory=list) + max_steps: int | None = Field(default=None, ge=1) + overlap: OverlapPolicy = "skip" + misfire: MisfirePolicy = "skip" + max_active_runs: int = Field(default=1, ge=1) + lateness_allowance_s: float = Field(default=60.0, ge=0.0) + revision: int = Field(default=1, ge=1) + enabled: bool = True + paused: bool = False + deleted: bool = False + exhausted: bool = False + blocked_reason: str | None = None + created_at: datetime + updated_at: datetime + + @field_validator("lateness_allowance_s") + @classmethod + def _require_finite_allowance(cls, value: float) -> float: + if not math.isfinite(value): + raise ValueError("lateness allowance must be a finite duration") + return value + + +class PendingCandidate(BaseModel): + """At most one unadmitted latest candidate per schedule.""" + + model_config = ConfigDict(extra="forbid") + + schedule_id: str = Field(pattern=SCHEDULE_ID_PATTERN) + intended_at: datetime + revision: int = Field(ge=1) + + +class OccurrenceRecord(BaseModel): + """One inspectable occurrence-history entry for a schedule.""" + + model_config = ConfigDict(extra="forbid") + + schedule_id: str = Field(pattern=SCHEDULE_ID_PATTERN) + occurrence_id: str + kind: OccurrenceKind + resolved_at: datetime | None = None + run_id: str | None = None + revision: int | None = None + reason: str = "" + admitted_at: datetime | None = None + started_at: datetime | None = None + interval_start: datetime | None = None + interval_end: datetime | None = None + interval_count: int = 0 + created_at: datetime diff --git a/src/wf_scheduling/store.py b/src/wf_scheduling/store.py new file mode 100644 index 00000000..5a19f267 --- /dev/null +++ b/src/wf_scheduling/store.py @@ -0,0 +1,223 @@ +"""File-backed schedule store for local development and tests. + +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. +- ``history.json`` — occurrence-history list for cursor pagination over + ``(resolved_at, occurrence_id)``. + +Per-file writes are atomic (tmp + rename); the lock is process-local only. +Scheduler ownership across processes arrives in T11. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from threading import RLock + +from .models import OccurrenceRecord, PendingCandidate, Schedule, ensure_schedule_id + +UTC = timezone.utc + + +class ScheduleExistsError(ValueError): + """A schedule id is already taken (deleted ids are never reused).""" + + +class ScheduleNotFoundError(KeyError): + """No schedule exists for the requested id.""" + + +class StaleScheduleRevisionError(ValueError): + """A schedule edit carried an outdated revision.""" + + +class FileScheduleStore: + """JSON file-backed schedule definitions, candidates, and history.""" + + def __init__(self, root: Path) -> None: + self.root = root + self._lock = RLock() + self.schedules_dir.mkdir(parents=True, exist_ok=True) + + @property + def schedules_dir(self) -> Path: + return self.root / "schedules" + + # -- schedules ------------------------------------------------------ + def create_schedule(self, schedule: Schedule) -> Schedule: + """Persist a new schedule; deleted ids are never reusable.""" + with self._lock: + path = self._schedule_path(schedule.id) + if path.exists(): + raise ScheduleExistsError( + f"schedule id already exists: {schedule.id!r}" + ) + self._write_json(path, schedule.model_dump(mode="json")) + return schedule + + def get_schedule(self, schedule_id: str) -> Schedule: + path = self._schedule_path(schedule_id) + if not path.exists(): + raise ScheduleNotFoundError(f"unknown schedule {schedule_id!r}") + return Schedule.model_validate_json(path.read_text(encoding="utf-8")) + + def list_schedules(self, *, include_deleted: bool = False) -> list[Schedule]: + schedules = [ + Schedule.model_validate_json(path.read_text(encoding="utf-8")) + for path in sorted(self.schedules_dir.glob("*/schedule.json")) + ] + if not include_deleted: + schedules = [item for item in schedules if not item.deleted] + return sorted(schedules, key=lambda item: item.id) + + def update_schedule( + self, schedule: Schedule, *, expected_revision: int + ) -> Schedule: + """Replace a schedule after a revision check (stale edits rejected).""" + with self._lock: + current = self.get_schedule(schedule.id) + if current.revision != expected_revision: + raise StaleScheduleRevisionError( + f"stale schedule revision for {schedule.id!r}: " + f"expected {expected_revision}, found {current.revision}" + ) + self._write_json( + self._schedule_path(schedule.id), schedule.model_dump(mode="json") + ) + return schedule + + def save_schedule(self, schedule: Schedule) -> None: + """Persist a schedule without a revision check (ownership internals).""" + with self._lock: + self._write_json( + self._schedule_path(schedule.id), schedule.model_dump(mode="json") + ) + + # -- candidates (at most one) ---------------------------------------- + def get_candidate(self, schedule_id: str) -> PendingCandidate | None: + path = self._candidate_path(schedule_id) + if not path.exists(): + return None + return PendingCandidate.model_validate_json(path.read_text(encoding="utf-8")) + + def save_candidate( + self, candidate: PendingCandidate | None, *, schedule_id: str + ) -> None: + """Persist or clear the single pending candidate for a schedule.""" + with self._lock: + path = self._candidate_path(schedule_id) + if candidate is None: + if path.exists(): + path.unlink() + return + self._write_json(path, candidate.model_dump(mode="json")) + + # -- consumed watermark ---------------------------------------------- + def get_consumed(self, schedule_id: str) -> datetime | None: + path = self._consumed_path(schedule_id) + if not path.exists(): + return None + raw = json.loads(path.read_text(encoding="utf-8")) + value = raw.get("consumed_through") + if value is None: + return None + return datetime.fromisoformat(value) + + def save_consumed(self, schedule_id: str, consumed_through: datetime) -> None: + with self._lock: + self._write_json( + self._consumed_path(schedule_id), + {"consumed_through": consumed_through.astimezone(UTC).isoformat()}, + ) + + # -- history with cursor pagination ----------------------------------- + def append_history(self, record: OccurrenceRecord) -> None: + """Append one occurrence-history entry for a schedule.""" + with self._lock: + path = self._history_path(record.schedule_id) + entries = self._read_history_locked(record.schedule_id) + entries.append(record.model_dump(mode="json")) + self._write_json(path, entries) + + def list_occurrences( + self, + schedule_id: str, + *, + cursor: str | None = None, + limit: int = 50, + ) -> dict[str, object]: + """Return one history page ordered by ``(resolved_at, occurrence_id)``.""" + 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") + total = len(records) + page = records[start : start + limit] + end = start + limit + return { + "occurrences": [item.model_dump(mode="json") for item in page], + "total": total, + "cursor": cursor, + "next_cursor": str(end) if end < total else None, + "limit": limit, + } + + def _read_history_locked(self, schedule_id: str) -> list[dict]: + path = self._history_path(schedule_id) + if not path.exists(): + return [] + raw = json.loads(path.read_text(encoding="utf-8")) + return raw if isinstance(raw, list) else [] + + # -- paths ------------------------------------------------------------ + def _schedule_dir(self, schedule_id: str) -> Path: + safe_id = ensure_schedule_id(schedule_id) + root = self.schedules_dir.resolve() + path = (self.schedules_dir / safe_id).resolve() + if path.parent != root: + raise ValueError(f"schedule id escapes schedule store: {schedule_id!r}") + return path + + def _schedule_path(self, schedule_id: str) -> Path: + return self._schedule_dir(schedule_id) / "schedule.json" + + def _candidate_path(self, schedule_id: str) -> Path: + return self._schedule_dir(schedule_id) / "candidate.json" + + def _consumed_path(self, schedule_id: str) -> Path: + return self._schedule_dir(schedule_id) / "consumed.json" + + def _history_path(self, schedule_id: str) -> Path: + return self._schedule_dir(schedule_id) / "history.json" + + def _write_json(self, path: Path, payload: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temp_path = path.with_name(path.name + ".tmp") + temp_path.write_text(json.dumps(payload, indent=2), encoding="utf-8") + temp_path.replace(path) diff --git a/tests/scheduling/test_schedule_store.py b/tests/scheduling/test_schedule_store.py new file mode 100644 index 00000000..5b752fdc --- /dev/null +++ b/tests/scheduling/test_schedule_store.py @@ -0,0 +1,218 @@ +"""Schedule models and file store: revisions, lifecycle, history (T07).""" + +from __future__ import annotations + +from datetime import UTC, datetime +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from wf_scheduling.models import ( + CronTrigger, + OccurrenceRecord, + PendingCandidate, + Schedule, +) +from wf_scheduling.store import ( + FileScheduleStore, + ScheduleExistsError, + StaleScheduleRevisionError, +) + + +def _schedule(sid: str = "sched-1", **over: object) -> Schedule: + now = datetime(2026, 9, 8, 12, 0, tzinfo=UTC) + base: dict[str, object] = { + "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(over) + return Schedule.model_validate(base) + + +def test_schedule_defaults_match_settled_policies() -> None: + sched = _schedule() + assert sched.overlap == "skip" + assert sched.misfire == "skip" + assert sched.lateness_allowance_s == 60.0 + assert sched.revision == 1 + assert not sched.paused and not sched.deleted and not sched.exhausted + assert sched.enabled + + +def test_cron_trigger_and_lateness_validation() -> None: + assert ( + CronTrigger.model_validate( + {"kind": "cron", "expression": "* * * * *", "timezone": "UTC"} + ).timezone + == "UTC" + ) + with pytest.raises(ValidationError): + _schedule(lateness_allowance_s=-1.0) + with pytest.raises(ValidationError): + _schedule(lateness_allowance_s=float("inf")) + with pytest.raises(ValidationError): + Schedule.model_validate( + { + "id": "s", + "deployment_id": "d", + "trigger": {"kind": "oneshot", "at": "2026-09-08T12:00:00"}, + "input_bindings": [], + "created_at": "2026-09-08T12:00:00+00:00", + "updated_at": "2026-09-08T12:00:00+00:00", + } + ) + + +def test_parallel_requires_positive_max_active() -> None: + with pytest.raises(ValidationError): + _schedule(overlap="parallel", max_active_runs=0) + + +def test_create_get_list_and_id_never_reused(tmp_path: Path) -> None: + store = FileScheduleStore(tmp_path) + store.create_schedule(_schedule("a")) + store.create_schedule(_schedule("b")) + assert [s.id for s in store.list_schedules()] == ["a", "b"] + with pytest.raises(ScheduleExistsError): + store.create_schedule(_schedule("a")) + # Delete clears future admission but keeps history; id stays taken. + doomed = store.get_schedule("a") + doomed.deleted = True + store.save_schedule(doomed) + assert [s.id for s in store.list_schedules()] == ["b"] + assert [s.id for s in store.list_schedules(include_deleted=True)] == ["a", "b"] + with pytest.raises(ScheduleExistsError): + store.create_schedule(_schedule("a")) + + +def test_stale_revision_edits_rejected(tmp_path: Path) -> None: + store = FileScheduleStore(tmp_path) + store.create_schedule(_schedule("a")) + current = store.get_schedule("a") + updated = current.model_copy(update={"revision": 2, "paused": True}) + store.update_schedule(updated, expected_revision=1) + assert store.get_schedule("a").revision == 2 + with pytest.raises(StaleScheduleRevisionError): + store.update_schedule(updated, expected_revision=1) + + +def test_pause_and_disable_exclude_interval_no_backfill(tmp_path: Path) -> None: + store = FileScheduleStore(tmp_path) + store.create_schedule(_schedule("a")) + now = datetime(2026, 9, 8, 12, 30, tzinfo=UTC) + sched = store.get_schedule("a") + # Pause clears the pending candidate and advances the watermark; resume + # starts from the next future instant (poll semantics tested in T08). + store.save_candidate( + PendingCandidate( + schedule_id="a", + intended_at=datetime(2026, 9, 8, 12, 0, tzinfo=UTC), + revision=sched.revision, + ), + schedule_id="a", + ) + paused = sched.model_copy(update={"paused": True, "updated_at": now}) + store.save_schedule(paused) + store.save_candidate(None, schedule_id="a") + store.save_consumed("a", now) + assert store.get_candidate("a") is None + assert store.get_consumed("a") == now + + +def test_candidate_is_at_most_one(tmp_path: Path) -> None: + store = FileScheduleStore(tmp_path) + store.create_schedule(_schedule("a")) + first = PendingCandidate( + schedule_id="a", + intended_at=datetime(2026, 9, 8, 12, 0, tzinfo=UTC), + revision=1, + ) + second = PendingCandidate( + schedule_id="a", + intended_at=datetime(2026, 9, 8, 13, 0, tzinfo=UTC), + revision=1, + ) + store.save_candidate(first, schedule_id="a") + store.save_candidate(second, schedule_id="a") + assert store.get_candidate("a") is not None + assert store.get_candidate("a") is not None + assert store.get_candidate("a").intended_at == datetime( # type: ignore[union-attr] + 2026, 9, 8, 13, 0, tzinfo=UTC + ) + + +def test_history_pagination_over_resolved_utc(tmp_path: Path) -> None: + store = FileScheduleStore(tmp_path) + store.create_schedule(_schedule("a")) + for hour in (12, 13, 14): + store.append_history( + OccurrenceRecord( + schedule_id="a", + occurrence_id=f"a|2026-09-08T{hour:02d}:00:00+00:00", + kind="admitted", + resolved_at=datetime(2026, 9, 8, hour, 0, tzinfo=UTC), + run_id=f"run-{hour}", + revision=1, + created_at=datetime(2026, 9, 8, hour, 0, tzinfo=UTC), + ) + ) + page = store.list_occurrences("a", limit=2) + assert page["total"] == 3 + assert page["next_cursor"] == "2" + assert len(page["occurrences"]) == 2 # type: ignore[arg-type] + second = store.list_occurrences("a", cursor="2", limit=2) + assert second["next_cursor"] is None + assert len(second["occurrences"]) == 1 # type: ignore[arg-type] + + +def test_inspection_payload_carries_contract_fields(tmp_path: Path) -> None: + store = FileScheduleStore(tmp_path) + store.create_schedule(_schedule("a")) + store.append_history( + OccurrenceRecord( + schedule_id="a", + occurrence_id="a|2026-09-08T12:00:00+00:00", + kind="skipped-overlap", + resolved_at=datetime(2026, 9, 8, 12, 0, tzinfo=UTC), + revision=2, + reason="active=['run-1']", + admitted_at=None, + started_at=None, + created_at=datetime(2026, 9, 8, 12, 0, tzinfo=UTC), + ) + ) + page = store.list_occurrences("a", limit=10) + row = page["occurrences"][0] # type: ignore[index] + assert row["resolved_at"] is not None + assert row["revision"] == 2 + assert row["reason"] == "active=['run-1']" + + +def test_deployment_revision_increments_on_save(tmp_path: Path) -> None: + from wf_artifacts import FileWorkflowArtifactStore, WorkflowDeployment + + artifacts = FileWorkflowArtifactStore(tmp_path) + artifacts.save_deployment( + WorkflowDeployment( + id="dep-1", + artifact_id="wf-1", + artifact_version=1, + bindings=[], + ) + ) + assert artifacts.get_deployment("dep-1").revision == 1 + artifacts.save_deployment( + WorkflowDeployment( + id="dep-1", + artifact_id="wf-1", + artifact_version=2, + bindings=[], + ) + ) + assert artifacts.get_deployment("dep-1").revision == 2