Backend: join reviewed runtime fixes (#251, #252, #266) onto the capture stack with join-verification journeys

backend/server/api/app.py

@@ -31,6 +31,7 @@
     uploads,
 )
 from server.common.config import Settings, load, resolve_state, resolve_storage
+from server.common.lease import LeaseBusy
 from server.common.logging import log_event, request_fields
 
 _MB = 1024 * 1024
@@ -202,6 +203,13 @@ async def send_now(response: JSONResponse):
     async def api_failure(_request: Request, failure: ApiFailure):
         return error_response(failure)
 
+    @app.exception_handler(LeaseBusy)
+    async def lease_busy(_request: Request, exc: LeaseBusy):
+        # The file-index merge never ran, so the registration is safe to retry.
+        return error_response(
+            fail(503, "lease_busy", "/", "file index is busy; retry the registration", **{"Retry-After": "1"})
+        )
+
     @app.exception_handler(RequestValidationError)
     async def validation_error(_request: Request, exc: RequestValidationError):
         items = []

backend/server/api/finalize.py

@@ -499,19 +499,26 @@ def _guard(deps: Deps, capture_id: str, future: Future[None]) -> None:
     try:
         _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)
+        #
+        # The worker slot stays occupied and the publication lock is held across
+        # the run_started check and the failure write: a concurrent finalize retry
+        # publishes its success under the same lock, so this older worker can
+        # never reset a healed retry or emit a contradictory terminal event after
+        # it.
+        with deps.locks.finish_sync(capture_id):
+            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)
+        _clear_slot(capture_id)
         future.set_exception(exc)
         return
     future.set_result(None)

backend/server/api/routes/files.py

@@ -289,58 +289,48 @@ async def commit_files(capture_id: str, body: CommitRequest, request: Request):
             newly: list[FileRecord] = []
             extra = extras(deps, capture_id)
             size_only = set(extra.get("size_only_paths") or [])
-            for item in body.files:
-                rec = deps.state.get_file(capture_id, item.path)
-                if rec is None:
-                    not_found.append(item.path)
-                    continue
-                if rec.state == "committed":
-                    committed.append(item.path)
-                    continue
-                spec = (extra.get("digests") or {}).get(item.path)
-                if spec and (rec.sha256 != spec["sha256"] or rec.bytes != spec["bytes"]):
-                    mismatch.append(item.path)
-                    continue
-                try:
-                    meta = deps.storage.head(upload_key(capture_id, item.path))
-                except FileNotFoundError:
-                    not_found.append(item.path)
-                    continue
-                except Exception as exc:
-                    log_event("storage.error", op="head", capture_id=capture_id, level="error")
-                    raise fail(503, "storage_unavailable", "/", "storage head failed", **{"Retry-After": "5"}) from exc
-                md5_bad = item.path not in size_only and meta.md5_b64 not in (None, rec.md5)
-                if meta.bytes != rec.bytes or md5_bad:
-                    rec.state = "mismatch"
+            # A failure partway through the batch (a busy index lease, a storage
+            # head error) must still announce the files already committed: the
+            # retry reads them as committed and would never announce them.
+            try:
+                for item in body.files:
+                    rec = deps.state.get_file(capture_id, item.path)
+                    if rec is None:
+                        not_found.append(item.path)
+                        continue
+                    if rec.state == "committed":
+                        committed.append(item.path)
+                        continue
+                    spec = (extra.get("digests") or {}).get(item.path)
+                    if spec and (rec.sha256 != spec["sha256"] or rec.bytes != spec["bytes"]):
+                        mismatch.append(item.path)
+                        continue
+                    try:
+                        meta = deps.storage.head(upload_key(capture_id, item.path))
+                    except FileNotFoundError:
+                        not_found.append(item.path)
+                        continue
+                    except Exception as exc:
+                        log_event("storage.error", op="head", capture_id=capture_id, level="error")
+                        raise fail(
+                            503, "storage_unavailable", "/", "storage head failed", **{"Retry-After": "5"}
+                        ) from exc
+                    md5_bad = item.path not in size_only and meta.md5_b64 not in (None, rec.md5)
+                    if meta.bytes != rec.bytes or md5_bad:
+                        rec.state = "mismatch"
+                        deps.state.upsert_file(rec)
+                        mismatch.append(item.path)
+                        continue
+                    rec.state = "committed"
+                    rec.generation = meta.generation
+                    rec.key = upload_key(capture_id, item.path)
+                    rec.committed_at = isoformat(deps.now())
                     deps.state.upsert_file(rec)
