From def8c71410ef79b1fdac50eb64bc37b17b220a8e Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Mon, 31 Aug 2026 18:19:04 +0300 Subject: [PATCH] perf(observatory): bound source destination planning --- .../observatory/worker_http_transport.py | 82 ++++++++++++++++++- .../test_observatory_worker_http_transport.py | 62 ++++++++++++++ 2 files changed, 142 insertions(+), 2 deletions(-) diff --git a/src/k1link/observatory/worker_http_transport.py b/src/k1link/observatory/worker_http_transport.py index c72ebc7..17423e7 100644 --- a/src/k1link/observatory/worker_http_transport.py +++ b/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" diff --git a/tests/test_observatory_worker_http_transport.py b/tests/test_observatory_worker_http_transport.py index ad74ea5..07bc826 100644 --- a/tests/test_observatory_worker_http_transport.py +++ b/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: