perf(observatory): avoid duplicate camera verification

src/k1link/observatory/worker_http_transport.py

@@ -347,21 +347,24 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport):
                 )
             destinations[destination] = member
         camera_members = _ordered_camera_epoch_members(members)
-        if not all(
+        camera_epoch_ready = all(
             _matches_file(
                 layout.destination(member),
                 member.sha256,
                 member.byte_length,
             )
             for member in camera_members
-        ):
+        )
-            self._download_camera_epoch_archive(
+        if not camera_epoch_ready:
+            camera_epoch_ready = self._download_camera_epoch_archive(
                 job=job,
                 context=context,
                 members=camera_members,
-                root=root,
+                layout=layout,
             )
         for destination, member in destinations.items():
+            if camera_epoch_ready and member.kind in {"camera-init", "camera-segment"}:
+                continue
             if _matches_file(destination, member.sha256, member.byte_length):
                 continue
             self._download_member(
@@ -526,8 +529,8 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport):
         job: SealedObservatoryRecordedJob,
         context: _ClaimContext,
         members: tuple[_SourceMember, ...],
-        root: Path,
+        layout: _SourceDestinationLayout,
-    ) -> None:
+    ) -> bool:
         transfer_root = _secure_directory(self._work_root / ".source-transfers")
         temporary = transfer_root / f".camera-epoch-{secrets.token_hex(16)}.tar"
         descriptor = -1
@@ -547,7 +550,7 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport):
                 ),
             ) as response:
                 if response.status_code == 404:
-                    return
+                    return False
                 self._raise_for_status(response)
                 content_encoding = response.headers.get("content-encoding")
                 if (
@@ -633,9 +636,10 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport):
                 )
             _extract_camera_epoch_archive(
                 temporary,
-                root=root,
+                layout=layout,
                 members=members,
             )
+            return True
         except ObservatoryWorkerHttpError:
             raise
         except httpx.HTTPError as exc:
@@ -1115,7 +1119,7 @@ def _camera_epoch_archive_byte_length(members: tuple[_SourceMember, ...]) -> int
 def _extract_camera_epoch_archive(
     archive_path: Path,
     *,
-    root: Path,
+    layout: _SourceDestinationLayout,
     members: tuple[_SourceMember, ...],
 ) -> None:
     expected_bytes = _camera_epoch_archive_byte_length(members)
