From c6586261d17d2a693321566492097c3cd69b225e Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Thu, 22 Oct 2026 12:26:06 -0500 Subject: [PATCH 1/8] test(backend): finalize refuses non-finite numbers and a second container's packet Two Z10 audit findings, as tests that fail on alpha (5 failed, 1 passed): - A packet with NaN, -Infinity or 1e999 in a still's gravity, or NaN in session.startedAtUptime, is accepted with 202 and reaches processing. - Two API containers that both load a capture before either accepts can both accept different packets; the first client is told about a run that never starts. The identical-packet replay test passes before and after; it guards the fix. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../server/api/tests/test_finalize_hostile.py | 119 ++++++++++++++++++ 1 file changed, 119 insertions(+) create mode 100644 backend/server/api/tests/test_finalize_hostile.py diff --git a/backend/server/api/tests/test_finalize_hostile.py b/backend/server/api/tests/test_finalize_hostile.py new file mode 100644 index 00000000..1765a57e --- /dev/null +++ b/backend/server/api/tests/test_finalize_hostile.py @@ -0,0 +1,119 @@ +"""Finalize input that the schema alone let through, found by the Z10 audit probes.""" + +from __future__ import annotations + +import json + +import pytest + +from server.api.deps import drain_finalize +from server.api.errors import ApiFailure +from server.api.finalize import accept_packet +from server.api.tests.conftest import photo_packet, sha256_hex +from server.api.tests.test_ingest import _create, _put_commit + +BLOB = b"jpeg" + + +def _ready(client, packet_id: str) -> tuple[str, dict]: + capture_id = _create(client, packet_id) + _put_commit(client, capture_id, "stills/s1.jpg", BLOB, meta={"still": "s1"}) + return capture_id, photo_packet(packet_id, [("stills/s1.jpg", BLOB)]) + + +def _post(client, capture_id: str, packet_id: str, raw: bytes): + return client.post( + f"/v1/captures/{capture_id}/finalize", + content=raw, + headers={"Idempotency-Key": packet_id, "Content-Type": "application/json"}, + ) + + +def _with_gravity(packet: dict, middle: str) -> bytes: + packet["stills"][0]["gravity"] = [0.0, 0.25, 0.0] + return json.dumps(packet).replace("0.25", middle, 1).encode() + + +def _with_uptime(packet: dict, value: str) -> bytes: + packet["session"]["startedAtUptime"] = 0.25 + return json.dumps(packet).replace("0.25", value, 1).encode() + + +# Python's json module reads NaN, Infinity and -Infinity, and turns an +# overflowing literal such as 1e999 into inf. None of these is JSON (RFC 8259), +# and before the fix each of these packets was accepted with 202 and went on to +# processing, because only poses and intrinsics are re-checked in the background. +@pytest.mark.parametrize( + ("build", "pointer"), + [ + (lambda pk: _with_gravity(pk, "NaN"), "/stills/0/gravity/1"), + (lambda pk: _with_gravity(pk, "-Infinity"), "/stills/0/gravity/1"), + (lambda pk: _with_gravity(pk, "1e999"), "/stills/0/gravity/1"), + (lambda pk: _with_uptime(pk, "NaN"), "/session/startedAtUptime"), + ], + ids=["gravity-nan", "gravity-neg-infinity", "gravity-overflow", "uptime-nan"], +) +def test_non_finite_numbers_are_refused_before_acceptance(client, build, pointer): + packet_id = "packet-nonfinite" + capture_id, packet = _ready(client, packet_id) + + response = _post(client, capture_id, packet_id, build(packet)) + + assert response.status_code == 422, response.text + errors = response.json()["errors"] + assert errors[0]["code"] == "schema" + assert errors[0]["pointer"] == pointer + record = client.app_state.state.get_capture(capture_id) + assert record.packet_sha256 is None + assert record.status == "uploading" + # Nothing was accepted, so the corrected packet still finalizes. + fixed = _post(client, capture_id, packet_id, json.dumps(packet).encode()) + assert fixed.status_code == 202, fixed.text + assert fixed.json()["status"] == "processing" + + +def test_a_second_container_cannot_accept_a_different_packet(client): + """Two API containers (modal_entry.py allows up to four) each load the capture + before either accepts. The per-capture lock is in-process, so both reach + accept_packet holding a record with no packet_sha256. Before the fix both + returned 202: the first client was told about a run that never started, and + the second body silently replaced the stored packet.json.""" + packet_id = "packet-two-containers" + capture_id, packet = _ready(client, packet_id) + deps = client.app_state + first_view = deps.state.get_capture(capture_id) + second_view = first_view.model_copy(deep=True) + first = json.dumps(packet).encode() + packet["createdAt"] = "2026-09-27T00:00:00.000Z" + second = json.dumps(packet).encode() + + accepted = accept_packet(deps, first_view, first) + with pytest.raises(ApiFailure) as refused: + accept_packet(deps, second_view, second) + drain_finalize(deps) + + assert accepted["status"] == "processing" + assert refused.value.status == 409 + assert refused.value.errors[0]["code"] == "packet_sha_conflict" + assert deps.storage.get_bytes(f"captures/{capture_id}/packet.json") == first + record = deps.state.get_capture(capture_id) + assert record.packet_sha256 == sha256_hex(first) + assert record.run_id == accepted["runId"] + + +def test_a_second_container_with_the_same_packet_still_replays(client): + """The claim must not turn an identical retry on another container into a conflict.""" + packet_id = "packet-same-twice" + capture_id, packet = _ready(client, packet_id) + deps = client.app_state + first_view = deps.state.get_capture(capture_id) + second_view = first_view.model_copy(deep=True) + raw = json.dumps(packet).encode() + + first = accept_packet(deps, first_view, raw) + second = accept_packet(deps, second_view, raw) + drain_finalize(deps) + + assert first["status"] == second["status"] == "processing" + assert first["runId"] == second["runId"] + assert deps.state.get_capture(capture_id).packet_sha256 == sha256_hex(raw) From 7f073804320c605828a307ae34b64869a4bd9cc7 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Thu, 22 Oct 2026 12:30:36 -0500 Subject: [PATCH 2/8] fix(backend): refuse non-finite packet numbers and claim packet acceptance Non-finite numbers: json.loads reads NaN, Infinity, -Infinity and turns 1e999 into inf, and the packet models take any float. A still's gravity or session.startedAtUptime carrying one was accepted and reached processing; poses and intrinsics were caught only by the background manifest check, after the 202 and the full upload. accept_packet now answers 422 schema with a pointer to the number before anything is stored. Acceptance claim: the capture lock is per process, but the API runs on up to four containers. Two containers that loaded the capture before either accepted could both accept different packets; the first client was told about a run that never started and its packet.json was overwritten. accept_packet now claims accepted_packet:: with put_if_absent before writing; a different digest gets the usual 409 packet_sha_conflict, the same digest proceeds as before. A storage write failure releases the claim so a corrected packet can still finalize. server/api/tests/test_finalize_hostile.py: 5 failed, 1 passed on alpha; 6 passed here. server/api/tests: 217 passed. Co-Authored-By: Claude Opus 5.5 (1M context) --- backend/server/api/finalize.py | 46 +++++++++++++++++++++++++++++++++- 1 file changed, 45 insertions(+), 1 deletion(-) diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py index c0ea6781..aacadbb1 100644 --- a/backend/server/api/finalize.py +++ b/backend/server/api/finalize.py @@ -6,6 +6,7 @@ import hashlib import io import json +import math import threading from concurrent.futures import Future @@ -73,6 +74,30 @@ def inflate(body: bytes, encoding: str | None) -> bytes: return body +def _non_finite_pointer(payload: object) -> str | None: + """JSON pointer to a NaN or infinite number in the parsed packet, or None. + + json.loads reads NaN, Infinity and -Infinity, and turns an overflowing literal + such as 1e999 into inf. None of them is JSON (RFC 8259), and the packet models + take any float, so without this check they reach storage and the pipeline. + The background manifest check catches them only in poses, intrinsics, + timestamps and taps, and only after the 202 and the whole upload. + """ + stack: list[tuple[object, str]] = [(payload, "")] + while stack: + value, pointer = stack.pop() + if isinstance(value, float): + if not math.isfinite(value): + return pointer or "/" + elif isinstance(value, dict): + stack.extend( + (item, f"{pointer}/{str(key).replace('~', '~0').replace('/', '~1')}") for key, item in value.items() + ) + elif isinstance(value, list): + stack.extend((item, f"{pointer}/{index}") for index, item in enumerate(value)) + return None + + def _schema_errors(exc: ValidationError) -> list[dict]: items = [] for err in exc.errors(): @@ -212,6 +237,9 @@ def accept_packet( raise fail(422, "schema", "/", "packet is not json") from exc if not isinstance(payload, dict): raise fail(422, "schema", "/", "packet must be an object") + non_finite = _non_finite_pointer(payload) + if non_finite is not None: + raise fail(422, "schema", non_finite, "number is not finite") try: packet = Packet.model_validate(payload) except ValidationError as exc: @@ -250,7 +278,23 @@ def accept_packet( log_event("finalize.rejected", codes=extra["last_422"], capture_id=record.capture_id) raise _multi(errors) key = packet_key(record.capture_id, amendment_id) - deps.storage.put_bytes(key, raw, "application/json") + # The capture lock is per process and the API runs on up to + # CAPTURE_API_MAX_CONTAINERS containers, so `record` and the amendment row can + # predate another container's acceptance. The first digest to claim wins; a + # different packet gets the same 409 it would get once packet_sha256 is visible. + # The same packet proceeds, because its writes are identical. + claim_key = f"accepted_packet:{record.capture_id}:{amendment_id or '-'}" + claimed_now = deps.state.put_if_absent(claim_key, {"packet_sha256": digest}) + if not claimed_now and (deps.state.get_kv(claim_key) or {}).get("packet_sha256") != digest: + what = "packet" if amendment_id is None else "amendment packet" + raise fail(409, "packet_sha_conflict", "/", f"a different {what} was already accepted") + try: + deps.storage.put_bytes(key, raw, "application/json") + except Exception: + # Nothing was accepted, so a corrected packet must not meet a 409 on retry. + if claimed_now: + deps.state.delete_kv(claim_key) + raise run_id = _run_id(record.capture_id, raw) paths = [ entry.path if amendment_id is None else f"amendments/{amendment_id}/{entry.path}" for entry in packet.files From 7d3a996efd0ba38fdbf5561a10547a124d820d97 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Thu, 22 Oct 2026 12:35:07 -0500 Subject: [PATCH 3/8] test(backend): a failed packet write releases the acceptance claim Guard for the claim added in the previous commit, not a failing-first test: alpha has no claim to release. With delete_kv made a no-op the test fails with 409 packet_sha_conflict; with the fix it passes. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../server/api/tests/test_finalize_hostile.py | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/backend/server/api/tests/test_finalize_hostile.py b/backend/server/api/tests/test_finalize_hostile.py index 1765a57e..6b5ddd53 100644 --- a/backend/server/api/tests/test_finalize_hostile.py +++ b/backend/server/api/tests/test_finalize_hostile.py @@ -117,3 +117,25 @@ def test_a_second_container_with_the_same_packet_still_replays(client): assert first["status"] == second["status"] == "processing" assert first["runId"] == second["runId"] assert deps.state.get_capture(capture_id).packet_sha256 == sha256_hex(raw) + + +def test_a_failed_packet_write_releases_the_claim(client, monkeypatch): + """Guard for the claim, not failing-first: alpha has no claim to release.""" + packet_id = "packet-write-fails" + capture_id, packet = _ready(client, packet_id) + deps = client.app_state + record = deps.state.get_capture(capture_id) + real_put = deps.storage.put_bytes + + def offline(*_args, **_kwargs): + raise RuntimeError("storage offline") + + monkeypatch.setattr(deps.storage, "put_bytes", offline) + with pytest.raises(RuntimeError): + accept_packet(deps, record.model_copy(deep=True), json.dumps(packet).encode()) + monkeypatch.setattr(deps.storage, "put_bytes", real_put) + + packet["createdAt"] = "2026-09-27T00:00:00.000Z" + corrected = accept_packet(deps, record.model_copy(deep=True), json.dumps(packet).encode()) + drain_finalize(deps) + assert corrected["status"] == "processing" From 7007328c27d0977727966195fc91f3905d7fc15c Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 23 Oct 2026 08:23:08 -0500 Subject: [PATCH 4/8] test(backend): cover lower severity audit findings Co-Authored-By: Claude Code --- backend/server/api/tests/test_ingest.py | 20 +++++++++++++++++++ .../placement/tests/test_scene_bridge.py | 4 ++++ 2 files changed, 24 insertions(+) diff --git a/backend/server/api/tests/test_ingest.py b/backend/server/api/tests/test_ingest.py index ee653ca7..d7b40dda 100644 --- a/backend/server/api/tests/test_ingest.py +++ b/backend/server/api/tests/test_ingest.py @@ -162,6 +162,26 @@ def test_keyframe_image_rejected(client): assert response.json()["errors"][0]["pointer"] == "/files/0/role" +def test_file_registration_rejects_non_finite_meta(client): + capture_id = _create(client, "packet-meta-nonfinite") + body = _file("stills/s1.jpg", b"abc", meta={"still": "s1"}) + body["meta"]["still"]["timestamp"] = float("nan") + + response = client.post( + f"/v1/captures/{capture_id}/files", + content=json.dumps({"files": [body]}).encode(), + headers={"Content-Type": "application/json"}, + ) + + assert response.status_code == 422, response.text + assert response.json()["errors"][0] == { + "code": "schema", + "pointer": "/files/0/meta/still/timestamp", + "message": "number is not finite", + } + assert client.app_state.state.files(capture_id) == [] + + def test_tamper_then_reregister(client): capture_id = _create(client, "packet-tamper") data = b"jpeg-bytes" diff --git a/backend/server/placement/tests/test_scene_bridge.py b/backend/server/placement/tests/test_scene_bridge.py index 21904e1e..d5e120b6 100644 --- a/backend/server/placement/tests/test_scene_bridge.py +++ b/backend/server/placement/tests/test_scene_bridge.py @@ -615,6 +615,10 @@ def test_invalid_refusals_carry_code_and_pointer() -> None: s_range["walls"][0]["s_range_m"] = [-2.0, 5.0] # geometry says [-3.0, 2.0] cases.append((s_range, "/walls/0/s_range_m")) + repeated_baseline = copy.deepcopy(document) + repeated_baseline["walls"][0]["baseline_m"] = [[1.0, 1.0], [1.0, 1.0]] + cases.append((repeated_baseline, "/walls/0/baseline_m")) + end_s = copy.deepcopy(document) end_s["chain"]["ends"]["right"]["s_m"] = 2.5 cases.append((end_s, "/chain/ends/right/s_m")) From c7094b1c5715c0c1ff997606fedc7ca9c99ed018 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 23 Oct 2026 08:24:01 -0500 Subject: [PATCH 5/8] fix(backend): reject non-finite file metadata and correct pointer Co-Authored-By: Claude Code --- backend/server/api/routes/files.py | 28 ++++++++++++++++++++++++ backend/server/placement/scene_bridge.py | 2 +- 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/backend/server/api/routes/files.py b/backend/server/api/routes/files.py index 1dbb69a0..24a45e1d 100644 --- a/backend/server/api/routes/files.py +++ b/backend/server/api/routes/files.py @@ -4,6 +4,7 @@ import asyncio import base64 +import math import posixpath import re @@ -72,6 +73,23 @@ def _meta_ok(role: str, meta) -> bool: return False +def _non_finite_pointer(value: object, pointer: str = "") -> str | None: + if isinstance(value, float): + return pointer if not math.isfinite(value) else None + if isinstance(value, dict): + for key, item in value.items(): + escaped = str(key).replace("~", "~0").replace("/", "~1") + found = _non_finite_pointer(item, f"{pointer}/{escaped}") + if found is not None: + return found + elif isinstance(value, list): + for index, item in enumerate(value): + found = _non_finite_pointer(item, f"{pointer}/{index}") + if found is not None: + return found + return None + + def _sign(deps, capture_id: str, rec: FileRecord) -> UploadTarget: if rec.bytes > deps.settings.HSS_INGEST_MAX_SINGLE_PUT_BYTES: raise fail(501, "multipart_not_implemented", "/files", "multipart upload is not implemented in this PR") @@ -146,6 +164,16 @@ async def register_files(capture_id: str, body: RegisterFilesRequest, request: R "message": "meta does not match the role", } ) + if item.meta is not None: + meta_pointer = _non_finite_pointer(item.meta.model_dump(mode="python"), "/meta") + if meta_pointer is not None: + errors.append( + { + "code": "schema", + "pointer": f"/files/{index}{meta_pointer}", + "message": "number is not finite", + } + ) if item.bytes > deps.settings.HSS_INGEST_MAX_FILE_BYTES: errors.append( { diff --git a/backend/server/placement/scene_bridge.py b/backend/server/placement/scene_bridge.py index aada91a6..d1f12b29 100644 --- a/backend/server/placement/scene_bridge.py +++ b/backend/server/placement/scene_bridge.py @@ -325,7 +325,7 @@ def _baseline_length(wall) -> float: ) length = math.hypot(x1 - x0, z1 - z0) if length <= 0.0: - _invalid("the baseline repeats its first point", "/walls/0/baseline") + _invalid("the baseline repeats its first point", "/walls/0/baseline_m") return length From 3c24e69a21bf4288c19874e55038977593a5114e Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 23 Oct 2026 08:26:08 -0500 Subject: [PATCH 6/8] fix(backend): restore cold restart receipt loader Co-Authored-By: Claude Code --- backend/scripts/replay_capture.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/backend/scripts/replay_capture.py b/backend/scripts/replay_capture.py index 65fae0dc..07d8df1e 100644 --- a/backend/scripts/replay_capture.py +++ b/backend/scripts/replay_capture.py @@ -101,6 +101,11 @@ def fire(self, step: str) -> None: raise SimulatedTimeout(step) +def load_receipt(path: Path) -> dict: + """Read a saved capture-flow receipt from disk.""" + return json.loads(path.read_text()) + + def receipt_problem(receipt: object) -> str | None: """A named reason the saved receipt cannot be replayed, or None. From 3864c43652c04d07e7c665a16fa92cfd069e36e2 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 23 Oct 2026 09:53:27 -0500 Subject: [PATCH 7/8] fix(backend): a same-packet retry that lost the claim writes nothing Review of #327 at 3c24e69a: an identical finalize on a second container passed the claim check and wrote packet.json again while the first finalize worker read it. The volume backend's write truncates before it writes, so the worker could read an empty file and fail with JSONDecodeError; test_a_second_container_with_the_same_packet_still_replays failed about one run in three. Only the claimant now writes packet.json and schedules the finish. A retry that finds its own digest already claimed replays the claimant's 202 (processing, or awaiting_files with the same missing list); the run id is derived from the bytes, so it matches. test_a_same_packet_retry_on_another_container_writes_nothing counts packet writes and finish jobs. At 3c24e69a it fails (2 writes); here it passes. The hostile file passed 20 of 20 runs; server/api/tests: 220 passed (OMP_NUM_THREADS=1). Co-Authored-By: Claude Opus 5.5 (1M context) --- backend/server/api/finalize.py | 30 +++++++++-------- .../server/api/tests/test_finalize_hostile.py | 33 +++++++++++++++++++ 2 files changed, 50 insertions(+), 13 deletions(-) diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py index aacadbb1..bd5067d2 100644 --- a/backend/server/api/finalize.py +++ b/backend/server/api/finalize.py @@ -282,19 +282,6 @@ def accept_packet( # CAPTURE_API_MAX_CONTAINERS containers, so `record` and the amendment row can # predate another container's acceptance. The first digest to claim wins; a # different packet gets the same 409 it would get once packet_sha256 is visible. - # The same packet proceeds, because its writes are identical. - claim_key = f"accepted_packet:{record.capture_id}:{amendment_id or '-'}" - claimed_now = deps.state.put_if_absent(claim_key, {"packet_sha256": digest}) - if not claimed_now and (deps.state.get_kv(claim_key) or {}).get("packet_sha256") != digest: - what = "packet" if amendment_id is None else "amendment packet" - raise fail(409, "packet_sha_conflict", "/", f"a different {what} was already accepted") - try: - deps.storage.put_bytes(key, raw, "application/json") - except Exception: - # Nothing was accepted, so a corrected packet must not meet a 409 on retry. - if claimed_now: - deps.state.delete_kv(claim_key) - raise run_id = _run_id(record.capture_id, raw) paths = [ entry.path if amendment_id is None else f"amendments/{amendment_id}/{entry.path}" for entry in packet.files @@ -304,6 +291,23 @@ def accept_packet( rec = deps.state.get_file(record.capture_id, path) if rec is None or rec.state != "committed": missing.append(entry.path) + claim_key = f"accepted_packet:{record.capture_id}:{amendment_id or '-'}" + claimed_now = deps.state.put_if_absent(claim_key, {"packet_sha256": digest}) + if not claimed_now: + if (deps.state.get_kv(claim_key) or {}).get("packet_sha256") != digest: + what = "packet" if amendment_id is None else "amendment packet" + raise fail(409, "packet_sha_conflict", "/", f"a different {what} was already accepted") + # The same packet, already claimed. Only the claimant writes packet.json and + # schedules the finish: the volume backend's write truncates before it writes, + # so a second write can hand the claimant's worker an empty file. Run ids are + # derived from the bytes, so this reply matches the claimant's. + return render_finalize("awaiting_files" if missing else "processing", missing, run_id) + try: + deps.storage.put_bytes(key, raw, "application/json") + except Exception: + # Nothing was accepted, so a corrected packet must not meet a 409 on retry. + deps.state.delete_kv(claim_key) + raise extra = extras(deps, record.capture_id) extra["accepted"] = { "packet_sha256": digest, diff --git a/backend/server/api/tests/test_finalize_hostile.py b/backend/server/api/tests/test_finalize_hostile.py index 6b5ddd53..83442843 100644 --- a/backend/server/api/tests/test_finalize_hostile.py +++ b/backend/server/api/tests/test_finalize_hostile.py @@ -119,6 +119,39 @@ def test_a_second_container_with_the_same_packet_still_replays(client): assert deps.state.get_capture(capture_id).packet_sha256 == sha256_hex(raw) +def test_a_same_packet_retry_on_another_container_writes_nothing(client, monkeypatch): + """Review of #327 at 3c24e69a: the retry rewrote packet.json while the first + finalize worker read it. The volume backend truncates before writing, so the + worker could read an empty file and fail with JSONDecodeError. Counting writes + makes the race deterministic: only the claimant may write or schedule.""" + packet_id = "packet-retry-no-write" + capture_id, packet = _ready(client, packet_id) + deps = client.app_state + first_view = deps.state.get_capture(capture_id) + second_view = first_view.model_copy(deep=True) + raw = json.dumps(packet).encode() + key = f"captures/{capture_id}/packet.json" + writes: list[str] = [] + real_put = deps.storage.put_bytes + + def counting_put(write_key, data, content_type): + writes.append(write_key) + return real_put(write_key, data, content_type) + + monkeypatch.setattr(deps.storage, "put_bytes", counting_put) + jobs_before = len(deps.finalize_jobs) + first = accept_packet(deps, first_view, raw) + second = accept_packet(deps, second_view, raw) + jobs = len(deps.finalize_jobs) - jobs_before + drain_finalize(deps) + + assert writes.count(key) == 1 + assert jobs == 1 + assert second == first + assert deps.storage.get_bytes(key) == raw + assert deps.state.get_capture(capture_id).status == "processing" + + def test_a_failed_packet_write_releases_the_claim(client, monkeypatch): """Guard for the claim, not failing-first: alpha has no claim to release.""" packet_id = "packet-write-fails" From 49904399bbb52037ca80e78ba47f3f1ba4d53205 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 23 Oct 2026 10:35:55 -0500 Subject: [PATCH 8/8] fix(backend): a same-packet retry takes over an abandoned acceptance claim Astra review of #327 at 3864c436: the claim became durable before packet.json was written or work was scheduled. A claimant that died in between left every same-digest retry answering processing with nothing scheduled, and the capture stayed uploading. The claim now records claimed_at and a token. A same-digest retry still replays without writing when the stored packet matches the digest, or when the claim is younger than web_function_timeout_s(), the timeout modal_entry.py gives the API function, so the claimant may still be running. Otherwise the claim is abandoned and the retry takes it over. The store has no compare-and-swap, so the takeover inserts :takeover: with put_if_absent: of all retries that read the same abandoned claim, exactly one inserts it, writes the packet and schedules the finish. Tests (server/api/tests/test_finalize_hostile.py): - test_a_retry_takes_over_an_abandoned_claim and test_concurrent_takeovers_schedule_once fail on 3864c436 with (0, 0) == (1, 1) packet writes and finish jobs; both pass here. The concurrent test holds both retries at the takeover until both have read the abandoned claim. - test_a_retry_leaves_a_fresh_claim_alone passes on 3864c436 as well, because that commit never schedules on a retry. With the bound replaced by -1 it fails with (1, 1) == (0, 0). The file passed 20 of 20 runs (11 passed each); server/api/tests: 223 passed with OMP_NUM_THREADS=1. ruff check and format clean. Co-Authored-By: Claude Opus 5.5 (1M context) --- backend/server/api/finalize.py | 44 +++++++- .../server/api/tests/test_finalize_hostile.py | 104 +++++++++++++++++- 2 files changed, 142 insertions(+), 6 deletions(-) diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py index bd5067d2..d235a433 100644 --- a/backend/server/api/finalize.py +++ b/backend/server/api/finalize.py @@ -9,6 +9,7 @@ import math import threading from concurrent.futures import Future +from datetime import datetime from pydantic import ValidationError @@ -19,7 +20,7 @@ from server.api.ids import require_contract_id from server.api.keys import packet_key from server.api.promotion import promote -from server.common.deploy_switch import finalize_eta_s +from server.common.deploy_switch import finalize_eta_s, web_function_timeout_s from server.common.logging import log_event from server.contracts.api import AmendmentRequest, FinalizeResponse, StageData from server.contracts.jobs import CaptureRecord, run_id_for_capture @@ -98,6 +99,30 @@ def _non_finite_pointer(payload: object) -> str | None: return None +def _new_claim(deps: Deps, digest: str) -> dict: + return {"packet_sha256": digest, "claimed_at": isoformat(deps.now()), "token": deps.new_id("clm_")} + + +def _claim_abandoned(deps: Deps, held: dict, key: str, digest: str) -> bool: + """True when the claimant cannot still be running and never wrote the packet. + + The bound is the API web function's own timeout (modal_entry.py passes + web_function_timeout_s() to Modal), so a claim older than that belongs to a + request Modal has already ended. A claim with no readable time counts as old. + """ + try: + stored = deps.storage.get_bytes(key) + except Exception: + stored = None + if stored is not None and _sha(stored) == digest: + return False + try: + claimed_at = datetime.fromisoformat(str(held["claimed_at"]).replace("Z", "+00:00")) + except (KeyError, ValueError): + return True + return (deps.now() - claimed_at).total_seconds() > web_function_timeout_s() + + def _schema_errors(exc: ValidationError) -> list[dict]: items = [] for err in exc.errors(): @@ -292,16 +317,25 @@ def accept_packet( if rec is None or rec.state != "committed": missing.append(entry.path) claim_key = f"accepted_packet:{record.capture_id}:{amendment_id or '-'}" - claimed_now = deps.state.put_if_absent(claim_key, {"packet_sha256": digest}) - if not claimed_now: - if (deps.state.get_kv(claim_key) or {}).get("packet_sha256") != digest: + claim = _new_claim(deps, digest) + if not deps.state.put_if_absent(claim_key, claim): + held = deps.state.get_kv(claim_key) or {} + if held.get("packet_sha256") != digest: what = "packet" if amendment_id is None else "amendment packet" raise fail(409, "packet_sha_conflict", "/", f"a different {what} was already accepted") # The same packet, already claimed. Only the claimant writes packet.json and # schedules the finish: the volume backend's write truncates before it writes, # so a second write can hand the claimant's worker an empty file. Run ids are # derived from the bytes, so this reply matches the claimant's. - return render_finalize("awaiting_files" if missing else "processing", missing, run_id) + if not _claim_abandoned(deps, held, key, digest): + return render_finalize("awaiting_files" if missing else "processing", missing, run_id) + # The claimant died before writing. put_if_absent on a key named after the + # abandoned claim's token is the compare-and-swap: of any number of retries + # that read the same abandoned claim, exactly one inserts it and takes over. + if not deps.state.put_if_absent(f"{claim_key}:takeover:{held.get('token')}", {"by": claim["token"]}): + return render_finalize("awaiting_files" if missing else "processing", missing, run_id) + deps.state.put_kv(claim_key, claim) + log_event("finalize.claim_taken_over", capture_id=record.capture_id, run_id=run_id) try: deps.storage.put_bytes(key, raw, "application/json") except Exception: diff --git a/backend/server/api/tests/test_finalize_hostile.py b/backend/server/api/tests/test_finalize_hostile.py index 83442843..42213bff 100644 --- a/backend/server/api/tests/test_finalize_hostile.py +++ b/backend/server/api/tests/test_finalize_hostile.py @@ -3,14 +3,17 @@ from __future__ import annotations import json +import threading +from datetime import timedelta import pytest -from server.api.deps import drain_finalize +from server.api.deps import drain_finalize, isoformat from server.api.errors import ApiFailure from server.api.finalize import accept_packet from server.api.tests.conftest import photo_packet, sha256_hex from server.api.tests.test_ingest import _create, _put_commit +from server.common.deploy_switch import web_function_timeout_s BLOB = b"jpeg" @@ -152,6 +155,105 @@ def counting_put(write_key, data, content_type): assert deps.state.get_capture(capture_id).status == "processing" +def _plant_claim(deps, capture_id: str, raw: bytes, age_s: float) -> None: + """A claim whose request died before it wrote packet.json or scheduled anything.""" + claimed_at = deps.now() - timedelta(seconds=age_s) + deps.state.put_kv( + f"accepted_packet:{capture_id}:-", + {"packet_sha256": sha256_hex(raw), "claimed_at": isoformat(claimed_at), "token": "clm_dead"}, + ) + + +def _count_writes_and_jobs(deps, monkeypatch, capture_id: str): + key = f"captures/{capture_id}/packet.json" + writes: list[str] = [] + real_put = deps.storage.put_bytes + + def counting_put(write_key, data, content_type): + writes.append(write_key) + return real_put(write_key, data, content_type) + + monkeypatch.setattr(deps.storage, "put_bytes", counting_put) + jobs_before = len(deps.finalize_jobs) + return lambda: (writes.count(key), len(deps.finalize_jobs) - jobs_before) + + +STALE_S = web_function_timeout_s() + 60 + + +def test_a_retry_takes_over_an_abandoned_claim(client, monkeypatch): + """Astra review at 3864c436: a claimant that died between put_if_absent and the + packet write left every same-digest retry answering processing with nothing + scheduled, so the capture stayed uploading forever.""" + packet_id = "packet-abandoned-claim" + capture_id, packet = _ready(client, packet_id) + deps = client.app_state + raw = json.dumps(packet).encode() + _plant_claim(deps, capture_id, raw, age_s=STALE_S) + counts = _count_writes_and_jobs(deps, monkeypatch, capture_id) + + response = accept_packet(deps, deps.state.get_capture(capture_id), raw) + writes, jobs = counts() + drain_finalize(deps) + + assert (writes, jobs) == (1, 1) + assert response["status"] == "processing" + assert deps.storage.get_bytes(f"captures/{capture_id}/packet.json") == raw + assert deps.state.get_capture(capture_id).status == "processing" + + +def test_a_retry_leaves_a_fresh_claim_alone(client, monkeypatch): + """Guard, not failing-first: 3864c436 never schedules on a retry, so it passes + there too. It fails if the bound is dropped (checked by setting it to -1).""" + packet_id = "packet-fresh-claim" + capture_id, packet = _ready(client, packet_id) + deps = client.app_state + raw = json.dumps(packet).encode() + _plant_claim(deps, capture_id, raw, age_s=5) + counts = _count_writes_and_jobs(deps, monkeypatch, capture_id) + + response = accept_packet(deps, deps.state.get_capture(capture_id), raw) + + assert counts() == (0, 0) + assert response["status"] == "processing" + + +def test_concurrent_takeovers_schedule_once(client, monkeypatch): + packet_id = "packet-two-takeovers" + capture_id, packet = _ready(client, packet_id) + deps = client.app_state + raw = json.dumps(packet).encode() + _plant_claim(deps, capture_id, raw, age_s=STALE_S) + counts = _count_writes_and_jobs(deps, monkeypatch, capture_id) + # Both retries must have read the abandoned claim before either tries to take it. + both_read = threading.Barrier(2, timeout=10) + real_insert = deps.state.put_if_absent + + def insert_after_both_read(key, value): + if ":takeover:" in key: + both_read.wait() + return real_insert(key, value) + + monkeypatch.setattr(deps.state, "put_if_absent", insert_after_both_read) + record = deps.state.get_capture(capture_id) + replies: list[dict] = [] + threads = [ + threading.Thread(target=lambda: replies.append(accept_packet(deps, record.model_copy(deep=True), raw))) + for _ in range(2) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + writes, jobs = counts() + drain_finalize(deps) + + assert len(replies) == 2 + assert (writes, jobs) == (1, 1) + assert {reply["runId"] for reply in replies} == {replies[0]["runId"]} + assert deps.state.get_capture(capture_id).status == "processing" + + def test_a_failed_packet_write_releases_the_claim(client, monkeypatch): """Guard for the claim, not failing-first: alpha has no claim to release.""" packet_id = "packet-write-fails"