diff --git a/docs/RELEASE_CONSOLE.md b/docs/RELEASE_CONSOLE.md index 6c09bd7..ca8c3a1 100644 --- a/docs/RELEASE_CONSOLE.md +++ b/docs/RELEASE_CONSOLE.md @@ -63,6 +63,16 @@ these digests detect accidental or manual corruption, not an attacker who can replace the private record and recompute its checksums. Changing a target, channel, or gate input requires a new run. +Creation requires a caller-generated `request_id`. Its SHA-256 fingerprint is +the private, workspace-scoped durable mapping to exactly one run; the raw +identifier is never persisted. Repeating the same identifier with the same +immutable inputs returns that run without rebuilding the plan, even if the +live dashboard has since drifted. Reusing it with different inputs fails +closed. The browser keeps an uncertain create identifier in session storage, +replays it after reload, and selects the known run returned by the server. A +successful create remains shown as saved if only the subsequent list refresh +fails. + Run records survive console restarts, but remain local operator state rather than a signed release artifact or the system audit log. The default location is `$XDG_STATE_HOME/govoplan/release-console/workspace-/release-runs/`, or @@ -79,8 +89,11 @@ rewritten nor automatically quarantined. The full resolved-workspace SHA-256 is part of both the private storage namespace and immutable input snapshot. Every list, read, and state transition checks it. An alternate workspace therefore cannot list or resume another -workspace's runs even when an embedding/test `run_state_root` override is -shared accidentally. +workspace's runs. This remains true for an embedding/test `run_state_root` +override: the server always appends +`workspace-/release-runs/` rather than treating the override as a +shared record directory. A corrupt record from one workspace therefore cannot +leak even its identifier or an integrity error into another workspace. Each frozen plan step has an explicit `pending`, `running`, `succeeded`, `failed`, or `interrupted` state. Plan order remains a prerequisite: a later @@ -94,6 +107,15 @@ enum event type, timestamp, step identifier, and bounded result code—never commands, process output, confirmation text, credentials, bearer tokens, or signing material. +Before a step can enter `running`, the store reserves the two command-ledger +slots needed for worst-case recovery. An explicit resume consumes one slot and +is accepted only while a persisted attempt is actually running. A mutating +attempt always retains its final `effect_absent` or `effect_succeeded` slot; +`unresolved` may be recorded at most once for that attempt and only when an +additional slot is available. Read-only interruption and known failure retain +the slot needed to prepare a retry. Capacity exhaustion is therefore detected +before starting an effect rather than stranding an ambiguous attempt. + An explicit resume after a process restart converts every persisted `running` step to `interrupted`; it never guesses whether an external effect happened. An interrupted read-only step can be prepared for retry. An interrupted @@ -105,10 +127,15 @@ effect, and `unresolved` keeps the run blocked. Each outcome emits a bounded, code-only state event. A known failed attempt can likewise be prepared for retry. The UI keeps unavailable controls visible and disabled. +The browser likewise retains the request identifier for an uncertain +resume/retry/reconciliation response and reuses it on the operator's retry. +Identifiers are cleared only after the server returns the resulting run state. + The run API is covered by the same local console token middleware as every other `/api/` route: -- `POST /api/release-runs` rebuilds and freezes a selective plan. +- `POST /api/release-runs` requires `request_id`, then idempotently rebuilds and + freezes a selective plan only when that creation is not already known. - `GET /api/release-runs` lists bounded summaries; unreadable entries are reported as unavailable instead of being silently omitted. - `GET /api/release-runs/{run_id}` reads and verifies one exact record. diff --git a/tests/test_release_run_state.py b/tests/test_release_run_state.py index f0c5cea..3cf646e 100644 --- a/tests/test_release_run_state.py +++ b/tests/test_release_run_state.py @@ -1,9 +1,13 @@ from __future__ import annotations from concurrent.futures import ThreadPoolExecutor +from itertools import count import json import os from pathlib import Path +import re +import shutil +import subprocess import sys import tempfile import threading @@ -11,6 +15,7 @@ import unittest from unittest.mock import patch from fastapi.testclient import TestClient +from filelock import FileLock META_ROOT = Path(__file__).resolve().parents[1] @@ -34,6 +39,7 @@ from server.app import ( # noqa: E402 WORKSPACE_FINGERPRINT = "a" * 64 +CREATE_REQUESTS = count(1) class ReleaseRunStoreTests(unittest.TestCase): @@ -49,7 +55,8 @@ class ReleaseRunStoreTests(unittest.TestCase): self.temporary.cleanup() def test_create_persists_private_immutable_plan_and_ordered_steps(self) -> None: - run = self.store.create( + run = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() ) @@ -60,7 +67,7 @@ class ReleaseRunStoreTests(unittest.TestCase): record_path = self.root / f"{run['run_id']}.json" self.assertEqual("govoplan.release-run", reopened["schema"]) - self.assertEqual(2, reopened["schema_version"]) + self.assertEqual(3, reopened["schema_version"]) self.assertEqual( {"govoplan-core": "1.2.3", "govoplan-files": "1.2.4"}, reopened["immutable"]["input"]["repo_versions"], @@ -78,10 +85,43 @@ class ReleaseRunStoreTests(unittest.TestCase): plan = release_plan() plan["dry_run_steps"] = [] with self.assertRaisesRegex(ReleaseRunConflict, "step collection"): - self.store.create(input_snapshot=run_input(), plan_snapshot=plan) + create_run(self.store, input_snapshot=run_input(), plan_snapshot=plan) + + def test_create_request_is_durable_idempotent_and_input_bound(self) -> None: + request_id = "durable-create-request-0001" + created = create_run(self.store, request_id=request_id) + changed_plan = release_plan() + changed_plan["dry_run_steps"][0]["title"] = "Dashboard drift" # type: ignore[index] + + replayed = create_run( + self.store, + request_id=request_id, + plan_snapshot=changed_plan, + ) + found = self.store.find_created_run( + request_id=request_id, + input_snapshot=run_input(), + ) + self.assertEqual(created["run_id"], replayed["run_id"]) + self.assertEqual(created["run_id"], found["run_id"] if found else None) + self.assertEqual( + "Inspect Core", + replayed["immutable"]["plan"]["dry_run_steps"][0]["title"], + ) + self.assertEqual(1, len(list(self.root.glob("*.json")))) + + changed_input = run_input() + changed_input["public_catalog"] = True + with self.assertRaisesRegex(ReleaseRunConflict, "different inputs"): + create_run( + self.store, + request_id=request_id, + input_snapshot=changed_input, + ) def test_attempt_transitions_are_idempotent_and_strictly_ordered(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] @@ -122,7 +162,8 @@ class ReleaseRunStoreTests(unittest.TestCase): self.assertTrue(finished["state"]["steps"][1]["available"]) def test_failed_attempt_and_delayed_start_replays_are_idempotent(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] attempt_id = "failed-attempt-replay-0001" @@ -155,77 +196,129 @@ class ReleaseRunStoreTests(unittest.TestCase): len(retried["state"]["events"]), len(delayed["state"]["events"]) ) - def test_delayed_resume_request_is_idempotent_for_the_run_lifetime(self) -> None: - run_id = self.store.create( + def test_resume_requires_a_running_attempt_and_only_records_an_effect(self) -> None: + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] original_request = "resume-original-request-0001" - self.store.resume(run_id, request_id=original_request) - for index in range(130): - self.store.resume( - run_id, request_id=f"resume-intervening-request-{index:04d}" - ) - running = self.store.start_step( + with self.assertRaisesRegex(ReleaseRunConflict, "only be resumed"): + self.store.resume(run_id, request_id=original_request) + before_start = self.store.get(run_id) + + self.store.start_step( run_id, "core:preflight", attempt_id="resume-delayed-attempt-0001", ) + resumed = self.store.resume(run_id, request_id=original_request) duplicate = self.store.resume(run_id, request_id=original_request) - self.assertEqual("running", duplicate["state"]["steps"][0]["state"]) + self.assertEqual("interrupted", duplicate["state"]["steps"][0]["state"]) self.assertEqual( - len(running["state"]["events"]), len(duplicate["state"]["events"]) + len(resumed["state"]["events"]), len(duplicate["state"]["events"]) ) + raw = json.loads( + (self.root / f"{run_id}.json").read_text(encoding="utf-8") + ) + self.assertEqual(1, len(raw["state"]["processed_requests"])) + self.assertLess(len(before_start["state"]["events"]), len(resumed["state"]["events"])) - def test_command_and_attempt_capacity_fail_closed_without_eviction(self) -> None: - run_id = self.store.create( - input_snapshot=run_input(), plan_snapshot=release_plan() - )["run_id"] - with patch("govoplan_release.release_run.MAX_COMMANDS", 2): - self.store.resume(run_id, request_id="capacity-resume-request-0001") - self.store.resume(run_id, request_id="capacity-resume-request-0002") - with self.assertRaisesRegex(ReleaseRunConflict, "safety limit"): - self.store.resume(run_id, request_id="capacity-resume-request-0003") - duplicate = self.store.resume( - run_id, request_id="capacity-resume-request-0001" - ) - self.assertEqual("planned", duplicate["state"]["status"]) - - attempt_run_id = self.store.create( + def test_attempt_start_reserves_capacity_for_ambiguous_effect_recovery(self) -> None: + saturated_run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] with patch("govoplan_release.release_run.MAX_COMMANDS", 1): - self.store.start_step( - attempt_run_id, - "core:preflight", - attempt_id="capacity-attempt-0001", - ) - self.store.finish_step( - attempt_run_id, - "core:preflight", - attempt_id="capacity-attempt-0001", - succeeded=False, - result_code="preflight_failed", - ) - self.store.retry_step( - attempt_run_id, - "core:preflight", - request_id="capacity-retry-request-0001", - ) - with self.assertRaisesRegex(ReleaseRunConflict, "attempt history"): + with self.assertRaisesRegex(ReleaseRunConflict, "recovery capacity"): self.store.start_step( - attempt_run_id, + saturated_run_id, "core:preflight", - attempt_id="capacity-attempt-0002", + attempt_id="capacity-prestart-attempt-0001", ) - delayed_attempt = self.store.start_step( - attempt_run_id, - "core:preflight", - attempt_id="capacity-attempt-0001", + untouched = self.store.get(saturated_run_id) + self.assertEqual("pending", untouched["state"]["steps"][0]["state"]) + self.assertEqual(0, untouched["state"]["steps"][0]["attempt_count"]) + + terminal_run_id = create_run( + self.store, + input_snapshot=run_input(), + plan_snapshot=mutating_first_plan(), + )["run_id"] + with patch("govoplan_release.release_run.MAX_COMMANDS", 2): + self.store.start_step( + terminal_run_id, "core:tag", attempt_id="capacity-terminal-attempt" ) - self.assertEqual("pending", delayed_attempt["state"]["steps"][0]["state"]) + self.store.resume( + terminal_run_id, request_id="capacity-terminal-resume-0001" + ) + with self.assertRaisesRegex(ReleaseRunConflict, "safety limit"): + self.store.reconcile_step( + terminal_run_id, + "core:tag", + request_id="capacity-unresolved-request-0001", + outcome="unresolved", + confirmation="RECONCILE", + ) + completed = self.store.reconcile_step( + terminal_run_id, + "core:tag", + request_id="capacity-terminal-request-0001", + outcome="effect_succeeded", + confirmation="RECONCILE", + ) + self.assertEqual("completed", completed["state"]["status"]) + + def test_unresolved_reconciliation_is_bounded_and_preserves_terminal_slot(self) -> None: + run_id = create_run( + self.store, + input_snapshot=run_input(), + plan_snapshot=mutating_first_plan(), + )["run_id"] + with patch("govoplan_release.release_run.MAX_COMMANDS", 3): + self.store.start_step( + run_id, "core:tag", attempt_id="capacity-unresolved-attempt" + ) + self.store.resume( + run_id, request_id="capacity-unresolved-resume-0001" + ) + unresolved = self.store.reconcile_step( + run_id, + "core:tag", + request_id="capacity-unresolved-request-0001", + outcome="unresolved", + confirmation="RECONCILE", + ) + duplicate = self.store.reconcile_step( + run_id, + "core:tag", + request_id="capacity-unresolved-request-0001", + outcome="unresolved", + confirmation="RECONCILE", + ) + self.assertEqual( + len(unresolved["state"]["events"]), + len(duplicate["state"]["events"]), + ) + with self.assertRaisesRegex(ReleaseRunConflict, "already recorded"): + self.store.reconcile_step( + run_id, + "core:tag", + request_id="capacity-unresolved-request-0002", + outcome="unresolved", + confirmation="RECONCILE", + ) + completed = self.store.reconcile_step( + run_id, + "core:tag", + request_id="capacity-final-request-0001", + outcome="effect_succeeded", + confirmation="RECONCILE", + ) + self.assertEqual("completed", completed["state"]["status"]) def test_reopen_and_resume_preserve_ambiguous_mutating_effect(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=mutating_first_plan() )["run_id"] self.store.start_step(run_id, "core:tag", attempt_id="attempt-tag-0001") @@ -281,7 +374,8 @@ class ReleaseRunStoreTests(unittest.TestCase): ) def test_reconciliation_can_record_unresolved_or_verified_success(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=mutating_first_plan() )["run_id"] self.store.start_step(run_id, "core:tag", attempt_id="reconcile-attempt-0001") @@ -322,7 +416,8 @@ class ReleaseRunStoreTests(unittest.TestCase): self.assertEqual("completed", succeeded["state"]["status"]) def test_tampered_record_fails_closed_and_is_not_rewritten(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] path = self.root / f"{run_id}.json" @@ -338,7 +433,8 @@ class ReleaseRunStoreTests(unittest.TestCase): self.assertEqual("unavailable", self.store.list()[0]["status"]) def test_broad_record_mode_and_symlinked_root_fail_closed(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] path = self.root / f"{run_id}.json" @@ -357,8 +453,58 @@ class ReleaseRunStoreTests(unittest.TestCase): expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, ).list() + def test_symlinked_lock_and_lock_io_errors_fail_closed(self) -> None: + self.store.list() + lock_path = self.root / ".release-runs.lock" + lock_path.unlink() + target = self.root / "lock-target" + target.write_text("do not follow", encoding="utf-8") + lock_path.symlink_to(target) + with self.assertRaisesRegex(ReleaseRunCorrupt, "lock path"): + self.store.list() + + lock_path.unlink() + with patch( + "govoplan_release.release_run.FileLock.acquire", + side_effect=PermissionError("denied"), + ): + with self.assertRaisesRegex(ReleaseRunCorrupt, "acquired safely"): + self.store.list() + + original_release = FileLock.release + + def fail_operator_release( + lock: FileLock, *args: object, **kwargs: object + ) -> None: + if kwargs.get("force") is True: + original_release(lock, *args, **kwargs) # type: ignore[arg-type] + return + raise PermissionError("denied") + + with patch.object( + FileLock, + "release", + autospec=True, + side_effect=fail_operator_release, + ): + with self.assertRaisesRegex(ReleaseRunCorrupt, "released safely"): + self.store.list() + + def test_atomic_write_oserror_is_normalized_and_leaves_no_record(self) -> None: + with patch( + "govoplan_release.release_run.os.replace", + side_effect=PermissionError("denied"), + ): + with self.assertRaisesRegex(ReleaseRunCorrupt, "persisted safely"): + create_run( + self.store, + request_id="atomic-write-create-request-0001", + ) + self.assertEqual([], list(self.root.glob("*.json"))) + def test_state_specific_fields_and_boolean_counts_fail_closed(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] path = self.root / f"{run_id}.json" @@ -372,7 +518,8 @@ class ReleaseRunStoreTests(unittest.TestCase): self.store.get(run_id) def test_valid_looking_mutable_state_tampering_fails_record_checksum(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=mutating_first_plan() )["run_id"] self.store.start_step(run_id, "core:tag", attempt_id="checksum-attempt-0001") @@ -396,7 +543,8 @@ class ReleaseRunStoreTests(unittest.TestCase): self.assertEqual(tampered, path.read_bytes()) def test_impossible_step_chronology_fails_even_with_recomputed_checksum(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] self.store.start_step( @@ -421,9 +569,13 @@ class ReleaseRunStoreTests(unittest.TestCase): self.store.get(run_id) def test_record_and_retained_event_timestamps_are_strictly_ordered(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] + self.store.start_step( + run_id, "core:preflight", attempt_id="timestamp-attempt-0001" + ) self.store.resume(run_id, request_id="timestamp-resume-request-0001") path = self.root / f"{run_id}.json" payload = json.loads(path.read_text(encoding="utf-8")) @@ -439,7 +591,8 @@ class ReleaseRunStoreTests(unittest.TestCase): self.store.get(run_id) def test_attempt_count_must_match_lifetime_attempt_history(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] self.store.start_step( @@ -456,7 +609,8 @@ class ReleaseRunStoreTests(unittest.TestCase): self.store.get(run_id) def test_workspace_fingerprint_is_enforced_for_every_store_operation(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] foreign = ReleaseRunStore( @@ -470,7 +624,11 @@ class ReleaseRunStoreTests(unittest.TestCase): with self.assertRaises(ReleaseRunNotFound): foreign.resume(run_id, request_id="foreign-resume-request-0001") with self.assertRaisesRegex(ReleaseRunConflict, "different workspace"): - foreign.create(input_snapshot=run_input(), plan_snapshot=release_plan()) + create_run( + foreign, + input_snapshot=run_input(), + plan_snapshot=release_plan(), + ) def test_list_limit_is_applied_after_foreign_workspace_records_are_skipped( self, @@ -479,7 +637,8 @@ class ReleaseRunStoreTests(unittest.TestCase): "govoplan_release.release_run._timestamp", return_value="2026-07-22T00:00:00Z", ): - local_run = self.store.create( + local_run = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() ) foreign_store = ReleaseRunStore( @@ -492,7 +651,8 @@ class ReleaseRunStoreTests(unittest.TestCase): "govoplan_release.release_run._timestamp", return_value="2026-07-22T00:00:01Z", ): - foreign_store.create( + create_run( + foreign_store, input_snapshot=foreign_input, plan_snapshot=release_plan() ) @@ -500,7 +660,8 @@ class ReleaseRunStoreTests(unittest.TestCase): self.assertEqual([local_run["run_id"]], [item["run_id"] for item in listed]) def test_concurrent_claims_have_one_winner_and_no_lost_update(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] barrier = threading.Barrier(2) @@ -535,7 +696,8 @@ class ReleaseRunStoreTests(unittest.TestCase): ) def test_event_history_is_bounded_and_command_history_is_not_evicted(self) -> None: - run_id = self.store.create( + run_id = create_run( + self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] for index in range(90): @@ -588,7 +750,7 @@ class ReleaseRunApiTests(unittest.TestCase): created_response = client.post( "/api/release-runs", headers=headers, - json=run_input(), + json=api_run_input(request_id="api-create-request-0001"), ) self.assertEqual(200, created_response.status_code) created = created_response.json() @@ -623,6 +785,59 @@ class ReleaseRunApiTests(unittest.TestCase): self.assertEqual(200, retried.status_code) self.assertEqual("pending", retried.json()["state"]["steps"][0]["state"]) + def test_create_requires_request_id_and_replays_before_plan_rebuild(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + app = create_app( + workspace_root=Path(temp_dir), + run_state_root=Path(temp_dir) / "runs", + token="token", + ) + headers = {"X-Release-Console-Token": "token"} + dashboard = patch("server.app.build_dashboard", return_value=object()) + planner = patch( + "server.app.build_selective_release_plan", + return_value=release_plan(), + ) + with dashboard as build_dashboard, planner as build_plan, TestClient( + app + ) as client: + missing = client.post( + "/api/release-runs", + headers=headers, + json={ + key: value + for key, value in api_run_input().items() + if key != "request_id" + }, + ) + payload = api_run_input(request_id="api-idempotent-create-0001") + created = client.post( + "/api/release-runs", headers=headers, json=payload + ) + self.assertEqual(200, created.status_code) + build_dashboard.reset_mock() + build_plan.reset_mock() + + replayed = client.post( + "/api/release-runs", headers=headers, json=payload + ) + mismatched = client.post( + "/api/release-runs", + headers=headers, + json=payload | {"public_catalog": True}, + ) + + self.assertEqual(422, missing.status_code) + self.assertEqual(200, replayed.status_code) + self.assertEqual(created.json()["run_id"], replayed.json()["run_id"]) + self.assertEqual(409, mismatched.status_code) + self.assertIn("different inputs", mismatched.json()["detail"]) + build_dashboard.assert_not_called() + build_plan.assert_not_called() + self.assertEqual( + 1, len(list(app.state.release_runs.root.glob("*.json"))) + ) + def test_create_rejects_unknown_or_unresolved_repository(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) / "runs" @@ -641,6 +856,7 @@ class ReleaseRunApiTests(unittest.TestCase): "/api/release-runs", headers={"X-Release-Console-Token": "token"}, json={ + "request_id": "api-unknown-create-request-0001", "channel": "stable", "repo_versions": {"govoplan-unknown": "1.0.0"}, "public_catalog": False, @@ -654,19 +870,20 @@ class ReleaseRunApiTests(unittest.TestCase): def test_api_rejects_tampered_record_without_rewriting_it(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) / "runs" - store = ReleaseRunStore( - root, - expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, - ) - run_id = store.create( - input_snapshot=run_input(), plan_snapshot=release_plan() - )["run_id"] - path = root / f"{run_id}.json" - path.write_text("{}", encoding="utf-8") - malformed = path.read_bytes() app = create_app( workspace_root=Path(temp_dir), run_state_root=root, token="token" ) + run_id = create_run( + app.state.release_runs, + request_id="api-tamper-create-request-0001", + input_snapshot=run_input( + workspace_fingerprint=app.state.workspace_fingerprint + ), + plan_snapshot=release_plan(), + )["run_id"] + path = app.state.release_runs.root / f"{run_id}.json" + path.write_text("{}", encoding="utf-8") + malformed = path.read_bytes() with TestClient(app) as client: response = client.get( f"/api/release-runs/{run_id}", @@ -712,7 +929,9 @@ class ReleaseRunApiTests(unittest.TestCase): TestClient(app) as client, ): created = client.post( - "/api/release-runs", headers=headers, json=run_input() + "/api/release-runs", + headers=headers, + json=api_run_input(request_id="api-reconcile-create-0001"), ).json() run_id = created["run_id"] app.state.release_runs.start_step( @@ -807,14 +1026,19 @@ class ReleaseRunApiTests(unittest.TestCase): response = client.post( "/api/release-runs", headers={"X-Release-Console-Token": "token"}, - json=run_input(), + json=api_run_input(request_id="api-explicit-create-0001"), ) self.assertEqual(200, response.status_code) self.assertEqual( release_workspace_fingerprint(first), response.json()["immutable"]["input"]["workspace_fingerprint"], ) - self.assertEqual(explicit_root.resolve(), app.state.release_runs.root) + self.assertEqual( + explicit_root.resolve() + / f"workspace-{release_workspace_fingerprint(first)}" + / "release-runs", + app.state.release_runs.root, + ) def test_shared_explicit_root_does_not_cross_workspace_boundary(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: @@ -837,9 +1061,16 @@ class ReleaseRunApiTests(unittest.TestCase): TestClient(first_app) as client, ): created = client.post( - "/api/release-runs", headers=headers, json=run_input() + "/api/release-runs", + headers=headers, + json=api_run_input(request_id="api-shared-create-0001"), ).json() + first_record = ( + first_app.state.release_runs.root / f"{created['run_id']}.json" + ) + first_record.write_text("{}", encoding="utf-8") + second_app = create_app( workspace_root=second, run_state_root=root, token="token" ) @@ -850,6 +1081,10 @@ class ReleaseRunApiTests(unittest.TestCase): ) self.assertEqual([], listed.json()["runs"]) self.assertEqual(404, loaded.status_code) + self.assertNotEqual( + first_app.state.release_runs.root, + second_app.state.release_runs.root, + ) def test_ui_and_runbook_state_tracking_boundary_are_explicit(self) -> None: webui = (RELEASE_ROOT / "webui" / "index.html").read_text(encoding="utf-8") @@ -866,13 +1101,163 @@ class ReleaseRunApiTests(unittest.TestCase): self.assertIn("effect_succeeded", webui) self.assertIn("renderReleaseRunStoreUnavailable", webui) self.assertIn('const runsRequest = api("/api/release-runs")', webui) + self.assertIn("pendingReleaseRunPayload", webui) + self.assertIn("recoverPendingReleaseRun", webui) + self.assertIn("inFlightRunCommandId", webui) + self.assertIn("Run saved; list refresh unavailable", webui) self.assertIn("This foundation tracks state only.", runbook) self.assertIn("effect_absent", runbook) self.assertIn("$XDG_STATE_HOME", runbook) self.assertIn("same-directory", runbook) + self.assertIn("caller-generated `request_id`", runbook) + self.assertIn("reserves the two command-ledger", runbook) + + def test_ui_reuses_uncertain_ids_and_separates_saved_run_list_failure( + self, + ) -> None: + node = shutil.which("node") + if node is None: + self.skipTest("Node.js is required for the Release Console UI check") + webui = (RELEASE_ROOT / "webui" / "index.html").read_text(encoding="utf-8") + function_names = ( + "pendingReleaseRunPayload", + "readPendingReleaseRun", + "clearPendingReleaseRun", + "recoverPendingReleaseRun", + "submitReleaseRunCreate", + "refreshReleaseRunListAfterCreate", + "readInFlightRunCommands", + "inFlightRunCommandId", + "clearInFlightRunCommand", + "requestId", + ) + functions = "\n".join( + javascript_function(webui, name) for name in function_names + ) + script = ( + """ +const assert = (condition, message) => { if (!condition) throw new Error(message); }; +const values = new Map(); +const sessionStorage = { + getItem: (key) => values.has(key) ? values.get(key) : null, + setItem: (key, value) => values.set(key, String(value)), + removeItem: (key) => values.delete(key), +}; +let uuidSequence = 0; +const window = { crypto: { randomUUID: () => `uuid-${++uuidSequence}` } }; +const pendingReleaseRunKey = "test-pending-create"; +const pendingRunCommandsKey = "test-pending-commands"; +const releaseState = { currentRun: null, runStoreAvailable: true }; +const elements = { + runStatus: { textContent: "" }, + runOutput: { + innerHTML: "", + insertAdjacentHTML: (_position, html) => { elements.runOutput.innerHTML = html + elements.runOutput.innerHTML; }, + }, + releaseRun: { value: "" }, +}; +const pill = (label) => label; +const escapeHtml = (value) => String(value); +const restoreReleaseRunSelection = () => {}; +const renderReleaseRun = (run) => { elements.runOutput.innerHTML = `RUN:${run.run_id}`; }; +const renderReleaseRunList = () => {}; +let failCreate = true; +let failList = true; +const postedRequestIds = []; +async function postJson(_path, payload) { + postedRequestIds.push(payload.request_id); + if (failCreate) throw new Error("uncertain transport failure"); + return { run_id: "rr-known", immutable: { input: {} }, state: { steps: [] } }; +} +async function api(_path) { + if (failList) throw new Error("list refresh failed"); + return { runs: [] }; +} +""" + + functions + + """ +(async () => { + const input = { repo_versions: { "govoplan-files": "1.2.4", "govoplan-core": "1.2.3" }, channel: "stable" }; + const first = pendingReleaseRunPayload(input); + const same = pendingReleaseRunPayload({ ...input, repo_versions: { "govoplan-core": "1.2.3", "govoplan-files": "1.2.4" } }); + assert(first.request_id === same.request_id, "same create input did not reuse its request ID"); + const uncertain = await submitReleaseRunCreate(first, { automatic: true }); + assert(uncertain === false, "uncertain create unexpectedly succeeded"); + assert(readPendingReleaseRun().payload.request_id === first.request_id, "uncertain create ID was cleared"); + + failCreate = false; + const recovered = await recoverPendingReleaseRun(); + assert(recovered === true, "pending create did not recover"); + assert(postedRequestIds.length === 2 && postedRequestIds.every((value) => value === first.request_id), "create replay changed request ID"); + assert(readPendingReleaseRun() === null, "successful replay did not clear pending create"); + assert(elements.runOutput.innerHTML.includes("Run saved; list refresh unavailable"), "list failure did not preserve saved state"); + assert(!elements.runOutput.innerHTML.includes("Run creation failed"), "list failure was labeled as create failure"); + assert(releaseState.currentRun.run_id === "rr-known", "saved run was not retained"); + + const commandOne = inFlightRunCommandId("retry:rr-known:step", "retry"); + const commandReplay = inFlightRunCommandId("retry:rr-known:step", "retry"); + assert(commandOne === commandReplay, "uncertain command did not reuse its request ID"); + clearInFlightRunCommand("retry:rr-known:step", commandOne); + const commandAfterSuccess = inFlightRunCommandId("retry:rr-known:step", "retry"); + assert(commandAfterSuccess !== commandOne, "successful command did not clear its request ID"); +})().catch((error) => { console.error(error.stack || error.message); process.exit(1); }); +""" + ) + result = subprocess.run( # noqa: S603 + [node, "-e", script], + check=False, + capture_output=True, + text=True, + ) + self.assertEqual(0, result.returncode, result.stderr) -def run_input() -> dict[str, object]: +def create_run( + store: ReleaseRunStore, + *, + request_id: str | None = None, + input_snapshot: dict[str, object] | None = None, + plan_snapshot: dict[str, object] | None = None, +) -> dict[str, object]: + return store.create( + request_id=request_id + or f"test-create-request-{next(CREATE_REQUESTS):06d}", + input_snapshot=input_snapshot if input_snapshot is not None else run_input(), + plan_snapshot=plan_snapshot if plan_snapshot is not None else release_plan(), + ) + + +def api_run_input( + *, request_id: str | None = None, **overrides: object +) -> dict[str, object]: + payload = run_input() + payload.pop("workspace_fingerprint") + payload["request_id"] = request_id or ( + f"api-create-request-{next(CREATE_REQUESTS):06d}" + ) + payload.update(overrides) + return payload + + +def javascript_function(source: str, name: str) -> str: + match = re.search( + rf"\n\s+(?:async\s+)?function\s+{re.escape(name)}\s*\(", source + ) + if match is None: + raise AssertionError(f"JavaScript function {name} was not found") + following = re.search( + r"\n\s+(?:async\s+)?function\s+[A-Za-z_$]", + source[match.end() :], + ) + if following is None: + raise AssertionError(f"JavaScript function after {name} was not found") + end = match.end() + following.start() + return source[match.start() : end].strip() + + +def run_input( + *, workspace_fingerprint: str = WORKSPACE_FINGERPRINT +) -> dict[str, object]: return { "channel": "stable", "repo_versions": { @@ -883,7 +1268,7 @@ def run_input() -> dict[str, object]: "remote_tags": False, "public_catalog": False, "include_migrations": True, - "workspace_fingerprint": WORKSPACE_FINGERPRINT, + "workspace_fingerprint": workspace_fingerprint, } diff --git a/tools/release/govoplan_release/release_run.py b/tools/release/govoplan_release/release_run.py index 792aa77..f85d0b2 100644 --- a/tools/release/govoplan_release/release_run.py +++ b/tools/release/govoplan_release/release_run.py @@ -19,20 +19,21 @@ import stat import tempfile import threading from typing import Any -import uuid from filelock import FileLock, Timeout as FileLockTimeout SCHEMA = "govoplan.release-run" -SCHEMA_VERSION = 2 +SCHEMA_VERSION = 3 MAX_RECORD_BYTES = 2 * 1024 * 1024 MAX_PLAN_STEPS = 512 MAX_REPOSITORY_VERSIONS = 128 MAX_EVENTS = 256 MAX_COMMANDS = 2_048 MAX_LIST_LIMIT = 100 -RUN_ID_PATTERN = re.compile(r"^rr-[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}$") +RUN_ID_PATTERN = re.compile( + r"^(?:rr-[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}|rr-request-[0-9a-f]{64})$" +) STEP_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@-]{0,159}$") REQUEST_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{7,127}$") CHANNEL_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$") @@ -57,6 +58,7 @@ EVENT_TYPES = { ATTEMPT_OUTCOMES = {"running", "succeeded", "failed", "interrupted"} RECONCILIATION_OUTCOMES = {"effect_absent", "effect_succeeded", "unresolved"} RECONCILIATION_CONFIRMATION = "RECONCILE" +START_RECOVERY_RESERVE = 2 class ReleaseRunError(RuntimeError): @@ -97,9 +99,11 @@ class ReleaseRunStore: def create( self, *, + request_id: str, input_snapshot: dict[str, Any], plan_snapshot: dict[str, Any], ) -> dict[str, Any]: + request_fingerprint = _request_fingerprint(request_id) immutable_input = _validated_input(input_snapshot) if ( immutable_input["workspace_fingerprint"] @@ -110,9 +114,7 @@ class ReleaseRunStore: ) immutable_plan = _validated_plan(plan_snapshot) now = _timestamp() - run_id = ( - f"rr-{now.replace('-', '').replace(':', '')[:15]}Z-{uuid.uuid4().hex[:12]}" - ) + run_id = _creation_run_id(request_fingerprint) immutable = { "input": immutable_input, "plan": immutable_plan, @@ -134,6 +136,7 @@ class ReleaseRunStore: "schema": SCHEMA, "schema_version": SCHEMA_VERSION, "run_id": run_id, + "create_request_fingerprint": request_fingerprint, "created_at": now, "updated_at": now, "immutable": immutable, @@ -152,9 +155,42 @@ class ReleaseRunStore: record["record_digest"] = _record_digest(record) with self._locked(): self._ensure_root() + existing = self._read_optional(run_id) + if existing is not None: + _validate_create_replay( + existing, + request_fingerprint=request_fingerprint, + input_snapshot=immutable_input, + ) + return _view(existing) self._write_new(record) return _view(record) + def find_created_run( + self, *, request_id: str, input_snapshot: dict[str, Any] + ) -> dict[str, Any] | None: + """Return an exact prior create command before rebuilding its plan.""" + + request_fingerprint = _request_fingerprint(request_id) + immutable_input = _validated_input(input_snapshot) + if ( + immutable_input["workspace_fingerprint"] + != self.expected_workspace_fingerprint + ): + raise ReleaseRunConflict( + "Release run input belongs to a different workspace." + ) + with self._locked(): + record = self._read_optional(_creation_run_id(request_fingerprint)) + if record is None: + return None + _validate_create_replay( + record, + request_fingerprint=request_fingerprint, + input_snapshot=immutable_input, + ) + return _view(record) + def get(self, run_id: str) -> dict[str, Any]: with self._locked(): record = self._read(run_id) @@ -214,10 +250,17 @@ class ReleaseRunStore: record = self._read(run_id) if _request_seen(record, request_fingerprint, "resume", None, None): return _view(record) + running_steps = [ + step + for step in record["state"]["steps"] + if step["state"] == "running" + ] + if not running_steps: + raise ReleaseRunConflict( + "A release run can only be resumed while a step is running." + ) _ensure_command_capacity(record) - for step in record["state"]["steps"]: - if step["state"] != "running": - continue + for step in running_steps: attempt = _attempt_for_fingerprint( record, step["attempt_fingerprint"], step_id=step["id"] ) @@ -232,7 +275,14 @@ class ReleaseRunStore: step_id=step["id"], result_code="process_interrupted", ) - _remember_request(record, request_fingerprint, "resume", None, None) + _remember_request( + record, + request_fingerprint, + "resume", + None, + None, + attempt_fingerprint=running_steps[0]["attempt_fingerprint"], + ) _append_event(record, event_type="run_resumed") self._persist_update(record) return _view(record) @@ -260,6 +310,7 @@ class ReleaseRunStore: "Record the observed effect of an interrupted mutating step before retry." ) _ensure_command_capacity(record) + attempt_fingerprint = step["attempt_fingerprint"] retry_reason = ( "known_failure" if step["state"] == "failed" @@ -274,7 +325,14 @@ class ReleaseRunStore: "result_code": None, } ) - _remember_request(record, request_fingerprint, "retry", step_id, None) + _remember_request( + record, + request_fingerprint, + "retry", + step_id, + None, + attempt_fingerprint=attempt_fingerprint, + ) _append_event( record, event_type="step_retry_requested", @@ -316,7 +374,20 @@ class ReleaseRunStore: raise ReleaseRunConflict( "Only an interrupted mutating step can be reconciled." ) - _ensure_command_capacity(record) + attempt_fingerprint = step["attempt_fingerprint"] + if outcome == "unresolved": + if any( + request["operation"] == "reconcile" + and request["parameter"] == "unresolved" + and request["attempt_fingerprint"] == attempt_fingerprint + for request in record["state"]["processed_requests"] + ): + raise ReleaseRunConflict( + "An unresolved outcome is already recorded for this attempt." + ) + _ensure_command_capacity(record, reserve_after=1) + else: + _ensure_command_capacity(record) if outcome == "effect_absent": step.update( { @@ -336,7 +407,12 @@ class ReleaseRunStore: } ) _remember_request( - record, request_fingerprint, "reconcile", step_id, outcome + record, + request_fingerprint, + "reconcile", + step_id, + outcome, + attempt_fingerprint=attempt_fingerprint, ) _append_event( record, @@ -374,6 +450,7 @@ class ReleaseRunStore: available, reason = _step_available(record, step_id) if not available: raise ReleaseRunConflict(reason) + _ensure_start_recovery_capacity(record) _ensure_attempt_capacity(record) step.update( { @@ -506,11 +583,22 @@ class ReleaseRunStore: raise ReleaseRunCorrupt("Release run record is unreadable.") from exc finally: if descriptor >= 0: - os.close(descriptor) + try: + os.close(descriptor) + except OSError as exc: + raise ReleaseRunCorrupt( + "Release run record could not be closed safely." + ) from exc _validate_record(payload, expected_run_id=run_id) self._assert_workspace(payload) return payload + def _read_optional(self, run_id: str) -> dict[str, Any] | None: + try: + return self._read(run_id) + except ReleaseRunNotFound: + return None + def _assert_workspace(self, record: dict[str, Any]) -> None: if ( record["immutable"]["input"]["workspace_fingerprint"] @@ -577,19 +665,48 @@ class ReleaseRunStore: "Release run record permissions could not be secured." ) _fsync_directory(self.root) + except ReleaseRunError: + raise + except OSError as exc: + raise ReleaseRunCorrupt( + "Release run record could not be persisted safely." + ) from exc finally: if descriptor >= 0: - os.close(descriptor) + try: + os.close(descriptor) + except OSError as exc: + raise ReleaseRunCorrupt( + "Release run record could not be persisted safely." + ) from exc if temporary_path is not None: try: os.unlink(temporary_path) - except FileNotFoundError: + except OSError: pass def _locked(self): # type: ignore[no-untyped-def] self._ensure_root() + self._validate_lock_path() return _CombinedLock(self._thread_lock, FileLock(str(self._lock_path))) + def _validate_lock_path(self) -> None: + try: + metadata = self._lock_path.lstat() + except FileNotFoundError: + return + except OSError as exc: + raise ReleaseRunCorrupt( + "Release run lock could not be inspected safely." + ) from exc + if ( + stat.S_ISLNK(metadata.st_mode) + or not stat.S_ISREG(metadata.st_mode) + or metadata.st_nlink != 1 + or (hasattr(os, "geteuid") and metadata.st_uid != os.geteuid()) + ): + raise ReleaseRunCorrupt("Release run lock path is unsafe.") + class _CombinedLock: def __init__(self, thread_lock: threading.RLock, file_lock: FileLock) -> None: @@ -605,13 +722,23 @@ class _CombinedLock: raise ReleaseRunConflict( "Release run storage is busy; retry the operation." ) from exc + except OSError as exc: + self.thread_lock.release() + raise ReleaseRunCorrupt( + "Release run lock could not be acquired safely." + ) from exc except BaseException: self.thread_lock.release() raise def __exit__(self, *exc_info: object) -> None: try: - self.file_lock.release() + try: + self.file_lock.release() + except OSError as exc: + raise ReleaseRunCorrupt( + "Release run lock could not be released safely." + ) from exc finally: self.thread_lock.release() @@ -755,6 +882,7 @@ def _validate_record(value: Any, *, expected_run_id: str | None = None) -> None: "schema", "schema_version", "run_id", + "create_request_fingerprint", "created_at", "updated_at", "immutable", @@ -773,6 +901,10 @@ def _validate_record(value: Any, *, expected_run_id: str | None = None) -> None: raise ValueError if expected_run_id is not None and run_id != expected_run_id: raise ValueError + if not _is_sha256(value.get("create_request_fingerprint")): + raise ValueError + if run_id != _creation_run_id(value["create_request_fingerprint"]): + raise ValueError for key in ("created_at", "updated_at"): if not _is_timestamp(value.get(key)): raise ValueError @@ -908,8 +1040,15 @@ def _validate_record(value: Any, *, expected_run_id: str | None = None) -> None: if ( not isinstance(request, dict) or set(request) - != {"fingerprint", "operation", "step_id", "parameter"} + != { + "fingerprint", + "operation", + "step_id", + "parameter", + "attempt_fingerprint", + } or not _is_sha256(request.get("fingerprint")) + or not _is_sha256(request.get("attempt_fingerprint")) or request.get("operation") not in {"resume", "retry", "reconcile"} ): @@ -961,6 +1100,29 @@ def _validate_record(value: Any, *, expected_run_id: str | None = None) -> None: attempts_by_fingerprint = { attempt["fingerprint"]: attempt for attempt in attempts } + for request in requests: + request_attempt = attempts_by_fingerprint.get( + request["attempt_fingerprint"] + ) + if request_attempt is None: + raise ValueError + if ( + request["step_id"] is not None + and request_attempt["step_id"] != request["step_id"] + ): + raise ValueError + if request["operation"] == "resume" and request_attempt[ + "outcome" + ] != "interrupted": + raise ValueError + if request["operation"] == "retry" and request_attempt[ + "outcome" + ] not in {"failed", "interrupted"}: + raise ValueError + if request["operation"] == "reconcile" and request_attempt[ + "outcome" + ] != "interrupted": + raise ValueError attempts_per_step = { step_id: sum( attempt["step_id"] == step_id for attempt in attempts @@ -1193,13 +1355,28 @@ def _append_event( record["state"]["events"] = record["state"]["events"][-MAX_EVENTS:] -def _ensure_command_capacity(record: dict[str, Any]) -> None: - if len(record["state"]["processed_requests"]) >= MAX_COMMANDS: +def _ensure_command_capacity( + record: dict[str, Any], *, reserve_after: int = 0 +) -> None: + if ( + len(record["state"]["processed_requests"]) + 1 + reserve_after + > MAX_COMMANDS + ): raise ReleaseRunConflict( "Release run command history reached its safety limit; create a new run." ) +def _ensure_start_recovery_capacity(record: dict[str, Any]) -> None: + if ( + len(record["state"]["processed_requests"]) + START_RECOVERY_RESERVE + > MAX_COMMANDS + ): + raise ReleaseRunConflict( + "Release run has insufficient reserved recovery capacity for a new attempt; create a new run." + ) + + def _ensure_attempt_capacity(record: dict[str, Any]) -> None: if len(record["state"]["processed_attempts"]) >= MAX_COMMANDS: raise ReleaseRunConflict( @@ -1213,6 +1390,8 @@ def _remember_request( operation: str, step_id: str | None, parameter: str | None, + *, + attempt_fingerprint: str | None, ) -> None: _ensure_command_capacity(record) record["state"]["processed_requests"].append( @@ -1221,6 +1400,7 @@ def _remember_request( "operation": operation, "step_id": step_id, "parameter": parameter, + "attempt_fingerprint": attempt_fingerprint, } ) @@ -1324,6 +1504,28 @@ def _record_digest(record: dict[str, Any]) -> str: ).hexdigest() +def _creation_run_id(request_fingerprint: str) -> str: + if not _is_sha256(request_fingerprint): + raise ReleaseRunConflict("Run creation request identifier is invalid.") + return f"rr-request-{request_fingerprint}" + + +def _validate_create_replay( + record: dict[str, Any], + *, + request_fingerprint: str, + input_snapshot: dict[str, Any], +) -> None: + if record["create_request_fingerprint"] != request_fingerprint: + raise ReleaseRunConflict( + "The run creation identifier is already mapped to another record." + ) + if record["immutable"]["input"] != input_snapshot: + raise ReleaseRunConflict( + "The run creation identifier was already used with different inputs." + ) + + def _request_fingerprint(request_id: str) -> str: if not isinstance(request_id, str) or not REQUEST_ID_PATTERN.fullmatch(request_id): raise ReleaseRunConflict("Run-state request identifier is invalid.") diff --git a/tools/release/server/app.py b/tools/release/server/app.py index 4f70da1..cfde3c5 100644 --- a/tools/release/server/app.py +++ b/tools/release/server/app.py @@ -100,6 +100,7 @@ class RepositoryTagRequest(BaseModel): class ReleaseRunCreateRequest(BaseModel): + request_id: str = Field(min_length=8, max_length=128) repo_versions: dict[str, str] = Field(default_factory=dict) channel: str = Field(default="stable", min_length=1, max_length=64) online: bool = False @@ -128,10 +129,15 @@ def create_app( app.state.workspace_root = workspace_root app.state.token = token app.state.workspace_fingerprint = release_workspace_fingerprint(workspace_root) - app.state.release_runs = ReleaseRunStore( - run_state_root + release_run_root = ( + workspace_release_run_root( + run_state_root, app.state.workspace_fingerprint + ) if run_state_root is not None - else default_release_run_root(workspace_root), + else default_release_run_root(workspace_root) + ) + app.state.release_runs = ReleaseRunStore( + release_run_root, expected_workspace_fingerprint=app.state.workspace_fingerprint, ) @@ -255,6 +261,24 @@ def create_app( raise HTTPException( status_code=400, detail="At least one repository version is required" ) + input_snapshot = { + "channel": request.channel, + "repo_versions": dict(request.repo_versions), + "online": request.online, + "remote_tags": request.remote_tags, + "public_catalog": request.public_catalog, + "include_migrations": request.include_migrations, + "workspace_fingerprint": app.state.workspace_fingerprint, + } + try: + existing = app.state.release_runs.find_created_run( + request_id=request.request_id, + input_snapshot=input_snapshot, + ) + except (ReleaseRunConflict, ReleaseRunCorrupt) as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + if existing is not None: + return existing snapshot = build_dashboard( workspace_root=app.state.workspace_root, online=request.online, @@ -294,15 +318,8 @@ def create_app( ) try: return app.state.release_runs.create( - input_snapshot={ - "channel": request.channel, - "repo_versions": dict(request.repo_versions), - "online": request.online, - "remote_tags": request.remote_tags, - "public_catalog": request.public_catalog, - "include_migrations": request.include_migrations, - "workspace_fingerprint": app.state.workspace_fingerprint, - }, + request_id=request.request_id, + input_snapshot=input_snapshot, plan_snapshot=plan_payload, ) except (ReleaseRunConflict, ReleaseRunCorrupt) as exc: @@ -492,6 +509,14 @@ def default_release_run_root(workspace_root: Path) -> Path: ) +def workspace_release_run_root(state_root: Path, workspace_fingerprint: str) -> Path: + return ( + state_root.expanduser() + / f"workspace-{workspace_fingerprint}" + / "release-runs" + ) + + def release_workspace_fingerprint(workspace_root: Path) -> str: canonical = str(workspace_root.expanduser().resolve()).encode("utf-8") return hashlib.sha256(canonical).hexdigest() diff --git a/tools/release/webui/index.html b/tools/release/webui/index.html index c95e74e..f27b54a 100644 --- a/tools/release/webui/index.html +++ b/tools/release/webui/index.html @@ -909,6 +909,8 @@ runs: [], runStoreAvailable: true, }; + const pendingReleaseRunKey = "govoplanReleaseConsolePendingRunCreate"; + const pendingRunCommandsKey = "govoplanReleaseConsolePendingRunCommands"; elements.refresh.addEventListener("click", load); elements.generateCandidate.addEventListener("click", generateCandidate); @@ -962,7 +964,9 @@ }).then((response) => { if (!response.ok) { return response.json().catch(() => ({})).then((payload) => { - throw new Error(payload.detail || `${response.status} ${response.statusText}`); + const error = new Error(payload.detail || `${response.status} ${response.statusText}`); + error.status = response.status; + throw error; }); } return response.json(); @@ -980,7 +984,9 @@ }).then((response) => { if (!response.ok) { return response.json().catch(() => ({})).then((payload) => { - throw new Error(payload.detail || `${response.status} ${response.statusText}`); + const error = new Error(payload.detail || `${response.status} ${response.statusText}`); + error.status = response.status; + throw error; }); } return response.json(); @@ -1000,9 +1006,10 @@ renderReleaseRunStoreUnavailable(runsResult.error); } else { renderReleaseRunList(runsResult.data?.runs || []); - if (releaseState.currentRun?.run_id) { + const pendingRecoveryResult = await recoverPendingReleaseRun(); + if (pendingRecoveryResult === null && releaseState.currentRun?.run_id) { await loadReleaseRun(releaseState.currentRun.run_id, { restoreSelection: false }); - } else { + } else if (pendingRecoveryResult === null) { renderIdlePlan(); } } @@ -1070,7 +1077,7 @@ setBusy([elements.createReleaseRun], true); elements.runStatus.textContent = "freezing plan..."; try { - const run = await postJson("/api/release-runs", { + const payload = pendingReleaseRunPayload({ repo_versions: repoVersions, channel: elements.channel.value.trim() || "stable", online: false, @@ -1078,12 +1085,7 @@ public_catalog: elements.publicCatalog.checked, include_migrations: elements.migrations.checked, }); - releaseState.currentRun = run; - restoreReleaseRunSelection(run); - renderReleaseRun(run); - const runs = await api("/api/release-runs"); - renderReleaseRunList(runs.runs || []); - elements.releaseRun.value = run.run_id; + await submitReleaseRunCreate(payload, { automatic: false }); } catch (error) { elements.runStatus.textContent = "error"; elements.runOutput.innerHTML = `

