From 64dfb7a96aeaa8b971f4eb7458e339b1c6f4d6a2 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:01:55 +0000 Subject: [PATCH] fix(backend): retryable finalize handoff and lease-guarded file-index merge - finish_ready: a failing on_finalized hook now raises ApiFailure(503) so the handoff stays retryable; run_started is only recorded after the hook succeeds. - _guard: exceptions escaping background finalize work flow through the normal failure-event and retry path instead of vanishing; the original exception is still set on the future for test callers. - ModalDictState.upsert_file: merge the per-capture path index under the shared state-store lease (claimed via the backing Dict's atomic skip_if_exists put) so concurrent containers cannot overwrite each other's index entries. - Regression tests for all three: failed run-start stays retryable, background exceptions emit failure events and leave retryable state, and two concurrent writers keep every index entry (the fake modal.Dict pairs concurrent readers onto the same snapshot to force the stale-read interleaving deterministically). --- backend/server/api/finalize.py | 22 +++- .../api/tests/test_finalize_background.py | 83 ++++++++++++ .../common/backends/state_modal_dict.py | 14 +- .../common/tests/test_modal_dict_files.py | 124 ++++++++++++++++++ 4 files changed, 238 insertions(+), 5 deletions(-) create mode 100644 backend/server/api/tests/test_finalize_background.py create mode 100644 backend/server/common/tests/test_modal_dict_files.py 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