fix(api): release finalize claims after failed acceptance

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]:

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