${pill("error", "block")} Run creation failed

${escapeHtml(error.message)}

`; @@ -1092,6 +1094,74 @@ } } + function pendingReleaseRunPayload(input) { + const normalized = { + ...input, + repo_versions: Object.fromEntries(Object.entries(input.repo_versions || {}).sort(([left], [right]) => left.localeCompare(right))), + }; + const signature = JSON.stringify(normalized); + const pending = readPendingReleaseRun(); + if (pending?.signature === signature) return pending.payload; + const payload = { ...normalized, request_id: requestId("create-run") }; + sessionStorage.setItem(pendingReleaseRunKey, JSON.stringify({ signature, payload })); + return payload; + } + + function readPendingReleaseRun() { + try { + const pending = JSON.parse(sessionStorage.getItem(pendingReleaseRunKey) || "null"); + if (!pending || typeof pending.signature !== "string" || typeof pending.payload?.request_id !== "string") return null; + return pending; + } catch (_error) { + sessionStorage.removeItem(pendingReleaseRunKey); + return null; + } + } + + function clearPendingReleaseRun(requestIdToClear) { + const pending = readPendingReleaseRun(); + if (pending?.payload?.request_id === requestIdToClear) { + sessionStorage.removeItem(pendingReleaseRunKey); + } + } + + async function recoverPendingReleaseRun() { + const pending = readPendingReleaseRun(); + if (!pending) return null; + if (!releaseState.runStoreAvailable) return false; + elements.runStatus.textContent = "recovering saved create..."; + return submitReleaseRunCreate(pending.payload, { automatic: true }); + } + + async function submitReleaseRunCreate(payload, options = {}) { + try { + const run = await postJson("/api/release-runs", payload); + clearPendingReleaseRun(payload.request_id); + releaseState.currentRun = run; + restoreReleaseRunSelection(run); + renderReleaseRun(run); + await refreshReleaseRunListAfterCreate(run); + return true; + } catch (error) { + const deterministicClientError = error.status >= 400 && error.status < 500 && error.status !== 409; + const conflictingReuse = error.status === 409 && error.message.includes("already used with different inputs"); + if (deterministicClientError || conflictingReuse) clearPendingReleaseRun(payload.request_id); + elements.runStatus.textContent = options.automatic ? "recovery pending" : "error"; + elements.runOutput.innerHTML = `

