From 13e0989cf6dcd1e6ca7a2ca024dcb606c9f464cc Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Thu, 22 Oct 2026 14:35:24 -0500 Subject: [PATCH 1/2] fix(api): release provisional finalize claims Co-Authored-By: Claude Opus 5.5 --- backend/server/api/finalize.py | 133 ++++++++++-------- .../server/api/tests/test_finalize_hostile.py | 58 ++++++++ 2 files changed, 135 insertions(+), 56 deletions(-) diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py index aacadbb1..9d697326 100644 --- a/backend/server/api/finalize.py +++ b/backend/server/api/finalize.py @@ -290,65 +290,86 @@ def accept_packet( raise fail(409, "packet_sha_conflict", "/", f"a different {what} was already accepted") try: 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 + ] + missing = [] + for path, entry in zip(paths, packet.files, strict=True): + rec = deps.state.get_file(record.capture_id, path) + if rec is None or rec.state != "committed": + missing.append(entry.path) + extra = extras(deps, record.capture_id) + extra["accepted"] = { + "packet_sha256": digest, + "run_id": run_id, + "at": isoformat(deps.now()), + "listed": len(packet.files), + } + extra["paths"] = paths + extra["digests"] = { + path: {"sha256": entry.sha256, "bytes": entry.bytes} + for path, entry in zip(paths, packet.files, strict=True) + } + if amendment_id is not None: + extra["amendment_id"] = amendment_id + # Completion belongs to the accepted packet, not the whole capture. A + # child must validate and promote its own files before starting its run. + for flag in ("run_started", "manifest_ok", "promoted_at"): + extra.pop(flag, None) + extra["missing"] = missing + extra.pop("last_422", None) + if amendment_id is None: + record.packet_sha256 = digest + record.run_id = run_id + else: + extra["parent_run_id"] = parent_run_id + amend = deps.state.get_kv(f"amend:{record.capture_id}:{amendment_id}") or {} + amend["packet_sha256"] = digest + amend["run_id"] = run_id + 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, + ) + if missing: + record.status = "awaiting_files" + deps.state.put_capture(record) + result = schedule_finish( + deps, + record, + digest, + key, + run_id, + parent_run_id=parent_run_id, + promote=False, + ) + else: + result = schedule_finish(deps, record, digest, key, run_id, parent_run_id=parent_run_id, promote=True) + return result 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) + # Persisting packet_sha256 on the capture (or amendment) is the commit + # point. Before it, a failed acceptance must allow a corrected packet. + # After it, keep the claim even if scheduling fails: identical retries + # resume the accepted packet, and stale containers must still conflict. + persisted = ( + deps.state.get_capture(record.capture_id) + if amendment_id is None + else deps.state.get_kv(f"amend:{record.capture_id}:{amendment_id}") + ) + persisted_digest = ( + getattr(persisted, "packet_sha256", None) + if amendment_id is None + else (persisted or {}).get("packet_sha256") + ) + if persisted_digest != digest: + 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 - ] - missing = [] - for path, entry in zip(paths, packet.files, strict=True): - rec = deps.state.get_file(record.capture_id, path) - if rec is None or rec.state != "committed": - missing.append(entry.path) - extra = extras(deps, record.capture_id) - extra["accepted"] = { - "packet_sha256": digest, - "run_id": run_id, - "at": isoformat(deps.now()), - "listed": len(packet.files), - } - extra["paths"] = paths - extra["digests"] = { - path: {"sha256": entry.sha256, "bytes": entry.bytes} for path, entry in zip(paths, packet.files, strict=True) - } - if amendment_id is not None: - extra["amendment_id"] = amendment_id - # Completion belongs to the accepted packet, not the whole capture. A - # child must validate and promote its own files before starting its run. - for flag in ("run_started", "manifest_ok", "promoted_at"): - extra.pop(flag, None) - extra["missing"] = missing - extra.pop("last_422", None) - if amendment_id is None: - record.packet_sha256 = digest - record.run_id = run_id - else: - extra["parent_run_id"] = parent_run_id - amend = deps.state.get_kv(f"amend:{record.capture_id}:{amendment_id}") or {} - amend["packet_sha256"] = digest - amend["run_id"] = run_id - 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 - ) - if missing: - record.status = "awaiting_files" - deps.state.put_capture(record) - return schedule_finish( - deps, - record, - digest, - key, - run_id, - 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) 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..7fa25e93 100644 --- a/backend/server/api/tests/test_finalize_hostile.py +++ b/backend/server/api/tests/test_finalize_hostile.py @@ -139,3 +139,61 @@ def offline(*_args, **_kwargs): 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", "schedule_finish"]) +def test_finalize_failure_releases_claim_for_a_different_packet(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() + 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}:-") == {"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 From 49fc2215a83ff946e2154ff4660f00f535f7d02f Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Thu, 22 Oct 2026 15:02:23 -0500 Subject: [PATCH 2/2] WIP: make finalize claims generational and append-only Co-Authored-By: Claude Opus 5.5 --- backend/server/api/finalize.py | 219 ++++++++++-------- .../server/api/tests/test_finalize_hostile.py | 155 ++++++++++++- 2 files changed, 278 insertions(+), 96 deletions(-) diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py index 9d697326..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,98 +336,77 @@ 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") - 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 - ] - missing = [] - for path, entry in zip(paths, packet.files, strict=True): - rec = deps.state.get_file(record.capture_id, path) - if rec is None or rec.state != "committed": - missing.append(entry.path) - extra = extras(deps, record.capture_id) - extra["accepted"] = { - "packet_sha256": digest, - "run_id": run_id, - "at": isoformat(deps.now()), - "listed": len(packet.files), - } - extra["paths"] = paths - extra["digests"] = { - path: {"sha256": entry.sha256, "bytes": entry.bytes} - for path, entry in zip(paths, packet.files, strict=True) - } - if amendment_id is not None: - extra["amendment_id"] = amendment_id - # Completion belongs to the accepted packet, not the whole capture. A - # child must validate and promote its own files before starting its run. - for flag in ("run_started", "manifest_ok", "promoted_at"): - extra.pop(flag, None) - extra["missing"] = missing - extra.pop("last_422", None) - if amendment_id is None: - record.packet_sha256 = digest - record.run_id = run_id - else: - extra["parent_run_id"] = parent_run_id - amend = deps.state.get_kv(f"amend:{record.capture_id}:{amendment_id}") or {} - amend["packet_sha256"] = digest - amend["run_id"] = run_id - 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, + # 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 + ] + missing = [] + for path, entry in zip(paths, packet.files, strict=True): + rec = deps.state.get_file(record.capture_id, path) + if rec is None or rec.state != "committed": + missing.append(entry.path) + extra = extras(deps, record.capture_id) + extra["accepted"] = { + "packet_sha256": digest, + "run_id": run_id, + "at": isoformat(deps.now()), + "listed": len(packet.files), + } + extra["paths"] = paths + extra["digests"] = { + path: {"sha256": entry.sha256, "bytes": entry.bytes} for path, entry in zip(paths, packet.files, strict=True) + } + if amendment_id is not None: + extra["amendment_id"] = amendment_id + # Completion belongs to the accepted packet, not the whole capture. A + # child must validate and promote its own files before starting its run. + for flag in ("run_started", "manifest_ok", "promoted_at"): + extra.pop(flag, None) + extra["missing"] = missing + extra.pop("last_422", None) + if amendment_id is None: + record.packet_sha256 = digest + record.run_id = run_id + else: + extra["parent_run_id"] = parent_run_id + amend = deps.state.get_kv(f"amend:{record.capture_id}:{amendment_id}") or {} + amend["packet_sha256"] = digest + amend["run_id"] = run_id + 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, + ) + if missing: + record.status = "awaiting_files" + 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, + key, + run_id, + parent_run_id=parent_run_id, + promote=False, ) - if missing: - record.status = "awaiting_files" - deps.state.put_capture(record) - result = schedule_finish( - deps, - record, - digest, - key, - run_id, - parent_run_id=parent_run_id, - promote=False, - ) - else: - result = schedule_finish(deps, record, digest, key, run_id, parent_run_id=parent_run_id, promote=True) - return result - except Exception: - if claimed_now: - # Persisting packet_sha256 on the capture (or amendment) is the commit - # point. Before it, a failed acceptance must allow a corrected packet. - # After it, keep the claim even if scheduling fails: identical retries - # resume the accepted packet, and stale containers must still conflict. - persisted = ( - deps.state.get_capture(record.capture_id) - if amendment_id is None - else deps.state.get_kv(f"amend:{record.capture_id}:{amendment_id}") - ) - persisted_digest = ( - getattr(persisted, "packet_sha256", None) - if amendment_id is None - else (persisted or {}).get("packet_sha256") - ) - if persisted_digest != digest: - deps.state.delete_kv(claim_key) - raise + 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 7fa25e93..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,14 +139,22 @@ 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", "schedule_finish"]) -def test_finalize_failure_releases_claim_for_a_different_packet(client, monkeypatch, failure): +@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}" @@ -164,6 +176,11 @@ def offline(*_args, **_kwargs): 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" @@ -187,7 +204,7 @@ def offline(*_args, **_kwargs): 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}:-") == {"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()) @@ -197,3 +214,131 @@ def offline(*_args, **_kwargs): 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