From d7becb7c416e17d1f5d890136da7c504c3993850 Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 15 Oct 2026 15:53:34 +0000 Subject: [PATCH] feat(capture): replay a capture after a client cold restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A device that dies mid-capture must recover from what it persisted and what the server remembers — not from its own cached state. PR #262's replay resumed a lost response inside one Python process: the client that retried was the interpreter that lost the reply. scripts/capture_cold_restart.py splits the client across two fresh Python processes against a real create_app() server (memory state, local volume storage, orchestrator stage stubs — no network, no credentials): - the primary client replays a saved capture_flow_e2e receipt and dies at a scripted interruption: a lost create/commit/finalize response or a stop before commit/finalize, persisting only a client journal of what it sent and which replies it saw. - the resume client reads the receipt and journal and recovers through the real routes: create again with the saved idempotency key, then register-upload-commit-finalize — or, when the dead process already finalized, straight through the finalize's idempotent replay, because the capture is closed to files and commits (the server 409s both). Across the restart it asserts one capture id (a 200 replay, never a second 201, still mapping to the same capture after the verdict), one deterministic run id (run_ + sha256 of captureId, NUL and the packet bytes, unchanged on the second finalize), committed hashes matching the receipt with no mismatch, and one terminal state (one terminal status, /result served once for that run, exactly one verdict_ready event). Every wait is bounded: server readiness, per-leg subprocess timeouts, per-request timeouts, and capped terminal/event polling. A receipt written before the replay fields existed fails before anything is sent, naming the missing field. tests/test_capture_cold_restart.py drives the CLI as a subprocess for all five interruption points, asserts the two legs ran as different fresh processes sharing only the receipt, journal and server, and checks pre-replay-field receipts fail loudly naming each missing field. Co-authored-by: Sam Gu --- backend/scripts/capture_cold_restart.py | 926 +++++++++++++++++++++ backend/tests/test_capture_cold_restart.py | 234 ++++++ 2 files changed, 1160 insertions(+) create mode 100644 backend/scripts/capture_cold_restart.py create mode 100644 backend/tests/test_capture_cold_restart.py diff --git a/backend/scripts/capture_cold_restart.py b/backend/scripts/capture_cold_restart.py new file mode 100644 index 000000000..f7df96286 --- /dev/null +++ b/backend/scripts/capture_cold_restart.py @@ -0,0 +1,926 @@ +#!/usr/bin/env python3 +"""Close the cold-restart gap PR #262 admitted: a client that dies and comes back. + +One repeatable command for reviewers: + + cd backend && uv run --extra api python scripts/capture_cold_restart.py \ + --receipt /receipt.json --stop commit-lost + +``scripts/replay_capture.py`` resumes a lost response inside one Python +process: the client that retries is the same interpreter that lost the reply, +so it still holds whatever it cached. A phone does not get that luxury — the +app dies, relaunches, and must recover from what it persisted and what the +server remembers. This command drives exactly that against the real app entry +point ``create_app()``, served by a local uvicorn process (memory state, local +volume storage, the documented dev handoff ``HSS_INGEST_HOOKS=auto`` with +orchestrator stage stubs — no GCP, no Modal, no network, no credentials). It +splits the client across two fresh Python processes: + +- the **primary client** is one Python process. It replays the saved journey + from a ``capture_flow_e2e`` receipt — create, register, PUT, commit, + finalize — and dies at the scripted interruption: right after a deliberately + lost response (``create-lost``, ``commit-lost``, ``finalize-lost``) or before + the next request (``pre-commit``, ``pre-finalize``). It persists only what a + device could: a journal of what it sent and which replies it actually saw. +- the **resume client** is a fresh Python process with fresh in-memory client + state. It reads the receipt and the journal and recovers through the real + capture routes: create again with the saved idempotency key, re-register to + learn the server's file states instead of trusting its own, re-send the + identical commit body, and replay the finalize for its stored run id. When + the dead process already finalized, the capture is closed to files and + commits, so recovery goes straight through the finalize replay; otherwise + the resume re-registers, uploads only what the server still lacks, commits, + and finalizes. It then reads status, result and events the way a device +does. + +Across the restart it asserts the four things a raw HTTP log cannot prove: + +- **one capture id** — the resume create answers 200 from the idempotency + store (never a second 201) with the capture id the dead process recorded, + and a final create after the verdict still maps to the same capture; +- **one deterministic run id** — the run id equals ``run_`` + sha256 of + captureId, NUL and the packet bytes, and a second finalize returns it + unchanged even though the first request came from a process that is gone; +- **the committed hashes** — the resumed commit reports the receipt's paths + and never a mismatch, on a first commit or a replayed one; +- **one terminal state** — the capture lands on one terminal status, served + once by ``GET /result`` with that run id and outcome, and the event feed + ends with exactly one terminal ``verdict_ready``. + +Nothing retries forever: the resume sends each step once more at most (the +second finalize is the determinism proof, not a retry), terminal and event +polling is capped at ``MAX_POLLS`` reads, each client leg is a subprocess with +a hard timeout, and the server readiness wait is bounded too. A receipt +written before the replay fields existed fails before anything is sent, +naming the missing field. + +It writes ``/cold-restart-report.json`` and exits 0 when every +assertion held, 1 otherwise. It never changes route contracts, never touches +the iOS tree, ``.github/`` or deployment, and leaves ``replay_capture.py``'s +owned files alone. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import socket +import subprocess +import sys +import tempfile +import time +import urllib.error +import urllib.request +from datetime import UTC, datetime +from pathlib import Path +from urllib.parse import urlsplit + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from scripts.replay_capture import ( # noqa: E402 + DEFAULT_PACKET, + TERMINAL_EVENTS, + TERMINAL_STATUSES, + check_fixture, + load_receipt, +) +from server.api.app import create_app # noqa: E402 +from server.common.config import Settings # noqa: E402 + +ROOT = Path(__file__).resolve().parents[1] + +# Same static test token the API test suite uses (server/api/tests/conftest.py). +TOKEN = "d" * 40 + +# Every wait in this command is bounded so a stuck server or leg fails loudly +# instead of hanging: readiness waits at most SERVER_TIMEOUT_S, each leg runs +# at most LEG_TIMEOUT_S, one HTTP request at most REQUEST_TIMEOUT_S, and the +# terminal status/events polls settle within MAX_POLLS * POLL_INTERVAL_S. +SERVER_TIMEOUT_S = 60.0 +LEG_TIMEOUT_S = 180.0 +REQUEST_TIMEOUT_S = 30.0 +MAX_POLLS = 60 +POLL_INTERVAL_S = 0.5 + +TERMINAL_EVENT = "verdict_ready" + +# Where the primary client dies. The lost-response points stop after a request +# the server processed but never answered to this client; the pre-* points stop +# before the next request, the way a crash between steps does. +STOP_POINTS = { + "create-lost": "the create reply was lost after the server processed it", + "pre-commit": "the client died before sending commit (all bytes were PUT)", + "commit-lost": "the commit reply was lost after the server processed it", + "pre-finalize": "the client died after commit was delivered, before finalize", + "finalize-lost": "the finalize reply was lost after the server processed it", +} + + +def _interruption(stop: str) -> dict: + lost_step = stop.removesuffix("-lost") + if lost_step != stop: + return {"kind": "lost_response", "step": lost_step} + return {"kind": "before_request", "step": stop.removesuffix("pre-")} + + +def missing_replay_fields(receipt: dict) -> list[str]: + """Name every replay field the saved receipt is missing, one by one. + + Receipts written before the replay fields existed fail here — loudly and + by name, never by replaying guessed hashes.""" + missing: list[str] = [] + keys = receipt.get("idempotencyKeys") + if not isinstance(keys, dict): + missing.append("idempotencyKeys") + else: + for name in ("create", "finalize"): + if not keys.get(name): + missing.append(f"idempotencyKeys.{name}") + files = receipt.get("files") + if not isinstance(files, list) or not files: + missing.append("files") + else: + for index, item in enumerate(files): + if not isinstance(item, dict) or not item.get("path") or not item.get("sha256"): + missing.append(f"files[{index}]") + break + if not receipt.get("packetSha256"): + missing.append("packetSha256") + return missing + + +def _resolve_packet_dir(receipt: dict, override: Path | None) -> Path: + if override is not None: + return override + saved = receipt.get("packetDir") + if saved and Path(saved).is_dir(): + return Path(saved) + return DEFAULT_PACKET + + +def _expected_run_id(capture_id: str, raw_packet: bytes) -> str: + return "run_" + hashlib.sha256(capture_id.encode() + b"\0" + raw_packet).hexdigest()[:16] + + +def _create_body(packet: dict) -> dict: + session = packet["session"] + return { + "packetId": packet["packetId"], + "formatVersion": packet["formatVersion"], + "tier": session["tier"], + "flow": "guided", + "device": { + "model": session["deviceModel"], + "systemVersion": session["systemVersion"], + "appVersion": session["appVersion"], + }, + } + + +class DeviceClient: + """A minimal capture-API client: one base URL, one bearer token, no memory. + + It keeps no state between requests on purpose — the cold restart must + prove recovery comes from the server and the persisted receipt, never from + a cached register response or a reused upload URL.""" + + def __init__(self, base_url: str, token: str = TOKEN) -> None: + self.base_url = base_url.rstrip("/") + + def request( + self, method: str, path: str, *, body: bytes | None = None, headers: dict | None = None + ) -> tuple[int, bytes]: + req = urllib.request.Request( + self.base_url + path, + data=body, + method=method, + headers={"Authorization": f"Bearer {TOKEN}", **(headers or {})}, + ) + try: + with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT_S) as response: + return response.status, response.read() + except urllib.error.HTTPError as error: # a non-2xx is still a server answer + return error.code, error.read() + + def post_json(self, path: str, payload: dict, headers: dict | None = None) -> tuple[int, dict]: + status, raw = self.request( + "POST", + path, + body=json.dumps(payload).encode(), + headers={"Content-Type": "application/json", **(headers or {})}, + ) + return status, (json.loads(raw) if raw else {}) + + def post_raw(self, path: str, payload: bytes, headers: dict | None = None) -> tuple[int, dict]: + status, raw = self.request("POST", path, body=payload, headers=headers) + return status, (json.loads(raw) if raw else {}) + + def get_json(self, path: str) -> tuple[int, dict]: + status, raw = self.request("GET", path) + return status, (json.loads(raw) if raw else {}) + + def put_signed(self, url: str, payload: bytes, headers: dict | None = None) -> tuple[int, bytes]: + # The signed URL names this server's host; the token it guards is in + # the path, so keep the path and put to the same local server. + return self.request("PUT", urlsplit(url).path, body=payload, headers=headers) + + +def _journal_path(workdir: Path) -> Path: + return workdir / "client-journal.json" + + +def _fresh_journal(stop: str) -> dict: + return { + "leg": "primary", + "stop": stop, + "pid": os.getpid(), + "captureId": None, + "createExecuted": False, + "createResponseSeen": False, + "registered": False, + "uploadedPaths": [], + "commitExecuted": False, + "commitResponseSeen": False, + "finalizeExecuted": False, + "finalizeResponseSeen": False, + } + + +def _write_json(path: Path, payload: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2) + "\n") + + +def _finish(report: dict, errors: list[str], detail: str | None) -> dict: + if detail: + report["errorDetail"] = detail + report["finishedAt"] = datetime.now(UTC).isoformat() + report["ok"] = not errors + return report + + +def run_primary_leg( + receipt_path: Path, workdir: Path, server_url: str, stop: str, *, packet_dir: Path | None = None +) -> dict: + """Walk the saved journey and die at the scripted interruption. + + Dying on time is success: the leg report is ``ok`` when the walk was clean + up to the stop point. The journal is the only state the next process gets. + """ + started_at = datetime.now(UTC).isoformat() + report: dict = { + "leg": "primary", + "stop": stop, + "interruption": _interruption(stop), + "pid": os.getpid(), + "freshProcess": True, + "serverUrl": server_url, + "startedAt": started_at, + "steps": [], + "captureId": None, + "errorCodes": [], + "ok": False, + } + errors = report["errorCodes"] + journal = _fresh_journal(stop) + + def record(step: str, **fields) -> None: + report["steps"].append({"step": step, **fields}) + + try: + client = DeviceClient(server_url) + receipt = load_receipt(receipt_path) + packet_dir = _resolve_packet_dir(receipt, packet_dir) + fixture = check_fixture(packet_dir, receipt, errors) + record("fixture", packetDir=str(packet_dir), files=len((fixture or {}).get("packet", {}).get("files", []))) + if fixture is None or errors: + raise RuntimeError("the saved receipt no longer matches the packet fixture") + packet = fixture["packet"] + packet_sha = fixture["packetSha256"] + keys = receipt["idempotencyKeys"] + saved_files = receipt["files"] + + # Create. With the response lost the client exits without reading the + # body: the capture id exists only on the server now. + status, body = client.post_json( + "/v1/captures", _create_body(packet), headers={"Idempotency-Key": keys["create"]} + ) + journal["createExecuted"] = True + if stop == "create-lost": + record("create", responseLost=True) + return _finish(report, errors, None) + journal["createResponseSeen"] = True + capture_id = body.get("captureId") if status in (200, 201) else None + report["captureId"] = capture_id + journal["captureId"] = capture_id + record("create", httpStatus=status, captureId=capture_id) + if status != 201: + errors.append(f"create_http_{status}") + if not capture_id: + raise RuntimeError(f"no captureId in create response: {body}") + + # Register, PUT the receipt's bytes, hash-checking each against it. + registered = client.post_json(f"/v1/captures/{capture_id}/files", {"files": saved_files})[1] + targets = registered.get("files", []) + record("register", files=len(targets)) + if registered.get("files") is None: + raise RuntimeError(f"register failed: {registered}") + journal["registered"] = True + for saved, target in zip(saved_files, targets, strict=True): + data = (packet_dir / saved["path"]).read_bytes() + if hashlib.sha256(data).hexdigest() != saved["sha256"]: + errors.append(f"upload_hash_drift:{saved['path']}") + put_status, _ = client.put_signed(target["upload"]["url"], data, target["upload"].get("headers") or {}) + if put_status == 200: + journal["uploadedPaths"].append(target["path"]) + else: + errors.append(f"upload_http_{put_status}:{target['path']}") + record("upload", uploaded=len(journal["uploadedPaths"]), expected=len(saved_files)) + + # Commit, unless the client dies before sending it. + if stop == "pre-commit": + record("stop", before="commit") + return _finish(report, errors, None) + commit_status, commit_body = client.post_json( + f"/v1/captures/{capture_id}/files:commit", + {"files": [{"path": item["path"], "sha256": item["sha256"]} for item in saved_files]}, + ) + journal["commitExecuted"] = True + if stop == "commit-lost": + record("commit", responseLost=True) + return _finish(report, errors, None) + journal["commitResponseSeen"] = True + record( + "commit", + httpStatus=commit_status, + committed=commit_body.get("committed", []), + notFound=commit_body.get("notFound", []), + mismatch=commit_body.get("mismatch", []), + ) + if commit_status != 200 or commit_body.get("notFound") or commit_body.get("mismatch"): + errors.append(f"commit_failed_http_{commit_status}") + + # Finalize, unless the client dies before sending it. + if stop == "pre-finalize": + record("stop", before="finalize") + return _finish(report, errors, None) + raw = (packet_dir / "packet.json").read_bytes() + finalize_status, finalize_body = client.post_raw( + f"/v1/captures/{capture_id}/finalize", + raw, + headers={ + "Idempotency-Key": keys["finalize"], + "Content-Type": "application/json", + "X-Packet-Sha256": packet_sha, + }, + ) + journal["finalizeExecuted"] = True + if stop == "finalize-lost": + record("finalize", responseLost=True) + return _finish(report, errors, None) + journal["finalizeResponseSeen"] = True + record("finalize", httpStatus=finalize_status, runId=finalize_body.get("runId")) + if finalize_status != 202: + errors.append(f"finalize_http_{finalize_status}") + except Exception as exc: # noqa: BLE001 - the report and journal must survive any failure + report["errorDetail"] = f"exception: {type(exc).__name__}: {exc}" + errors.append(f"exception:{type(exc).__name__}") + finally: + # The dying client's last act: persist what it knows, nothing more. + report["journal"] = journal + _write_json(_journal_path(workdir), journal) + return _finish(report, errors, report.get("errorDetail")) + + +def run_resume_leg(receipt_path: Path, workdir: Path, server_url: str, *, packet_dir: Path | None = None) -> dict: + """Recover the journey from a fresh process and assert the invariants. + + The client shares nothing with the dead one: no cached responses, no + upload URLs, no committed set. Everything it needs it re-reads from the + receipt and the journal, and re-asks the server through the real routes. + """ + started_at = datetime.now(UTC).isoformat() + report: dict = { + "leg": "resume", + "pid": os.getpid(), + "freshProcess": True, + "serverUrl": server_url, + "startedAt": started_at, + "steps": [], + "captureId": None, + "createReplayHttpStatus": None, + "recreatedSameCaptureId": None, + "serverFileStates": {}, + "recoveryPath": None, + "recommit": None, + "commitCommitted": [], + "commitMismatch": [], + "filesCommitted": None, + "runId": None, + "expectedRunId": None, + "finalizeReplayRunId": None, + "finalCaptureStatus": None, + "statusPolls": None, + "resultOutcomeKind": None, + "resultRunId": None, + "terminalCreateCaptureId": None, + "terminalEvents": [], + "errorCodes": [], + "ok": False, + } + errors = report["errorCodes"] + + def record(step: str, **fields) -> None: + report["steps"].append({"step": step, **fields}) + + try: + journal_path = _journal_path(workdir) + if not journal_path.is_file(): + raise RuntimeError(f"no client journal at {journal_path}: the primary leg never ran") + journal = json.loads(journal_path.read_text()) + client = DeviceClient(server_url) + receipt = load_receipt(receipt_path) + packet_dir = _resolve_packet_dir(receipt, packet_dir) + fixture = check_fixture(packet_dir, receipt, errors) + record("fixture", packetDir=str(packet_dir), files=len((fixture or {}).get("packet", {}).get("files", []))) + if fixture is None or errors: + raise RuntimeError("the saved receipt no longer matches the packet fixture") + packet = fixture["packet"] + packet_sha = fixture["packetSha256"] + keys = receipt["idempotencyKeys"] + saved_files = receipt["files"] + + # Create again with the saved key. The dead process's create really + # executed, so this must answer 200 from the idempotency store with + # the same capture id — a 201 here minted a second capture. + status, body = client.post_json( + "/v1/captures", _create_body(packet), headers={"Idempotency-Key": keys["create"]} + ) + capture_id = body.get("captureId") if status == 200 else None + report["captureId"] = capture_id + report["createReplayHttpStatus"] = status + report["recreatedSameCaptureId"] = journal.get("captureId") in (None, capture_id) + record( + "create", + httpStatus=status, + captureId=capture_id, + journalCaptureId=journal.get("captureId"), + ) + if status != 200: + errors.append(f"resume_create_not_stored:{status}") + if not report["recreatedSameCaptureId"]: + errors.append("resume_capture_id_changed") + if not capture_id: + raise RuntimeError(f"create replay did not yield a capture id: {body}") + + # Register to learn the server's file states — the resume never + # trusts the journal's idea of what was uploaded; the server says. + # But a finalize the dead process already sent changes the recovery: + # the capture is closed to files and commits (the server answers 409 + # to both), so the fresh client recovers through the finalize's + # idempotent replay — the stored run id — and never re-registers. + if journal.get("finalizeExecuted"): + report["recoveryPath"] = "finalize_replay" + report["recommit"] = False # the resume sends no commit on this path + record("register", skipped="the capture is closed to files after finalize") + else: + report["recoveryPath"] = "register_commit" + register_status, registered = client.post_json(f"/v1/captures/{capture_id}/files", {"files": saved_files}) + record("register", httpStatus=register_status, files=len(saved_files)) + if register_status != 200: + raise RuntimeError(f"register failed: {registered}") + states = {item["path"]: item.get("state") for item in registered.get("files", [])} + report["serverFileStates"] = states + + # PUT only what the server still lacks; the bytes hash against + # the receipt before anything is sent. + uploaded: list[str] = [] + for saved, target in zip(saved_files, registered.get("files", []), strict=True): + if target.get("state") == "committed": + continue # already on the server: a resume needs no second PUT + data = (packet_dir / saved["path"]).read_bytes() + if hashlib.sha256(data).hexdigest() != saved["sha256"]: + errors.append(f"upload_hash_drift:{saved['path']}") + put_status, _ = client.put_signed(target["upload"]["url"], data, target["upload"].get("headers") or {}) + if put_status == 200: + uploaded.append(target["path"]) + else: + errors.append(f"upload_http_{put_status}:{target['path']}") + record( + "upload", + uploaded=len(uploaded), + alreadyCommitted=sorted(path for path, state in states.items() if state == "committed"), + ) + + # The identical commit body — first send or replay, the server + # answers the receipt's hashes with the receipt's paths and no + # mismatch. + commit_status, commit_body = client.post_json( + f"/v1/captures/{capture_id}/files:commit", + {"files": [{"path": item["path"], "sha256": item["sha256"]} for item in saved_files]}, + ) + report["recommit"] = bool(journal.get("commitExecuted")) + report["commitCommitted"] = commit_body.get("committed", []) if commit_status == 200 else [] + report["commitMismatch"] = commit_body.get("mismatch", []) + record( + "commit", + httpStatus=commit_status, + replayed=report["recommit"], + committed=report["commitCommitted"], + notFound=commit_body.get("notFound", []), + mismatch=report["commitMismatch"], + ) + if commit_status != 200 or commit_body.get("notFound") or report["commitMismatch"]: + errors.append(f"commit_failed_http_{commit_status}") + if sorted(report["commitCommitted"]) != sorted(item["path"] for item in saved_files): + errors.append("commit_incomplete") + + # Finalize twice. The first response may be the stored replay of the + # dead process's request; the second must return the same deterministic + # run id — the run is derived from the capture and the packet bytes. + raw = (packet_dir / "packet.json").read_bytes() + expected_run_id = _expected_run_id(capture_id, raw) + report["expectedRunId"] = expected_run_id + finalize_status, finalize_body = client.post_raw( + f"/v1/captures/{capture_id}/finalize", + raw, + headers={ + "Idempotency-Key": keys["finalize"], + "Content-Type": "application/json", + "X-Packet-Sha256": packet_sha, + }, + ) + run_id = finalize_body.get("runId") if finalize_status == 202 else None + report["runId"] = run_id + record("finalize", httpStatus=finalize_status, replayed=bool(journal.get("finalizeExecuted")), runId=run_id) + if finalize_status != 202: + errors.append(f"finalize_http_{finalize_status}") + raise RuntimeError(f"finalize failed: {finalize_body}") + if finalize_body.get("missing"): + errors.append("finalize_missing_files") + replay_status, replay_body = client.post_raw( + f"/v1/captures/{capture_id}/finalize", + raw, + headers={ + "Idempotency-Key": keys["finalize"], + "Content-Type": "application/json", + "X-Packet-Sha256": packet_sha, + }, + ) + report["finalizeReplayRunId"] = replay_body.get("runId") if replay_status == 202 else None + record("finalize-replay", httpStatus=replay_status, runId=report["finalizeReplayRunId"]) + if replay_status != 202 or report["finalizeReplayRunId"] != run_id: + errors.append("finalize_replay_run_id_changed") + if run_id != expected_run_id: + errors.append("run_id_not_deterministic") + + # Terminal status, bounded: the stub pipeline runs in the server, so + # the client waits with a capped poll, never forever. + status_payload: dict = {} + polls = 0 + while polls < MAX_POLLS: + polls += 1 + poll_status, status_payload = client.get_json(f"/v1/captures/{capture_id}") + if poll_status == 200 and status_payload.get("status") in TERMINAL_STATUSES: + break + time.sleep(POLL_INTERVAL_S) + terminal = status_payload.get("status") + report["finalCaptureStatus"] = terminal + report["statusPolls"] = polls + record("status", finalCaptureStatus=terminal, polls=polls, filesCommitted=status_payload.get("filesCommitted")) + report["filesCommitted"] = status_payload.get("filesCommitted") + if terminal not in TERMINAL_STATUSES: + errors.append(f"not_terminal:{terminal}") + + # The result is served once, for the one run, agreeing with the status. + result_payload: dict = {} + for _ in range(MAX_POLLS): + got_status, result_payload = client.get_json(f"/v1/captures/{capture_id}/result") + if got_status == 200 and result_payload.get("outcome"): + break + time.sleep(POLL_INTERVAL_S) + report["resultOutcomeKind"] = (result_payload or {}).get("outcome", {}).get("kind") + report["resultRunId"] = (result_payload or {}).get("runId") + record("result", outcomeKind=report["resultOutcomeKind"], runId=report["resultRunId"]) + if not result_payload.get("outcome"): + errors.append("result_not_served") + else: + if report["resultRunId"] != run_id: + errors.append("result_run_id_mismatch") + if report["resultOutcomeKind"] != terminal: + errors.append("result_outcome_status_mismatch") + + # A final create with the saved key still maps to the one capture. + final_status, final_body = client.post_json( + "/v1/captures", _create_body(packet), headers={"Idempotency-Key": keys["create"]} + ) + report["terminalCreateCaptureId"] = final_body.get("captureId") + record("create-terminal", httpStatus=final_status, captureId=report["terminalCreateCaptureId"]) + if final_status != 200 or report["terminalCreateCaptureId"] != capture_id: + errors.append("terminal_create_duplicated_capture") + + # The event feed ends with exactly one terminal verdict for the one + # run, read with the same bounded poll as the status. + terminals: list[str] = [] + polls = 0 + while polls < MAX_POLLS: + polls += 1 + events: list[dict] = [] + after = 0 + while True: + _, page = client.get_json(f"/v1/captures/{capture_id}/events?after={after}&wait=0") + events.extend(page.get("events", [])) + nxt = page.get("next") + if not page.get("events") or nxt in (None, after): + break + after = nxt + terminals = [event["type"] for event in events if event["type"] in TERMINAL_EVENTS] + if terminals == [TERMINAL_EVENT]: + break + time.sleep(POLL_INTERVAL_S) + report["terminalEvents"] = terminals + record("events", polls=polls, terminalEvents=terminals) + if terminals != [TERMINAL_EVENT]: + errors.append(f"terminal_events:{terminals}") + except Exception as exc: # noqa: BLE001 - the report must survive any failure + report["errorDetail"] = f"exception: {type(exc).__name__}: {exc}" + errors.append(f"exception:{type(exc).__name__}") + return _finish(report, errors, report.get("errorDetail")) + + +def _invariants(receipt: dict, primary: dict, resume: dict) -> dict: + """The four cross-restart claims, recomputed from both legs' evidence.""" + paths = sorted(item["path"] for item in receipt["files"]) + committed_ok = ( + sorted(resume.get("commitCommitted") or []) == paths and not (resume.get("commitMismatch") or []) + ) or ( + # On the finalize-replay path the commit was the dead process's; the + # server's status count is the evidence every file is committed. + resume.get("recoveryPath") == "finalize_replay" and resume.get("filesCommitted") == len(paths) + ) + return { + "oneCaptureId": bool(resume.get("captureId")) + and resume.get("createReplayHttpStatus") == 200 + and resume.get("recreatedSameCaptureId") is True + and resume.get("terminalCreateCaptureId") == resume.get("captureId"), + "oneDeterministicRunId": bool(resume.get("runId")) + and resume.get("runId") == resume.get("expectedRunId") == resume.get("finalizeReplayRunId"), + "committedHashesMatchReceipt": committed_ok, + "oneTerminalEvent": resume.get("terminalEvents") == [TERMINAL_EVENT], + "oneTerminalResult": bool(resume.get("runId")) + and resume.get("resultRunId") == resume.get("runId") + and resume.get("resultOutcomeKind") == resume.get("finalCaptureStatus") + and resume.get("finalCaptureStatus") in TERMINAL_STATUSES, + } + + +def _free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def _child_env() -> dict: + env = os.environ.copy() + env["PYTHONPATH"] = str(ROOT) + (os.pathsep + env["PYTHONPATH"] if env.get("PYTHONPATH") else "") + # The orchestrator reads this from the environment when the API binds it. + env["HSS_ORCH_STAGE_STUBS"] = "1" + return env + + +def _wait_for_server(server_url: str) -> bool: + deadline = time.monotonic() + SERVER_TIMEOUT_S + while time.monotonic() < deadline: + try: + with urllib.request.urlopen(server_url + "/v1/healthz", timeout=2) as response: + if response.status == 200: + return True + except Exception: # noqa: BLE001 - not ready yet; bounded by the deadline + time.sleep(0.2) + return False + + +def run_cold_restart( + receipt_path: Path, workdir: Path, *, stop: str = "commit-lost", packet_dir: Path | None = None +) -> dict: + """Orchestrate the cold restart and return the report. Never raises.""" + started_at = datetime.now(UTC).isoformat() + report: dict = { + "coldRestart": "capture_cold_restart", + "startedAt": started_at, + "receiptPath": str(receipt_path), + "stop": stop, + "workdir": str(workdir), + "processModel": { + "legs": ["primary", "resume"], + "freshPythonProcessPerLeg": True, + "sharedState": "saved receipt + client journal on disk only", + }, + "server": { + "entry": "create_app()", + "transport": "local uvicorn", + "stateBackend": "memory", + "storageBackend": "volume", + "hooks": "auto", + "stageStubs": True, + }, + "missingReceiptFields": [], + "invariants": {}, + "captureId": None, + "runId": None, + "finalCaptureStatus": None, + "terminalEvents": [], + "errorCodes": [], + "ok": False, + } + errors = report["errorCodes"] + receipt = load_receipt(receipt_path) + + # Fail before anything is sent: a pre-replay-field receipt is a caller + # mistake, and the report must name the field instead of guessing. + missing = missing_replay_fields(receipt) + if missing: + report["missingReceiptFields"] = missing + errors.extend(f"receipt_missing_field:{name}" for name in missing) + return _finish( + report, errors, "the saved receipt predates the replay fields; rerun capture_flow_e2e.py to rewrite it" + ) + packet_dir = _resolve_packet_dir(receipt, packet_dir) + fixture_errors: list[str] = [] + fixture = check_fixture(packet_dir, receipt, fixture_errors) + if fixture is None or fixture_errors: + errors.extend(fixture_errors) + return _finish(report, errors, "the saved receipt no longer matches the packet fixture") + + workdir.mkdir(parents=True, exist_ok=True) + port = _free_port() + server_url = f"http://127.0.0.1:{port}" + script = str(Path(__file__).resolve()) + server_log = (workdir / "server.log").open("wb") + server = subprocess.Popen( + [sys.executable, script, "--serve", "--workdir", str(workdir), "--port", str(port)], + env=_child_env(), + stdout=server_log, + stderr=server_log, + ) + try: + if not _wait_for_server(server_url): + errors.append("server_not_ready") + return _finish(report, errors, f"the local app never answered /v1/healthz within {SERVER_TIMEOUT_S:g}s") + report["server"]["url"] = server_url + + for leg, extra in (("primary", ["--stop", stop]), ("resume", [])): + try: + completed = subprocess.run( + [ + sys.executable, + script, + "--leg", + leg, + "--receipt", + str(receipt_path), + "--workdir", + str(workdir), + "--server-url", + server_url, + *extra, + ], + cwd=str(ROOT), + env=_child_env(), + capture_output=True, + text=True, + timeout=LEG_TIMEOUT_S, + check=False, + ) + if completed.returncode != 0: + errors.append(f"{leg}_leg_failed") + report[f"{leg}Stderr"] = completed.stderr[-500:] + except subprocess.TimeoutExpired: + errors.append(f"{leg}_leg_timeout") + + leg_reports: dict[str, dict] = {} + for leg in ("primary", "resume"): + path = workdir / f"leg-{leg}.json" + try: + leg_reports[leg] = json.loads(path.read_text()) + except (OSError, json.JSONDecodeError): + errors.append(f"{leg}_leg_report_missing") + if leg_reports: + primary = leg_reports.get("primary", {}) + resume = leg_reports.get("resume", {}) + report["primary"] = primary + report["resume"] = resume + if not primary.get("ok"): + errors.append("primary_leg_not_ok") + if not resume.get("ok"): + errors.append("resume_leg_not_ok") + report["invariants"] = _invariants(receipt, primary, resume) + if not all(report["invariants"].values()): + errors.append("invariants_broken") + report["captureId"] = resume.get("captureId") + report["runId"] = resume.get("runId") + report["finalCaptureStatus"] = resume.get("finalCaptureStatus") + report["terminalEvents"] = resume.get("terminalEvents", []) + return _finish(report, errors, report.get("errorDetail")) + finally: + server.terminate() + try: + server.wait(timeout=10) + except subprocess.TimeoutExpired: + server.kill() + server.wait(timeout=10) + server_log.close() + + +def serve(workdir: Path, port: int) -> int: + """The local app: one uvicorn process holding the memory state and the + volume the legs resume against. Internal mode of this script.""" + import uvicorn + + os.environ["VOLUME_ROOT"] = str(workdir / "vol") + os.environ["HSS_ORCH_STAGE_STUBS"] = "1" + os.environ.setdefault("STATE_BACKEND", "memory") + settings = Settings( + SMP_ENVIRONMENT="dev", + HSS_AUTH_MODE="static", + CAPTURE_API_TOKENS=f"device:{TOKEN}", + UPLOAD_URL_HMAC_SECRET="h" * 32, + STORAGE_BACKEND="volume", + STATE_BACKEND="memory", + VOLUME_ROOT=str(workdir / "vol"), + CAPTURE_API_PUBLIC_URL=f"http://127.0.0.1:{port}", + HSS_INGEST_HOOKS="auto", + VISION_EXTERNAL_CALLS="off", + GIT_SHA="0123456789abcdef", + SMP_SERVICE_NAME="capture-api", + ) + uvicorn.run(create_app(settings=settings), host="127.0.0.1", port=port, log_level="warning") + return 0 + + +def _summarize(report: dict) -> dict: + summary = { + "ok": report["ok"], + "stop": report.get("stop"), + "captureId": report.get("captureId"), + "runId": report.get("runId"), + "finalCaptureStatus": report.get("finalCaptureStatus"), + "errorCodes": report.get("errorCodes", []), + } + if report.get("missingReceiptFields"): + summary["missingReceiptFields"] = report["missingReceiptFields"] + if "errorDetail" in report: + summary["errorDetail"] = report["errorDetail"] + return summary + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--receipt", type=Path, help="saved capture_flow_e2e receipt to recover") + parser.add_argument("--workdir", type=Path, help="run directory (default: a fresh temp dir)") + parser.add_argument( + "--packet", type=Path, help="packet fixture (default: the receipt's packet dir, else the committed mini-arkit)" + ) + parser.add_argument("--report", type=Path, help="report path (default: /cold-restart-report.json)") + parser.add_argument( + "--stop", choices=sorted(STOP_POINTS), default="commit-lost", help="where the primary client dies" + ) + parser.add_argument("--leg", choices=["primary", "resume"], help=argparse.SUPPRESS) + parser.add_argument("--server-url", help=argparse.SUPPRESS) + parser.add_argument("--serve", action="store_true", help=argparse.SUPPRESS) + parser.add_argument("--port", type=int, default=0, help=argparse.SUPPRESS) + args = parser.parse_args(argv) + + if args.serve: + return serve(args.workdir or Path(tempfile.mkdtemp(prefix="capture-cold-restart-")), args.port) + + if args.leg: + if not (args.receipt and args.workdir and args.server_url): + parser.error(f"--leg {args.leg} needs --receipt, --workdir and --server-url") + if args.leg == "primary": + report = run_primary_leg(args.receipt, args.workdir, args.server_url, args.stop, packet_dir=args.packet) + else: + report = run_resume_leg(args.receipt, args.workdir, args.server_url, packet_dir=args.packet) + _write_json(args.workdir / f"leg-{args.leg}.json", report) + print(json.dumps({"ok": report["ok"], "leg": args.leg, "errorCodes": report["errorCodes"]})) + return 0 if report["ok"] else 1 + + if not args.receipt: + parser.error("--receipt is required") + + owned = args.workdir is None + workdir = args.workdir or Path(tempfile.mkdtemp(prefix="capture-cold-restart-")) + workdir.mkdir(parents=True, exist_ok=True) + report = run_cold_restart(args.receipt, workdir, stop=args.stop, packet_dir=args.packet) + report_path = args.report or workdir / "cold-restart-report.json" + _write_json(report_path, report) + print(json.dumps({**_summarize(report), "report": str(report_path)})) + if owned: + print(f"workdir: {workdir}") + return 0 if report["ok"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/tests/test_capture_cold_restart.py b/backend/tests/test_capture_cold_restart.py new file mode 100644 index 000000000..19e16e4c3 --- /dev/null +++ b/backend/tests/test_capture_cold_restart.py @@ -0,0 +1,234 @@ +"""Checks for ``scripts/capture_cold_restart.py``: recovery across a client cold restart. + +Each test first runs the real runner (``scripts/capture_flow_e2e.py``) to +produce a saved receipt, then drives the cold-restart command and asserts the +journey recovers when the client that resumes is a different Python process +from the one that lost the response. The subprocess tests are the point: the +primary client is a process that exits before the resume client starts, so the +only state the resume inherits is the saved receipt, the journal file, and +whatever the server remembers. Everything runs against the local memory/volume +backends with orchestrator stage stubs; no network, no Modal, no credentials. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +CLI_TIMEOUT_S = 600 +STOP_POINTS = ["create-lost", "pre-commit", "commit-lost", "pre-finalize", "finalize-lost"] + + +@pytest.fixture +def runner(): + import scripts.capture_flow_e2e as mod + + return mod + + +@pytest.fixture +def cold(): + import scripts.capture_cold_restart as mod + + return mod + + +@pytest.fixture +def saved_receipt(runner, tmp_path) -> Path: + """A receipt the runner wrote to disk, the way a prior session leaves it.""" + workdir = tmp_path / "original" + receipt = runner.run_capture_flow(workdir) + assert receipt["ok"] is True, receipt.get("errorDetail") + path = workdir / "receipt.json" + runner.write_receipt(receipt, path) + return path + + +def _run_cli(receipt: Path, workdir: Path, *extra: str) -> subprocess.CompletedProcess: + env = os.environ.copy() + env["PYTHONPATH"] = str(ROOT) + return subprocess.run( + [ + sys.executable, + "scripts/capture_cold_restart.py", + "--receipt", + str(receipt), + "--workdir", + str(workdir), + *extra, + ], + cwd=ROOT, + check=False, + capture_output=True, + text=True, + env=env, + timeout=CLI_TIMEOUT_S, + ) + + +def test_missing_replay_fields_names_each_field(cold) -> None: + assert cold.missing_replay_fields({}) == ["idempotencyKeys", "files", "packetSha256"] + assert cold.missing_replay_fields({"idempotencyKeys": {}}) == [ + "idempotencyKeys.create", + "idempotencyKeys.finalize", + "files", + "packetSha256", + ] + assert cold.missing_replay_fields( + {"idempotencyKeys": {"create": "k", "finalize": "k"}, "files": [{"path": "p"}]} + ) == [ + "files[0]", + "packetSha256", + ] + + +@pytest.mark.parametrize("field", ["idempotencyKeys", "files", "packetSha256"]) +def test_receipt_missing_replay_field_fails_loudly_naming_the_field( + cold, saved_receipt, tmp_path, capsys, field +) -> None: + receipt = json.loads(saved_receipt.read_text()) + receipt.pop(field) + stripped = tmp_path / "stripped-receipt.json" + stripped.write_text(json.dumps(receipt)) + + code = cold.main(["--receipt", str(stripped), "--workdir", str(tmp_path / "run")]) + + assert code == 1 + out = capsys.readouterr().out + assert f"receipt_missing_field:{field}" in out + report = json.loads((tmp_path / "run" / "cold-restart-report.json").read_text()) + assert report["missingReceiptFields"] == [field] + assert report["captureId"] is None # nothing reached the API + + +def test_cli_missing_replay_field_exits_nonzero(cold, saved_receipt, tmp_path) -> None: + receipt = json.loads(saved_receipt.read_text()) + receipt.pop("packetSha256") + stripped = tmp_path / "stripped-receipt.json" + stripped.write_text(json.dumps(receipt)) + + completed = _run_cli(stripped, tmp_path / "cli") + + assert completed.returncode == 1, completed.stdout + completed.stderr + assert "receipt_missing_field:packetSha256" in completed.stdout + + +@pytest.mark.parametrize("stop", STOP_POINTS) +def test_subprocess_recovery_after_client_process_exit(cold, saved_receipt, tmp_path, stop) -> None: + completed = _run_cli(saved_receipt, tmp_path / "run", "--stop", stop) + + assert completed.returncode == 0, completed.stdout + completed.stderr + report = json.loads((tmp_path / "run" / "cold-restart-report.json").read_text()) + assert report["ok"] is True, report.get("errorDetail") + assert report["errorCodes"] == [] + assert all(report["invariants"].values()), report["invariants"] + + # Two fresh client processes: the resume shares nothing with the primary + # but the receipt, the journal and the server. + assert report["processModel"]["legs"] == ["primary", "resume"] + assert report["processModel"]["freshPythonProcessPerLeg"] is True + assert report["primary"]["pid"] != report["resume"]["pid"] + assert report["primary"]["freshProcess"] and report["resume"]["freshProcess"] + assert report["primary"]["stop"] == stop + + assert report["finalCaptureStatus"] in cold.TERMINAL_STATUSES + assert report["terminalEvents"] == ["verdict_ready"] + + +def test_commit_lost_resume_rebuilds_state_from_the_server(cold, saved_receipt, tmp_path) -> None: + completed = _run_cli(saved_receipt, tmp_path / "run", "--stop", "commit-lost") + + assert completed.returncode == 0, completed.stdout + completed.stderr + report = json.loads((tmp_path / "run" / "cold-restart-report.json").read_text()) + primary, resume = report["primary"], report["resume"] + + # The primary died after the commit executed, without seeing the reply. + assert primary["interruption"] == {"kind": "lost_response", "step": "commit"} + assert primary["journal"]["commitExecuted"] is True + assert primary["journal"]["commitResponseSeen"] is False + assert primary["journal"]["captureId"] + + # The fresh client learned the commit outcome from the server, not from + # itself: create replayed from the idempotency store, re-register reported + # every file committed, and the identical commit body committed again. + assert resume["recoveryPath"] == "register_commit" + assert resume["createReplayHttpStatus"] == 200 + assert resume["recreatedSameCaptureId"] is True + assert resume["captureId"] == primary["journal"]["captureId"] + assert resume["serverFileStates"] and set(resume["serverFileStates"].values()) == {"committed"} + assert resume["recommit"] is True + assert resume["commitMismatch"] == [] + + # One deterministic run id across the restart, and one terminal state. + assert resume["runId"] == resume["expectedRunId"] == resume["finalizeReplayRunId"] + assert resume["finalCaptureStatus"] in cold.TERMINAL_STATUSES + assert resume["resultRunId"] == resume["runId"] + assert resume["resultOutcomeKind"] == resume["finalCaptureStatus"] + assert resume["terminalEvents"] == ["verdict_ready"] + assert resume["terminalCreateCaptureId"] == resume["captureId"] + assert report["invariants"] == { + "oneCaptureId": True, + "oneDeterministicRunId": True, + "committedHashesMatchReceipt": True, + "oneTerminalEvent": True, + "oneTerminalResult": True, + } + + +def test_finalize_lost_resume_recovers_the_run_id_the_dead_process_never_saw(cold, saved_receipt, tmp_path) -> None: + completed = _run_cli(saved_receipt, tmp_path / "run", "--stop", "finalize-lost") + + assert completed.returncode == 0, completed.stdout + completed.stderr + report = json.loads((tmp_path / "run" / "cold-restart-report.json").read_text()) + primary, resume = report["primary"], report["resume"] + assert primary["journal"]["finalizeExecuted"] is True + assert primary["journal"]["finalizeResponseSeen"] is False + + # The run id the dead process never learned is the deterministic one, and + # the resume's second finalize returns it unchanged. Recovery went through + # the finalize replay: the resume never re-registered (the capture is + # closed to files) and never re-committed. + assert resume["recoveryPath"] == "finalize_replay" + assert resume["serverFileStates"] == {} + assert resume["recommit"] is False + assert resume["runId"] == resume["expectedRunId"] == resume["finalizeReplayRunId"] + assert report["invariants"]["oneDeterministicRunId"] is True + assert report["invariants"]["oneCaptureId"] is True + + +def test_create_lost_resume_recovers_a_capture_id_that_never_reached_the_client(cold, saved_receipt, tmp_path) -> None: + completed = _run_cli(saved_receipt, tmp_path / "run", "--stop", "create-lost") + + assert completed.returncode == 0, completed.stdout + completed.stderr + report = json.loads((tmp_path / "run" / "cold-restart-report.json").read_text()) + primary, resume = report["primary"], report["resume"] + assert primary["journal"]["captureId"] is None + + # The store answered the saved key with the one capture, and the fresh + # client walked the whole remaining journey to the terminal result. + assert resume["createReplayHttpStatus"] == 200 + assert resume["captureId"] + assert resume["recreatedSameCaptureId"] is True + assert report["invariants"]["oneCaptureId"] is True + assert all(report["invariants"].values()), report["invariants"] + + +def test_pre_commit_resume_uploads_and_commits_for_the_first_time(cold, saved_receipt, tmp_path) -> None: + completed = _run_cli(saved_receipt, tmp_path / "run", "--stop", "pre-commit") + + assert completed.returncode == 0, completed.stdout + completed.stderr + report = json.loads((tmp_path / "run" / "cold-restart-report.json").read_text()) + primary, resume = report["primary"], report["resume"] + assert primary["journal"]["commitExecuted"] is False + + # The commit was never sent before the restart, so the fresh client is the + # first to commit — and the server still answers with the receipt's hashes. + assert resume["recommit"] is False + assert report["invariants"]["committedHashesMatchReceipt"] is True + assert all(report["invariants"].values()), report["invariants"]