${pill(options.automatic ? "pending" : "error", options.automatic ? "warn" : "block")} ${options.automatic ? "Saved run recovery is still pending" : "Run creation failed"}

${escapeHtml(error.message)}

The same private request identifier will be reused after an uncertain response.

`; + return false; + } + } + + async function refreshReleaseRunListAfterCreate(run) { + try { + const runs = await api("/api/release-runs"); + renderReleaseRunList(runs.runs || []); + elements.releaseRun.value = run.run_id; + } catch (error) { + elements.runOutput.insertAdjacentHTML("afterbegin", `

${pill("saved", "warn")} Run saved; list refresh unavailable

${escapeHtml(error.message)}

The persisted run remains selected and can be reloaded after refresh.

`); + } + } + async function loadReleaseRun(runId, options = {}) { if (!runId) { releaseState.currentRun = null; @@ -1183,9 +1253,12 @@ async function resumeReleaseRun() { const run = releaseState.currentRun; if (!run) return; + const commandKey = `resume:${run.run_id}`; + const commandRequestId = inFlightRunCommandId(commandKey, "resume"); setBusy([elements.resumeReleaseRun], true); try { - const resumed = await postJson(`/api/release-runs/${encodeURIComponent(run.run_id)}/resume`, { request_id: requestId("resume") }); + const resumed = await postJson(`/api/release-runs/${encodeURIComponent(run.run_id)}/resume`, { request_id: commandRequestId }); + clearInFlightRunCommand(commandKey, commandRequestId); releaseState.currentRun = resumed; renderReleaseRun(resumed); } catch (error) { @@ -1199,8 +1272,11 @@ async function retryReleaseRunStep(stepId) { const run = releaseState.currentRun; if (!run || !stepId) return; + const commandKey = `retry:${run.run_id}:${stepId}`; + const commandRequestId = inFlightRunCommandId(commandKey, "retry"); try { - const retried = await postJson(`/api/release-runs/${encodeURIComponent(run.run_id)}/steps/${encodeURIComponent(stepId)}/retry`, { request_id: requestId("retry") }); + const retried = await postJson(`/api/release-runs/${encodeURIComponent(run.run_id)}/steps/${encodeURIComponent(stepId)}/retry`, { request_id: commandRequestId }); + clearInFlightRunCommand(commandKey, commandRequestId); releaseState.currentRun = retried; renderReleaseRun(retried); } catch (error) { @@ -1211,13 +1287,16 @@ async function reconcileReleaseRunStep(stepId, outcome, confirmation, button) { const run = releaseState.currentRun; if (!run || !stepId || !outcome || confirmation !== "RECONCILE") return; + const commandKey = `reconcile:${run.run_id}:${stepId}:${outcome}`; + const commandRequestId = inFlightRunCommandId(commandKey, "reconcile"); setBusy([button], true); try { const reconciled = await postJson(`/api/release-runs/${encodeURIComponent(run.run_id)}/steps/${encodeURIComponent(stepId)}/reconcile`, { - request_id: requestId("reconcile"), + request_id: commandRequestId, outcome, confirm: confirmation, }); + clearInFlightRunCommand(commandKey, commandRequestId); releaseState.currentRun = reconciled; renderReleaseRun(reconciled); } catch (error) { @@ -1226,6 +1305,33 @@ } } + function readInFlightRunCommands() { + try { + const commands = JSON.parse(sessionStorage.getItem(pendingRunCommandsKey) || "{}"); + return commands && typeof commands === "object" && !Array.isArray(commands) ? commands : {}; + } catch (_error) { + sessionStorage.removeItem(pendingRunCommandsKey); + return {}; + } + } + + function inFlightRunCommandId(commandKey, prefix) { + const commands = readInFlightRunCommands(); + if (typeof commands[commandKey] === "string") return commands[commandKey]; + const identifier = requestId(prefix); + commands[commandKey] = identifier; + sessionStorage.setItem(pendingRunCommandsKey, JSON.stringify(commands)); + return identifier; + } + + function clearInFlightRunCommand(commandKey, requestIdToClear) { + const commands = readInFlightRunCommands(); + if (commands[commandKey] !== requestIdToClear) return; + delete commands[commandKey]; + if (Object.keys(commands).length) sessionStorage.setItem(pendingRunCommandsKey, JSON.stringify(commands)); + else sessionStorage.removeItem(pendingRunCommandsKey); + } + function requestId(prefix) { const identifier = window.crypto?.randomUUID ? window.crypto.randomUUID() : `${Date.now()}-${Math.random().toString(16).slice(2)}`; return `${prefix}-${identifier}`;