@@ -1192,7 +1196,7 @@ def _extract_camera_epoch_archive(
         for temporary, member in zip(staged, members, strict=True):
             _publish_local_file(
                 temporary,
-                _source_destination(root, member),
+                layout.destination(member),
                 member.sha256,
                 member.byte_length,
             )
@@ -1225,6 +1229,7 @@ def _copy_camera_archive_member(
     )
     digest = hashlib.sha256()
     byte_length = 0
+    verified = False
     try:
         with os.fdopen(descriptor, "wb") as stream:
             descriptor = -1
@@ -1242,10 +1247,11 @@ def _copy_camera_archive_member(
             raise ObservatoryWorkerHttpError(
                 "camera epoch archive member content changed"
             )
+        verified = True
     finally:
         if descriptor >= 0:
             os.close(descriptor)
-        if not _matches_file(destination, expected_sha256, expected_byte_length):
+        if not verified:
             with suppress(OSError):
                 destination.unlink()
 
@@ -1295,27 +1301,6 @@ def _secure_source_subdirectory(root: Path, candidate: Path) -> Path:
     return resolved
 
 
-def _source_destination(root: Path, member: _SourceMember) -> Path:
-    if member.kind == "source-bundle":
-        return root / "source-bundle.json"
-    if member.kind == "source-capability":
-        return root / "source-capability.json"
-    if member.kind == "spatial-replay":
-        if member.primary:
-            return root / "mqtt.raw.k1mqtt"
-        return _secure_directory(root / "spatial") / member.member_id
-    if member.kind == "spatial-replay-metadata":
-        return root / "mqtt.metadata.jsonl"
-    if member.camera_epoch is None:
-        raise ObservatoryWorkerHttpError("camera source member has no epoch")
-    epoch = _secure_directory(root / "camera" / f"epoch-{member.camera_epoch}")
-    if member.kind == "camera-init":
-        return epoch / "init.mp4"
-    if member.camera_sequence is None:
-        raise ObservatoryWorkerHttpError("camera segment has no sequence")
-    return _secure_directory(epoch / "segments") / f"{member.camera_sequence}.m4s"
-
-
 def _upload_members(
     document: Mapping[str, object],
     job: SealedObservatoryRecordedJob,

tests/test_observatory_worker_http_transport.py

@@ -562,6 +562,7 @@ def test_large_camera_inventory_has_bounded_secure_destination_planning(
 
 def test_http_gateway_packed_camera_epoch_bounds_requests_and_header_timeout(
     tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     job = _job(
         bundle_sha256=hashlib.sha256(b"source-bundle").hexdigest(),
@@ -576,6 +577,35 @@ def test_http_gateway_packed_camera_epoch_bounds_requests_and_header_timeout(
     artifact_requests: list[httpx.Request] = []
     camera_member_requests: list[str] = []
     archive_read_timeouts: list[float] = []
+    secure_directory_calls: list[Path] = []
+    existing_hash_paths: list[Path] = []
+    original_secure_directory = worker_http_transport_module._secure_directory
+    original_hash_local_regular_file = (
+        worker_http_transport_module._hash_local_regular_file
+    )
+
+    def counting_secure_directory(path: Path) -> Path:
+        secure_directory_calls.append(path)
+        return original_secure_directory(path)
+
+    def counting_hash_local_regular_file(
+        path: Path,
+        maximum_bytes: int,
+    ) -> tuple[str, int]:
+        if path.exists():
+            existing_hash_paths.append(path)
+        return original_hash_local_regular_file(path, maximum_bytes)
+
+    monkeypatch.setattr(
+        worker_http_transport_module,
+        "_secure_directory",
+        counting_secure_directory,
+    )
+    monkeypatch.setattr(
+        worker_http_transport_module,
+        "_hash_local_regular_file",
+        counting_hash_local_regular_file,
+    )
     rows = manifest["members"]
     assert isinstance(rows, list)
     camera_member_ids = {
@@ -627,12 +657,43 @@ def test_http_gateway_packed_camera_epoch_bounds_requests_and_header_timeout(
     ) as gateway:
         _cache_claim(gateway)
         stage = gateway.materialize(job)
+        first_materialization_secure_calls = tuple(secure_directory_calls)
+        assert existing_hash_paths == []
+        existing_hash_paths.clear()
+        resumed_stage = gateway.materialize(job)
 
+    assert resumed_stage == stage
     assert archive_read_timeouts == [
         WORKER_HTTP_CAMERA_ARCHIVE_READ_TIMEOUT_SECONDS
     ]
     assert camera_member_requests == []
-    assert len(artifact_requests) == 6
+    assert len(artifact_requests) == 7
+    camera_root = stage.root / "camera"
+    camera_hash_paths = [
+        path for path in existing_hash_paths if path.is_relative_to(camera_root)
+    ]
+    assert len(camera_hash_paths) == 102
+    assert set(camera_hash_paths) == {
+        stage.root / "camera/epoch-1/init.mp4",
+        *(
+            stage.root / f"camera/epoch-1/segments/{sequence}.m4s"
+            for sequence in range(1, 102)
+        ),
+    }
+    assert (
+        sum(
+            path == stage.root / "camera/epoch-1"
+            for path in first_materialization_secure_calls
+        )
+        == 1
+    )
+    assert (
+        sum(
+            path == stage.root / "camera/epoch-1/segments"
+            for path in first_materialization_secure_calls
+        )
+        == 1
+    )
     assert (
         stage.root / "camera/epoch-1/segments/101.m4s"
     ).read_bytes() == b"sealed-camera-segment-101"