diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py index aacadbb1..e3db74b2 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 UTC, datetime, timedelta from pydantic import ValidationError @@ -28,6 +29,13 @@ _INFLATE_CAP = 48 * 1024 * 1024 _GZIP_CAP = 8 * 1024 * 1024 +# There is no evidence behind this value; it is a bounded recovery window for a +# process that claimed a packet and then disappeared. Claims are never deleted, +# so a later generation preserves the audit trail and cannot erase an in-flight +# owner's decision. +_ACCEPTED_CLAIM_STALE_AFTER = timedelta(minutes=5) + + _TIER_MARKERS = ( "ARKit", "photo_only", @@ -144,6 +152,56 @@ def _multi(errors: list[dict]): return ApiFailure(422, errors) +def _claim_prefix(capture_id: str, amendment_id: str | None) -> str: + if amendment_id is None: + return f"accepted_packet:{capture_id}" + return f"accepted_packet:{capture_id}:amendment:{amendment_id}" + + +def _claim_generation_key(capture_id: str, amendment_id: str | None, generation: int) -> str: + return f"{_claim_prefix(capture_id, amendment_id)}:{generation}" + + +def _claim_is_stale(deps: Deps, claim: dict) -> bool: + claimed_at = claim.get("claimed_at") + if not isinstance(claimed_at, str): + return False + try: + when = datetime.fromisoformat(claimed_at.replace("Z", "+00:00")) + except ValueError: + return False + if when.tzinfo is None: + when = when.replace(tzinfo=UTC) + return deps.now() - when > _ACCEPTED_CLAIM_STALE_AFTER + + +def _claim_packet(deps: Deps, capture_id: str, amendment_id: str | None, digest: str) -> int: + # Walk immutable, contiguous generations rather than updating a current-key + # pointer: a crash between insertion and pointer publication would strand it. + generation = 0 + while True: + key = _claim_generation_key(capture_id, amendment_id, generation) + claim = deps.state.get_kv(key) + if claim is None: + candidate = { + "packet_sha256": digest, + "owner": deps.new_id("finalize-"), + "claimed_at": isoformat(deps.now()), + } + if deps.state.put_if_absent(key, candidate): + return generation + continue + if deps.state.get_kv(_claim_generation_key(capture_id, amendment_id, generation + 1)) is not None: + generation += 1 + continue + if claim.get("packet_sha256") == digest: + return generation + if deps.state.get_kv(f"{key}:committed") or not _claim_is_stale(deps, claim): + what = "packet" if amendment_id is None else "amendment packet" + raise fail(409, "packet_sha_conflict", "/", f"a different {what} was already accepted") + generation += 1 + + class _FinishSlot: def __init__(self) -> None: self.inflight = False @@ -278,23 +336,13 @@ 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) - # 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 + # Claims are append-only because StateStore cannot compare-and-delete. A + # generation records its owner and age; only an uncommitted stale generation + # can be superseded, while identical requests remain safe to replay. + generation = _claim_packet(deps, record.capture_id, amendment_id, digest) + claim_key = _claim_generation_key(record.capture_id, amendment_id, generation) + committed_key = f"{claim_key}:committed" + deps.storage.put_bytes(key, raw, "application/json") 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 @@ -334,12 +382,20 @@ def accept_packet( deps.state.put_kv(f"amend:{record.capture_id}:{amendment_id}", amend) save_extras(deps, record.capture_id, extra) log_event( - "finalize.accepted", listed=len(packet.files), missing=len(missing), run_id=run_id, capture_id=record.capture_id + "finalize.accepted", + listed=len(packet.files), + missing=len(missing), + run_id=run_id, + capture_id=record.capture_id, ) if missing: record.status = "awaiting_files" - deps.state.put_capture(record) - return schedule_finish( + deps.state.put_capture(record) + # Both the capture and any amendment row are durable before commitment. + # Scheduling may fail afterward; an identical request resumes that packet. + deps.state.put_if_absent(committed_key, {"committed_at": isoformat(deps.now())}) + if missing: + result = schedule_finish( deps, record, digest, @@ -348,7 +404,9 @@ def accept_packet( parent_run_id=parent_run_id, promote=False, ) - return schedule_finish(deps, record, digest, key, run_id, parent_run_id=parent_run_id, promote=True) + else: + result = schedule_finish(deps, record, digest, key, run_id, parent_run_id=parent_run_id, promote=True) + return result def _amendment_manifest(deps: Deps, record: CaptureRecord, packet: Packet, amendment_id: str) -> list[dict]: diff --git a/backend/server/api/tests/test_finalize_hostile.py b/backend/server/api/tests/test_finalize_hostile.py index 6b5ddd53..07708d84 100644 --- a/backend/server/api/tests/test_finalize_hostile.py +++ b/backend/server/api/tests/test_finalize_hostile.py @@ -3,6 +3,10 @@ from __future__ import annotations import json +from asyncio import CancelledError +from concurrent.futures import ThreadPoolExecutor +from datetime import UTC, datetime, timedelta +from threading import Event, current_thread import pytest @@ -119,8 +123,8 @@ 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_failed_packet_write_releases_the_claim(client, monkeypatch): - """Guard for the claim, not failing-first: alpha has no claim to release.""" +def test_a_failed_packet_write_keeps_claim_until_stale(client, monkeypatch): + """A failed owner remains visible while an identical retry can recover it.""" packet_id = "packet-write-fails" capture_id, packet = _ready(client, packet_id) deps = client.app_state @@ -135,7 +139,206 @@ def offline(*_args, **_kwargs): accept_packet(deps, record.model_copy(deep=True), json.dumps(packet).encode()) monkeypatch.setattr(deps.storage, "put_bytes", real_put) + raw = json.dumps(packet).encode() + claim = deps.state.get_kv(f"accepted_packet:{capture_id}:0") + assert claim and claim["packet_sha256"] == sha256_hex(raw) packet["createdAt"] = "2026-09-27T00:00:00.000Z" + with pytest.raises(ApiFailure) as refused: + accept_packet(deps, record.model_copy(deep=True), json.dumps(packet).encode()) + assert refused.value.status == 409 + now = deps.now() + deps.clock = lambda: now + timedelta(minutes=5, seconds=1) corrected = accept_packet(deps, record.model_copy(deep=True), json.dumps(packet).encode()) drain_finalize(deps) assert corrected["status"] == "processing" + + +@pytest.mark.parametrize("failure", ["save_extras", "put_capture"]) +def test_finalize_failure_keeps_claim_for_a_different_packet_until_stale(client, monkeypatch, failure): + import server.api.finalize as finalize_module + + packet_id = f"packet-{failure}" + capture_id, packet = _ready(client, packet_id) + deps = client.app_state + raw = json.dumps(packet).encode() + + def offline(*_args, **_kwargs): + raise RuntimeError(failure) + + with monkeypatch.context() as patch: + target = deps.state if failure == "put_capture" else finalize_module + patch.setattr(target, failure, offline) + with pytest.raises(RuntimeError, match=failure): + accept_packet(deps, deps.state.get_capture(capture_id), raw) + + # The packet bytes made it to storage, but acceptance was not persisted. + assert deps.storage.get_bytes(f"captures/{capture_id}/packet.json") == raw + assert deps.state.get_capture(capture_id).packet_sha256 is None + packet["createdAt"] = "2026-09-27T00:00:00.000Z" + corrected_raw = json.dumps(packet).encode() + with pytest.raises(ApiFailure) as refused: + accept_packet(deps, deps.state.get_capture(capture_id), corrected_raw) + assert refused.value.status == 409 + now = deps.now() + deps.clock = lambda: now + timedelta(minutes=5, seconds=1) + corrected = accept_packet(deps, deps.state.get_capture(capture_id), corrected_raw) + drain_finalize(deps) + assert corrected["status"] == "processing" + assert deps.state.get_capture(capture_id).packet_sha256 == sha256_hex(corrected_raw) + + +def test_failure_after_capture_persistence_keeps_claim_and_identical_replay(client, monkeypatch): + import server.api.finalize as finalize_module + + capture_id, packet = _ready(client, "packet-persisted-failure") + deps = client.app_state + stale_record = deps.state.get_capture(capture_id) + raw = json.dumps(packet).encode() + + def offline(*_args, **_kwargs): + raise RuntimeError("event offline") + + with monkeypatch.context() as patch: + patch.setattr(finalize_module, "_emit_validate", offline) + with pytest.raises(RuntimeError, match="event offline"): + accept_packet(deps, stale_record.model_copy(deep=True), raw) + + assert deps.state.get_capture(capture_id).packet_sha256 == sha256_hex(raw) + assert deps.state.get_kv(f"accepted_packet:{capture_id}:0")["packet_sha256"] == sha256_hex(raw) + packet["createdAt"] = "2026-09-27T00:00:00.000Z" + with pytest.raises(ApiFailure) as refused: + accept_packet(deps, stale_record, json.dumps(packet).encode()) + assert refused.value.status == 409 + assert refused.value.errors[0]["code"] == "packet_sha_conflict" + replay = accept_packet(deps, deps.state.get_capture(capture_id), raw) + drain_finalize(deps) + assert replay["status"] == "processing" + assert replay["runId"] == deps.state.get_capture(capture_id).run_id + + +@pytest.mark.parametrize("interruption", [CancelledError, SystemExit], ids=["cancelled", "crash"]) +def test_interrupted_claim_can_be_superseded_only_after_staleness_bound(client, monkeypatch, interruption): + capture_id, packet = _ready(client, "packet-interrupted") + deps = client.app_state + stale_record = deps.state.get_capture(capture_id) + raw = json.dumps(packet).encode() + now = datetime(2026, 10, 22, 19, 0, tzinfo=UTC) + deps.clock = lambda: now + + def interrupted(*_args, **_kwargs): + raise interruption() + + with monkeypatch.context() as patch: + patch.setattr(deps.storage, "put_bytes", interrupted) + with pytest.raises(interruption): + accept_packet(deps, stale_record.model_copy(deep=True), raw) + + first_key = f"accepted_packet:{capture_id}:0" + claim = deps.state.get_kv(first_key) + assert claim["packet_sha256"] == sha256_hex(raw) + assert claim["owner"] + assert claim["claimed_at"] == "2026-10-22T19:00:00.000Z" + assert deps.state.get_kv(f"{first_key}:committed") is None + packet["createdAt"] = "2026-09-27T00:00:00.000Z" + replacement = json.dumps(packet).encode() + for elapsed in (timedelta(0), timedelta(minutes=5)): + deps.clock = lambda elapsed=elapsed: now + elapsed + with pytest.raises(ApiFailure) as refused: + accept_packet(deps, stale_record.model_copy(deep=True), replacement) + assert refused.value.status == 409 + assert deps.state.get_kv(f"accepted_packet:{capture_id}:1") is None + + deps.clock = lambda: now + timedelta(minutes=5, microseconds=1) + accepted = accept_packet(deps, stale_record.model_copy(deep=True), replacement) + drain_finalize(deps) + assert accepted["status"] == "processing" + assert deps.state.get_kv(first_key) == claim + assert deps.state.get_kv(f"accepted_packet:{capture_id}:1")["packet_sha256"] == sha256_hex(replacement) + assert deps.state.get_kv(f"accepted_packet:{capture_id}:1:committed") + + +def test_identical_retry_completes_failed_claim_without_deleting_it(client, monkeypatch): + capture_id, packet = _ready(client, "packet-identical-retry") + deps = client.app_state + stale_record = deps.state.get_capture(capture_id) + raw = json.dumps(packet).encode() + + def offline(*_args, **_kwargs): + raise RuntimeError("storage offline") + + with monkeypatch.context() as patch: + patch.setattr(deps.storage, "put_bytes", offline) + with pytest.raises(RuntimeError, match="storage offline"): + accept_packet(deps, stale_record.model_copy(deep=True), raw) + + claim_key = f"accepted_packet:{capture_id}:0" + claim = deps.state.get_kv(claim_key) + assert claim is not None + accepted = accept_packet(deps, stale_record.model_copy(deep=True), raw) + drain_finalize(deps) + assert accepted["status"] == "processing" + assert deps.state.get_kv(claim_key) == claim + assert deps.state.get_kv(f"{claim_key}:committed") + assert deps.state.get_kv(f"accepted_packet:{capture_id}:1") is None + + +def test_failed_owner_cannot_release_identical_inflight_writers_claim(client, monkeypatch): + capture_id, packet = _ready(client, "packet-inflight-identical") + deps = client.app_state + stale_record = deps.state.get_capture(capture_id) + raw = json.dumps(packet).encode() + packet["createdAt"] = "2026-09-27T00:00:00.000Z" + different = json.dumps(packet).encode() + owner_writing, retry_writing = Event(), Event() + fail_owner, finish_retry = Event(), Event() + real_put = deps.storage.put_bytes + + def gated_put(*args, **kwargs): + if current_thread().name.startswith("claim-owner"): + owner_writing.set() + assert fail_owner.wait(5), "owner was never released" + raise RuntimeError("owner write failed") + if current_thread().name.startswith("claim-retry"): + retry_writing.set() + assert finish_retry.wait(5), "retry was never released" + return real_put(*args, **kwargs) + + monkeypatch.setattr(deps.storage, "put_bytes", gated_put) + with ThreadPoolExecutor(max_workers=1, thread_name_prefix="claim-owner") as owner_pool: + with ThreadPoolExecutor(max_workers=1, thread_name_prefix="claim-retry") as retry_pool: + try: + owner = owner_pool.submit(accept_packet, deps, stale_record.model_copy(deep=True), raw) + assert owner_writing.wait(5), "owner did not reach storage" + retry = retry_pool.submit(accept_packet, deps, stale_record.model_copy(deep=True), raw) + assert retry_writing.wait(5), "identical retry did not reach storage" + fail_owner.set() + with pytest.raises(RuntimeError, match="owner write failed"): + owner.result(timeout=5) + # B is still writing after A failed. A must not open a hole for C. + with pytest.raises(ApiFailure) as refused: + accept_packet(deps, stale_record.model_copy(deep=True), different) + assert refused.value.status == 409 + finally: + fail_owner.set() + finish_retry.set() + assert retry.result(timeout=5)["status"] == "processing" + drain_finalize(deps) + assert deps.storage.get_bytes(f"captures/{capture_id}/packet.json") == raw + assert deps.state.get_capture(capture_id).packet_sha256 == sha256_hex(raw) + + +def test_committed_generation_never_expires_even_for_stale_capture_view(client): + capture_id, packet = _ready(client, "packet-committed-generation") + deps = client.app_state + stale_record = deps.state.get_capture(capture_id) + raw = json.dumps(packet).encode() + accept_packet(deps, stale_record.model_copy(deep=True), raw) + drain_finalize(deps) + now = deps.now() + deps.clock = lambda: now + timedelta(days=30) + packet["createdAt"] = "2026-09-27T00:00:00.000Z" + with pytest.raises(ApiFailure) as refused: + accept_packet(deps, stale_record, json.dumps(packet).encode()) + assert refused.value.status == 409 + assert deps.state.get_kv(f"accepted_packet:{capture_id}:0:committed") + assert deps.state.get_kv(f"accepted_packet:{capture_id}:1") is None