diff --git a/src/wf_scheduling/models.py b/src/wf_scheduling/models.py index 69835d80..be75a2ed 100644 --- a/src/wf_scheduling/models.py +++ b/src/wf_scheduling/models.py @@ -164,6 +164,16 @@ class OccurrenceRecord(BaseModel): "checkpoint id and therefore a new entry." ), ) + seq: int | None = Field( + default=None, + description=( + "History-entry ordinal assigned at append time, for stable " + "pagination across multiple entries of one occurrence " + "(admission and stopped entries share resolved_at and " + "occurrence_id). Rows persisted before this field existed " + "carry None and order by file position." + ), + ) interval_start: datetime | None = None interval_end: datetime | None = None interval_count: int = 0 diff --git a/src/wf_scheduling/store.py b/src/wf_scheduling/store.py index 893b4fb4..a5673e73 100644 --- a/src/wf_scheduling/store.py +++ b/src/wf_scheduling/store.py @@ -7,8 +7,9 @@ Layout per schedule ``/schedules//``: - ``candidate.json`` — at most one pending latest candidate. - ``consumed.json`` — ``{"consumed_through": }`` watermark. - ``history.json`` — occurrence-history list for cursor pagination over - ``(resolved_at, occurrence_id)`` (keyset; integer offsets accepted as a - legacy fallback). + ``(resolved_at, occurrence_id, seq)`` (keyset; integer offsets accepted as a + legacy fallback, as are two-part ``"|"`` cursors issued before + the entry ordinal existed). - ``_poll_cursor.json`` — round-robin fairness cursor (persisted). Per-file writes are atomic (tmp + rename); the lock is process-local only. @@ -161,10 +162,26 @@ class FileScheduleStore: # -- history with cursor pagination ----------------------------------- def append_history(self, record: OccurrenceRecord) -> None: - """Append one occurrence-history entry for a schedule.""" + """Append one occurrence-history entry for a schedule. + + Stamps the entry ordinal (one past the highest effective ordinal so + far) so pagination distinguishes multiple entries sharing one + occurrence's ``(resolved_at, occurrence_id)`` — admission and each + stopped result each keep their own position. + """ with self._lock: path = self._history_path(record.schedule_id) entries = self._read_history_locked(record.schedule_id) + highest = -1 + for index, item in enumerate(entries): + ordinal = item.get("seq") + effective = ( + ordinal if isinstance(ordinal, int) else index + ) + if effective > highest: + highest = effective + if record.seq is None: + record = record.model_copy(update={"seq": highest + 1}) entries.append(record.model_dump(mode="json")) self._write_json(path, entries) @@ -193,12 +210,23 @@ class FileScheduleStore: @staticmethod def _sort_key( - item: OccurrenceRecord, - ) -> tuple[bool, datetime, str]: + item: tuple[OccurrenceRecord, int], + ) -> tuple[bool, datetime, str, float]: + """Order one history entry behind its stable entry ordinal. + + Admission and stopped entries for one occurrence share + ``(resolved_at, occurrence_id)`` by construction; the ordinal (the + stamped ``seq``, or the file position for rows persisted before it + existed) keeps every entry at its own position so a keyset cursor + can resume between tied entries instead of skipping them. + """ + record, index = item + ordinal = record.seq if record.seq is not None else index return ( - item.resolved_at is None, - item.resolved_at or datetime.max.replace(tzinfo=UTC), - item.occurrence_id, + record.resolved_at is None, + record.resolved_at or datetime.max.replace(tzinfo=UTC), + record.occurrence_id, + float(ordinal), ) def list_occurrences( @@ -208,37 +236,61 @@ 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, seq)``. - Keyset cursors are ``"|"`` (empty iso - for rows without a resolved instant); plain integer offsets remain - accepted as a legacy fallback. + Keyset cursors are ``"||"`` + (empty iso for rows without a resolved instant). Two-part + ``"|"`` cursors issued before the entry ordinal existed + remain accepted and resume exactly as they did then (after the + whole tied group); 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) + indexed = [ + (OccurrenceRecord.model_validate(item), index) + for index, item in enumerate(self._read_history_locked(schedule_id)) ] - records.sort(key=self._sort_key) + indexed.sort(key=self._sort_key) + records = [record for record, _ in indexed] + keys = [self._sort_key(item) for item in indexed] total = len(records) start_index = 0 if cursor is not None: if "|" in cursor: - iso_part, _, oid_part = cursor.partition("|") - key = ( + # The occurrence id itself contains "|" separators, so split + # the trailing ordinal off the right: a new cursor ends in + # an integer seq, anything else is a pre-ordinal cursor + # (whose id tail is always an ISO instant, never an int). + head, _, seq_part = cursor.rpartition("|") + try: + ordinal = float(int(seq_part)) + except ValueError: + # Pre-ordinal cursor: resume exactly as before, after + # the whole tied group (no duplicates, no crash). + iso_part, _, oid_part = cursor.partition("|") + ordinal = float("inf") + else: + iso_part, _, oid_part = head.partition("|") + key: tuple[bool, datetime, str, float] = ( True, datetime.max.replace(tzinfo=UTC), oid_part, + ordinal, ) if iso_part: try: - key = (False, datetime.fromisoformat(iso_part), oid_part) + key = ( + False, + datetime.fromisoformat(iso_part), + oid_part, + ordinal, + ) 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: + for index, item_key in enumerate(keys): + if item_key > key: start_index = index break else: @@ -248,15 +300,19 @@ class FileScheduleStore: start_index = int(cursor) except ValueError as exc: raise ValueError( - "cursor must be a keyset '|' or integer offset" + "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] + page_keys = keys[start_index : start_index + limit] if start_index + limit < total and page: last = page[-1] + _, _, _, last_ordinal = page_keys[-1] iso = "" if last.resolved_at is None else last.resolved_at.isoformat() - next_cursor: str | None = f"{iso}|{last.occurrence_id}" + next_cursor: str | None = ( + f"{iso}|{last.occurrence_id}|{int(last_ordinal)}" + ) else: next_cursor = None return { diff --git a/tests/scheduling/test_schedule_store.py b/tests/scheduling/test_schedule_store.py index 736aeca0..06057c86 100644 --- a/tests/scheduling/test_schedule_store.py +++ b/tests/scheduling/test_schedule_store.py @@ -222,3 +222,92 @@ def test_deployment_revision_increments_on_save(tmp_path: Path) -> None: ) ) assert artifacts.get_deployment("dep-1").revision == 2 + + +def _tied_entry( + kind: str, + checkpoint_id: str | None, + created_minute: int, +) -> OccurrenceRecord: + """One history entry for the shared 12:00 occurrence of schedule a.""" + return OccurrenceRecord( + schedule_id="a", + occurrence_id="a|2026-09-08T12:00:00+00:00", + kind=kind, # type: ignore[arg-type] + resolved_at=datetime(2026, 9, 8, 12, 0, tzinfo=UTC), + run_id="run-1", + revision=1, + checkpoint_id=checkpoint_id, + created_at=datetime(2026, 9, 8, 12, created_minute, tzinfo=UTC), + ) + + +def _traverse(store: FileScheduleStore, limit: int) -> list[dict]: + """Walk every page to the end, returning all rows in visit order.""" + rows: list[dict] = [] + cursor: str | None = None + while True: + page = store.list_occurrences("a", cursor=cursor, limit=limit) + rows.extend(page["occurrences"]) # type: ignore[arg-type] + cursor = page["next_cursor"] # type: ignore[assignment] + if cursor is None: + assert page["total"] == len(rows) + return rows + + +def test_history_pagination_visits_every_tied_entry_once(tmp_path: Path) -> None: + """Admission + repeated interruptions + completion share one cursor tie. + + Every stored entry must appear exactly once across small-page + traversal (B4): the cursor carries the entry ordinal, not just the + shared ``(resolved_at, occurrence_id)`` tie. + """ + store = FileScheduleStore(tmp_path) + store.create_schedule(_schedule("a")) + store.append_history(_tied_entry("admitted", None, 0)) + store.append_history(_tied_entry("interrupted", "run-1.000001", 5)) + store.append_history(_tied_entry("interrupted", "run-1.000002", 9)) + store.append_history(_tied_entry("completed", "run-1.000003", 14)) + + rows = _traverse(store, limit=1) + assert [(row["kind"], row["checkpoint_id"]) for row in rows] == [ + ("admitted", None), + ("interrupted", "run-1.000001"), + ("interrupted", "run-1.000002"), + ("completed", "run-1.000003"), + ] + first = store.list_occurrences("a", limit=1) + assert first["next_cursor"] is not None + assert len(str(first["next_cursor"]).split("|")) >= 3 + + +def test_history_pagination_legacy_rows_keep_file_order(tmp_path: Path) -> None: + """Rows persisted before the entry ordinal order by file position. + + Real persisted data without ``seq`` must still traverse exactly once; + pre-ordinal two-part cursors stay accepted and resume after the tied + group exactly as they did before (no duplicates, no crash). + """ + import json + + store = FileScheduleStore(tmp_path) + store.create_schedule(_schedule("a")) + history_path = tmp_path / "schedules" / "a" / "history.json" + history_path.parent.mkdir(parents=True, exist_ok=True) + legacy = [ + _tied_entry("admitted", None, 0).model_dump(mode="json"), + _tied_entry("interrupted", "run-1.000001", 5).model_dump(mode="json"), + ] + for item in legacy: + del item["seq"] + history_path.write_text(json.dumps(legacy), encoding="utf-8") + + rows = _traverse(store, limit=1) + assert [row["kind"] for row in rows] == ["admitted", "interrupted"] + + legacy_cursor = "2026-09-08T12:00:00+00:00|a|2026-09-08T12:00:00+00:00" + resumed = store.list_occurrences("a", cursor=legacy_cursor, limit=1) + assert resumed["occurrences"] == [] + assert resumed["next_cursor"] is None + with pytest.raises(ValueError): + store.list_occurrences("a", cursor="a|b|c|d", limit=1) diff --git a/tests/wf_api/test_schedules.py b/tests/wf_api/test_schedules.py index 471186d3..f251587f 100644 --- a/tests/wf_api/test_schedules.py +++ b/tests/wf_api/test_schedules.py @@ -767,6 +767,74 @@ async def test_occurrences_pending_synthesis_first_page_only(tmp_path: Path) -> assert all(row["kind"] != "pending" for row in plain["occurrences"]) +async def test_occurrences_pagination_visits_tied_entries_once(tmp_path: Path) -> None: + """One occurrence's admission + stops each appear exactly once (B4). + + Small-page traversal through the public API must not lose the tied + stopped entries sharing the occurrence's ``(resolved_at, + occurrence_id)``. + """ + api, sched_store, _, _, _ = _harness(tmp_path / "tied") + await api.create_schedule( + schedule_id="s", deployment_id="dep.personal", trigger=_cron() + ) + base = ts(2026, 9, 8, 12, 0) + recorder = FileScheduleHistoryRecorder(sched_store) + recorder.record( + HistoryEntry( + schedule_id="s", + kind="admitted", + resolved_at=base, + run_id="run-1", + revision=1, + reason="rev=1", + created_at=base, + ) + ) + recorder.record( + HistoryEntry( + schedule_id="s", + kind="interrupted", + resolved_at=base, + run_id="run-1", + revision=1, + reason="fresh-result", + checkpoint_id="run-1.000001", + created_at=base + timedelta(minutes=5), + ) + ) + recorder.record( + HistoryEntry( + schedule_id="s", + kind="completed", + resolved_at=base, + run_id="run-1", + revision=1, + reason="reconciled-on-recovery", + checkpoint_id="run-1.000002", + created_at=base + timedelta(minutes=9), + ) + ) + + seen: list[tuple[str, str | None]] = [] + cursor: str | None = None + while True: + page = await api.list_schedule_occurrences( + schedule_id="s", cursor=cursor, limit=1 + ) + assert page["total"] == 3 + for row in page["occurrences"]: + seen.append((row["kind"], row["checkpoint_id"])) + cursor = page["next_cursor"] + if cursor is None: + break + assert seen == [ + ("admitted", None), + ("interrupted", "run-1.000001"), + ("completed", "run-1.000002"), + ] + + async def test_inspect_admitted_run_without_fabrication(tmp_path: Path) -> None: api, sched_store, run_store, artifact_store, context = _harness( tmp_path / "admitted" diff --git a/tests/wf_transport_rpc_http/test_schedules_rpc.py b/tests/wf_transport_rpc_http/test_schedules_rpc.py index cacae534..0fcfb0c1 100644 --- a/tests/wf_transport_rpc_http/test_schedules_rpc.py +++ b/tests/wf_transport_rpc_http/test_schedules_rpc.py @@ -265,6 +265,58 @@ async def test_rpc_schedule_occurrences_pagination_and_pending(tmp_path) -> None assert all(row["kind"] != "pending" for row in plain["occurrences"]) +async def test_rpc_schedule_occurrences_tied_entries_traverse_once(tmp_path) -> None: + """Tied admission/stopped entries each arrive exactly once over RPC (B4).""" + server = await _seed_server(tmp_path) + client, http_client = _client_for(server) + store = FileScheduleStore(tmp_path / "store") + base = datetime(2026, 9, 8, 12, 0, tzinfo=UTC) + async with http_client: + await client.create_schedule( + schedule_id="s", + deployment_id="dep.personal", + trigger=_cron(), + ) + recorder = FileScheduleHistoryRecorder(store) + recorder.record( + HistoryEntry( + schedule_id="s", + kind="admitted", + resolved_at=base, + run_id="run-1", + revision=1, + reason="rev=1", + created_at=base, + ) + ) + recorder.record( + HistoryEntry( + schedule_id="s", + kind="interrupted", + resolved_at=base, + run_id="run-1", + revision=1, + reason="fresh-result", + checkpoint_id="run-1.000001", + created_at=base + timedelta(minutes=5), + ) + ) + + seen: list[tuple[str, object]] = [] + cursor: object = None + while True: + page = await client.list_schedule_occurrences( + schedule_id="s", cursor=cursor, limit=1 # type: ignore[arg-type] + ) + assert page["total"] == 2 + for row in page["occurrences"]: + seen.append((row["kind"], row["checkpoint_id"])) + cursor = page["next_cursor"] + if cursor is None: + break + assert seen == [("admitted", None), ("interrupted", "run-1.000001")] + + async def test_rpc_schedule_error_mapping(tmp_path) -> None: server = await _seed_server(tmp_path) client, http_client = _client_for(server)