diff --git a/src/k1link/observatory/worker_http_transport.py b/src/k1link/observatory/worker_http_transport.py index 17423e7..e2a00bf 100644 --- a/src/k1link/observatory/worker_http_transport.py +++ b/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, - ) -> None: + layout: _SourceDestinationLayout, + ) -> 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, diff --git a/tests/test_observatory_worker_http_transport.py b/tests/test_observatory_worker_http_transport.py index 07bc826..6445c60 100644 --- a/tests/test_observatory_worker_http_transport.py +++ b/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"