diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py index 7b4f0c45d..42ecadd7d 100644 --- a/backend/server/api/finalize.py +++ b/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) diff --git a/backend/server/api/tests/test_finalize_background.py b/backend/server/api/tests/test_finalize_background.py new file mode 100644 index 000000000..ff886dc8a --- /dev/null +++ b/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" diff --git a/backend/server/common/backends/state_modal_dict.py b/backend/server/common/backends/state_modal_dict.py index c6e002f43..8633e0a61 100644 --- a/backend/server/common/backends/state_modal_dict.py +++ b/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) diff --git a/backend/server/common/tests/test_modal_dict_files.py b/backend/server/common/tests/test_modal_dict_files.py new file mode 100644 index 000000000..8f5ee2762 --- /dev/null +++ b/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