perf(observatory): bound source destination planning

src/k1link/observatory/worker_http_transport.py

@@ -109,6 +109,38 @@ class _SourceMember:
     camera_sequence: int | None
 
 
+@dataclass(frozen=True, slots=True)
+class _SourceDestinationLayout:
+    root: Path
+    spatial: Path | None
+    camera_epoch: int
+    camera_epoch_root: Path
+    camera_segments: Path
+
+    def destination(self, member: _SourceMember) -> Path:
+        if member.kind == "source-bundle":
+            return self.root / "source-bundle.json"
+        if member.kind == "source-capability":
+            return self.root / "source-capability.json"
+        if member.kind == "spatial-replay":
+            if member.primary:
+                return self.root / "mqtt.raw.k1mqtt"
+            if self.spatial is None:
+                raise ObservatoryWorkerHttpError(
+                    "spatial source destination is unavailable"
+                )
+            return self.spatial / member.member_id
+        if member.kind == "spatial-replay-metadata":
+            return self.root / "mqtt.metadata.jsonl"
+        if member.camera_epoch != self.camera_epoch:
+            raise ObservatoryWorkerHttpError("camera source member has another epoch")
+        if member.kind == "camera-init":
+            return self.camera_epoch_root / "init.mp4"
+        if member.camera_sequence is None:
+            raise ObservatoryWorkerHttpError("camera segment has no sequence")
+        return self.camera_segments / f"{member.camera_sequence}.m4s"
+
+
 @dataclass(frozen=True, slots=True)
 class _UploadMember:
     member_id: str
@@ -305,9 +337,10 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport):
             / f"generation-{job.claim_generation}"
             / job.source_bundle_sha256
         )
+        layout = _prepare_source_destination_layout(root, members)
         destinations: dict[Path, _SourceMember] = {}
         for member in members:
-            destination = _source_destination(root, member)
+            destination = layout.destination(member)
             if destination in destinations:
                 raise ObservatoryWorkerHttpError(
                     "source materialization members select the same local role"
@@ -316,7 +349,7 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport):
         camera_members = _ordered_camera_epoch_members(members)
         if not all(
             _matches_file(
-                _source_destination(root, member),
+                layout.destination(member),
                 member.sha256,
                 member.byte_length,
             )
@@ -1217,6 +1250,51 @@ def _copy_camera_archive_member(
                 destination.unlink()
 
 
+def _prepare_source_destination_layout(
+    root: Path,
+    members: tuple[_SourceMember, ...],
+) -> _SourceDestinationLayout:
+    camera_inits = tuple(
+        member for member in members if member.kind == "camera-init"
+    )
+    if len(camera_inits) != 1 or camera_inits[0].camera_epoch is None:
+        raise ObservatoryWorkerHttpError(
+            "source materialization camera epoch is incomplete"
+        )
+    camera_epoch = camera_inits[0].camera_epoch
+    camera_root = _secure_source_subdirectory(root, root / "camera")
+    camera_epoch_root = _secure_source_subdirectory(
+        root,
+        camera_root / f"epoch-{camera_epoch}",
+    )
+    camera_segments = _secure_source_subdirectory(
+        root,
+        camera_epoch_root / "segments",
+    )
+    spatial = None
+    if any(
+        member.kind == "spatial-replay" and not member.primary
+        for member in members
+    ):
+        spatial = _secure_source_subdirectory(root, root / "spatial")
+    return _SourceDestinationLayout(
+        root=root,
+        spatial=spatial,
+        camera_epoch=camera_epoch,
+        camera_epoch_root=camera_epoch_root,
+        camera_segments=camera_segments,
+    )
+
+
+def _secure_source_subdirectory(root: Path, candidate: Path) -> Path:
+    resolved = _secure_directory(candidate)
+    if not resolved.is_relative_to(root):
+        raise ObservatoryWorkerHttpError(
+            "source materialization destination escapes its root"
+        )
+    return resolved
+
+
 def _source_destination(root: Path, member: _SourceMember) -> Path:
     if member.kind == "source-bundle":
         return root / "source-bundle.json"

tests/test_observatory_worker_http_transport.py

@@ -11,6 +11,7 @@ from pathlib import Path
 import httpx
 import pytest
 
+import k1link.observatory.worker_http_transport as worker_http_transport_module
 from k1link.observatory.portable_artifact_transport import (
     PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER,
     PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE,
@@ -498,6 +499,67 @@ def test_artifact_get_rejects_non_identity_content_encoding(
             gateway.materialize(job)
 
 
+def test_large_camera_inventory_has_bounded_secure_destination_planning(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    job = _job(
+        bundle_sha256=hashlib.sha256(b"source-bundle").hexdigest(),
+        capability_sha256=hashlib.sha256(b"source-capability").hexdigest(),
+    )
+    manifest, _payloads = _source_contract(job, camera_segment_count=2048)
+    rows = manifest["members"]
+    assert isinstance(rows, list)
+    for ordinal in range(128):
+        row, _payload = _source_member(
+            job,
+            kind="spatial-replay",
+            payload=f"secondary-spatial-{ordinal}".encode("ascii"),
+            artifact_id=f"secondary-spatial-{ordinal}",
+            media_type="application/x-nodedc-k1mqtt",
+        )
+        rows.append(row)
+    rows.sort(key=lambda row: str(row["member_id"]))
+    archive_requested = False
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal archive_requested
+        if request.url.path.endswith("/claims"):
+            return _claim_response()
+        if request.url.path.endswith("/source-materialization"):
+            return httpx.Response(200, json=manifest)
+        if request.url.path.endswith("/source-camera-epoch-archive"):
+            archive_requested = True
+            return httpx.Response(503)
+        pytest.fail("destination planning must finish before member fallback")
+
+    with ObservatoryWorkerHttpGateway(
+        base_url="http://127.0.0.1:18080",
+        bearer_token=BEARER_TOKEN,
+        work_root=tmp_path / "worker",
+        transport=httpx.MockTransport(handler),
+    ) as gateway:
+        _cache_claim(gateway)
+        secure_directory_calls: list[Path] = []
+        original_secure_directory = worker_http_transport_module._secure_directory
+
+        def counting_secure_directory(path: Path) -> Path:
+            secure_directory_calls.append(path)
+            return original_secure_directory(path)
+
+        monkeypatch.setattr(
+            worker_http_transport_module,
+            "_secure_directory",
+            counting_secure_directory,
+        )
+        with pytest.raises(ObservatoryWorkerHttpError, match="HTTP 503"):
+            gateway.materialize(job)
+
+    assert archive_requested is True
+    assert len(secure_directory_calls) <= 7
+    assert sum(path.name == "spatial" for path in secure_directory_calls) == 1
+
+
 def test_http_gateway_packed_camera_epoch_bounds_requests_and_header_timeout(
     tmp_path: Path,
 ) -> None: