fix(backend): retryable finalize handoff and lease-guarded file-index merge

backend/server/api/finalize.py

@@ -15,7 +15,6 @@
 from server.api.errors import ApiFailure, fail, pointer_from_loc
 from server.api.events import emit
 from server.api.finalize_checks import manifest_errors
-from server.api.hooks import safe_hook
 from server.api.ids import require_contract_id
 from server.api.keys import packet_key
 from server.api.promotion import promote
@@ -345,7 +344,14 @@ def finish_ready(
         if record.status == "failed":
             raise fail(409, "invalid_state", "/", "promotion failed")
     if extra.get("started_run_id") != run_id:
-        safe_hook("on_finalized", deps.hooks.on_finalized, record.capture_id, digest, key, parent_run_id=parent_run_id)
+        try:
+            deps.hooks.on_finalized(record.capture_id, digest, key, parent_run_id=parent_run_id)
+        except Exception as exc:
+            log_event("hook.error", hook="on_finalized", error_code=type(exc).__name__, level="error")
+            # The run did not start. Leaving run_started unset keeps the handoff
+            # retryable: the background failure path restores the previous status,
+            # and a later finalize request schedules the run again.
+            raise fail(503, "run_start_failed", "/", "run start failed; finalize again") from exc
         log_event("run.started", run_id=run_id, capture_id=record.capture_id)
         extra = extras(deps, record.capture_id)
         extra["run_started"] = True
@@ -494,6 +500,18 @@ def _guard(deps: Deps, capture_id: str, future: Future[None]) -> None:
         _finish_loop(deps, capture_id)
     except Exception as exc:
         _clear_slot(capture_id)
+        # The 202 has already returned and the capture reads as processing, so a
+        # failure stored only on this Future would strand the client: no failure
+        # event, no retry, no run for the sweepers to recover. Route the exception
+        # through the normal failure path instead. Once the run has started, the
+        # handoff succeeded; later errors must not undo it.
+        if not extras(deps, capture_id).get("run_started"):
+            failure = (
+                exc
+                if isinstance(exc, ApiFailure)
+                else fail(503, "finalize_failed", "/", f"finalize background work failed ({type(exc).__name__})")
+            )
+            _on_background_failure(deps, capture_id, failure)
         future.set_exception(exc)
         return
     future.set_result(None)

backend/server/api/tests/test_finalize_background.py

@@ -0,0 +1,83 @@
+"""Escaped background finalize failures stay visible and retryable."""
+
+from __future__ import annotations
+
+import pytest
+
+from server.api.deps import drain_finalize, extras
+from server.api.tests.conftest import photo_packet
+from server.api.tests.test_ingest import _create, _finalize, _put_commit
+
+
+def _ready_capture(client, packet_id: str):
+    """A capture with every packet file committed, ready for a full finalize."""
+    capture_id = _create(client, packet_id)
+    blob = b"jpeg"
+    _put_commit(client, capture_id, "stills/s1.jpg", blob, meta={"still": "s1"})
+    packet = photo_packet(packet_id, [("stills/s1.jpg", blob)])
+    return capture_id, packet
+
+
+def test_failed_run_start_stays_retryable(client, monkeypatch):
+    capture_id, packet = _ready_capture(client, "packet-hook-fail")
+    deps = client.app_state
+
+    def boom(*args, **kwargs):
+        raise RuntimeError("modal start_run failed")
+
+    monkeypatch.setattr(deps.hooks, "on_finalized", boom)
+    client.hold_finalize = True
+    try:
+        response, _raw = _finalize(client, capture_id, packet)
+        assert response.status_code == 202, response.text
+        # finish_ready raises ApiFailure(503); _work_once handles it through the
+        # normal background failure path, so the drain itself completes.
+        drain_finalize(deps)
+    finally:
+        client.hold_finalize = False
+
+    # The handoff failed, so the run must not be marked started.
+    assert not extras(deps, capture_id).get("run_started")
+    assert extras(deps, capture_id).get("started_run_id") is None
+    events = client.get(f"/v1/captures/{capture_id}/events").json()["events"]
+    failures = [e for e in events if e["type"] == "failed" and e["data"].get("code") == "run_start_failed"]
+    assert failures and failures[0]["data"]["next"] == "retry_finalize"
+    assert deps.state.get_capture(capture_id).status == "uploading"
+
+    # With the hook healthy again, a later finalize starts the run.
+    monkeypatch.undo()
+    second, _raw = _finalize(client, capture_id, packet)
+    assert second.status_code == 202, second.text
+    assert second.json()["status"] == "processing"
+    assert extras(deps, capture_id).get("run_started") is True
+
+
+def test_background_exception_fails_retryably(client, monkeypatch):
+    capture_id, packet = _ready_capture(client, "packet-bg-fail")
+    deps = client.app_state
+
+    def offline(_key):
+        raise RuntimeError("volume offline")
+
+    monkeypatch.setattr(deps.storage, "get_bytes", offline)
+    client.hold_finalize = True
+    try:
+        response, _raw = _finalize(client, capture_id, packet)
+        assert response.status_code == 202, response.text
+        with pytest.raises(RuntimeError):
+            drain_finalize(deps)
+    finally:
+        client.hold_finalize = False
+
+    # The storage error must surface as a failure event and a retryable status,
+    # not as a capture stuck on processing with the exception on a dead Future.
+    events = client.get(f"/v1/captures/{capture_id}/events").json()["events"]
+    failures = [e for e in events if e["type"] == "failed" and e["data"].get("code") == "finalize_failed"]
+    assert failures and failures[0]["data"]["next"] == "retry_finalize"
+    assert deps.state.get_capture(capture_id).status == "uploading"
+
+    # Storage healthy again: a later finalize completes the handoff.
+    monkeypatch.undo()
+    second, _raw = _finalize(client, capture_id, packet)
+    assert second.status_code == 202, second.text
+    assert second.json()["status"] == "processing"

backend/server/common/backends/state_modal_dict.py

@@ -336,9 +336,17 @@ def delete_capture(self, capture_id: str) -> None:
 
     def upsert_file(self, rec: FileRecord) -> None:
         self._kv[f"file:{rec.capture_id}:{rec.path}"] = rec.model_dump(mode="json")
-        paths = set(self._kv.get(f"files:{rec.capture_id}") or [])
-        paths.add(rec.path)
-        self._kv[f"files:{rec.capture_id}"] = sorted(paths)
+
+        def merge() -> None:
+            paths = set(self._kv.get(f"files:{rec.capture_id}") or [])
+            paths.add(rec.path)
+            self._kv[f"files:{rec.capture_id}"] = sorted(paths)
+
+        # Two containers can register files in parallel; the route's capture lock
+        # only guards one process. Merge the index under the shared Dict lease so
+        # a stale read in one container cannot drop the other's entry. The lease
+        # claims its slot with the Dict's atomic skip_if_exists put.
+        with_lease(self, f"files:{rec.capture_id}", merge)
 
     def get_file(self, capture_id: str, path: str) -> FileRecord | None:
         return self._model(f"file:{capture_id}:{path}", FileRecord)

backend/server/common/tests/test_modal_dict_files.py

@@ -0,0 +1,124 @@
+"""Concurrent upsert_file writers keep every index entry."""
+
+from __future__ import annotations
+
+import threading
+import time
+
+from server.common.backends.state_modal_dict import ModalDictState
+from server.contracts.jobs import FileRecord
+
+
+class _ReadGate:
+    """Hold each concurrent pair of readers until both have entered.
+
+    A thread arriving alone waits briefly and proceeds (the leased, serialized
+    path); two threads arriving together are released with the same snapshot,
+    which is exactly the stale-read interleaving the index merge must survive.
+    """
+
+    def __init__(self) -> None:
+        self._cond = threading.Condition()
+        self._waiting = 0
+        self.enabled = True
+
+    def hold(self) -> None:
+        if not self.enabled:
+            return
+        with self._cond:
+            self._waiting += 1
+            self._cond.notify_all()
+            if self._waiting >= 2:
+                self._cond.notify_all()
+                return
+            deadline = time.monotonic() + 0.05
+            while self._waiting < 2 and time.monotonic() < deadline:
+                self._cond.wait(deadline - time.monotonic())
+            self._waiting = max(0, self._waiting - 1)
+
+
+class _FakeModalDict:
+    """modal.Dict stand-in: point get, write-through setitem, atomic skip_if_exists put."""
+
+    def __init__(self) -> None:
+        self.store: dict = {}
+        self._put_lock = threading.Lock()
+        self.gate = _ReadGate()
+
+    def get(self, key):
+        # Capture the snapshot BEFORE the GIL is released by the gate hold: the
+        # unfixed read-modify-write has no GIL release between read and write, so
+        # a hold placed after the read would never produce a stale snapshot.
+        value = self.store.get(key)
+        if key.startswith("files:"):
+            self.gate.hold()
+        return value
+
+    def __contains__(self, key) -> bool:
+        return key in self.store
+
+    def __setitem__(self, key, value) -> None:
+        self.store[key] = value
+
+    def __delitem__(self, key) -> None:
+        del self.store[key]
+
+    def pop(self, key, default=None):
+        if key not in self.store:
+            return default
+        value = self.store[key]
+        del self.store[key]
+        return value
+
+    def put(self, key, value, skip_if_exists: bool = False) -> bool:
+        with self._put_lock:
+            if skip_if_exists:
+                if key in self.store:
+                    return False
+                self.store[key] = value
+                return True
+            self.store[key] = value
+            return True
+
+    def keys(self):
+        return list(self.store.keys())
+
+
+def _file(capture_id: str, path: str) -> FileRecord:
+    return FileRecord(
+        capture_id=capture_id,
+        path=path,
+        role="still",
+        bytes=3,
+        sha256="a" * 64,
+        md5="b",
+        content_type="image/jpeg",
+        state="committed",
+    )
+
+
+def test_two_concurrent_writers_keep_every_index_entry() -> None:
+    mapping = _FakeModalDict()
+    left = ModalDictState(mapping)
+    right = ModalDictState(mapping)
+    start = threading.Barrier(2)
+
+    def register(state: ModalDictState, tag: str) -> None:
+        start.wait()
+        for index in range(20):
+            state.upsert_file(_file("cap_1", f"{tag}/s{index}.jpg"))
+
+    threads = [
+        threading.Thread(target=register, args=(left, "left")),
+        threading.Thread(target=register, args=(right, "right")),
+    ]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    mapping.gate.enabled = False
+    paths = [rec.path for rec in left.files("cap_1")]
+    expected = {f"{tag}/s{index}.jpg" for tag in ("left", "right") for index in range(20)}
+    assert set(paths) == expected
+    assert len(paths) == 40