From e9bbfb9a412593e6359bab855daf64316c27597e Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Thu, 30 Jul 2026 21:52:10 +0300 Subject: [PATCH] perf(data): bound lidar readers and lab session loading --- .../observation/useObservationSessions.ts | 6 +- .../test/observationSessions.test.mjs | 3 + src/k1link/compute/lidar_field_review.py | 19 +- src/k1link/compute/lidar_local_surface.py | 124 +++++-- src/k1link/web/app.py | 6 + src/k1link/web/lidar_api.py | 103 ++---- src/k1link/web/lidar_local_surface_service.py | 346 ++++++++++++++++++ src/k1link/web/session_api.py | 12 - tests/test_lidar_local_surface.py | 85 +++++ tests/test_session_api.py | 10 +- 10 files changed, 580 insertions(+), 134 deletions(-) create mode 100644 src/k1link/web/lidar_local_surface_service.py diff --git a/apps/control-station/src/core/observation/useObservationSessions.ts b/apps/control-station/src/core/observation/useObservationSessions.ts index 729790d..62bc612 100644 --- a/apps/control-station/src/core/observation/useObservationSessions.ts +++ b/apps/control-station/src/core/observation/useObservationSessions.ts @@ -385,6 +385,7 @@ export function clearObservationReplayPreparation( export function useObservationSessions({ limit = 100, scope = "all", + pollingEnabled = true, replayEnabled = true, onReplayBegin, onReplayAccepted, @@ -393,6 +394,7 @@ export function useObservationSessions({ }: { limit?: number; scope?: ObservationSessionScope; + pollingEnabled?: boolean; replayEnabled?: boolean; /** Called only after the archive is ready, immediately before replacing the old viewer. */ onReplayBegin?: ( @@ -475,7 +477,7 @@ export function useObservationSessions({ }, [refresh]); useEffect(() => { - if (state !== "ready") return; + if (!pollingEnabled || state !== "ready") return; let disposed = false; let timer = 0; const poll = async () => { @@ -488,7 +490,7 @@ export function useObservationSessions({ disposed = true; window.clearTimeout(timer); }; - }, [loadCatalog, state]); + }, [loadCatalog, pollingEnabled, state]); const executeReplay = useCallback(async ( session: ObservationSessionSummary, diff --git a/apps/control-station/test/observationSessions.test.mjs b/apps/control-station/test/observationSessions.test.mjs index df6ff83..3c0b6a9 100644 --- a/apps/control-station/test/observationSessions.test.mjs +++ b/apps/control-station/test/observationSessions.test.mjs @@ -206,6 +206,9 @@ test("data recordings keep the compact session dropdown and laboratory results s assert.match(laboratorySource, /e29-camera-geometry/); assert.match(laboratorySource, /laboratoryWorkOrdinal\(right\.label\)/); assert.match(laboratorySource, /initialWorkSelectedRef/); + assert.match(laboratorySource, /pollingEnabled:\s*false/); + assert.match(laboratorySource, /useAdvancedLaboratoryCatalog/); + assert.doesNotMatch(laboratorySource, /fetchAdvancedLaboratoryResults/); }); test("recording preparation statuses share the viewer's left alignment", async () => { diff --git a/src/k1link/compute/lidar_field_review.py b/src/k1link/compute/lidar_field_review.py index 76698c0..7aac008 100644 --- a/src/k1link/compute/lidar_field_review.py +++ b/src/k1link/compute/lidar_field_review.py @@ -113,6 +113,17 @@ class E10LidarFieldSource: """Strict reader for the immutable, intensity-free RAVNOVES00 E10 pack.""" def __init__(self, root: Path) -> None: + self._open(root, verify_content=True) + + @classmethod + def _restore_validated_generation(cls, root: Path) -> E10LidarFieldSource: + """Open a generation already admitted by the host validation cache.""" + + instance = cls.__new__(cls) + instance._open(root, verify_content=False) + return instance + + def _open(self, root: Path, *, verify_content: bool) -> None: candidate = root.expanduser().absolute() if candidate.is_symlink(): raise LidarGroundError("E10 LiDAR source cannot be a symlink") @@ -140,12 +151,16 @@ class E10LidarFieldSource: arrays_path.is_symlink() or not arrays_path.is_file() or arrays_path.stat().st_size != artifact.get("byte_length") - or _sha256(arrays_path) != artifact.get("sha256") + or ( + verify_content + and _sha256(arrays_path) != artifact.get("sha256") + ) ): raise LidarGroundError("E10 LiDAR source artifact is invalid") self.arrays = np.load(arrays_path, allow_pickle=False) try: - self._validate_arrays() + if verify_content: + self._validate_arrays() except BaseException: self.close() raise diff --git a/src/k1link/compute/lidar_local_surface.py b/src/k1link/compute/lidar_local_surface.py index 3b5013e..7730907 100644 --- a/src/k1link/compute/lidar_local_surface.py +++ b/src/k1link/compute/lidar_local_surface.py @@ -224,6 +224,17 @@ class K1LocalSurfaceV1: """Strict reader for source-aligned, passive K1 local-surface evidence.""" def __init__(self, root: Path) -> None: + self._open(root, verify_content=True) + + @classmethod + def _restore_validated_generation(cls, root: Path) -> K1LocalSurfaceV1: + """Open a generation already admitted by the host validation cache.""" + + instance = cls.__new__(cls) + instance._open(root, verify_content=False) + return instance + + def _open(self, root: Path, *, verify_content: bool) -> None: candidate = root.expanduser().absolute() if candidate.is_symlink(): raise LidarGroundError("K1 local-surface artifact cannot be a symlink") @@ -243,11 +254,18 @@ class K1LocalSurfaceV1: or self.manifest.get("model_id") != self.root.name ): raise LidarGroundError("K1 local-surface identity is invalid") - artifacts = _validate_artifacts(self.root, self.manifest.get("artifacts")) + artifacts = _validate_artifacts( + self.root, + self.manifest.get("artifacts"), + verify_digests=verify_content, + ) self.arrays = np.load(artifacts["local-surface"], allow_pickle=False) self.report = _read_json(artifacts["local-surface-report"]) try: - self._validate() + if verify_content: + self._validate() + else: + self._restore_capabilities() except BaseException: self.close() raise @@ -256,6 +274,34 @@ class K1LocalSurfaceV1: def close(self) -> None: self.arrays.close() + def _restore_capabilities(self) -> None: + """Restore derived reader flags without touching large array payloads.""" + + files = set(self.arrays.files) + qualification = { + "prediction_available", + "prediction_cell_count", + "prediction_residual_p50_m", + "prediction_residual_p95_m", + "prediction_inlier_fraction", + "height_delta_m", + "slope_delta_deg", + "roughness_delta_m", + "temporal_compared", + "temporal_jump", + "step_candidate_cell_count", + "step_candidate_point_count", + "point_step_candidate", + } + prediction_evidence = { + "prediction_prior_plane_coefficients_map", + "prediction_cell_offsets", + "prediction_cell_points_map", + "prediction_cell_signed_residual_m", + } + self.has_temporal_qualification = qualification <= files + self.has_prediction_evidence = prediction_evidence <= files + def _validate(self) -> None: frame_count = _nonnegative_int(self.identity.get("frame_count"), "frame count") point_count = _nonnegative_int(self.identity.get("point_count"), "point count") @@ -835,21 +881,30 @@ class K1LocalSurfaceV1: height_threshold = float(criteria["surface_height_jump_m"]) slope_threshold = float(criteria["surface_slope_jump_deg"]) roughness_threshold = float(criteria["surface_roughness_jump_m"]) + prediction_available_values = self.arrays["prediction_available"] + prediction_p50_values = self.arrays["prediction_residual_p50_m"] + prediction_p95_values = self.arrays["prediction_residual_p95_m"] + prediction_inlier_values = self.arrays["prediction_inlier_fraction"] + temporal_compared_values = self.arrays["temporal_compared"] + height_delta_values = self.arrays["height_delta_m"] + slope_delta_values = self.arrays["slope_delta_deg"] + roughness_delta_values = self.arrays["roughness_delta_m"] + sensor_height_values = self.arrays["sensor_height_m"] + slope_values = self.arrays["slope_deg"] + roughness_values = self.arrays["roughness_m"] + confidence_values = self.arrays["confidence"] + step_point_values = self.arrays["step_candidate_point_count"] + source_frame_indices = source.arrays["source_frame_indices"] + session_seconds = source.arrays["session_seconds"] chronological: list[dict[str, object]] = [] last_review_frame: int | None = None episode_index = 0 for frame_index in range(source.frame_count): reasons: list[str] = [] ratios: list[float] = [] - prediction_available = bool( - self.arrays["prediction_available"][frame_index] - ) - prediction_p95 = float( - self.arrays["prediction_residual_p95_m"][frame_index] - ) - prediction_inlier = float( - self.arrays["prediction_inlier_fraction"][frame_index] - ) + prediction_available = bool(prediction_available_values[frame_index]) + prediction_p95 = float(prediction_p95_values[frame_index]) + prediction_inlier = float(prediction_inlier_values[frame_index]) if prediction_available and prediction_p95 >= tail_threshold: reasons.append("prediction-tail") ratios.append(prediction_p95 / tail_threshold) @@ -857,10 +912,10 @@ class K1LocalSurfaceV1: reasons.append("prediction-inlier-drop") ratios.append((1.0 - prediction_inlier) / (1.0 - inlier_floor)) - temporal_compared = bool(self.arrays["temporal_compared"][frame_index]) - height_delta = float(self.arrays["height_delta_m"][frame_index]) - slope_delta = float(self.arrays["slope_delta_deg"][frame_index]) - roughness_delta = float(self.arrays["roughness_delta_m"][frame_index]) + temporal_compared = bool(temporal_compared_values[frame_index]) + height_delta = float(height_delta_values[frame_index]) + slope_delta = float(slope_delta_values[frame_index]) + roughness_delta = float(roughness_delta_values[frame_index]) if temporal_compared and height_delta >= height_threshold: reasons.append("surface-height-jump") ratios.append(height_delta / height_threshold) @@ -882,12 +937,8 @@ class K1LocalSurfaceV1: { "rank": 0, "frame_index": frame_index, - "source_frame_index": int( - source.arrays["source_frame_indices"][frame_index] - ), - "session_seconds": float( - source.arrays["session_seconds"][frame_index] - ), + "source_frame_index": int(source_frame_indices[frame_index]), + "session_seconds": float(session_seconds[frame_index]), "episode_id": f"episode-{episode_index:02d}", "attention": ( "high" @@ -898,9 +949,7 @@ class K1LocalSurfaceV1: "reasons": reasons, "prediction": { "available": prediction_available, - "residual_p50_m": float( - self.arrays["prediction_residual_p50_m"][frame_index] - ), + "residual_p50_m": float(prediction_p50_values[frame_index]), "residual_p95_m": prediction_p95, "inlier_fraction": prediction_inlier, }, @@ -911,20 +960,12 @@ class K1LocalSurfaceV1: "roughness_delta_m": roughness_delta, }, "surface": { - "sensor_height_m": float( - self.arrays["sensor_height_m"][frame_index] - ), - "slope_deg": float(self.arrays["slope_deg"][frame_index]), - "roughness_m": float( - self.arrays["roughness_m"][frame_index] - ), - "confidence": float( - self.arrays["confidence"][frame_index] - ), + "sensor_height_m": float(sensor_height_values[frame_index]), + "slope_deg": float(slope_values[frame_index]), + "roughness_m": float(roughness_values[frame_index]), + "confidence": float(confidence_values[frame_index]), }, - "step_candidate_point_count": int( - self.arrays["step_candidate_point_count"][frame_index] - ), + "step_candidate_point_count": int(step_point_values[frame_index]), } ) items = sorted( @@ -2025,7 +2066,12 @@ def _logical_sha256(arrays: Mapping[str, npt.NDArray[Any]]) -> str: return digest.hexdigest() -def _validate_artifacts(root: Path, value: object) -> dict[str, Path]: +def _validate_artifacts( + root: Path, + value: object, + *, + verify_digests: bool = True, +) -> dict[str, Path]: artifacts = _list(value, "K1 local-surface artifacts") resolved: dict[str, Path] = {} for value in artifacts: @@ -2048,7 +2094,7 @@ def _validate_artifacts(root: Path, value: object) -> dict[str, Path]: path.is_symlink() or not path.is_file() or path.stat().st_size != item["byte_length"] - or _sha256(path) != item["sha256"] + or (verify_digests and _sha256(path) != item["sha256"]) ): raise LidarGroundError("K1 local-surface artifact is invalid") resolved[role] = path diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index a5bc81b..7a1cb71 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -43,6 +43,7 @@ from k1link.web.e30_review_api import build_e30_review_router from k1link.web.environment_api import build_environment_router from k1link.web.laboratory_api import build_laboratory_router from k1link.web.lidar_api import build_lidar_router +from k1link.web.lidar_local_surface_service import K1LocalSurfaceReadService from k1link.web.map_api import ( MapGatewayConfiguration, MapGatewayProxy, @@ -90,6 +91,9 @@ plugin_catalog: DevicePluginCatalog = plugin_environment.catalog plugin_dispatcher: DevicePluginDispatcher = plugin_environment.dispatcher session_store = SessionStore(REPOSITORY_ROOT) session_artifact_gateway = configured_artifact_gateway(session_store.data_dir) +lidar_local_surface_read_service = K1LocalSurfaceReadService( + session_store.data_dir / "lidar-read-cache" +) session_recording_materializer = SessionRecordingMaterializer( session_store.data_dir, exporters=plugin_environment.recording_exporters, @@ -308,6 +312,7 @@ async def app_lifespan(_: FastAPI) -> AsyncIterator[None]: with suppress(asyncio.CancelledError): await reconciler await asyncio.to_thread(session_recording_preparation_manager.close) + await asyncio.to_thread(lidar_local_surface_read_service.close) plugin_environment.close() @@ -493,6 +498,7 @@ app.include_router( dataset_ground_preview_provider=lambda: ( REPOSITORY_ROOT / ".runtime" / "dataset-gateway" / "ground-comparison.json" ), + local_surface_read_service=lidar_local_surface_read_service, ) ) app.include_router( diff --git a/src/k1link/web/lidar_api.py b/src/k1link/web/lidar_api.py index 1f97522..d89ac77 100644 --- a/src/k1link/web/lidar_api.py +++ b/src/k1link/web/lidar_api.py @@ -9,14 +9,11 @@ from typing import Annotated, Any, Final from fastapi import APIRouter, HTTPException, Query, Response from k1link.compute import ( - E10LidarFieldSource, - K1LocalSurfaceV1, LidarFieldReviewV1, LidarGroundBenchmarkV1, LidarGroundError, LidarReplayError, LidarReplayPackV2, - k1_local_surface_catalog_item, lidar_field_review_catalog_item, lidar_ground_benchmark_catalog_item, lidar_ground_frame_detail, @@ -33,6 +30,10 @@ from k1link.datasets import ( read_dataset_ground_preview, read_dataset_native_scan_preview, ) +from k1link.web.lidar_local_surface_service import ( + K1LocalSurfaceReadService, + LocalSurfaceSourceUnavailable, +) LIDAR_CATALOG_SCHEMA: Final = "missioncore.lidar-replay-pack-catalog/v1" LIDAR_GROUND_CATALOG_SCHEMA: Final = "missioncore.lidar-ground-benchmark-catalog/v1" @@ -42,7 +43,6 @@ _PACK_ID = re.compile(r"^lidar-replay-pack-[a-f0-9]{64}$") _BENCHMARK_ID = re.compile(r"^ground-benchmark-[a-f0-9]{64}$") _FIELD_REVIEW_ID = re.compile(r"^lidar-field-review-[a-f0-9]{64}$") _LOCAL_SURFACE_ID = re.compile(r"^k1-local-surface-[a-f0-9]{64}$") -_E10_PACK_ID = re.compile(r"^e10-lidar-pack-[a-f0-9]{64}$") RootProvider = Callable[[], Path | None] DatasetArtifactProvider = Callable[[], Path | None] @@ -84,8 +84,10 @@ def build_lidar_router( dataset_rellis_preview_provider: DatasetArtifactProvider = lambda: None, dataset_rellis_admission_provider: DatasetArtifactProvider = lambda: None, dataset_ground_preview_provider: DatasetArtifactProvider = configured_dataset_ground_preview, + local_surface_read_service: K1LocalSurfaceReadService | None = None, ) -> APIRouter: router = APIRouter(prefix="/api/v1/lidar", tags=["lidar"]) + surface_reader = local_surface_read_service or K1LocalSurfaceReadService() @router.get("/dataset-gateway") def get_dataset_gateway() -> dict[str, object]: @@ -531,11 +533,7 @@ def build_lidar_router( if len(items) >= limit: break try: - model = K1LocalSurfaceV1(candidate) - try: - items.append(k1_local_surface_catalog_item(model)) - finally: - model.close() + items.append(surface_reader.catalog_item(candidate)) except (LidarGroundError, OSError): invalid_total += 1 return { @@ -575,29 +573,12 @@ def build_lidar_router( detail="K1 local-surface model не найден", ) try: - model = K1LocalSurfaceV1(model_path) - try: - source_pack_id = model.identity.get("source_pack_id") - if ( - not isinstance(source_pack_id, str) - or _E10_PACK_ID.fullmatch(source_pack_id) is None - ): - raise LidarGroundError("K1 local-surface source id is invalid") - source_path = source_root / source_pack_id - if not source_path.is_dir(): - raise HTTPException( - status_code=404, - detail="Связанный E10 LiDAR source не найден", - ) - source = E10LidarFieldSource(source_path) - try: - return model.timeline_detail(source) - finally: - source.close() - finally: - model.close() - except HTTPException: - raise + return surface_reader.timeline_detail(model_path, source_root) + except LocalSurfaceSourceUnavailable as exc: + raise HTTPException( + status_code=404, + detail="Связанный E10 LiDAR source не найден", + ) from exc except (LidarGroundError, OSError) as exc: raise HTTPException( status_code=409, @@ -630,29 +611,12 @@ def build_lidar_router( detail="K1 local-surface model не найден", ) try: - model = K1LocalSurfaceV1(model_path) - try: - source_pack_id = model.identity.get("source_pack_id") - if ( - not isinstance(source_pack_id, str) - or _E10_PACK_ID.fullmatch(source_pack_id) is None - ): - raise LidarGroundError("K1 local-surface source id is invalid") - source_path = source_root / source_pack_id - if not source_path.is_dir(): - raise HTTPException( - status_code=404, - detail="Связанный E10 LiDAR source не найден", - ) - source = E10LidarFieldSource(source_path) - try: - return model.review_detail(source) - finally: - source.close() - finally: - model.close() - except HTTPException: - raise + return surface_reader.review_detail(model_path, source_root) + except LocalSurfaceSourceUnavailable as exc: + raise HTTPException( + status_code=404, + detail="Связанный E10 LiDAR source не найден", + ) from exc except (LidarGroundError, OSError) as exc: raise HTTPException( status_code=409, @@ -688,34 +652,17 @@ def build_lidar_router( detail="K1 local-surface model не найден", ) try: - model = K1LocalSurfaceV1(model_path) - try: - source_pack_id = model.identity.get("source_pack_id") - if ( - not isinstance(source_pack_id, str) - or _E10_PACK_ID.fullmatch(source_pack_id) is None - ): - raise LidarGroundError("K1 local-surface source id is invalid") - source_path = source_root / source_pack_id - if not source_path.is_dir(): - raise HTTPException( - status_code=404, - detail="Связанный E10 LiDAR source не найден", - ) - source = E10LidarFieldSource(source_path) - try: - return model.frame_detail(source, frame_index) - finally: - source.close() - finally: - model.close() + return surface_reader.frame_detail(model_path, source_root, frame_index) except IndexError as exc: raise HTTPException( status_code=404, detail="K1 local-surface frame не найден", ) from exc - except HTTPException: - raise + except LocalSurfaceSourceUnavailable as exc: + raise HTTPException( + status_code=404, + detail="Связанный E10 LiDAR source не найден", + ) from exc except (LidarGroundError, OSError) as exc: raise HTTPException( status_code=409, diff --git a/src/k1link/web/lidar_local_surface_service.py b/src/k1link/web/lidar_local_surface_service.py new file mode 100644 index 0000000..da82a9e --- /dev/null +++ b/src/k1link/web/lidar_local_surface_service.py @@ -0,0 +1,346 @@ +"""Bounded, restart-safe readers for immutable E28 laboratory evidence.""" + +from __future__ import annotations + +import json +import os +import re +import stat +import threading +from collections import OrderedDict +from dataclasses import dataclass +from pathlib import Path +from typing import Final +from uuid import uuid4 + +from k1link.compute.lidar_field_review import E10LidarFieldSource +from k1link.compute.lidar_ground import LidarGroundError +from k1link.compute.lidar_local_surface import ( + K1_LOCAL_SURFACE_ARRAYS_NAME, + K1_LOCAL_SURFACE_MANIFEST_NAME, + K1_LOCAL_SURFACE_REPORT_NAME, + K1LocalSurfaceV1, + k1_local_surface_catalog_item, +) + +VALIDATION_CACHE_SCHEMA: Final = "missioncore.lidar-read-validation-cache/v1" +E10_LIDAR_ARRAYS_NAME: Final = "lidar-pack.npz" +E10_LIDAR_MANIFEST_NAME: Final = "manifest.json" +DEFAULT_READER_CACHE_ENTRIES: Final = 2 + +_E10_PACK_ID = re.compile(r"^e10-lidar-pack-[a-f0-9]{64}$") +_MAX_PROOF_BYTES = 64 * 1024 +_FileIdentity = tuple[int, int, int, int, int] +_Generation = tuple[tuple[str, _FileIdentity], ...] + + +class LocalSurfaceSourceUnavailable(LidarGroundError): + """The immutable model points to a source pack absent from this host.""" + + +@dataclass(slots=True) +class _ModelEntry: + generation: _Generation + reader: K1LocalSurfaceV1 + + +@dataclass(slots=True) +class _SourceEntry: + generation: _Generation + reader: E10LidarFieldSource + + +class K1LocalSurfaceReadService: + """Own strict admission, durable validation proofs, and bounded NPZ handles. + + A changed generation is always read by the strict compute reader first. + Once admitted, later process starts may restore that exact inode/stat + generation using the private proof without hashing or scanning the large + arrays again. Public read operations remain serialized so parallel LAB + bootstrap requests cannot multiply disk and memory pressure. + """ + + def __init__( + self, + cache_root: Path | None = None, + *, + max_entries: int = DEFAULT_READER_CACHE_ENTRIES, + ) -> None: + if max_entries < 1: + raise ValueError("LiDAR reader cache must retain at least one entry") + self.cache_root = ( + cache_root.expanduser().absolute() if cache_root is not None else None + ) + self.max_entries = max_entries + self._lock = threading.RLock() + self._models: OrderedDict[Path, _ModelEntry] = OrderedDict() + self._sources: OrderedDict[Path, _SourceEntry] = OrderedDict() + + def close(self) -> None: + with self._lock: + for entry in self._models.values(): + entry.reader.close() + for source_entry in self._sources.values(): + source_entry.reader.close() + self._models.clear() + self._sources.clear() + + def catalog_item(self, model_path: Path) -> dict[str, object]: + with self._lock: + return k1_local_surface_catalog_item(self._model(model_path)) + + def timeline_detail( + self, + model_path: Path, + source_root: Path, + ) -> dict[str, object]: + with self._lock: + model, source = self._bound_readers(model_path, source_root) + return model.timeline_detail(source) + + def review_detail( + self, + model_path: Path, + source_root: Path, + ) -> dict[str, object]: + with self._lock: + model, source = self._bound_readers(model_path, source_root) + return model.review_detail(source) + + def frame_detail( + self, + model_path: Path, + source_root: Path, + frame_index: int, + ) -> dict[str, object]: + with self._lock: + model, source = self._bound_readers(model_path, source_root) + return model.frame_detail(source, frame_index) + + def _bound_readers( + self, + model_path: Path, + source_root: Path, + ) -> tuple[K1LocalSurfaceV1, E10LidarFieldSource]: + model = self._model(model_path) + source_pack_id = model.identity.get("source_pack_id") + if ( + not isinstance(source_pack_id, str) + or _E10_PACK_ID.fullmatch(source_pack_id) is None + ): + raise LidarGroundError("K1 local-surface source id is invalid") + source_path = source_root / source_pack_id + if not source_path.is_dir(): + raise LocalSurfaceSourceUnavailable("linked E10 LiDAR source is unavailable") + return model, self._source(source_path) + + def _model(self, path: Path) -> K1LocalSurfaceV1: + root = path.expanduser().absolute() + generation = _generation( + root, + ( + K1_LOCAL_SURFACE_MANIFEST_NAME, + K1_LOCAL_SURFACE_REPORT_NAME, + K1_LOCAL_SURFACE_ARRAYS_NAME, + ), + ) + cached = self._models.get(root) + if cached is not None and cached.generation == generation: + self._models.move_to_end(root) + return cached.reader + if cached is not None: + cached.reader.close() + del self._models[root] + + proof_hit = self._proof_matches("models", root.name, generation) + reader = ( + K1LocalSurfaceV1._restore_validated_generation(root) + if proof_hit + else K1LocalSurfaceV1(root) + ) + try: + stable_generation = _generation( + root, + ( + K1_LOCAL_SURFACE_MANIFEST_NAME, + K1_LOCAL_SURFACE_REPORT_NAME, + K1_LOCAL_SURFACE_ARRAYS_NAME, + ), + ) + if stable_generation != generation: + raise LidarGroundError("K1 local-surface generation changed during admission") + if not proof_hit: + self._publish_proof("models", root.name, stable_generation) + except BaseException: + reader.close() + raise + self._models[root] = _ModelEntry(stable_generation, reader) + self._evict_models() + return reader + + def _source(self, path: Path) -> E10LidarFieldSource: + root = path.expanduser().absolute() + generation = _generation( + root, + (E10_LIDAR_MANIFEST_NAME, E10_LIDAR_ARRAYS_NAME), + ) + cached = self._sources.get(root) + if cached is not None and cached.generation == generation: + self._sources.move_to_end(root) + return cached.reader + if cached is not None: + cached.reader.close() + del self._sources[root] + + proof_hit = self._proof_matches("sources", root.name, generation) + reader = ( + E10LidarFieldSource._restore_validated_generation(root) + if proof_hit + else E10LidarFieldSource(root) + ) + try: + stable_generation = _generation( + root, + (E10_LIDAR_MANIFEST_NAME, E10_LIDAR_ARRAYS_NAME), + ) + if stable_generation != generation: + raise LidarGroundError("E10 LiDAR source generation changed during admission") + if not proof_hit: + self._publish_proof("sources", root.name, stable_generation) + except BaseException: + reader.close() + raise + self._sources[root] = _SourceEntry(stable_generation, reader) + self._evict_sources() + return reader + + def _proof_matches( + self, + kind: str, + artifact_id: str, + generation: _Generation, + ) -> bool: + if self.cache_root is None: + return False + path = self.cache_root / kind / f"{artifact_id}.json" + try: + metadata = path.lstat() + if ( + stat.S_ISLNK(metadata.st_mode) + or not stat.S_ISREG(metadata.st_mode) + or metadata.st_size > _MAX_PROOF_BYTES + ): + return False + document: object = json.loads(path.read_text(encoding="utf-8")) + except (FileNotFoundError, OSError, json.JSONDecodeError): + return False + return document == _proof_document(kind, artifact_id, generation) + + def _publish_proof( + self, + kind: str, + artifact_id: str, + generation: _Generation, + ) -> None: + if self.cache_root is None: + return + root = _private_directory(self.cache_root) + destination_root = _private_directory(root / kind) + destination = destination_root / f"{artifact_id}.json" + temporary = destination_root / f".{artifact_id}.{uuid4().hex}.tmp" + payload = json.dumps( + _proof_document(kind, artifact_id, generation), + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + descriptor = os.open( + temporary, + os.O_WRONLY | os.O_CREAT | os.O_EXCL, + 0o600, + ) + try: + with os.fdopen(descriptor, "wb") as stream: + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, destination) + finally: + temporary.unlink(missing_ok=True) + + def _evict_models(self) -> None: + while len(self._models) > self.max_entries: + _, entry = self._models.popitem(last=False) + entry.reader.close() + + def _evict_sources(self) -> None: + while len(self._sources) > self.max_entries: + _, entry = self._sources.popitem(last=False) + entry.reader.close() + + +def _proof_document( + kind: str, + artifact_id: str, + generation: _Generation, +) -> dict[str, object]: + return { + "schema_version": VALIDATION_CACHE_SCHEMA, + "kind": kind, + "artifact_id": artifact_id, + "files": [ + { + "name": name, + "device": identity[0], + "inode": identity[1], + "byte_length": identity[2], + "mtime_ns": identity[3], + "ctime_ns": identity[4], + } + for name, identity in generation + ], + } + + +def _generation(root: Path, names: tuple[str, ...]) -> _Generation: + try: + root_metadata = root.lstat() + except OSError as exc: + raise LidarGroundError("LiDAR evidence root is unavailable") from exc + if stat.S_ISLNK(root_metadata.st_mode) or not stat.S_ISDIR(root_metadata.st_mode): + raise LidarGroundError("LiDAR evidence root is unsafe") + return tuple((name, _regular_file_identity(root / name)) for name in names) + + +def _regular_file_identity(path: Path) -> _FileIdentity: + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(path, flags) + except OSError as exc: + raise LidarGroundError("LiDAR evidence file is unavailable or unsafe") from exc + try: + value = os.fstat(descriptor) + current = os.lstat(path) + if ( + not stat.S_ISREG(value.st_mode) + or stat.S_ISLNK(current.st_mode) + or (current.st_dev, current.st_ino) != (value.st_dev, value.st_ino) + ): + raise LidarGroundError("LiDAR evidence file changed during no-follow open") + return ( + value.st_dev, + value.st_ino, + value.st_size, + value.st_mtime_ns, + value.st_ctime_ns, + ) + finally: + os.close(descriptor) + + +def _private_directory(path: Path) -> Path: + path.mkdir(mode=0o700, parents=True, exist_ok=True) + metadata = path.lstat() + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode): + raise OSError("LiDAR validation cache root is unsafe") + return path diff --git a/src/k1link/web/session_api.py b/src/k1link/web/session_api.py index f43186a..cd7cdd3 100644 --- a/src/k1link/web/session_api.py +++ b/src/k1link/web/session_api.py @@ -353,7 +353,6 @@ def build_session_router( **( { "preparation": _catalog_preparation_document( - store, recording_preparation_manager, item.session_id, item.replayable, @@ -1530,7 +1529,6 @@ def _require_matching_recording_generation( def _catalog_preparation_document( - store: SessionStore, manager: SessionRecordingPreparationManager | None, session_id: str, replayable: bool, @@ -1538,16 +1536,6 @@ def _catalog_preparation_document( if manager is None or not replayable: return None snapshot = manager.status(session_id) - if snapshot is None: - try: - snapshot = manager.restore_published(store.prepare_replay(session_id)) - except ( - SessionNotFoundError, - SessionNotReplayableError, - SessionIntegrityError, - ValueError, - ): - return None if snapshot is None: return None document: dict[str, Any] = { diff --git a/tests/test_lidar_local_surface.py b/tests/test_lidar_local_surface.py index 0e22644..4186733 100644 --- a/tests/test_lidar_local_surface.py +++ b/tests/test_lidar_local_surface.py @@ -2,6 +2,7 @@ from __future__ import annotations import hashlib import json +import os import time from pathlib import Path @@ -26,6 +27,7 @@ from k1link.compute.lidar_local_surface_geometry import ( DEFAULT_K1_LOCAL_SURFACE_PROFILE as WORKER_LOCAL_SURFACE_PROFILE, ) from k1link.web.lidar_api import build_lidar_router +from k1link.web.lidar_local_surface_service import K1LocalSurfaceReadService def _canonical_json(value: object) -> bytes: @@ -139,6 +141,7 @@ def _endpoint(router: APIRouter, path: str) -> object: def test_k1_local_surface_is_dynamic_source_bound_and_read_only( tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, ) -> None: source_path = _source_pack(tmp_path / "source") source_artifact = source_path / "lidar-pack.npz" @@ -217,12 +220,14 @@ def test_k1_local_surface_is_dynamic_source_bound_and_read_only( source.close() model.close() + read_service = K1LocalSurfaceReadService(tmp_path / "validation-cache") router = build_lidar_router( root_provider=lambda: None, ground_root_provider=lambda: None, field_review_root_provider=lambda: None, local_surface_root_provider=lambda: output.parent, e10_source_root_provider=lambda: source_path.parent, + local_surface_read_service=read_service, ) catalog_route = _endpoint(router, "/api/v1/lidar/local-surfaces") frame_route = _endpoint( @@ -252,6 +257,86 @@ def test_k1_local_surface_is_dynamic_source_bound_and_read_only( assert review["review_profile_id"] == "missioncore-local-surface-attention/v1" assert review["access"] == "read-only" assert str(tmp_path) not in repr(review) + read_service.close() + + def forbidden_strict_open(*_args: object, **_kwargs: object) -> None: + raise AssertionError("an unchanged admitted generation must restore without strict scan") + + monkeypatch.setattr(K1LocalSurfaceV1, "__init__", forbidden_strict_open) + monkeypatch.setattr(E10LidarFieldSource, "__init__", forbidden_strict_open) + restored_service = K1LocalSurfaceReadService(tmp_path / "validation-cache") + restored_router = build_lidar_router( + root_provider=lambda: None, + ground_root_provider=lambda: None, + field_review_root_provider=lambda: None, + local_surface_root_provider=lambda: output.parent, + e10_source_root_provider=lambda: source_path.parent, + local_surface_read_service=restored_service, + ) + restored_catalog = _endpoint(restored_router, "/api/v1/lidar/local-surfaces") + restored_timeline = _endpoint( + restored_router, + "/api/v1/lidar/local-surfaces/{model_id}/timeline", + ) + try: + assert restored_catalog(limit=1)["items"][0]["model_id"] == output.name + assert restored_timeline(model_id=output.name)["frame_count"] == 8 + finally: + restored_service.close() + + source_arrays = source_path / "lidar-pack.npz" + source_metadata = source_arrays.stat() + os.utime( + source_arrays, + ns=(source_metadata.st_atime_ns, source_metadata.st_mtime_ns + 1), + ) + invalidated_source_service = K1LocalSurfaceReadService( + tmp_path / "validation-cache" + ) + invalidated_source_router = build_lidar_router( + root_provider=lambda: None, + ground_root_provider=lambda: None, + field_review_root_provider=lambda: None, + local_surface_root_provider=lambda: output.parent, + e10_source_root_provider=lambda: source_path.parent, + local_surface_read_service=invalidated_source_service, + ) + invalidated_timeline = _endpoint( + invalidated_source_router, + "/api/v1/lidar/local-surfaces/{model_id}/timeline", + ) + try: + with pytest.raises(AssertionError, match="strict scan"): + invalidated_timeline(model_id=output.name) + finally: + invalidated_source_service.close() + + model_manifest = output / "manifest.json" + model_metadata = model_manifest.stat() + os.utime( + model_manifest, + ns=(model_metadata.st_atime_ns, model_metadata.st_mtime_ns + 1), + ) + invalidated_model_service = K1LocalSurfaceReadService( + tmp_path / "validation-cache" + ) + invalidated_model_router = build_lidar_router( + root_provider=lambda: None, + ground_root_provider=lambda: None, + field_review_root_provider=lambda: None, + local_surface_root_provider=lambda: output.parent, + e10_source_root_provider=lambda: source_path.parent, + local_surface_read_service=invalidated_model_service, + ) + invalidated_catalog = _endpoint( + invalidated_model_router, + "/api/v1/lidar/local-surfaces", + ) + try: + with pytest.raises(AssertionError, match="strict scan"): + invalidated_catalog(limit=1) + finally: + invalidated_model_service.close() def _shadow_input( diff --git a/tests/test_session_api.py b/tests/test_session_api.py index c26a68b..5560ef4 100644 --- a/tests/test_session_api.py +++ b/tests/test_session_api.py @@ -851,7 +851,10 @@ def test_replay_post_retries_terminal_camera_finalization_failure( manager.close() -def test_catalog_read_never_prepares_a_cold_historical_session(tmp_path: Path) -> None: +def test_catalog_read_never_prepares_a_cold_historical_session( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: repository = tmp_path / "repo" sessions = repository / "sessions" session = make_legacy_session(sessions, "20260716T205632Z_viewer_live") @@ -866,6 +869,11 @@ def test_catalog_read_never_prepares_a_cold_historical_session(tmp_path: Path) - materializer = SessionRecordingMaterializer(store.data_dir, exporter=exporter) manager = SessionRecordingPreparationManager(materializer) + + def forbidden_restore(*_args: object, **_kwargs: object) -> None: + raise AssertionError("catalog reads must not restore published recordings") + + monkeypatch.setattr(manager, "restore_published", forbidden_restore) router = build_session_router( store, recording_materializer=materializer,