From 75e902cf699ba5d5890b2bac49f18d4e7572e2cc Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 10:03:33 +0700 Subject: [PATCH] sched: add occurrence identity and UTC helpers (T02) --- src/wf_scheduling/occurrences.py | 83 ++++++++++++++++++++++++++ tests/scheduling/test_occurrences.py | 88 ++++++++++++++++++++++++++++ 2 files changed, 171 insertions(+) create mode 100644 src/wf_scheduling/occurrences.py create mode 100644 tests/scheduling/test_occurrences.py diff --git a/src/wf_scheduling/occurrences.py b/src/wf_scheduling/occurrences.py new file mode 100644 index 00000000..8c0e029c --- /dev/null +++ b/src/wf_scheduling/occurrences.py @@ -0,0 +1,83 @@ +"""Occurrence identity and UTC persistence helpers. + +Occurrence identity is ``(schedule_id, resolved UTC instant)``. The +``occurrence_id`` string is derived deterministically from that pair as +``f"{schedule_id}|{resolved_utc.isoformat()}"`` with the instant normalized +to UTC; the definition's cron expression, time-zone name, and admission +snapshots are retained separately, but no finer intended-wall-time +provenance is claimed beyond what the calendar library supplies. + +Clock use is split: a monotonic clock drives sleeping, and a wall clock +drives calendar eligibility. Clock rollback cannot admit an already-consumed +instant again: any resolved instant ``<= consumed_through`` is never +re-admitted. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from datetime import datetime, timezone + +UTC = timezone.utc + + +def _require_aware(instant: datetime, *, op: str) -> datetime: + if instant.tzinfo is None or instant.utcoffset() is None: + raise ValueError(f"{op} requires an aware datetime with an offset") + return instant + + +def occurrence_id(schedule_id: str, resolved_utc: datetime) -> str: + """Derive a deterministic occurrence id from schedule id and UTC instant.""" + _require_aware(resolved_utc, op="occurrence_id") + normalized = resolved_utc.astimezone(UTC) + return f"{schedule_id}|{normalized.isoformat()}" + + +def scheduled_at_str(resolved: datetime) -> str: + """Serialize a resolved instant as a UTC RFC 3339 string.""" + _require_aware(resolved, op="scheduled_at") + return resolved.astimezone(UTC).isoformat() + + +def is_already_consumed(resolved_utc: datetime, consumed_through: datetime) -> bool: + """Return whether ``resolved_utc`` was already decided (rollback guard).""" + _require_aware(resolved_utc, op="is_already_consumed") + _require_aware(consumed_through, op="is_already_consumed") + return resolved_utc.astimezone(UTC) <= consumed_through.astimezone(UTC) + + +def wall_now_utc() -> datetime: + """Return the current wall-clock time in UTC for calendar eligibility.""" + return datetime.now(UTC) + + +def monotonic_ns() -> int: + """Return a monotonic timestamp for sleeping between polls.""" + return time.monotonic_ns() + + +@dataclass(frozen=True) +class Occurrence: + """One resolved schedule occurrence with its durable identity.""" + + schedule_id: str + scheduled_at: datetime + + def __post_init__(self) -> None: + _require_aware(self.scheduled_at, op="Occurrence") + object.__setattr__(self, "scheduled_at", self.scheduled_at.astimezone(UTC)) + + @property + def occurrence_id(self) -> str: + """Return the deterministic ``(schedule_id, resolved UTC)`` identity.""" + return occurrence_id(self.schedule_id, self.scheduled_at) + + def to_binding_dict(self) -> dict[str, str]: + """Return the schedule environment exposed to occurrence expressions.""" + return { + "schedule_id": self.schedule_id, + "occurrence_id": self.occurrence_id, + "scheduled_at": scheduled_at_str(self.scheduled_at), + } diff --git a/tests/scheduling/test_occurrences.py b/tests/scheduling/test_occurrences.py new file mode 100644 index 00000000..db6c5956 --- /dev/null +++ b/tests/scheduling/test_occurrences.py @@ -0,0 +1,88 @@ +"""Occurrence identity and UTC persistence helpers (T02).""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from wf_scheduling.calendar import CronSource +from wf_scheduling.occurrences import ( + Occurrence, + is_already_consumed, + monotonic_ns, + occurrence_id, + scheduled_at_str, + wall_now_utc, +) + + +def utc(*args: int) -> datetime: + return datetime(*args, tzinfo=timezone.utc) + + +def test_occurrence_id_is_deterministic_utc_pair() -> None: + at = utc(2026, 9, 8, 13, 0, 0) + assert occurrence_id("sched-1", at) == occurrence_id("sched-1", at) + assert occurrence_id("sched-1", at) != occurrence_id("sched-2", at) + assert occurrence_id("sched-1", at) != occurrence_id( + "sched-1", utc(2026, 9, 8, 14, 0, 0) + ) + assert "sched-1" in occurrence_id("sched-1", at) + + +def test_occurrence_id_rejects_naive() -> None: + with pytest.raises(ValueError, match="aware"): + occurrence_id("s", datetime(2026, 9, 8, 12, 0)) + + +def test_scheduled_at_serializes_as_utc_rfc3339() -> None: + from datetime import timedelta + + at = datetime(2026, 9, 8, 12, 0, tzinfo=timezone(timedelta(hours=2))) + assert scheduled_at_str(at) == "2026-09-08T10:00:00+00:00" + with pytest.raises(ValueError, match="aware"): + scheduled_at_str(datetime(2026, 9, 8, 12, 0)) + + +def test_occurrence_model_exposes_contract_fields() -> None: + occ = Occurrence(schedule_id="s", scheduled_at=utc(2026, 9, 8, 13, 0, 0)) + assert occ.schedule_id == "s" + assert occ.occurrence_id == occurrence_id("s", utc(2026, 9, 8, 13, 0, 0)) + assert occ.scheduled_at == utc(2026, 9, 8, 13, 0, 0) + payload = occ.to_binding_dict() + assert payload["schedule_id"] == "s" + assert payload["scheduled_at"] == "2026-09-08T13:00:00+00:00" + + +def test_calendar_identities_are_unique_and_increasing() -> None: + src = CronSource("30 9 * * *", "UTC") + cursor = utc(2026, 9, 1, 0, 0, 0) + seen: set[str] = set() + prev: datetime | None = None + for _ in range(10): + nxt = src.next_after(cursor) + assert nxt is not None + ident = occurrence_id("s", nxt) + assert ident not in seen + seen.add(ident) + if prev is not None: + assert nxt > prev + prev = nxt + cursor = nxt + + +def test_rollback_guard_never_readmits_consumed() -> None: + consumed = utc(2026, 9, 8, 12, 20, 0) + assert is_already_consumed(utc(2026, 9, 8, 12, 0, 0), consumed) + assert is_already_consumed(consumed, consumed) + assert not is_already_consumed(utc(2026, 9, 8, 12, 30, 0), consumed) + + +def test_wall_and_monotonic_clocks_split() -> None: + wall = wall_now_utc() + assert wall.tzinfo is not None + assert wall.utcoffset() is not None + first = monotonic_ns() + second = monotonic_ns() + assert second >= first