diff --git a/src/wf_api/runs.py b/src/wf_api/runs.py index 9bd619e7..845d2ebc 100644 --- a/src/wf_api/runs.py +++ b/src/wf_api/runs.py @@ -188,7 +188,14 @@ class WorkflowRunApi: trace_range: TraceRangeLike | None, ) -> RunResult: trace_values = _trace_range_values(trace_range) - record, stopped_run = restore_interrupted_run(self._run_store(), run_id) + store = self._run_store() + pre_attempt = store.get_resume_attempt(run_id) + if pre_attempt is not None and pre_attempt.state == "ACTIVE": + raise ValueError( + f"workflow run {run_id!r} has an ambiguous active resume attempt; " + "recovery must fail it closed before retry" + ) + record, stopped_run = restore_interrupted_run(store, run_id) environment = record.environment diagnostics = validate_pinned_resume_environment( record=record, diff --git a/src/wf_artifacts/runs/store.py b/src/wf_artifacts/runs/store.py index e2c742fd..4dc20e74 100644 --- a/src/wf_artifacts/runs/store.py +++ b/src/wf_artifacts/runs/store.py @@ -55,6 +55,15 @@ class RunStore: def get_resume_attempt(self, run_id: str) -> ResumeAttempt | None: raise NotImplementedError + def mark_pending_dispatch(self, run_id: str) -> None: + raise NotImplementedError + + def clear_pending_dispatch(self, run_id: str) -> None: + raise NotImplementedError + + def is_pending_dispatch(self, run_id: str) -> bool: + raise NotImplementedError + class FileRunStore(RunStore): """JSON file-backed admitted- and stopped-run store for local dev/tests. @@ -186,6 +195,26 @@ class FileRunStore(RunStore): return None return ResumeAttempt.model_validate_json(path.read_text(encoding="utf-8")) + def mark_pending_dispatch(self, run_id: str) -> None: + """Atomically flag a materialized view as pending-dispatch.""" + with self._lock: + self._write_json(self._pending_path(run_id), {"pending": True}) + + def clear_pending_dispatch(self, run_id: str) -> None: + with self._lock: + try: + path = self._pending_path(run_id) + except ValueError: + return + if path.exists(): + path.unlink() + + def is_pending_dispatch(self, run_id: str) -> bool: + try: + return self._pending_path(run_id).exists() + except ValueError: + return False + def _write_json(self, path: Path, payload: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) temp_path = path.with_suffix(".json.tmp") @@ -209,5 +238,8 @@ class FileRunStore(RunStore): def _resume_attempt_path(self, run_id: str) -> Path: return self._run_directory(run_id) / "resume_attempt.json" + def _pending_path(self, run_id: str) -> Path: + return self._run_directory(run_id) / "pending_dispatch.json" + def _checkpoint_path(self, run_id: str, sequence: int) -> Path: return self._run_directory(run_id) / "checkpoints" / f"{sequence:06d}.json" diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index 9aa3d9b8..24bd2707 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -310,8 +310,21 @@ class Scheduler: self.schedule_store.save_consumed(sched.id, intended) from wf_api.run_lifecycle import materialize_admitted_view + # Dispatch mark precedes the view so a crash after admission but + # before/during first dispatch stays pending (not abandoned). Cleared + # after dispatch returns regardless of outcome (hang still dispatched). + try: + self.run_store.mark_pending_dispatch(run_id) + except AttributeError: + pass materialize_admitted_view(store=self.run_store, admission=admission) - self._dispatch(run_id, now) + try: + self._dispatch(run_id, now) + finally: + try: + self.run_store.clear_pending_dispatch(run_id) + except AttributeError: + pass return run_id def _dispatch(self, run_id: str, now: datetime) -> None: diff --git a/src/wf_scheduling/recovery.py b/src/wf_scheduling/recovery.py index 55013701..c44bc094 100644 --- a/src/wf_scheduling/recovery.py +++ b/src/wf_scheduling/recovery.py @@ -97,18 +97,31 @@ def recover( elif status == StoredRunStatus.COMPLETED.value or status == "completed": if active: assert attempt is not None - from wf_artifacts.runs.models import ResumeAttempt + try: + latest = run_store.get_latest_checkpoint(run.id) + completed_attempt = latest.attempt_id + except KeyError: + completed_attempt = None + if ( + completed_attempt is not None + and completed_attempt == attempt.attempt_id + ): + from wf_artifacts.runs.models import ResumeAttempt - run_store.save_resume_attempt( - ResumeAttempt( - run_id=run.id, - attempt_id=attempt.attempt_id, - state="DONE", - created_at=attempt.created_at, - updated_at=now, + run_store.save_resume_attempt( + ResumeAttempt( + run_id=run.id, + attempt_id=attempt.attempt_id, + state="DONE", + created_at=attempt.created_at, + updated_at=now, + ) ) - ) - diags.append(f"{run.id}:attempt-reconciled") + diags.append(f"{run.id}:attempt-reconciled") + else: + _fail_run(run_store, run, AMBIGUOUS_REASON, now, record_history) + diags.append(f"{run.id}:failed-closed") + continue if record_history is not None and not _has_terminal( schedule_store, run.id, "completed" ): @@ -154,23 +167,20 @@ def _has_terminal(schedule_store: Any, run_id: str, kind: str) -> bool: def _mark_pending(run_store: Any, run_id: str) -> None: - path = run_store._run_directory(run_id) / "pending_dispatch" - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text("pending", encoding="utf-8") + run_store.mark_pending_dispatch(run_id) def _is_pending(run_store: Any, run_id: str) -> bool: try: - return (run_store._run_directory(run_id) / "pending_dispatch").exists() - except ValueError: + return bool(run_store.is_pending_dispatch(run_id)) + except AttributeError: + # Legacy stores without the pending protocol: treat as not pending. return False def clear_pending(run_store: Any, run_id: str) -> None: """Clear the pending-dispatch marker after the poll sweep dispatches.""" try: - path = run_store._run_directory(run_id) / "pending_dispatch" - except ValueError: + run_store.clear_pending_dispatch(run_id) + except AttributeError: return - if path.exists(): - path.unlink()