diff --git a/src/wf_api/schedules.py b/src/wf_api/schedules.py index 919e4783..bfd64d5a 100644 --- a/src/wf_api/schedules.py +++ b/src/wf_api/schedules.py @@ -41,6 +41,11 @@ from .saved_subgraphs import resolve_saved_subgraph_tree _PROJECT_SCHEDULE = JsonProjector(ScheduleResult) _PROJECT_LIST_SCHEDULES = JsonProjector(ListSchedulesResult) _PROJECT_OCCURRENCE_PAGE = JsonProjector(OccurrencePage) +# The pending projection is not stored, so a limit-one page needs a cursor +# that means "start the stored history at its first row". A regular store +# cursor cannot express a position before that row without hiding it behind a +# cursor that points after it. +_PENDING_PAGE_CURSOR = "__pending__" def _serialize_schedule_write(method: Any) -> Any: @@ -457,19 +462,47 @@ class WorkflowScheduleApi: is prepended: ``occurrence_id`` derives from ``(schedule_id, intended instant)``, ``resolved_at`` is the intended instant, ``revision`` is the candidate revision, and - ``reason`` is empty; ``total`` grows by one. Later pages never + ``reason`` is empty; ``total`` grows by one. The stored page is + fetched with one fewer row so the response still honors ``limit``. + For ``limit=1``, the synthetic ``next_cursor`` is an opaque marker + that resumes stored history from its first row. Later pages never carry the row (a repeat fetch after the candidate admits therefore - shows the durable ``admitted`` entry instead of a duplicate - pending projection). + shows the durable ``admitted`` entry instead of a duplicate pending + projection). """ store = self._schedule_store() # KeyError first: an unknown schedule must not leak pagination. store.get_schedule(schedule_id) if limit < 1 or limit > 100: raise ValueError("limit must be between 1 and 100") - page = store.list_occurrences(schedule_id, cursor=cursor, limit=limit) + first_page = cursor in (None, "0") candidate = store.get_candidate(schedule_id) - if candidate is not None and cursor in (None, "0"): + include_pending = candidate is not None and first_page + stored_cursor = None if cursor == _PENDING_PAGE_CURSOR else cursor + stored_limit = limit - 1 if include_pending else limit + if stored_limit: + page = store.list_occurrences( + schedule_id, + cursor=stored_cursor, + limit=stored_limit, + ) + else: + # Fetch only enough metadata to know whether a stored row must be + # visited after the pending-only page; do not expose that row or + # use its after-row cursor, which would make it unreachable. + stored_page = store.list_occurrences( + schedule_id, + cursor=stored_cursor, + limit=1, + ) + page = { + "occurrences": [], + "total": stored_page["total"], + "cursor": stored_cursor, + "next_cursor": (_PENDING_PAGE_CURSOR if stored_page["total"] else None), + "limit": limit, + } + if include_pending: now = datetime.now(UTC) intended = candidate.intended_at pending = OccurrenceRecord( @@ -492,12 +525,15 @@ class WorkflowScheduleApi: "occurrences": [ pending.model_dump(mode="json"), *page["occurrences"], - ], + ][:limit], "total": page["total"] + 1, "cursor": page["cursor"], "next_cursor": page["next_cursor"], - "limit": page["limit"], + "limit": limit, } + elif cursor == _PENDING_PAGE_CURSOR: + page["cursor"] = cursor + page["limit"] = limit return _PROJECT_OCCURRENCE_PAGE(page) diff --git a/tests/wf_api/test_schedules.py b/tests/wf_api/test_schedules.py index 104a3049..6596e764 100644 --- a/tests/wf_api/test_schedules.py +++ b/tests/wf_api/test_schedules.py @@ -787,6 +787,65 @@ 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_pending_limit_one_traverses_every_row( + tmp_path: Path, +) -> None: + """A synthesized pending row must not make the first stored row unreachable.""" + api, sched_store, _, _, _ = _harness(tmp_path / "pending_limit_one") + await api.create_schedule( + schedule_id="s", deployment_id="dep.personal", trigger=_cron() + ) + base = ts(2026, 9, 8, 12, 0) + recorder = FileScheduleHistoryRecorder(sched_store) + for index in range(3): + instant = base + timedelta(minutes=index) + recorder.record( + HistoryEntry( + schedule_id="s", + kind="admitted", + resolved_at=instant, + run_id=f"run-{index}", + revision=1, + reason="rev=1", + created_at=instant, + ) + ) + sched_store.save_candidate( + PendingCandidate( + schedule_id="s", + intended_at=base + timedelta(hours=1), + revision=1, + ), + schedule_id="s", + ) + + rows: list[dict[str, Any]] = [] + cursor: str | None = None + for _ in range(5): + page = await api.list_schedule_occurrences( + schedule_id="s", cursor=cursor, limit=1 + ) + assert len(page["occurrences"]) <= 1 + rows.extend(page["occurrences"]) + cursor = page["next_cursor"] + if cursor is None: + break + else: + pytest.fail("occurrence pagination did not terminate") + + assert [row["kind"] for row in rows] == [ + "pending", + "admitted", + "admitted", + "admitted", + ] + assert [row["run_id"] for row in rows[1:]] == [ + "run-0", + "run-1", + "run-2", + ] + + async def test_occurrences_pagination_visits_tied_entries_once(tmp_path: Path) -> None: """One occurrence's admission + stops each appear exactly once (B4). diff --git a/tests/wf_transport_rpc_http/test_schedules_rpc.py b/tests/wf_transport_rpc_http/test_schedules_rpc.py index cba8c3ac..86be7620 100644 --- a/tests/wf_transport_rpc_http/test_schedules_rpc.py +++ b/tests/wf_transport_rpc_http/test_schedules_rpc.py @@ -242,8 +242,8 @@ async def test_rpc_schedule_occurrences_pagination_and_pending(tmp_path) -> None assert first["total"] == 4 assert first["cursor"] is None assert first["limit"] == 2 - # The pending synthesis prepends one row to the stored page. - assert len(first["occurrences"]) == 3 + # The pending synthesis consumes one slot from the stored page. + assert len(first["occurrences"]) == 2 pending = first["occurrences"][0] assert pending["kind"] == "pending" assert pending["schedule_id"] == "s"