From b0943850773fdededd6f069045d9ce6036976957 Mon Sep 17 00:00:00 2001 From: lda Date: Wed, 9 Sep 2026 17:27:36 +0700 Subject: [PATCH] sched: record resumed stopped results live and idempotently (B3) --- src/wf_api/runs.py | 12 ++ src/wf_scheduling/poll.py | 48 +++++ src/wf_scheduling/resume_gate.py | 43 ++++- tests/scheduling/test_poll.py | 86 +++++++++ tests/wf_server/test_scheduler_integration.py | 171 ++++++++++++++++++ 5 files changed, 357 insertions(+), 3 deletions(-) diff --git a/src/wf_api/runs.py b/src/wf_api/runs.py index 97c99705..9b825d46 100644 --- a/src/wf_api/runs.py +++ b/src/wf_api/runs.py @@ -320,6 +320,18 @@ class WorkflowRunApi: updated_at=cleared_at, ) ) + # Live occurrence history for scheduled resumes: record the resumed + # stopped result now (completion, failure, or re-interruption each + # carry their own checkpoint id, so each records exactly once and + # repeats dedup). Manual runs and scheduler-off resumes skip + # quietly — restart recovery reconciles those instead. + gate = self.resume_slot_gate + if gate is not None: + gate.note_resumed_result( + run_id, + status_value=run.status.value, + checkpoint_id=next_record.latest_checkpoint_id, + ) return _run_payload( deployment=environment.deployment, artifact=environment.root_artifact, diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index 92ab7e91..990f6772 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -504,6 +504,54 @@ class Scheduler: checkpoint_id=stopped.latest_checkpoint_id, ) + def record_resumed_stopped_result( + self, + run_id: str, + *, + status_value: str, + checkpoint_id: str | None, + now: datetime, + ) -> bool: + """Record one live resumed stopped result, idempotently. + + The run API persists the resumed stopped checkpoint itself; this + only appends the matching occurrence-history entry through the + same ``(run_id, kind, checkpoint_id)`` idempotency as dispatch + and recovery, so repeated polls and restart recovery never + duplicate it. Schedule flags are deliberately not consulted: + pausing or deleting a schedule never suppresses retained run + history. Returns whether an entry was appended. + """ + self._require_ownership() + kind = { + "completed": "completed", + "interrupted": "interrupted", + "failed": "failed", + }.get(status_value) + if kind is None: + return False + try: + admission = self.run_store.get_admission(run_id) + except KeyError: + return False + sched_id = admission.schedule_id + if sched_id is None: + return False + if self.history.has_terminal(sched_id, run_id, kind, checkpoint_id): + return False + self._record( + kind=kind, # type: ignore[arg-type] + sched_id=sched_id, + intended=admission.scheduled_at, + run_id=run_id, + revision=admission.schedule_revision, + reason="resumed", + now=now, + started_at=now, + checkpoint_id=checkpoint_id, + ) + return True + # -- polling -------------------------------------------------------- def poll(self, now: datetime) -> dict[str, str]: self._require_ownership() diff --git a/src/wf_scheduling/resume_gate.py b/src/wf_scheduling/resume_gate.py index 77ff35f2..48882a50 100644 --- a/src/wf_scheduling/resume_gate.py +++ b/src/wf_scheduling/resume_gate.py @@ -20,6 +20,8 @@ from __future__ import annotations from typing import Any +from wf_scheduling.ownership import SecondOwnerError + class ScheduledCapacityBusyError(ValueError): """A scheduled resume found every server execution slot occupied.""" @@ -42,15 +44,15 @@ class SchedulerResumeGate: Returns the schedule admission when the slot is held (the caller must :meth:`release` it), or ``None`` when this run is genuinely - manual or the scheduler is not live — both keep the legacy - resume path unchanged. Raises + manual, not resumable, or the scheduler is not live — all keep + the legacy resume path unchanged. Raises :class:`ScheduledCapacityBusyError` when every slot is occupied; nothing is marked in that case, in particular no ACTIVE resume attempt. """ service = self._service with service._lock: - if not service.running: + if not service._started: return None try: admission = service.run_store.get_admission(run_id) @@ -89,3 +91,38 @@ class SchedulerResumeGate: with service._lock: service.run_store.clear_executing(run_id) service._live_resumes.discard(run_id) + + def note_resumed_result( + self, + run_id: str, + *, + status_value: str, + checkpoint_id: str | None, + ) -> bool: + """Record one live resumed stopped result for a scheduled run. + + No-op (``False``) for manual runs or when the scheduler is not + live — restart recovery still reconciles those through its own + idempotent path. Live scheduled resumes append exactly once + through the scheduler's recording, sharing idempotency with + dispatch and recovery. A lost-ownership race (shutdown releasing + mid-note) also skips quietly: the resume result itself is already + durable, and recovery reconciles the entry at the next start — + the note must never break a completed resume. + """ + service = self._service + with service._lock: + if not service._started: + return False + scheduler = service._scheduler + if scheduler is None: + return False + try: + return scheduler.record_resumed_stopped_result( + run_id, + status_value=status_value, + checkpoint_id=checkpoint_id, + now=service.clock(), + ) + except (SecondOwnerError, OSError, ValueError): + return False diff --git a/tests/scheduling/test_poll.py b/tests/scheduling/test_poll.py index 0d2c1ff8..65580c72 100644 --- a/tests/scheduling/test_poll.py +++ b/tests/scheduling/test_poll.py @@ -384,3 +384,89 @@ def test_manual_and_other_schedule_runs_are_overlap_independent( assert result["a"].startswith("admit:run-") assert result["b"] == "exhausted" sched.ownership.release() + + +def test_record_resumed_stopped_result_is_idempotent_and_attributed( + tmp_path: Path, +) -> None: + """Live resumed results record exactly once with schedule attribution (B3).""" + from wf_api.run_lifecycle import persist_admission + + from tests.scheduling.controlled import fixture_environment + + sched, store, runs, sources = _harness(tmp_path, script={"*": "interrupt"}) + t0 = ts(2026, 9, 8, 12, 0) + _add(sched, store, sources, "a", OneShotSource(t0), t0 - timedelta(seconds=1)) + assert sched.poll(t0)["a"].startswith("admit:run-") + [run] = runs.list_runs() + assert run.status.value == "interrupted" + assert len([r for r in _history(store, "a") if r["kind"] == "interrupted"]) == 1 + + now = t0 + timedelta(minutes=1) + resumed_ckpt = f"{run.id}.000002" + assert ( + sched.record_resumed_stopped_result( + run.id, + status_value="completed", + checkpoint_id=resumed_ckpt, + now=now, + ) + is True + ) + # Repeats (poll retries, restart recovery) never duplicate the entry. + assert ( + sched.record_resumed_stopped_result( + run.id, + status_value="completed", + checkpoint_id=resumed_ckpt, + now=now, + ) + is False + ) + completed = [r for r in _history(store, "a") if r["kind"] == "completed"] + assert len(completed) == 1 + assert completed[0]["run_id"] == run.id + assert completed[0]["checkpoint_id"] == resumed_ckpt + assert completed[0]["revision"] == 1 + assert completed[0]["reason"] == "resumed" + # A resumed re-interruption is a new result (new checkpoint), not a dup. + assert ( + sched.record_resumed_stopped_result( + run.id, + status_value="interrupted", + checkpoint_id=f"{run.id}.000003", + now=now, + ) + is True + ) + assert len([r for r in _history(store, "a") if r["kind"] == "interrupted"]) == 2 + # Unknown runs, manual runs, and bad statuses record nothing. + assert ( + sched.record_resumed_stopped_result( + "run-999999", + status_value="completed", + checkpoint_id=None, + now=now, + ) + is False + ) + assert ( + sched.record_resumed_stopped_result( + run.id, status_value="bogus", checkpoint_id=None, now=now + ) + is False + ) + manual = persist_admission( + store=runs, + run_id=runs.allocate_run_id(), + environment=fixture_environment(object()), + resolved_input={}, + max_steps=None, + ) + assert ( + sched.record_resumed_stopped_result( + manual.id, status_value="completed", checkpoint_id=None, now=now + ) + is False + ) + sched.ownership.release() diff --git a/tests/wf_server/test_scheduler_integration.py b/tests/wf_server/test_scheduler_integration.py index 0449f015..eec88f7a 100644 --- a/tests/wf_server/test_scheduler_integration.py +++ b/tests/wf_server/test_scheduler_integration.py @@ -1260,3 +1260,174 @@ async def test_service_stop_drains_inflight_scheduled_resume(tmp_path: Path) -> finally: _gate_open.set() await service.stop() + + +async def _kinds(root: Path, schedule_id: str) -> list[str]: + page = FileScheduleStore(root).list_occurrences(schedule_id, limit=100) + rows = cast(list[dict[str, Any]], page["occurrences"]) + return [row["kind"] for row in rows] + + +async def test_scheduled_resume_records_completion_live(tmp_path: Path) -> None: + """A resumed completion lands in occurrence history live (B3). + + No restart recovery is involved: after the API resume plus ordinary + polls, history reads admitted/interrupted/completed, and repeated + polling plus a restart never duplicate the entries. + """ + root = tmp_path / "store" + server = build_local_static_workflow_server(root) + await _seed(server, "ask1", "ask1.default", _single_interrupt_plan("ask1"), ["submitted"]) + store = FileScheduleStore(root) + store.create_schedule(_one_shot("asker", "ask1.default", datetime.now(UTC))) + service = _scheduler(server, capacity=2) + try: + await service.start() + await _wait_for(lambda: len(_run_ids(root)) == 1) + ask_id = _run_ids(root)[0] + await _wait_for( + lambda: FileRunStore(root).get_run(ask_id).status.value == "interrupted" + ) + resumed = await server.api.resume_run( + run_id=ask_id, resume_payload={}, resume_outcome="submitted" + ) + assert resumed["status"] == "completed" + assert await _kinds(root, "asker") == ["admitted", "interrupted", "completed"] + completed = _entries(root, "asker", "completed") + assert len(completed) == 1 + assert completed[0]["run_id"] == ask_id + assert ( + completed[0]["checkpoint_id"] + == FileRunStore(root).get_run(ask_id).latest_checkpoint_id + ) + assert completed[0]["revision"] == 1 + await service.poll_once(datetime.now(UTC)) + await service.poll_once(datetime.now(UTC)) + assert await _kinds(root, "asker") == ["admitted", "interrupted", "completed"] + finally: + await service.stop() + + server_b = build_local_static_workflow_server(root) + revived = _scheduler(server_b) + try: + await revived.start() + await revived.poll_once(datetime.now(UTC)) + assert await _kinds(root, "asker") == ["admitted", "interrupted", "completed"] + finally: + await revived.stop() + + +async def test_scheduled_resume_records_failure_and_reinterruption_live( + tmp_path: Path, +) -> None: + """Resumed failures and re-interruptions each record live (B3).""" + root = tmp_path / "store" + server = build_local_static_workflow_server(root) + await _seed( + server, + "ask_twice", + "ask_twice.default", + _double_interrupt_plan("ask_twice"), + ["submitted"], + ) + await _seed( + server, + "ask_fail", + "ask_fail.default", + _interrupt_then_fail_plan("ask_fail"), + ["submitted"], + ) + store = FileScheduleStore(root) + store.create_schedule(_one_shot("twicer", "ask_twice.default", datetime.now(UTC))) + store.create_schedule(_one_shot("failer", "ask_fail.default", datetime.now(UTC))) + service = _scheduler(server, capacity=2) + try: + await service.start() + await _wait_for(lambda: len(_run_ids(root)) == 2) + twice_id = fail_id = "" + for rid in _run_ids(root): + admission = FileRunStore(root).get_admission(rid) + if admission.schedule_id == "twicer": + twice_id = rid + else: + fail_id = rid + assert twice_id and fail_id + await _wait_for( + lambda: FileRunStore(root).get_run(twice_id).status.value == "interrupted" + ) + await _wait_for( + lambda: FileRunStore(root).get_run(fail_id).status.value == "interrupted" + ) + first = await server.api.resume_run( + run_id=twice_id, resume_payload={}, resume_outcome="submitted" + ) + assert first["status"] == "interrupted" + second = await server.api.resume_run( + run_id=twice_id, resume_payload={}, resume_outcome="submitted" + ) + assert second["status"] == "completed" + failed = await server.api.resume_run( + run_id=fail_id, resume_payload={}, resume_outcome="submitted" + ) + assert failed["status"] == "failed" + + assert await _kinds(root, "twicer") == [ + "admitted", + "interrupted", + "interrupted", + "completed", + ] + twicer_page = FileScheduleStore(root).list_occurrences("twicer", limit=100) + twicer_ckpts = [ + row["checkpoint_id"] + for row in cast(list[dict[str, Any]], twicer_page["occurrences"]) + if row["kind"] in ("interrupted", "completed") + ] + assert len(set(twicer_ckpts)) == 3 + assert await _kinds(root, "failer") == ["admitted", "interrupted", "failed"] + fail_rows = _entries(root, "failer", "failed") + assert len(fail_rows) == 1 + assert fail_rows[0]["run_id"] == fail_id + finally: + await service.stop() + + +async def test_paused_and_deleted_schedules_keep_resume_history( + tmp_path: Path, +) -> None: + """Pausing/deleting never suppresses retained resume history (B3).""" + root = tmp_path / "store" + server = build_local_static_workflow_server(root) + await _seed(server, "ask1", "ask1.default", _single_interrupt_plan("ask1"), ["submitted"]) + store = FileScheduleStore(root) + store.create_schedule(_one_shot("pausable", "ask1.default", datetime.now(UTC))) + store.create_schedule(_one_shot("doomed", "ask1.default", datetime.now(UTC))) + service = _scheduler(server, capacity=2) + try: + await service.start() + await _wait_for(lambda: len(_run_ids(root)) == 2) + ids = { + FileRunStore(root).get_admission(rid).schedule_id or "": rid + for rid in _run_ids(root) + } + await _wait_for( + lambda: all( + FileRunStore(root).get_run(rid).status.value == "interrupted" + for rid in ids.values() + ) + ) + paused = store.get_schedule("pausable") + paused.paused = True + store.save_schedule(paused) + gone = store.get_schedule("doomed") + gone.deleted = True + store.save_schedule(gone) + for rid in ids.values(): + resumed = await server.api.resume_run( + run_id=rid, resume_payload={}, resume_outcome="submitted" + ) + assert resumed["status"] == "completed" + assert await _kinds(root, "pausable") == ["admitted", "interrupted", "completed"] + assert await _kinds(root, "doomed") == ["admitted", "interrupted", "completed"] + finally: + await service.stop()