-                    mismatch.append(item.path)
-                    continue
-                rec.state = "committed"
-                rec.generation = meta.generation
-                rec.key = upload_key(capture_id, item.path)
-                rec.committed_at = isoformat(deps.now())
-                deps.state.upsert_file(rec)
-                committed.append(item.path)
-                newly.append(rec)
-            if newly:
-                listed = (extras(deps, capture_id).get("accepted") or {}).get("listed")
-                payload = FilesCommittedData(
-                    paths=[rec.path for rec in newly], committed=len(newly), listed=listed
-                ).model_dump(mode="json")
-                emit(deps, capture_id, "files_committed", payload)
-                hooked = [
-                    CommittedFile(
-                        path=rec.path,
-                        role=rec.role,
-                        sha256=rec.sha256,
-                        bytes=rec.bytes,
-                        generation=rec.generation or "",
-                        key=rec.key or upload_key(capture_id, rec.path),
-                        meta=rec.meta,
-                    )
-                    for rec in newly
-                ]
-                safe_hook("on_files_committed", deps.hooks.on_files_committed, capture_id, hooked)
+                    committed.append(item.path)
+                    newly.append(rec)
+            finally:
+                if newly:
+                    _announce_committed(deps, capture_id, newly)
             maybe_finish_after_commit(deps, deps.state.get_capture(capture_id) or record)
             log_event(
                 "files.committed", n=len(newly), mismatch=len(mismatch), not_found=len(not_found), capture_id=capture_id
@@ -356,3 +346,24 @@ async def complete_multipart(capture_id: str, body: CompleteMultipartRequest, re
         async with deps.locks.acquire(capture_id):
             load_owned(deps, capture_id, principal, write=True)
             raise fail(501, "multipart_not_implemented", "/", "multipart upload is not implemented in this PR")
+
+
+def _announce_committed(deps, capture_id: str, newly: list[FileRecord]) -> None:
+    listed = (extras(deps, capture_id).get("accepted") or {}).get("listed")
+    payload = FilesCommittedData(paths=[rec.path for rec in newly], committed=len(newly), listed=listed).model_dump(
+        mode="json"
+    )
+    emit(deps, capture_id, "files_committed", payload)
+    hooked = [
+        CommittedFile(
+            path=rec.path,
+            role=rec.role,
+            sha256=rec.sha256,
+            bytes=rec.bytes,
+            generation=rec.generation or "",
+            key=rec.key or upload_key(capture_id, rec.path),
+            meta=rec.meta,
+        )
+        for rec in newly
+    ]
+    safe_hook("on_files_committed", deps.hooks.on_files_committed, capture_id, hooked)

backend/server/api/tests/test_commit_lease.py

@@ -0,0 +1,107 @@
+"""A busy file-index lease must not strand committed files unannounced.
+
+ModalDictState merges the per-capture file index under a strict lease; a lease
+that stays busy raises LeaseBusy and the commit route answers a retryable 503.
+Whatever the failed request already committed must still reach
+``files_committed`` and ``on_files_committed``, and whatever it did not commit
+must stay pending so the retry commits and announces it.
+"""
+
+from __future__ import annotations
+
+import time
+from urllib.parse import urlparse
+
+import pytest
+
+from server.api.tests.conftest import sha256_hex
+from server.api.tests.test_ingest import _create, _file
+from server.common.backends import state_modal_dict
+from server.common.backends.state_modal_dict import ModalDictState
+from server.common.lease import LeaseBusy
+
+_BLOBS = {"stills/s1.jpg": b"jpeg-one", "stills/s2.jpg": b"jpeg-two"}
+
+
+@pytest.fixture
+def shared(client, monkeypatch):
+    deps = client.app_state
+    deps.state = ModalDictState()
+    monkeypatch.setattr(state_modal_dict, "_INDEX_LEASE_WAIT_S", 0.05)
+    announced: list[list[str]] = []
+    monkeypatch.setattr(
+        deps.hooks, "on_files_committed", lambda _capture_id, files: announced.append([f.path for f in files])
+    )
+    return deps, announced
+
+
+def _upload_all(client, capture_id: str) -> None:
+    files = [_file(path, data, meta={"still": f"s{index + 1}"}) for index, (path, data) in enumerate(_BLOBS.items())]
+    registered = client.post(f"/v1/captures/{capture_id}/files", json={"files": files})
+    assert registered.status_code == 200, registered.text
+    for item, data in zip(registered.json()["files"], _BLOBS.values(), strict=True):
+        target = item["upload"]
+        put = client.put(urlparse(target["url"]).path, content=data, headers=target["headers"])
+        assert put.status_code == 200, put.text
+
+
+def _commit(client, capture_id: str):
+    return client.post(
+        f"/v1/captures/{capture_id}/files:commit",
+        json={"files": [{"path": path, "sha256": sha256_hex(data)} for path, data in _BLOBS.items()]},
+    )
+
+
+def _committed_events(client, capture_id: str) -> list[str]:
+    events = client.get(f"/v1/captures/{capture_id}/events").json()["events"]
+    return sorted(path for e in events if e["type"] == "files_committed" for path in e["data"]["paths"])
+
+
+def test_a_busy_lease_saves_nothing_and_the_retry_announces(client, shared):
+    deps, announced = shared
+    capture_id = _create(client, "packet-lease-busy")
+    _upload_all(client, capture_id)
+    # Another container holds a fresh lease on this capture's file index.
+    deps.state.put_if_absent(f"lease:files:{capture_id}", {"id": "other", "at": time.time()})
+
+    busy = _commit(client, capture_id)
+    assert busy.status_code == 503, busy.text
+    assert busy.json()["errors"][0]["code"] == "lease_busy"
+    for path in _BLOBS:
+        assert deps.state.get_file(capture_id, path).state == "pending", "nothing may commit without the lease"
+
+    deps.state.delete_kv(f"lease:files:{capture_id}")
+    retried = _commit(client, capture_id)
+    assert retried.status_code == 200, retried.text
+    assert _committed_events(client, capture_id) == sorted(_BLOBS)
+    assert sorted(path for batch in announced for path in batch) == sorted(_BLOBS)
+    assert sorted(rec.path for rec in deps.state.files(capture_id)) == sorted(_BLOBS)
+
+
+def test_files_committed_before_a_busy_lease_are_still_announced(client, shared, monkeypatch):
+    deps, announced = shared
+    capture_id = _create(client, "packet-lease-partial")
+    _upload_all(client, capture_id)
+
+    real_with_lease = state_modal_dict.with_lease
+    calls = {"n": 0}
+
+    def second_commit_busy(state, name, fn, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 2:
+            raise LeaseBusy(f"lease {name!r} stayed busy")
+        return real_with_lease(state, name, fn, **kwargs)
+
+    monkeypatch.setattr(state_modal_dict, "with_lease", second_commit_busy)
+    busy = _commit(client, capture_id)
+    assert busy.status_code == 503, busy.text
+    # The first file committed before the lease went busy: it is announced now,
+    # because the retry will read it as already committed.
+    assert _committed_events(client, capture_id) == ["stills/s1.jpg"]
+    assert deps.state.get_file(capture_id, "stills/s2.jpg").state == "pending"
+
+    monkeypatch.setattr(state_modal_dict, "with_lease", real_with_lease)
+    retried = _commit(client, capture_id)
+    assert retried.status_code == 200, retried.text
+    assert _committed_events(client, capture_id) == sorted(_BLOBS)
+    assert sorted(path for batch in announced for path in batch) == sorted(_BLOBS)

backend/server/api/tests/test_finalize_background.py

@@ -22,8 +22,10 @@ def test_failed_run_start_stays_retryable(client, monkeypatch):
     capture_id, packet = _ready_capture(client, "packet-hook-fail")
     deps = client.app_state
 
+    # A local fake of any orchestrator start_run failure — Modal is the
+    # production transport; this suite pins the retryable handoff contract.
     def boom(*args, **kwargs):
-        raise RuntimeError("modal start_run failed")
+        raise RuntimeError("orchestrator start_run failed")
 
     monkeypatch.setattr(deps.hooks, "on_finalized", boom)
     client.hold_finalize = True
@@ -76,8 +78,56 @@ def offline(_key):
     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.
+
+def test_escaped_failure_publishes_before_releasing_the_worker_slot(client, monkeypatch):
+    """Finding: _guard — the slot was released before the failure was published.
+
+    A finalize retry that lands in the gap sees a free slot, publishes its
+    success, and is then overwritten by the older worker's failure — the healed
+    capture resets to uploading behind a failure event for a run that started.
+    With publication and the run_started check serialized under the finish lock,
+    the older worker observes the retry's success and defers; either order ends
+    consistent. This pins the ordering half: at the moment the failure is
+    published, this worker still owns the slot.
+    """
+    import server.api.finalize as finalize_module
+
+    capture_id, packet = _ready_capture(client, "packet-bg-order")
+    deps = client.app_state
+
+    def offline(_key):
+        raise RuntimeError("volume offline")
+
+    publication_slot_state: list[bool] = []
+    original_failure_path = finalize_module._on_background_failure
+
+    def probing_failure_path(*args, **kwargs):
+        publication_slot_state.append(finalize_module._slots[capture_id].inflight)
+        return original_failure_path(*args, **kwargs)
+
+    monkeypatch.setattr(deps.storage, "get_bytes", offline)
+    monkeypatch.setattr(finalize_module, "_on_background_failure", probing_failure_path)
+    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
+
+    # One storage exception publishes one failure, while this worker still holds
+    # the slot: a retry that starts after the slot frees sees the failure already
+    # published (or the run_started flag of a healed retry), never both.
+    assert publication_slot_state == [True]
+    assert finalize_module._slots[capture_id].inflight is False
+    events = client.get(f"/v1/captures/{capture_id}/events").json()["events"]
+    assert len([e for e in events if e["type"] == "failed" and e["data"].get("code") == "finalize_failed"]) == 1
+    assert deps.state.get_capture(capture_id).status == "uploading"
+
+    # And the healed retry still lands one healthy run.
     monkeypatch.undo()
     second, _raw = _finalize(client, capture_id, packet)
     assert second.status_code == 202, second.text
+    assert extras(deps, capture_id).get("run_started") is True
     assert second.json()["status"] == "processing"

backend/server/common/backends/state_modal_dict.py

@@ -20,6 +20,11 @@
     StageStatus,
 )
 
+# How long upsert_file waits for the per-capture index lease before failing.
+# A module constant so tests can shorten it without sleeping through the real
+# budget.
+_INDEX_LEASE_WAIT_S = 3.0
+
 _OPEN = frozenset({"uploading", "awaiting_files"})
 _APPEND_GUARD = threading.Lock()
 _APPEND_LOCKS: dict[int, threading.Lock] = {}
@@ -335,9 +340,10 @@ def delete_capture(self, capture_id: str) -> None:
         self._kv[tomb] = {"capture_id": capture_id}
 
     def upsert_file(self, rec: FileRecord) -> None:
-        self._kv[f"file:{rec.capture_id}:{rec.path}"] = rec.model_dump(mode="json")
-
         def merge() -> None:
+            # The record is written inside the lease too: a busy lease then saves
+            # nothing, so the caller's retry still sees the file in its old state.
+            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)
@@ -346,7 +352,11 @@ def merge() -> None:
         # 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)
+        #
+        # The merge runs strictly under the lease: a busy lease raises LeaseBusy
+        # (the route turns it into a retryable 503) instead of merging unlocked
+        # while an abandoned-but-valid lease holds the slot.
+        with_lease(self, f"files:{rec.capture_id}", merge, wait_s=_INDEX_LEASE_WAIT_S, strict=True)
 
     def get_file(self, capture_id: str, path: str) -> FileRecord | None:
         return self._model(f"file:{capture_id}:{path}", FileRecord)