From cf8d28f1ff307c26cc227e93df8495b7b3909e7c Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 10:27:37 +0700 Subject: [PATCH] sched: address R2 fail-closed and fault-proof findings --- src/wf_api/run_lifecycle.py | 25 +++++++++--- src/wf_api/runs.py | 21 +++++++--- src/wf_artifacts/runs/store.py | 16 +++++--- tests/artifacts/test_run_admission.py | 52 ++++++++++++++++++++++--- tests/wf_api/test_run_admission_path.py | 29 ++++++++++---- 5 files changed, 113 insertions(+), 30 deletions(-) diff --git a/src/wf_api/run_lifecycle.py b/src/wf_api/run_lifecycle.py index eb2a6852..ddb56482 100644 --- a/src/wf_api/run_lifecycle.py +++ b/src/wf_api/run_lifecycle.py @@ -28,6 +28,7 @@ from wf_core import ( load_run_state, load_run_state_with_upgrade, ) +from wf_core.models.json_values import validate_strict_json_value def create_pinned_environment( @@ -62,8 +63,6 @@ def persist_admission( single authority for the run identity and frozen invocation. A failed durable admission must never dispatch. """ - from wf_core.models.json_values import validate_strict_json_value - frozen = validate_strict_json_value(dict(resolved_input)) if not isinstance(frozen, dict): raise ValueError("resolved workflow input must be a JSON object") @@ -91,6 +90,8 @@ def materialize_admitted_view( The view carries no checkpoint, trace, output, or step counts: the outcome is unknown until dispatch completes and persists a stopped checkpoint. + A view without a matching admission is corrupt and fails closed instead + of being silently returned (recovery authority is the admission record). """ now = datetime.now(UTC) try: @@ -98,6 +99,11 @@ def materialize_admitted_view( except KeyError: existing = None if existing is not None: + stored = store.get_admission(admission.id) + if stored.id != existing.id or stored.environment != existing.environment: + raise ValueError( + f"run view {admission.id!r} contradicts its admission record" + ) return existing record = WorkflowRunRecord( id=admission.id, @@ -116,12 +122,18 @@ def recover_admission_view(*, store: RunStore, run_id: str) -> WorkflowRunRecord """Reconcile a missing run view from its admission record. Recovery never executes work: it only completes the missing view so a - later poll can dispatch the captured invocation exactly once. + later poll can dispatch the captured invocation exactly once. A run view + without an admission record is corrupt and fails closed. """ try: - return store.get_run(run_id) + existing = store.get_run(run_id) except KeyError: - pass + existing = None + if existing is not None: + # Fail closed on a corrupt view-without-admission (F5): do not + # silently return it to clear overlap. + store.get_admission(run_id) + return existing admission = store.get_admission(run_id) return materialize_admitted_view(store=store, admission=admission) @@ -143,6 +155,9 @@ def persist_stopped_run( f"cannot persist active workflow run with status {run.status!s}" ) + # New admissions use store-backed run-###### ids (allocate_run_id); the + # run_ hex fallback only serves pre-admission stopped runs and legacy + # tests that persist without an admission record. key = run_id or f"run_{uuid4().hex}" now = datetime.now(UTC) sequence = 1 diff --git a/src/wf_api/runs.py b/src/wf_api/runs.py index f4cc7d3c..4444f481 100644 --- a/src/wf_api/runs.py +++ b/src/wf_api/runs.py @@ -100,10 +100,13 @@ class WorkflowRunApi: max_steps=limits.max_steps, ) - # Durable admission ordering: recheck -> allocate/freeze -> persist - # admission -> materialize view -> dispatch captured -> persist - # stopped -> reconcile. A failed durable admission never dispatches, - # and dispatch never re-resolves the deployment. + # Durable admission ordering for manual runs: deployment recheck -> + # allocate/freeze -> persist admission -> materialize view -> dispatch + # captured -> persist stopped. A failed durable admission never + # dispatches, and dispatch never re-resolves the deployment. + # TODO(T11): hold the single-owner admission lock around this sequence + # once scheduler ownership lands; manual recheck here is only + # deployment validation (no schedule/capacity/overlap yet). store = self._run_store() run_id = store.allocate_run_id() environment = create_pinned_environment( @@ -119,6 +122,9 @@ class WorkflowRunApi: max_steps=limits.max_steps, ) materialize_admitted_view(store=store, admission=admission) + # TODO(T10): record a dispatch mark between materialize and execute so + # crash-after-dispatch (abandoned, failed without replay) is + # distinguishable from pending-dispatch (safe to dispatch later). plan = raw_plan_from_artifact(admission.environment.root_artifact) captured_tree = saved_subgraph_tree_from_snapshots( admission.environment.child_artifacts @@ -282,7 +288,12 @@ class WorkflowRunApi: } async def inspect_run(self, *, run_id: str) -> RunResult: - """Return one durable stopped-run summary without debug trace entries.""" + """Return one durable stopped-run summary without debug trace entries. + + Admitted runs with no stopped checkpoint fail closed here (no + fabricated trace/output); checkpoint-free inspection arrives with + the scheduling administration surface (T13). + """ record, run = load_stored_run(self._run_store(), run_id) environment = record.environment return _run_payload( diff --git a/src/wf_artifacts/runs/store.py b/src/wf_artifacts/runs/store.py index d8ac5965..6bbb003f 100644 --- a/src/wf_artifacts/runs/store.py +++ b/src/wf_artifacts/runs/store.py @@ -42,7 +42,7 @@ class RunStore: class FileRunStore(RunStore): - """JSON file-backed stopped-run store for local development and tests. + """JSON file-backed admitted- and stopped-run store for local dev/tests. The internal lock protects individual writes inside one process only. `WorkflowRunApi.resume_run()` owns the same-process read/execute/write @@ -125,11 +125,15 @@ class FileRunStore(RunStore): seq = 0 if seq_path.exists(): try: - seq = int( - json.loads(seq_path.read_text(encoding="utf-8")).get("seq", 0) - ) - except ValueError, AttributeError: - seq = 0 + raw = json.loads(seq_path.read_text(encoding="utf-8")) + seq_value = raw.get("seq", 0) if isinstance(raw, dict) else None + if not isinstance(seq_value, int) or seq_value < 0: + raise ValueError( + f"corrupt run-id sequence at {seq_path}: {raw!r}" + ) + seq = seq_value + except (ValueError, AttributeError, TypeError) as exc: + raise ValueError(f"corrupt run-id sequence at {seq_path}") from exc seq += 1 self._write_json(seq_path, {"seq": seq}) return f"run-{seq:06d}" diff --git a/tests/artifacts/test_run_admission.py b/tests/artifacts/test_run_admission.py index 4be9a104..558f3ce5 100644 --- a/tests/artifacts/test_run_admission.py +++ b/tests/artifacts/test_run_admission.py @@ -11,6 +11,7 @@ from wf_api.run_lifecycle import ( load_stored_run, materialize_admitted_view, persist_admission, + recover_admission_view, ) from wf_artifacts import ( PinnedRunEnvironment, @@ -102,8 +103,6 @@ def test_run_ids_are_store_backed_across_restart(tmp_path: Path) -> None: def test_fault_before_admission_persists_nothing(tmp_path: Path, monkeypatch) -> None: - from wf_artifacts.runs import store as store_module - store = FileRunStore(tmp_path) run_id = store.allocate_run_id() @@ -122,13 +121,13 @@ def test_fault_before_admission_persists_nothing(tmp_path: Path, monkeypatch) -> schedule_id=None, schedule_revision=None, ) - assert store_module.FileRunStore(tmp_path).list_admissions() == [] + assert FileRunStore(tmp_path).list_admissions() == [] with pytest.raises(KeyError): store.get_admission(run_id) def test_fault_between_admission_and_view_recovers_without_dispatch( - tmp_path: Path, monkeypatch + tmp_path: Path, ) -> None: store = FileRunStore(tmp_path) run_id = store.allocate_run_id() @@ -147,8 +146,6 @@ def test_fault_between_admission_and_view_recovers_without_dispatch( with pytest.raises(KeyError): store.get_run(run_id) # Recovery materializes the missing view but never executes work. - from wf_api.run_lifecycle import recover_admission_view - record = recover_admission_view(store=store, run_id=run_id) assert record.status is StoredRunStatus.ADMITTED assert store.get_run(run_id).id == run_id @@ -179,3 +176,46 @@ def test_admitted_run_has_no_fabricated_checkpoint_or_output( # fabricate trace/output/step counts. with pytest.raises(ValueError, match="admitted"): load_stored_run(store, run_id) + + +def test_corrupt_run_id_sequence_fails_closed(tmp_path: Path) -> None: + FileRunStore(tmp_path) + seq_path = tmp_path / "runs" / "_run_id_seq.json" + seq_path.write_text('{"seq": "not-an-int"}', encoding="utf-8") + with pytest.raises(ValueError, match="corrupt run-id sequence"): + FileRunStore(tmp_path).allocate_run_id() + + +def test_view_without_admission_fails_closed(tmp_path: Path) -> None: + store = FileRunStore(tmp_path) + run_id = store.allocate_run_id() + admission = persist_admission( + store=store, + run_id=run_id, + environment=_env(), + resolved_input={}, + max_steps=None, + scheduled_at=None, + schedule_id=None, + schedule_revision=None, + ) + materialize_admitted_view(store=store, admission=admission) + (tmp_path / "runs" / run_id / "admission.json").unlink() + with pytest.raises(KeyError, match="unknown run admission"): + recover_admission_view(store=store, run_id=run_id) + + +def test_non_json_resolved_input_rejected_before_dispatch(tmp_path: Path) -> None: + store = FileRunStore(tmp_path) + run_id = store.allocate_run_id() + with pytest.raises(ValueError, match="must be finite"): + persist_admission( + store=store, + run_id=run_id, + environment=_env(), + resolved_input={"x": float("inf")}, # type: ignore[dict-item] + max_steps=None, + scheduled_at=None, + schedule_id=None, + schedule_revision=None, + ) diff --git a/tests/wf_api/test_run_admission_path.py b/tests/wf_api/test_run_admission_path.py index 39c07fcc..f54c517d 100644 --- a/tests/wf_api/test_run_admission_path.py +++ b/tests/wf_api/test_run_admission_path.py @@ -65,15 +65,26 @@ def test_manual_run_persists_admission_before_dispatch(tmp_path: Path) -> None: def test_fault_before_admission_never_dispatches(tmp_path: Path, monkeypatch) -> None: + import wf_api.runs as runs_module + api, store = _api_with_echo(tmp_path / "fault-before") + real_write = store._write_json + dispatched: list[str] = [] + real_plan = runs_module.raw_plan_from_artifact - def _fail_save(admission) -> None: # type: ignore[no-untyped-def] - raise OSError("injected admission failure") + def _fail_on_admission(path: Path, payload: object) -> None: + if path.name == "admission.json": + raise OSError("injected admission failure") + real_write(path, payload) - # Admission persist precedes dispatch in run_deployment ordering: a failure - # here must propagate before any run view exists, so no dispatch could - # have produced a stopped checkpoint. - monkeypatch.setattr(store, "save_admission", _fail_save) + def _spy_plan(artifact): # type: ignore[no-untyped-def] + dispatched.append("dispatch") + return real_plan(artifact) + + # Real serialization fault at the admission file: nothing after it + # (plan building, dispatch, stopped persist) may run. + monkeypatch.setattr(store, "_write_json", _fail_on_admission) + monkeypatch.setattr(runs_module, "raw_plan_from_artifact", _spy_plan) try: asyncio.run( api.run_deployment( @@ -84,8 +95,10 @@ def test_fault_before_admission_never_dispatches(tmp_path: Path, monkeypatch) -> raise AssertionError("fault must propagate") except OSError: pass - assert store.list_admissions() == [] - assert store.list_runs() == [] + assert dispatched == [] + fresh = FileRunStore(store.root) + assert fresh.list_admissions() == [] + assert fresh.list_runs() == [] def test_captured_invocation_freezes_input(tmp_path: Path) -> None: