From e56fa0c0743b8a14790f71c54bdabf97491d28af Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Thu, 30 Jul 2026 10:05:48 +0300 Subject: [PATCH] feat(storage): add portable session artifact gateway --- docs/09_OBSERVATION_SESSIONS.md | 7 + ...ession-artifacts-and-bounded-host-cache.md | 159 +++ src/k1link/artifact_cli.py | 248 +++++ src/k1link/artifact_gateway.py | 954 ++++++++++++++++++ src/k1link/compute/integrated_perception.py | 60 ++ src/k1link/device_plugins/xgrids_k1/cli.py | 2 + src/k1link/sessions/recording.py | 119 ++- src/k1link/web/app.py | 4 + tests/test_artifact_gateway.py | 209 ++++ tests/test_e10_integrated_perception.py | 76 ++ tests/test_session_recording.py | 53 + 11 files changed, 1889 insertions(+), 2 deletions(-) create mode 100644 docs/adr/0033-central-session-artifacts-and-bounded-host-cache.md create mode 100644 src/k1link/artifact_cli.py create mode 100644 src/k1link/artifact_gateway.py create mode 100644 tests/test_artifact_gateway.py diff --git a/docs/09_OBSERVATION_SESSIONS.md b/docs/09_OBSERVATION_SESSIONS.md index fef27c4..eb19014 100644 --- a/docs/09_OBSERVATION_SESSIONS.md +++ b/docs/09_OBSERVATION_SESSIONS.md @@ -230,6 +230,13 @@ export MISSIONCORE_RRD_FREE_SPACE_RESERVE_BYTES=2147483648 uv run k1link serve ``` +Portable session products can additionally use the content-addressed central +artifact gateway described in +[`ADR 0033`](adr/0033-central-session-artifacts-and-bounded-host-cache.md). +This does not move or delete the source-of-record evidence. It adds verified +central manifests plus a quota-bound, pinnable local working set for exact base +RRDs and perception overlays. + The directories and derived recording cache use owner-only permissions where the host filesystem permits them. A new acquisition writer is assigned only a direct child of the private evidence root; it no longer writes a new run to the diff --git a/docs/adr/0033-central-session-artifacts-and-bounded-host-cache.md b/docs/adr/0033-central-session-artifacts-and-bounded-host-cache.md new file mode 100644 index 0000000..2fb78b9 --- /dev/null +++ b/docs/adr/0033-central-session-artifacts-and-bounded-host-cache.md @@ -0,0 +1,159 @@ +# ADR 0033: Central session artifacts and bounded operator-host cache + +Date: 2026-07-30 +Status: accepted and implemented + +## Decision + +Mission Core stores portable recorded-session products in a provider-neutral, +content-addressed artifact store. The current central provider is the existing +Synology SMB mount: + +```text +/Volumes/docker/nodedc-mission-core/artifact-store +``` + +This is product runtime data. It is not a deploy artifact, a Map Gateway cache, +a Device Plane package, or a new Docker service. Authentication remains owned +by the operating-system SMB mount; Mission Core does not receive a NAS password. + +Every immutable object is stored under: + +```text +objects/sha256// +``` + +Canonical JSON manifests are themselves addressed by SHA-256. Mutable names +exist only as small, atomically replaced references: + +```text +refs//.json -> manifest SHA-256 +``` + +An operator host never serves a central SMB file directly. It first streams the +object into a temporary local file, proves the expected SHA-256 and byte length, +and atomically publishes it into: + +```text +/artifact-cache +``` + +The local cache has: + +- an 8 GiB default byte quota; +- a 2 GiB default free-space reserve; +- persistent SQLite metadata; +- persistent pin sets; +- LRU eviction of unpinned objects only; +- no automatic deletion of source evidence; +- offline resolution through locally sealed reference and manifest snapshots. + +The limits are configurable with: + +```text +MISSIONCORE_ARTIFACT_CACHE_MAX_BYTES +MISSIONCORE_ARTIFACT_CACHE_FREE_SPACE_RESERVE_BYTES +``` + +The central root is configured with: + +```text +MISSIONCORE_ARTIFACT_STORE_ROOT +``` + +## Recorded playback behavior + +The base RRD and integrated E10 overlay are separate immutable roles in one +recorded-session manifest. + +On a normal cache hit, recorded playback uses the existing private RRD cache. +On a private-cache miss it may restore the exact `base-rrd` role from the +bounded artifact cache. Restoration is admitted only when: + +- the manifest type and subject match the requested session; +- the RRD content digest and length match the manifest; +- the local source evidence still hashes to `base-source-sha256`; +- the manifest carries the exact session timeline bounds. + +The restored RRD is republished through the existing private recording-cache +transaction and receives a fresh sidecar bound to the current local evidence +stats. The exporter is not run. + +The integrated-perception endpoint resolves +`integrated-overlay:` before scanning or validating local worker +results. A sealed central/local CAS hit therefore needs neither Worker 006 nor +the original E10 result tree. Invalid central metadata fails closed; a genuinely +absent central role can still use the existing local materialization path. + +## Persistent pins and offline use + +The CLI surface is: + +```bash +k1link artifact publish ... +k1link artifact pin ... +k1link artifact unpin ... +k1link artifact resolve ... +k1link artifact status ... +``` + +Pinning is a local operator-host decision. A pin downloads and verifies every +member of the selected manifest. If any member fails, the new pin is removed; +objects already downloaded remain ordinary unpinned cache entries. + +When the central root is unavailable: + +- a sealed local reference, manifest, and object resolve normally; +- an absent local object returns an explicit offline-cache error; +- Mission Core does not silently fetch from Worker 006 or start a heavy AI run. + +## RAVNOVES00 migration acceptance + +The source session is `20260720T065719Z_viewer_live`, displayed as +`RAVNOVES00`. + +The accepted central reference currently points to: + +```text +092f758c09965be98fd76d8191474c528ee20540dce5f4761c2a087f6056a6d5 +``` + +It contains nine logical roles totaling `721626746` bytes: + +- the source evidence archive; +- the base RRD and its original audit sidecar; +- the accepted E10 result archive; +- the compute-job manifest; +- the LiDAR pack and its manifest; +- the sealed integrated overlay and its sidecar. + +The host pin is: + +```text +session:20260720T065719Z_viewer_live +``` + +Acceptance facts: + +- all nine roles are cached and pinned; +- base RRD SHA-256: + `4db5f2519caa61aca3683499094329b35cf3c24d1cc242953c54d2a228762e13`; +- base source SHA-256: + `70da0edad5cbf0e89b6e2355c0aac294f8c36a67249cd33b8918353fd88c83af`; +- overlay SHA-256: + `c2bb73fca8bb616a3929df236c6dc63d071b6466a6ca57d06487ef18ed985b57`; +- an offline overlay resolve completed in 0.76 seconds; +- an isolated offline base restore produced the exact RRD without invoking the + exporter; +- an isolated offline overlay materialization succeeded with nonexistent local + worker-result, job, and LiDAR roots. + +The migration was copy-only. Existing evidence, derived recordings, accepted +worker results, and historical overlay generations were not deleted. + +## Resource boundary + +Artifact publication, fetch, hashing, and restoration are sequential and use a +1 MiB streaming buffer. They do not start Docker or parallel workers. This is +the required operating envelope for the 14-inch 2023 MacBook Pro with 18 GiB +RAM used as the current combined operator and test host. diff --git a/src/k1link/artifact_cli.py b/src/k1link/artifact_cli.py new file mode 100644 index 0000000..c5ea8d9 --- /dev/null +++ b/src/k1link/artifact_cli.py @@ -0,0 +1,248 @@ +"""Operator commands for the Mission Core central artifact gateway.""" + +from __future__ import annotations + +import json +import mimetypes +import os +from pathlib import Path +from typing import Annotated + +import typer +from rich.console import Console + +from k1link.artifact_gateway import ( + CACHE_MAX_BYTES_ENV, + CACHE_RESERVE_BYTES_ENV, + DEFAULT_CACHE_MAX_BYTES, + DEFAULT_FREE_SPACE_RESERVE_BYTES, + STORE_ROOT_ENV, + ArtifactGateway, + ArtifactGatewayError, + CentralArtifactStore, + LocalArtifactCache, +) +from k1link.sessions.store import resolve_missioncore_data_dir + +app = typer.Typer( + help="Publish, pin, inspect, and resolve verified central Mission Core artifacts.", + no_args_is_help=True, +) +console = Console() + + +def _positive_environment_bytes(name: str, default: int) -> int: + value = os.environ.get(name, "").strip() + if not value: + return default + if not value.isascii() or not value.isdecimal() or int(value) <= 0: + raise ValueError(f"{name} must be a positive integer") + return int(value) + + +def _non_negative_environment_bytes(name: str, default: int) -> int: + value = os.environ.get(name, "").strip() + if not value: + return default + if not value.isascii() or not value.isdecimal(): + raise ValueError(f"{name} must be a non-negative integer") + return int(value) + + +def _gateway(repository_root: Path, *, create_store: bool = False) -> ArtifactGateway: + configured = os.environ.get(STORE_ROOT_ENV, "").strip() + if not configured: + raise ValueError(f"{STORE_ROOT_ENV} is not configured") + data_dir = resolve_missioncore_data_dir(repository_root) + return ArtifactGateway( + CentralArtifactStore(Path(configured), create=create_store), + LocalArtifactCache( + data_dir / "artifact-cache", + max_bytes=_positive_environment_bytes( + CACHE_MAX_BYTES_ENV, + DEFAULT_CACHE_MAX_BYTES, + ), + free_space_reserve_bytes=_non_negative_environment_bytes( + CACHE_RESERVE_BYTES_ENV, + DEFAULT_FREE_SPACE_RESERVE_BYTES, + ), + ), + ) + + +def _parse_assignment(value: str, *, label: str) -> tuple[str, str]: + key, separator, item = value.partition("=") + if not separator or not key or not item: + raise ValueError(f"{label} must use KEY=VALUE") + return key, item + + +def _media_type(path: Path) -> str: + explicit = { + ".rrd": "application/vnd.rerun.rrd", + ".jsonl": "application/x-ndjson", + ".npz": "application/x-numpy", + ".k1mqtt": "application/vnd.nodedc.k1mqtt", + ".tar": "application/x-tar", + ".tgz": "application/gzip", + ".m4s": "video/iso.segment", + }.get(path.suffix.lower()) + if explicit is not None: + return explicit + guessed, _encoding = mimetypes.guess_type(path.name) + return guessed or "application/octet-stream" + + +@app.command("publish") +def publish( + namespace: Annotated[str, typer.Option("--namespace")], + key: Annotated[str, typer.Option("--key")], + artifact_type: Annotated[str, typer.Option("--artifact-type")], + subject_id: Annotated[str, typer.Option("--subject-id")], + member: Annotated[ + list[str], + typer.Option( + "--member", + help="Repeat ROLE=/absolute/file/path for every immutable bundle member.", + ), + ], + metadata: Annotated[ + list[str] | None, + typer.Option("--metadata", help="Optional repeatable KEY=VALUE metadata."), + ] = None, +) -> None: + """Publish exact files, then atomically advance one named central reference.""" + + repository_root = Path(__file__).resolve().parents[2] + try: + sources = [] + for raw in member: + role, raw_path = _parse_assignment(raw, label="member") + path = Path(raw_path).expanduser().resolve(strict=True) + sources.append((role, _media_type(path), path)) + metadata_values = dict( + _parse_assignment(raw, label="metadata") for raw in (metadata or []) + ) + manifest = _gateway(repository_root, create_store=True).publish( + namespace=namespace, + key=key, + artifact_type=artifact_type, + subject_id=subject_id, + sources=sources, + metadata=metadata_values, + ) + except (ArtifactGatewayError, OSError, ValueError) as exc: + console.print(f"[red]Artifact publication failed:[/red] {exc}") + raise typer.Exit(code=2) from exc + total_bytes = sum(item.byte_length for item in manifest.members) + console.print( + "[green]Central artifact published.[/green] " + f"reference={namespace}/{key}; manifest={manifest.manifest_id}; " + f"objects={len(manifest.members)}; logical_bytes={total_bytes}" + ) + + +@app.command("pin") +def pin( + namespace: Annotated[str, typer.Option("--namespace")], + key: Annotated[str, typer.Option("--key")], + pin_id: Annotated[str | None, typer.Option("--pin-id")] = None, +) -> None: + """Fetch and persistently pin every object in one named bundle.""" + + repository_root = Path(__file__).resolve().parents[2] + selected_pin = pin_id or f"{namespace}:{key}" + try: + gateway = _gateway(repository_root) + manifest = gateway.pin_reference(namespace, key, pin_id=selected_pin) + status = gateway.cache.status() + except (ArtifactGatewayError, OSError, ValueError) as exc: + console.print(f"[red]Artifact pin failed:[/red] {exc}") + raise typer.Exit(code=2) from exc + console.print( + "[green]Artifact bundle pinned.[/green] " + f"manifest={manifest.manifest_id}; pin={selected_pin}; " + f"pinned_objects={status.pinned_object_count}; pinned_bytes={status.pinned_bytes}" + ) + + +@app.command("unpin") +def unpin(pin_id: Annotated[str, typer.Option("--pin-id")]) -> None: + """Remove one persistent pin and immediately enforce local LRU limits.""" + + repository_root = Path(__file__).resolve().parents[2] + try: + gateway = _gateway(repository_root) + gateway.cache.unpin(pin_id) + status = gateway.cache.status() + except (ArtifactGatewayError, OSError, ValueError) as exc: + console.print(f"[red]Artifact unpin failed:[/red] {exc}") + raise typer.Exit(code=2) from exc + console.print( + "[green]Artifact pin removed.[/green] " + f"pin={pin_id}; cached_bytes={status.total_bytes}; " + f"pinned_bytes={status.pinned_bytes}" + ) + + +@app.command("resolve") +def resolve( + namespace: Annotated[str, typer.Option("--namespace")], + key: Annotated[str, typer.Option("--key")], + role: Annotated[str, typer.Option("--role")], + json_output: Annotated[bool, typer.Option("--json")] = False, +) -> None: + """Resolve one role to a verified local path, fetching it when online.""" + + repository_root = Path(__file__).resolve().parents[2] + try: + artifact = _gateway(repository_root).resolve_role(namespace, key, role) + except (ArtifactGatewayError, OSError, ValueError) as exc: + console.print(f"[red]Artifact resolution failed:[/red] {exc}") + raise typer.Exit(code=2) from exc + document = { + "manifest_id": artifact.manifest.manifest_id, + "role": artifact.member.role, + "sha256": artifact.member.sha256, + "byte_length": artifact.member.byte_length, + "path": str(artifact.path), + "cache_hit": artifact.cache_hit, + "central_available": artifact.central_available, + } + if json_output: + typer.echo(json.dumps(document, ensure_ascii=False, indent=2)) + return + console.print( + "[green]Artifact resolved.[/green] " + f"role={role}; path={artifact.path}; cache_hit={artifact.cache_hit}; " + f"central_available={artifact.central_available}" + ) + + +@app.command("status") +def status(json_output: Annotated[bool, typer.Option("--json")] = False) -> None: + """Report local working-set quota and persistent pin usage.""" + + repository_root = Path(__file__).resolve().parents[2] + try: + cache_status = _gateway(repository_root).cache.status() + except (ArtifactGatewayError, OSError, ValueError) as exc: + console.print(f"[red]Artifact status failed:[/red] {exc}") + raise typer.Exit(code=2) from exc + document = { + "object_count": cache_status.object_count, + "total_bytes": cache_status.total_bytes, + "pinned_object_count": cache_status.pinned_object_count, + "pinned_bytes": cache_status.pinned_bytes, + "cache_max_bytes": cache_status.cache_max_bytes, + "free_space_reserve_bytes": cache_status.free_space_reserve_bytes, + } + if json_output: + typer.echo(json.dumps(document, ensure_ascii=False, indent=2)) + return + console.print( + f"objects={cache_status.object_count}; bytes={cache_status.total_bytes}; " + f"pinned_objects={cache_status.pinned_object_count}; " + f"pinned_bytes={cache_status.pinned_bytes}; " + f"quota={cache_status.cache_max_bytes}" + ) diff --git a/src/k1link/artifact_gateway.py b/src/k1link/artifact_gateway.py new file mode 100644 index 0000000..e3ebec4 --- /dev/null +++ b/src/k1link/artifact_gateway.py @@ -0,0 +1,954 @@ +"""Content-addressed central artifacts with a bounded, offline-capable host cache.""" + +from __future__ import annotations + +import fcntl +import hashlib +import json +import os +import re +import shutil +import sqlite3 +import stat +import threading +import time +from collections.abc import Iterator, Mapping, Sequence +from contextlib import contextmanager, suppress +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Final +from uuid import uuid4 + +from k1link.artifacts import utc_now_iso + +MANIFEST_SCHEMA: Final = "missioncore.artifact-manifest/v1" +REFERENCE_SCHEMA: Final = "missioncore.artifact-reference/v1" +DEFAULT_CACHE_MAX_BYTES: Final = 8 * 1024 * 1024 * 1024 +DEFAULT_FREE_SPACE_RESERVE_BYTES: Final = 2 * 1024 * 1024 * 1024 +STORE_ROOT_ENV: Final = "MISSIONCORE_ARTIFACT_STORE_ROOT" +CACHE_MAX_BYTES_ENV: Final = "MISSIONCORE_ARTIFACT_CACHE_MAX_BYTES" +CACHE_RESERVE_BYTES_ENV: Final = "MISSIONCORE_ARTIFACT_CACHE_FREE_SPACE_RESERVE_BYTES" + +_SHA256 = re.compile(r"^[a-f0-9]{64}$") +_SAFE_COMPONENT = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") +_MAX_MANIFEST_BYTES = 8 * 1024 * 1024 +_COPY_CHUNK_BYTES = 1024 * 1024 + + +class ArtifactGatewayError(RuntimeError): + """Base class for central artifact and host-cache failures.""" + + +class ArtifactStoreUnavailable(ArtifactGatewayError): + """The configured central store cannot currently be reached.""" + + +class ArtifactNotFound(ArtifactGatewayError): + """An exact reference, manifest, role, or object does not exist.""" + + +class ArtifactIntegrityError(ArtifactGatewayError): + """An artifact violates its content-addressed identity.""" + + +class ArtifactCacheCapacityError(ArtifactGatewayError): + """The local host cannot safely reserve enough cache capacity.""" + + +@dataclass(frozen=True, slots=True) +class ArtifactMember: + role: str + media_type: str + sha256: str + byte_length: int + + +@dataclass(frozen=True, slots=True) +class ArtifactManifest: + manifest_id: str + artifact_type: str + subject_id: str + created_at_utc: str + members: tuple[ArtifactMember, ...] + metadata: Mapping[str, str] + + def member(self, role: str) -> ArtifactMember: + matches = tuple(item for item in self.members if item.role == role) + if len(matches) != 1: + raise ArtifactNotFound(f"artifact role is unavailable: {role}") + return matches[0] + + +@dataclass(frozen=True, slots=True) +class ResolvedArtifact: + manifest: ArtifactManifest + member: ArtifactMember + path: Path + cache_hit: bool + central_available: bool + + +@dataclass(frozen=True, slots=True) +class ArtifactCacheStatus: + object_count: int + total_bytes: int + pinned_object_count: int + pinned_bytes: int + cache_max_bytes: int + free_space_reserve_bytes: int + + +class CentralArtifactStore: + """Immutable SHA-256 objects/manifests plus atomically replaceable named refs.""" + + def __init__(self, root: Path, *, create: bool = False) -> None: + self.root = root.expanduser().absolute() + if create: + try: + self.root.mkdir(parents=True, exist_ok=True) + except OSError as exc: + raise ArtifactStoreUnavailable("central artifact store cannot be created") from exc + self._require_root() + + def publish_file(self, source: Path) -> ArtifactMember: + source_path = _regular_source(source) + digest, byte_length = _hash_file(source_path) + destination = self.object_path(digest) + try: + destination.parent.mkdir(parents=True, exist_ok=True) + if destination.exists() or destination.is_symlink(): + _verify_object(destination, digest, byte_length) + else: + temporary = destination.with_name(f".{digest}.{uuid4().hex}.tmp") + try: + copied_digest, copied_bytes = _copy_and_hash(source_path, temporary) + if copied_digest != digest or copied_bytes != byte_length: + raise ArtifactIntegrityError("source changed while it was published") + os.replace(temporary, destination) + finally: + temporary.unlink(missing_ok=True) + except ArtifactGatewayError: + raise + except OSError as exc: + raise ArtifactStoreUnavailable("central artifact object cannot be published") from exc + return ArtifactMember( + role="", + media_type="application/octet-stream", + sha256=digest, + byte_length=byte_length, + ) + + def publish_manifest( + self, + *, + artifact_type: str, + subject_id: str, + members: Sequence[ArtifactMember], + metadata: Mapping[str, str] | None = None, + created_at_utc: str | None = None, + ) -> ArtifactManifest: + _validate_component(artifact_type, "artifact type") + _validate_component(subject_id, "artifact subject") + normalized_members = tuple(sorted(members, key=lambda item: item.role)) + _validate_members(normalized_members) + normalized_metadata = dict(sorted((metadata or {}).items())) + for key, value in normalized_metadata.items(): + _validate_component(key, "metadata key") + if not isinstance(value, str) or len(value) > 4096: + raise ValueError("artifact metadata value is invalid") + document = { + "schema_version": MANIFEST_SCHEMA, + "artifact_type": artifact_type, + "subject_id": subject_id, + "created_at_utc": created_at_utc or utc_now_iso(), + "members": [ + { + "role": item.role, + "media_type": item.media_type, + "sha256": item.sha256, + "byte_length": item.byte_length, + } + for item in normalized_members + ], + "metadata": normalized_metadata, + } + payload = _canonical_json(document) + manifest_id = hashlib.sha256(payload).hexdigest() + destination = self.manifest_path(manifest_id) + try: + destination.parent.mkdir(parents=True, exist_ok=True) + if destination.exists() or destination.is_symlink(): + if destination.read_bytes() != payload: + raise ArtifactIntegrityError("central artifact manifest digest is inconsistent") + else: + _write_bytes_atomic(destination, payload) + except ArtifactGatewayError: + raise + except OSError as exc: + raise ArtifactStoreUnavailable("central artifact manifest cannot be published") from exc + return _parse_manifest(document, manifest_id) + + def set_reference(self, namespace: str, key: str, manifest_id: str) -> None: + _validate_component(namespace, "artifact namespace") + _validate_component(key, "artifact reference key") + _validate_sha256(manifest_id) + self.read_manifest(manifest_id) + document = { + "schema_version": REFERENCE_SCHEMA, + "namespace": namespace, + "key": key, + "manifest_id": manifest_id, + "updated_at_utc": utc_now_iso(), + } + try: + _write_bytes_atomic(self.reference_path(namespace, key), _canonical_json(document)) + except OSError as exc: + raise ArtifactStoreUnavailable( + "central artifact reference cannot be published" + ) from exc + + def resolve_reference(self, namespace: str, key: str) -> ArtifactManifest: + reference = self.read_reference_document(namespace, key) + return self.read_manifest(str(reference["manifest_id"])) + + def read_reference_document(self, namespace: str, key: str) -> dict[str, Any]: + _validate_component(namespace, "artifact namespace") + _validate_component(key, "artifact reference key") + path = self.reference_path(namespace, key) + document = _read_json_document(path, unavailable_message="central reference is unavailable") + if ( + document.get("schema_version") != REFERENCE_SCHEMA + or document.get("namespace") != namespace + or document.get("key") != key + or not isinstance(document.get("manifest_id"), str) + or _SHA256.fullmatch(str(document["manifest_id"])) is None + or not isinstance(document.get("updated_at_utc"), str) + ): + raise ArtifactIntegrityError("central artifact reference is invalid") + return document + + def read_manifest(self, manifest_id: str) -> ArtifactManifest: + _validate_sha256(manifest_id) + path = self.manifest_path(manifest_id) + document = _read_json_document(path, unavailable_message="central manifest is unavailable") + payload = _canonical_json(document) + if hashlib.sha256(payload).hexdigest() != manifest_id: + raise ArtifactIntegrityError("central artifact manifest digest changed") + return _parse_manifest(document, manifest_id) + + def object_path(self, digest: str) -> Path: + _validate_sha256(digest) + return self.root / "objects" / "sha256" / digest[:2] / digest + + def manifest_path(self, manifest_id: str) -> Path: + _validate_sha256(manifest_id) + return self.root / "manifests" / "sha256" / manifest_id[:2] / f"{manifest_id}.json" + + def reference_path(self, namespace: str, key: str) -> Path: + _validate_component(namespace, "artifact namespace") + _validate_component(key, "artifact reference key") + return self.root / "refs" / namespace / f"{key}.json" + + def _require_root(self) -> None: + try: + metadata = self.root.lstat() + except OSError as exc: + raise ArtifactStoreUnavailable("central artifact store is unavailable") from exc + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode): + raise ArtifactIntegrityError("central artifact store root is invalid") + + +class LocalArtifactCache: + """Persistent local CAS with durable pins and unpinned LRU eviction.""" + + def __init__( + self, + root: Path, + *, + max_bytes: int = DEFAULT_CACHE_MAX_BYTES, + free_space_reserve_bytes: int = DEFAULT_FREE_SPACE_RESERVE_BYTES, + ) -> None: + if max_bytes <= 0: + raise ValueError("artifact cache maximum bytes must be positive") + if free_space_reserve_bytes < 0: + raise ValueError("artifact cache free-space reserve must be non-negative") + self.root = root.expanduser().absolute() + self.root.mkdir(mode=0o700, parents=True, exist_ok=True) + root_metadata = self.root.lstat() + if stat.S_ISLNK(root_metadata.st_mode) or not stat.S_ISDIR(root_metadata.st_mode): + raise ArtifactIntegrityError("local artifact cache root is invalid") + with suppress(OSError): + self.root.chmod(0o700) + self.objects_root = self.root / "objects" / "sha256" + self.metadata_root = self.root / "metadata" + self.objects_root.mkdir(mode=0o700, parents=True, exist_ok=True) + self.metadata_root.mkdir(mode=0o700, parents=True, exist_ok=True) + self.database_path = self.root / "cache.sqlite3" + self.lock_path = self.root / ".cache.lock" + self.max_bytes = max_bytes + self.free_space_reserve_bytes = free_space_reserve_bytes + self._thread_lock = threading.RLock() + self._initialize_database() + + def get(self, member: ArtifactMember) -> Path | None: + with self._locked(): + return self._get_locked(member, touch=True) + + def fetch( + self, + store: CentralArtifactStore, + member: ArtifactMember, + *, + pin_id: str | None = None, + ) -> tuple[Path, bool]: + _validate_member(member) + if pin_id is not None: + _validate_component(pin_id, "artifact pin id") + with self._locked(): + cached = self._get_locked(member, touch=True) + if cached is not None: + if pin_id is not None: + self._pin_locked(pin_id, member.sha256) + return cached, True + self._reserve_locked(member.byte_length, protected_digest=member.sha256) + source = store.object_path(member.sha256) + destination = self.object_path(member.sha256) + destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + temporary = destination.with_name(f".{member.sha256}.{uuid4().hex}.tmp") + try: + copied_digest, copied_bytes = _copy_and_hash(source, temporary) + if copied_digest != member.sha256 or copied_bytes != member.byte_length: + raise ArtifactIntegrityError("central artifact object failed verification") + os.chmod(temporary, 0o600) + os.replace(temporary, destination) + except ArtifactGatewayError: + raise + except OSError as exc: + raise ArtifactStoreUnavailable("central artifact object is unavailable") from exc + finally: + temporary.unlink(missing_ok=True) + metadata = destination.stat() + with self._connect() as connection: + connection.execute( + """ + INSERT INTO objects( + sha256, byte_length, last_access_ns, mtime_ns, inode + ) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(sha256) DO UPDATE SET + byte_length=excluded.byte_length, + last_access_ns=excluded.last_access_ns, + mtime_ns=excluded.mtime_ns, + inode=excluded.inode + """, + ( + member.sha256, + member.byte_length, + time.time_ns(), + metadata.st_mtime_ns, + metadata.st_ino, + ), + ) + if pin_id is not None: + connection.execute( + "INSERT OR IGNORE INTO pins(pin_id, sha256) VALUES (?, ?)", + (pin_id, member.sha256), + ) + self._evict_locked(protected_digest=member.sha256) + return destination.resolve(strict=True), False + + def pin(self, pin_id: str, members: Sequence[ArtifactMember]) -> tuple[Path, ...]: + _validate_component(pin_id, "artifact pin id") + paths: list[Path] = [] + try: + for member in members: + path = self.get(member) + if path is None: + raise ArtifactNotFound( + f"artifact object must be fetched before pinning: {member.role}" + ) + with self._locked(): + self._pin_locked(pin_id, member.sha256) + paths.append(path) + except BaseException: + self.unpin(pin_id) + raise + return tuple(paths) + + def unpin(self, pin_id: str) -> None: + _validate_component(pin_id, "artifact pin id") + with self._locked(), self._connect() as connection: + connection.execute("DELETE FROM pins WHERE pin_id = ?", (pin_id,)) + with self._locked(): + self._evict_locked(protected_digest=None) + + def save_snapshot( + self, + namespace: str, + key: str, + reference: Mapping[str, Any], + manifest: ArtifactManifest, + ) -> None: + _validate_component(namespace, "artifact namespace") + _validate_component(key, "artifact reference key") + manifest_document = _manifest_document(manifest) + reference_document = dict(reference) + _validate_reference_snapshot(reference_document, namespace, key) + if reference_document["manifest_id"] != manifest.manifest_id: + raise ArtifactIntegrityError("artifact reference and manifest do not match") + manifest_path = ( + self.metadata_root + / "manifests" + / manifest.manifest_id[:2] + / f"{manifest.manifest_id}.json" + ) + reference_path = self.metadata_root / "refs" / namespace / f"{key}.json" + with self._locked(): + _write_bytes_atomic(manifest_path, _canonical_json(manifest_document), mode=0o600) + _write_bytes_atomic(reference_path, _canonical_json(reference_document), mode=0o600) + + def load_snapshot(self, namespace: str, key: str) -> ArtifactManifest: + _validate_component(namespace, "artifact namespace") + _validate_component(key, "artifact reference key") + reference_path = self.metadata_root / "refs" / namespace / f"{key}.json" + try: + reference = _read_json_document( + reference_path, + unavailable_message="local artifact reference is unavailable", + ) + except ArtifactStoreUnavailable as exc: + raise ArtifactNotFound("local artifact reference is unavailable") from exc + _validate_reference_snapshot(reference, namespace, key) + manifest_id = str(reference["manifest_id"]) + manifest_path = ( + self.metadata_root / "manifests" / manifest_id[:2] / f"{manifest_id}.json" + ) + try: + document = _read_json_document( + manifest_path, + unavailable_message="local artifact manifest is unavailable", + ) + except ArtifactStoreUnavailable as exc: + raise ArtifactNotFound("local artifact manifest is unavailable") from exc + if hashlib.sha256(_canonical_json(document)).hexdigest() != manifest_id: + raise ArtifactIntegrityError("local artifact manifest digest changed") + return _parse_manifest(document, manifest_id) + + def status(self) -> ArtifactCacheStatus: + with self._locked(), self._connect() as connection: + row = connection.execute( + """ + SELECT + COUNT(*), + COALESCE(SUM(byte_length), 0), + COUNT(DISTINCT pins.sha256), + COALESCE(SUM( + CASE WHEN pins.sha256 IS NOT NULL THEN objects.byte_length ELSE 0 END + ), 0) + FROM objects + LEFT JOIN (SELECT DISTINCT sha256 FROM pins) AS pins USING (sha256) + """ + ).fetchone() + assert row is not None + return ArtifactCacheStatus( + object_count=int(row[0]), + total_bytes=int(row[1]), + pinned_object_count=int(row[2]), + pinned_bytes=int(row[3]), + cache_max_bytes=self.max_bytes, + free_space_reserve_bytes=self.free_space_reserve_bytes, + ) + + def object_path(self, digest: str) -> Path: + _validate_sha256(digest) + return self.objects_root / digest[:2] / digest + + def _initialize_database(self) -> None: + with self._connect() as connection: + connection.executescript( + """ + PRAGMA journal_mode = WAL; + PRAGMA foreign_keys = ON; + CREATE TABLE IF NOT EXISTS objects ( + sha256 TEXT PRIMARY KEY, + byte_length INTEGER NOT NULL CHECK (byte_length >= 0), + last_access_ns INTEGER NOT NULL, + mtime_ns INTEGER NOT NULL, + inode INTEGER NOT NULL + ); + CREATE TABLE IF NOT EXISTS pins ( + pin_id TEXT NOT NULL, + sha256 TEXT NOT NULL REFERENCES objects(sha256) ON DELETE CASCADE, + PRIMARY KEY(pin_id, sha256) + ); + CREATE INDEX IF NOT EXISTS objects_lru ON objects(last_access_ns, sha256); + CREATE INDEX IF NOT EXISTS pins_digest ON pins(sha256); + """ + ) + with suppress(OSError): + self.database_path.chmod(0o600) + + def _connect(self) -> sqlite3.Connection: + connection = sqlite3.connect(self.database_path, timeout=30.0) + connection.execute("PRAGMA foreign_keys = ON") + return connection + + @contextmanager + def _locked(self) -> Iterator[None]: + with self._thread_lock: + descriptor = os.open(self.lock_path, os.O_CREAT | os.O_RDWR, 0o600) + try: + fcntl.flock(descriptor, fcntl.LOCK_EX) + yield + finally: + fcntl.flock(descriptor, fcntl.LOCK_UN) + os.close(descriptor) + + def _get_locked(self, member: ArtifactMember, *, touch: bool) -> Path | None: + _validate_member(member) + path = self.object_path(member.sha256) + with self._connect() as connection: + row = connection.execute( + "SELECT byte_length, mtime_ns, inode FROM objects WHERE sha256 = ?", + (member.sha256,), + ).fetchone() + if row is None: + return None + try: + metadata = path.lstat() + except OSError: + connection.execute("DELETE FROM objects WHERE sha256 = ?", (member.sha256,)) + return None + if ( + stat.S_ISLNK(metadata.st_mode) + or not stat.S_ISREG(metadata.st_mode) + or metadata.st_size != member.byte_length + or int(row[0]) != member.byte_length + ): + path.unlink(missing_ok=True) + connection.execute("DELETE FROM objects WHERE sha256 = ?", (member.sha256,)) + return None + if metadata.st_mtime_ns != int(row[1]) or metadata.st_ino != int(row[2]): + digest, byte_length = _hash_file(path) + if digest != member.sha256 or byte_length != member.byte_length: + path.unlink(missing_ok=True) + connection.execute("DELETE FROM objects WHERE sha256 = ?", (member.sha256,)) + return None + connection.execute( + "UPDATE objects SET mtime_ns = ?, inode = ? WHERE sha256 = ?", + (metadata.st_mtime_ns, metadata.st_ino, member.sha256), + ) + if touch: + connection.execute( + "UPDATE objects SET last_access_ns = ? WHERE sha256 = ?", + (time.time_ns(), member.sha256), + ) + return path.resolve(strict=True) + + def _pin_locked(self, pin_id: str, digest: str) -> None: + with self._connect() as connection: + connection.execute( + "INSERT OR IGNORE INTO pins(pin_id, sha256) VALUES (?, ?)", + (pin_id, digest), + ) + + def _reserve_locked(self, required_bytes: int, *, protected_digest: str) -> None: + if required_bytes < 0: + raise ArtifactCacheCapacityError("artifact cache reservation is invalid") + if required_bytes > self.max_bytes: + raise ArtifactCacheCapacityError("artifact exceeds the local cache quota") + self._evict_locked( + protected_digest=protected_digest, + incoming_bytes=required_bytes, + ) + with self._connect() as connection: + row = connection.execute( + "SELECT COALESCE(SUM(byte_length), 0) FROM objects" + ).fetchone() + total_bytes = int(row[0]) if row is not None else 0 + free_bytes = shutil.disk_usage(self.root).free + if ( + total_bytes + required_bytes > self.max_bytes + or free_bytes - required_bytes < self.free_space_reserve_bytes + ): + raise ArtifactCacheCapacityError( + "local artifact cache cannot preserve its quota and free-space reserve" + ) + + def _evict_locked( + self, + *, + protected_digest: str | None, + incoming_bytes: int = 0, + ) -> None: + while True: + with self._connect() as connection: + total_row = connection.execute( + "SELECT COALESCE(SUM(byte_length), 0) FROM objects" + ).fetchone() + total_bytes = int(total_row[0]) if total_row is not None else 0 + free_bytes = shutil.disk_usage(self.root).free + if ( + total_bytes + incoming_bytes <= self.max_bytes + and free_bytes - incoming_bytes >= self.free_space_reserve_bytes + ): + return + row = connection.execute( + """ + SELECT objects.sha256 + FROM objects + LEFT JOIN pins ON pins.sha256 = objects.sha256 + WHERE pins.sha256 IS NULL AND objects.sha256 != COALESCE(?, '') + ORDER BY objects.last_access_ns ASC, objects.sha256 ASC + LIMIT 1 + """, + (protected_digest,), + ).fetchone() + if row is None: + return + digest = str(row[0]) + self.object_path(digest).unlink(missing_ok=True) + connection.execute("DELETE FROM objects WHERE sha256 = ?", (digest,)) + + +class ArtifactGateway: + """Resolve named central manifests through a verified local working set.""" + + def __init__(self, store: CentralArtifactStore, cache: LocalArtifactCache) -> None: + self.store = store + self.cache = cache + + def publish( + self, + *, + namespace: str, + key: str, + artifact_type: str, + subject_id: str, + sources: Sequence[tuple[str, str, Path]], + metadata: Mapping[str, str] | None = None, + ) -> ArtifactManifest: + members: list[ArtifactMember] = [] + seen_roles: set[str] = set() + for role, media_type, source in sources: + _validate_component(role, "artifact role") + if role in seen_roles: + raise ValueError(f"duplicate artifact role: {role}") + seen_roles.add(role) + published = self.store.publish_file(source) + members.append( + ArtifactMember( + role=role, + media_type=media_type, + sha256=published.sha256, + byte_length=published.byte_length, + ) + ) + manifest = self.store.publish_manifest( + artifact_type=artifact_type, + subject_id=subject_id, + members=members, + metadata=metadata, + ) + self.store.set_reference(namespace, key, manifest.manifest_id) + reference = self.store.read_reference_document(namespace, key) + self.cache.save_snapshot(namespace, key, reference, manifest) + return manifest + + def resolve_role(self, namespace: str, key: str, role: str) -> ResolvedArtifact: + manifest, central_available = self._resolve_manifest(namespace, key) + member = manifest.member(role) + cached = self.cache.get(member) + if cached is not None: + return ResolvedArtifact( + manifest=manifest, + member=member, + path=cached, + cache_hit=True, + central_available=central_available, + ) + if not central_available: + raise ArtifactStoreUnavailable( + f"artifact role is not in the offline cache: {namespace}/{key}/{role}" + ) + path, cache_hit = self.cache.fetch(self.store, member) + return ResolvedArtifact( + manifest=manifest, + member=member, + path=path, + cache_hit=cache_hit, + central_available=True, + ) + + def pin_reference(self, namespace: str, key: str, *, pin_id: str) -> ArtifactManifest: + manifest, central_available = self._resolve_manifest(namespace, key) + fetched: list[ArtifactMember] = [] + try: + for member in manifest.members: + if self.cache.get(member) is None: + if not central_available: + raise ArtifactStoreUnavailable( + f"artifact role is not in the offline cache: {member.role}" + ) + self.cache.fetch(self.store, member, pin_id=pin_id) + else: + with self.cache._locked(): + self.cache._pin_locked(pin_id, member.sha256) + fetched.append(member) + except BaseException: + self.cache.unpin(pin_id) + raise + return manifest + + def _resolve_manifest( + self, + namespace: str, + key: str, + ) -> tuple[ArtifactManifest, bool]: + try: + reference = self.store.read_reference_document(namespace, key) + manifest = self.store.read_manifest(str(reference["manifest_id"])) + except ArtifactStoreUnavailable: + return self.cache.load_snapshot(namespace, key), False + self.cache.save_snapshot(namespace, key, reference, manifest) + return manifest, True + + +def configured_artifact_gateway(data_dir: Path) -> ArtifactGateway | None: + """Build the optional production gateway from environment-only configuration.""" + + configured_root = os.environ.get(STORE_ROOT_ENV, "").strip() + if not configured_root: + return None + max_bytes = _configured_non_negative_int(CACHE_MAX_BYTES_ENV, DEFAULT_CACHE_MAX_BYTES) + reserve_bytes = _configured_non_negative_int( + CACHE_RESERVE_BYTES_ENV, + DEFAULT_FREE_SPACE_RESERVE_BYTES, + ) + if max_bytes <= 0: + raise ArtifactCacheCapacityError("artifact cache maximum bytes must be positive") + return ArtifactGateway( + CentralArtifactStore(Path(configured_root), create=False), + LocalArtifactCache( + data_dir / "artifact-cache", + max_bytes=max_bytes, + free_space_reserve_bytes=reserve_bytes, + ), + ) + + +def _configured_non_negative_int(name: str, default: int) -> int: + raw = os.environ.get(name, "").strip() + if not raw: + return default + if not raw.isascii() or not raw.isdecimal(): + raise ValueError(f"{name} must be a non-negative integer") + return int(raw) + + +def _regular_source(path: Path) -> Path: + try: + metadata = path.lstat() + resolved = path.expanduser().resolve(strict=True) + except OSError as exc: + raise ArtifactNotFound("artifact source is unavailable") from exc + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): + raise ArtifactIntegrityError("artifact source must be a regular file") + return resolved + + +def _hash_file(path: Path) -> tuple[str, int]: + digest = hashlib.sha256() + byte_length = 0 + try: + with path.open("rb") as stream: + while chunk := stream.read(_COPY_CHUNK_BYTES): + digest.update(chunk) + byte_length += len(chunk) + except OSError as exc: + raise ArtifactStoreUnavailable("artifact object is unavailable") from exc + return digest.hexdigest(), byte_length + + +def _copy_and_hash(source: Path, destination: Path) -> tuple[str, int]: + digest = hashlib.sha256() + byte_length = 0 + with source.open("rb") as input_stream, destination.open("xb") as output_stream: + while chunk := input_stream.read(_COPY_CHUNK_BYTES): + output_stream.write(chunk) + digest.update(chunk) + byte_length += len(chunk) + output_stream.flush() + os.fsync(output_stream.fileno()) + return digest.hexdigest(), byte_length + + +def _verify_object(path: Path, digest: str, byte_length: int) -> None: + metadata = path.lstat() + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): + raise ArtifactIntegrityError("content-addressed object path is invalid") + actual_digest, actual_bytes = _hash_file(path) + if actual_digest != digest or actual_bytes != byte_length: + raise ArtifactIntegrityError("content-addressed object is corrupt") + + +def _validate_members(members: Sequence[ArtifactMember]) -> None: + if not members: + raise ValueError("artifact manifest must contain at least one member") + roles: set[str] = set() + for member in members: + _validate_member(member) + if member.role in roles: + raise ValueError(f"duplicate artifact role: {member.role}") + roles.add(member.role) + + +def _validate_member(member: ArtifactMember) -> None: + _validate_component(member.role, "artifact role") + if not isinstance(member.media_type, str) or not 1 <= len(member.media_type) <= 255: + raise ValueError("artifact media type is invalid") + _validate_sha256(member.sha256) + if not isinstance(member.byte_length, int) or isinstance(member.byte_length, bool): + raise ValueError("artifact byte length is invalid") + if member.byte_length < 0: + raise ValueError("artifact byte length is invalid") + + +def _parse_manifest(document: Mapping[str, Any], manifest_id: str) -> ArtifactManifest: + if ( + document.get("schema_version") != MANIFEST_SCHEMA + or not isinstance(document.get("artifact_type"), str) + or not isinstance(document.get("subject_id"), str) + or not isinstance(document.get("created_at_utc"), str) + or not isinstance(document.get("members"), list) + or not isinstance(document.get("metadata"), dict) + ): + raise ArtifactIntegrityError("artifact manifest shape is invalid") + _validate_component(str(document["artifact_type"]), "artifact type") + _validate_component(str(document["subject_id"]), "artifact subject") + raw_metadata = document["metadata"] + assert isinstance(raw_metadata, dict) + metadata: dict[str, str] = {} + for key, value in raw_metadata.items(): + if not isinstance(key, str) or not isinstance(value, str): + raise ArtifactIntegrityError("artifact manifest metadata is invalid") + _validate_component(key, "metadata key") + metadata[key] = value + raw_members = document["members"] + assert isinstance(raw_members, list) + members: list[ArtifactMember] = [] + for raw in raw_members: + if not isinstance(raw, dict): + raise ArtifactIntegrityError("artifact manifest member is invalid") + try: + member = ArtifactMember( + role=str(raw["role"]), + media_type=str(raw["media_type"]), + sha256=str(raw["sha256"]), + byte_length=int(raw["byte_length"]), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ArtifactIntegrityError("artifact manifest member is invalid") from exc + _validate_member(member) + members.append(member) + _validate_members(members) + if tuple(members) != tuple(sorted(members, key=lambda item: item.role)): + raise ArtifactIntegrityError("artifact manifest members are not canonical") + return ArtifactManifest( + manifest_id=manifest_id, + artifact_type=str(document["artifact_type"]), + subject_id=str(document["subject_id"]), + created_at_utc=str(document["created_at_utc"]), + members=tuple(members), + metadata=metadata, + ) + + +def _manifest_document(manifest: ArtifactManifest) -> dict[str, Any]: + return { + "schema_version": MANIFEST_SCHEMA, + "artifact_type": manifest.artifact_type, + "subject_id": manifest.subject_id, + "created_at_utc": manifest.created_at_utc, + "members": [ + { + "role": item.role, + "media_type": item.media_type, + "sha256": item.sha256, + "byte_length": item.byte_length, + } + for item in manifest.members + ], + "metadata": dict(sorted(manifest.metadata.items())), + } + + +def _validate_reference_snapshot( + document: Mapping[str, Any], + namespace: str, + key: str, +) -> None: + if ( + document.get("schema_version") != REFERENCE_SCHEMA + or document.get("namespace") != namespace + or document.get("key") != key + or not isinstance(document.get("manifest_id"), str) + or _SHA256.fullmatch(str(document["manifest_id"])) is None + or not isinstance(document.get("updated_at_utc"), str) + ): + raise ArtifactIntegrityError("artifact reference snapshot is invalid") + + +def _read_json_document(path: Path, *, unavailable_message: str) -> dict[str, Any]: + try: + metadata = path.lstat() + if ( + stat.S_ISLNK(metadata.st_mode) + or not stat.S_ISREG(metadata.st_mode) + or not 0 < metadata.st_size <= _MAX_MANIFEST_BYTES + ): + raise ArtifactIntegrityError("artifact metadata path is invalid") + value = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise ArtifactStoreUnavailable(unavailable_message) from exc + except OSError as exc: + raise ArtifactStoreUnavailable(unavailable_message) from exc + except json.JSONDecodeError as exc: + raise ArtifactIntegrityError("artifact metadata is not valid JSON") from exc + if not isinstance(value, dict): + raise ArtifactIntegrityError("artifact metadata is not an object") + return value + + +def _write_bytes_atomic(path: Path, payload: bytes, *, mode: int | None = None) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{uuid4().hex}.tmp") + try: + with temporary.open("xb") as stream: + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + if mode is not None: + os.chmod(temporary, mode) + os.replace(temporary, path) + finally: + temporary.unlink(missing_ok=True) + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +def _validate_component(value: str, label: str) -> None: + if not isinstance(value, str) or _SAFE_COMPONENT.fullmatch(value) is None: + raise ValueError(f"{label} is invalid") + + +def _validate_sha256(value: str) -> None: + if not isinstance(value, str) or _SHA256.fullmatch(value) is None: + raise ValueError("artifact SHA-256 is invalid") diff --git a/src/k1link/compute/integrated_perception.py b/src/k1link/compute/integrated_perception.py index fd5b80b..dfe1bbe 100644 --- a/src/k1link/compute/integrated_perception.py +++ b/src/k1link/compute/integrated_perception.py @@ -22,6 +22,11 @@ import numpy as np import rerun as rr from rerun.components import FillMode +from k1link.artifact_gateway import ( + ArtifactGateway, + ArtifactNotFound, + ArtifactStoreUnavailable, +) from k1link.artifacts import write_json_atomic from k1link.sessions import SessionIntegrityError @@ -377,12 +382,14 @@ class IntegratedPerceptionOverlayStore: lidar_packs_root: Path, cache_root: Path, ffmpeg_path: Path, + artifact_gateway: ArtifactGateway | None = None, ) -> None: self.jobs_root = jobs_root.expanduser().absolute() self.results_root = results_root.expanduser().absolute() self.lidar_packs_root = lidar_packs_root.expanduser().absolute() self.cache_root = cache_root.expanduser().absolute() self.ffmpeg_path = ffmpeg_path.expanduser().absolute() + self.artifact_gateway = artifact_gateway self._catalog_lock = threading.Lock() self._descriptor_catalog_generation: tuple[int, int] | None = None self._descriptors_by_session: dict[str, tuple[_ResultDescriptor, ...]] = {} @@ -444,6 +451,15 @@ class IntegratedPerceptionOverlayStore: self._set_status(key, state="preparing", phase="cache-lookup") try: with self._single_flight(key): + central = self._read_gateway_cache(session_id, recording_id) + if central is not None: + self._set_status( + key, + state="ready", + phase="ready", + byte_length=central.byte_length, + ) + return central cached = self._read_admitted_cache(session_id, recording_id) if cached is not None: self._set_status( @@ -534,6 +550,50 @@ class IntegratedPerceptionOverlayStore: self._set_status(key, state="error", phase="error") raise + def _read_gateway_cache( + self, + session_id: str, + recording_id: str, + ) -> RecordedPerceptionOverlayArtifact | None: + if self.artifact_gateway is None: + return None + role = f"integrated-overlay:{recording_id}" + try: + resolved = self.artifact_gateway.resolve_role("sessions", session_id, role) + except (ArtifactNotFound, ArtifactStoreUnavailable): + return None + expected_result_id = resolved.manifest.metadata.get("integrated-result-id") + if ( + resolved.member.media_type != "application/vnd.rerun.rrd" + or expected_result_id is None + or _SAFE_RESULT_ID.fullmatch(expected_result_id) is None + ): + raise RecordedPerceptionOverlayError( + "central integrated perception artifact metadata is invalid" + ) + try: + metadata = resolved.path.lstat() + with resolved.path.open("rb") as stream: + magic = stream.read(4) + except OSError as exc: + raise RecordedPerceptionOverlayError( + "central integrated perception artifact became unavailable" + ) from exc + if ( + stat.S_ISLNK(metadata.st_mode) + or not stat.S_ISREG(metadata.st_mode) + or metadata.st_size != resolved.member.byte_length + or magic != b"RRF2" + ): + raise RecordedPerceptionOverlayError( + "central integrated perception artifact is invalid" + ) + return RecordedPerceptionOverlayArtifact( + path=resolved.path.resolve(strict=True), + byte_length=resolved.member.byte_length, + sha256=resolved.member.sha256, + ) + def status(self, session_id: str, *, recording_id: str) -> dict[str, Any]: if ( _SAFE_RECORDING_ID.fullmatch(session_id) is None diff --git a/src/k1link/device_plugins/xgrids_k1/cli.py b/src/k1link/device_plugins/xgrids_k1/cli.py index 93587a9..28ffc4f 100644 --- a/src/k1link/device_plugins/xgrids_k1/cli.py +++ b/src/k1link/device_plugins/xgrids_k1/cli.py @@ -16,6 +16,7 @@ from rich.console import Console from rich.table import Table from k1link import __version__ +from k1link.artifact_cli import app as artifact_app from k1link.artifacts import write_json_atomic from k1link.compute import ( prepare_camera_compute_job, @@ -90,6 +91,7 @@ app.add_typer(analyze_app, name="analyze") app.add_typer(authority_app, name="authority") app.add_typer(compute_app, name="compute") app.add_typer(lab_app, name="lab") +app.add_typer(artifact_app, name="artifact") class ToolStatus(TypedDict): diff --git a/src/k1link/sessions/recording.py b/src/k1link/sessions/recording.py index 93e3740..81f01bd 100644 --- a/src/k1link/sessions/recording.py +++ b/src/k1link/sessions/recording.py @@ -16,6 +16,13 @@ from pathlib import Path from typing import Any, cast from uuid import uuid4 +from k1link.artifact_gateway import ( + ArtifactGateway, + ArtifactGatewayError, + ArtifactNotFound, + ArtifactStoreUnavailable, +) + from .models import ReplayArtifact, ReplayCommand from .plugin_contract import ( PluginRecordingExportCancelled, @@ -139,6 +146,7 @@ class SessionRecordingMaterializer: exporters: Mapping[str, RecordingExporter] | None = None, cache_max_bytes: int | None = None, free_space_reserve_bytes: int | None = None, + artifact_gateway: ArtifactGateway | None = None, ) -> None: private_root = data_dir.expanduser().resolve() private_root.mkdir(mode=0o700, parents=True, exist_ok=True) @@ -166,6 +174,7 @@ class SessionRecordingMaterializer: environment_name="MISSIONCORE_RRD_FREE_SPACE_RESERVE_BYTES", default=DEFAULT_FREE_SPACE_RESERVE_BYTES, ) + self.artifact_gateway = artifact_gateway self._locks_guard = threading.Lock() self._session_locks: dict[str, threading.Lock] = {} self._memory_guard = threading.Lock() @@ -349,6 +358,11 @@ class SessionRecordingMaterializer: if cached is not None: _report_progress(progress_callback, "ready", 1.0) return cached + _report_progress(progress_callback, "restoring", 0.15) + restored = self._restore_gateway_recording_locked(session_id, source) + if restored is not None: + _report_progress(progress_callback, "ready", 1.0) + return restored _report_progress(progress_callback, "exporting", 0.2) return self._export_recording_locked( session_id, @@ -373,13 +387,114 @@ class SessionRecordingMaterializer: self._scavenge_export_artifacts_locked() source = _validate_source(command) recording = self._load_cached_recording(session_id, source, pin=True) + if recording is None: + recording = self._restore_gateway_recording_locked(session_id, source) if recording is None: recording = self._export_recording_locked(session_id, source) - with self._cache_guard: - self._increment_pin_locked(session_id) + with self._cache_guard: + self._increment_pin_locked(session_id) return recording, self._release_callback(session_id) + def _restore_gateway_recording_locked( + self, + session_id: str, + source: _ValidatedSource, + ) -> MaterializedRecording | None: + if self.artifact_gateway is None: + return None + try: + resolved = self.artifact_gateway.resolve_role( + "sessions", + session_id, + "base-rrd", + ) + except (ArtifactNotFound, ArtifactStoreUnavailable): + return None + except ArtifactGatewayError as exc: + raise RecordingMaterializationError( + "central recording artifact failed validation" + ) from exc + metadata = resolved.manifest.metadata + if ( + resolved.manifest.artifact_type != "recorded-session" + or resolved.manifest.subject_id != session_id + or resolved.member.media_type != RERUN_RECORDING_MEDIA_TYPE + or metadata.get("base-timeline") != RERUN_SESSION_TIMELINE + ): + raise RecordingMaterializationError("central recording manifest is invalid") + source_sha256 = metadata.get("base-source-sha256") + try: + timeline_start_ns = int(metadata["base-timeline-start-ns"]) + timeline_end_ns = int(metadata["base-timeline-end-ns"]) + except (KeyError, TypeError, ValueError) as exc: + raise RecordingMaterializationError( + "central recording timeline metadata is invalid" + ) from exc + artifact_digests = _validated_artifact_digests(source) + if ( + not isinstance(source_sha256, str) + or source_sha256 != artifact_digests[source.primary_artifact_id] + or timeline_start_ns < 0 + or timeline_end_ns < timeline_start_ns + ): + raise RecordingMaterializationError( + "central recording does not match the local source evidence" + ) + with self._cache_guard: + session_root, recording_path, sidecar_path = self._cache_paths(session_id) + self._ensure_cache_capacity( + required_bytes=resolved.member.byte_length + 4 * 1024, + protected_session_id=session_id, + ) + candidate = session_root / f".scene.{uuid4().hex}.candidate.rrd" + try: + digest = hashlib.sha256() + byte_length = 0 + with resolved.path.open("rb") as source_stream, candidate.open("xb") as output: + while chunk := source_stream.read(1024 * 1024): + output.write(chunk) + digest.update(chunk) + byte_length += len(chunk) + output.flush() + os.fsync(output.fileno()) + if ( + digest.hexdigest() != resolved.member.sha256 + or byte_length != resolved.member.byte_length + ): + raise RecordingMaterializationError( + "central recording changed during local restoration" + ) + _chmod_best_effort(candidate, 0o600) + os.replace(candidate, recording_path) + _fsync_directory(session_root) + recording_stat = _regular_file_stat(recording_path, "derived recording") + recording = MaterializedRecording( + session_id=session_id, + path=recording_path, + media_type=RERUN_RECORDING_MEDIA_TYPE, + byte_length=byte_length, + sha256=resolved.member.sha256, + source_sha256=source_sha256, + timeline=RERUN_SESSION_TIMELINE, + timeline_start_ns=timeline_start_ns, + timeline_end_ns=timeline_end_ns, + ) + _write_json_atomic( + sidecar_path, + _cache_document(recording, source, recording_stat), + ) + _chmod_best_effort(sidecar_path, 0o600) + self._remember(recording, source, recording_stat) + _touch_lru(session_root) + return recording + except OSError as exc: + raise RecordingMaterializationError( + "central recording could not be restored locally" + ) from exc + finally: + candidate.unlink(missing_ok=True) + def _release_callback(self, session_id: str) -> Callable[[], None]: released = False release_guard = threading.Lock() diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index 2d927e3..a5bc81b 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -15,6 +15,7 @@ from fastapi.staticfiles import StaticFiles from pydantic import ValidationError from k1link import __version__ +from k1link.artifact_gateway import configured_artifact_gateway from k1link.compute import ( IntegratedPerceptionOverlayStore, RecordedCalibratedFusionStore, @@ -88,9 +89,11 @@ plugin_environment = load_installed_device_plugins(REPOSITORY_ROOT) 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) session_recording_materializer = SessionRecordingMaterializer( session_store.data_dir, exporters=plugin_environment.recording_exporters, + artifact_gateway=session_artifact_gateway, ) session_recorded_media_inspector = RecordedMediaInspector( session_store.data_dir / "recorded-media-preparations" @@ -136,6 +139,7 @@ session_integrated_perception_store = ( ), cache_root=session_store.data_dir / "integrated-perception-overlays", ffmpeg_path=_ffmpeg, + artifact_gateway=session_artifact_gateway, ) if _ffmpeg is not None else None diff --git a/tests/test_artifact_gateway.py b/tests/test_artifact_gateway.py new file mode 100644 index 0000000..2e2e729 --- /dev/null +++ b/tests/test_artifact_gateway.py @@ -0,0 +1,209 @@ +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from k1link.artifact_gateway import ( + ArtifactCacheCapacityError, + ArtifactGateway, + ArtifactIntegrityError, + ArtifactStoreUnavailable, + CentralArtifactStore, + LocalArtifactCache, +) + + +def _gateway( + tmp_path: Path, + *, + max_bytes: int = 1024, +) -> tuple[ArtifactGateway, Path, Path]: + central_root = tmp_path / "central" + local_root = tmp_path / "local" + return ( + ArtifactGateway( + CentralArtifactStore(central_root, create=True), + LocalArtifactCache( + local_root, + max_bytes=max_bytes, + free_space_reserve_bytes=0, + ), + ), + central_root, + local_root, + ) + + +def test_gateway_publishes_deduplicated_objects_and_resolves_through_local_cache( + tmp_path: Path, +) -> None: + gateway, central_root, _local_root = _gateway(tmp_path) + first = tmp_path / "first.rrd" + duplicate = tmp_path / "duplicate.rrd" + first.write_bytes(b"RRF2same-content") + duplicate.write_bytes(first.read_bytes()) + + manifest = gateway.publish( + namespace="sessions", + key="session-1", + artifact_type="recorded-session", + subject_id="session-1", + sources=( + ("base-rrd", "application/vnd.rerun.rrd", first), + ("overlay-rrd", "application/vnd.rerun.rrd", duplicate), + ), + metadata={"display-name": "Session 1"}, + ) + + assert len(manifest.members) == 2 + assert manifest.members[0].sha256 == manifest.members[1].sha256 + objects = tuple((central_root / "objects" / "sha256").glob("*/*")) + assert len(objects) == 1 + + cold = gateway.resolve_role("sessions", "session-1", "overlay-rrd") + assert cold.path.read_bytes() == first.read_bytes() + assert cold.cache_hit is False + assert cold.central_available is True + + warm = gateway.resolve_role("sessions", "session-1", "overlay-rrd") + assert warm.path == cold.path + assert warm.cache_hit is True + + +def test_persistent_pin_survives_restart_and_resolves_with_central_offline( + tmp_path: Path, +) -> None: + gateway, central_root, local_root = _gateway(tmp_path) + source = tmp_path / "overlay.rrd" + source.write_bytes(b"RRF2offline-ready") + gateway.publish( + namespace="sessions", + key="session-1", + artifact_type="recorded-session", + subject_id="session-1", + sources=(("overlay-rrd", "application/vnd.rerun.rrd", source),), + ) + gateway.pin_reference("sessions", "session-1", pin_id="session-1") + offline_root = tmp_path / "central-offline" + os.replace(central_root, offline_root) + + restarted = ArtifactGateway( + gateway.store, + LocalArtifactCache( + local_root, + max_bytes=1024, + free_space_reserve_bytes=0, + ), + ) + + resolved = restarted.resolve_role("sessions", "session-1", "overlay-rrd") + assert resolved.path.read_bytes() == source.read_bytes() + assert resolved.cache_hit is True + assert resolved.central_available is False + status = restarted.cache.status() + assert status.pinned_object_count == 1 + assert status.pinned_bytes == len(source.read_bytes()) + + +def test_offline_cache_miss_is_explicit(tmp_path: Path) -> None: + gateway, central_root, local_root = _gateway(tmp_path) + source = tmp_path / "overlay.rrd" + source.write_bytes(b"RRF2not-fetched") + gateway.publish( + namespace="sessions", + key="session-1", + artifact_type="recorded-session", + subject_id="session-1", + sources=(("overlay-rrd", "application/vnd.rerun.rrd", source),), + ) + os.replace(central_root, tmp_path / "central-offline") + + restarted = ArtifactGateway( + gateway.store, + LocalArtifactCache(local_root, max_bytes=1024, free_space_reserve_bytes=0), + ) + + with pytest.raises(ArtifactStoreUnavailable, match="offline cache"): + restarted.resolve_role("sessions", "session-1", "overlay-rrd") + + +def test_unpinned_lru_is_evicted_but_pinned_object_is_preserved(tmp_path: Path) -> None: + gateway, _central_root, _local_root = _gateway(tmp_path, max_bytes=18) + first = tmp_path / "first.bin" + second = tmp_path / "second.bin" + third = tmp_path / "third.bin" + first.write_bytes(b"a" * 9) + second.write_bytes(b"b" * 9) + third.write_bytes(b"c" * 9) + for key, source in (("first", first), ("second", second), ("third", third)): + gateway.publish( + namespace="objects", + key=key, + artifact_type="test-object", + subject_id=key, + sources=(("payload", "application/octet-stream", source),), + ) + + gateway.pin_reference("objects", "first", pin_id="keep-first") + second_resolved = gateway.resolve_role("objects", "second", "payload") + assert second_resolved.path.is_file() + gateway.resolve_role("objects", "third", "payload") + + first_member = gateway.store.resolve_reference("objects", "first").member("payload") + second_member = gateway.store.resolve_reference("objects", "second").member("payload") + third_member = gateway.store.resolve_reference("objects", "third").member("payload") + assert gateway.cache.get(first_member) is not None + assert gateway.cache.get(second_member) is None + assert gateway.cache.get(third_member) is not None + + +def test_cache_refuses_object_larger_than_its_quota(tmp_path: Path) -> None: + gateway, _central_root, _local_root = _gateway(tmp_path, max_bytes=8) + source = tmp_path / "large.bin" + source.write_bytes(b"x" * 9) + gateway.publish( + namespace="objects", + key="large", + artifact_type="test-object", + subject_id="large", + sources=(("payload", "application/octet-stream", source),), + ) + + with pytest.raises(ArtifactCacheCapacityError, match="exceeds"): + gateway.resolve_role("objects", "large", "payload") + + +def test_same_size_local_corruption_is_detected_after_stat_identity_changes( + tmp_path: Path, +) -> None: + gateway, _central_root, _local_root = _gateway(tmp_path) + source = tmp_path / "payload.bin" + source.write_bytes(b"original") + gateway.publish( + namespace="objects", + key="payload", + artifact_type="test-object", + subject_id="payload", + sources=(("payload", "application/octet-stream", source),), + ) + resolved = gateway.resolve_role("objects", "payload", "payload") + resolved.path.write_bytes(b"tampered") + os.utime(resolved.path, ns=(resolved.path.stat().st_atime_ns, 1)) + + member = gateway.store.resolve_reference("objects", "payload").member("payload") + assert gateway.cache.get(member) is None + repaired = gateway.resolve_role("objects", "payload", "payload") + assert repaired.path.read_bytes() == b"original" + + +def test_store_rejects_corrupt_existing_content_addressed_object(tmp_path: Path) -> None: + store = CentralArtifactStore(tmp_path / "central", create=True) + source = tmp_path / "source.bin" + source.write_bytes(b"original") + published = store.publish_file(source) + store.object_path(published.sha256).write_bytes(b"tampered") + + with pytest.raises(ArtifactIntegrityError, match="corrupt"): + store.publish_file(source) diff --git a/tests/test_e10_integrated_perception.py b/tests/test_e10_integrated_perception.py index 7572180..96d8280 100644 --- a/tests/test_e10_integrated_perception.py +++ b/tests/test_e10_integrated_perception.py @@ -9,11 +9,18 @@ import time from concurrent.futures import ThreadPoolExecutor from pathlib import Path from types import SimpleNamespace +from typing import cast import numpy as np import pytest import k1link.compute.integrated_perception as integrated_module +from k1link.artifact_gateway import ( + ArtifactGateway, + ArtifactManifest, + ArtifactMember, + ResolvedArtifact, +) from k1link.compute.integrated_perception import ( IntegratedPerceptionOverlayStore, _CuboidPresentationState, @@ -227,6 +234,75 @@ def test_integrated_overlay_serializes_heavy_materialization_across_sessions( assert maximum_active == 1 +def test_integrated_overlay_uses_verified_central_cache_before_local_result_scan( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + for name in ("jobs", "results", "lidar-packs"): + (tmp_path / name).mkdir() + recording_id = "recording-1" + result_id = f"e10-integrated-perception-{'a' * 64}" + payload = b"RRF2central-overlay" + artifact_path = tmp_path / "artifact-cache" / "overlay.rrd" + artifact_path.parent.mkdir() + artifact_path.write_bytes(payload) + member = ArtifactMember( + role=f"integrated-overlay:{recording_id}", + media_type="application/vnd.rerun.rrd", + sha256=hashlib.sha256(payload).hexdigest(), + byte_length=len(payload), + ) + manifest = ArtifactManifest( + manifest_id="b" * 64, + artifact_type="recorded-session", + subject_id="session-1", + created_at_utc="2026-07-30T00:00:00Z", + members=(member,), + metadata={"integrated-result-id": result_id}, + ) + + class FakeGateway: + def resolve_role(self, namespace: str, key: str, role: str) -> ResolvedArtifact: + assert (namespace, key, role) == ( + "sessions", + "session-1", + f"integrated-overlay:{recording_id}", + ) + return ResolvedArtifact( + manifest=manifest, + member=member, + path=artifact_path, + cache_hit=False, + central_available=True, + ) + + store = IntegratedPerceptionOverlayStore( + jobs_root=tmp_path / "jobs", + results_root=tmp_path / "results", + lidar_packs_root=tmp_path / "lidar-packs", + cache_root=tmp_path / "cache", + ffmpeg_path=tmp_path / "ffmpeg", + artifact_gateway=cast(ArtifactGateway, FakeGateway()), + ) + monkeypatch.setattr( + store, + "_latest_descriptor", + lambda _session_id: (_ for _ in ()).throw( + AssertionError("central cache hit must not scan local results") + ), + ) + + artifact = store.materialize( + "session-1", + application_id="nodedc_mission_core_recorded", + recording_id=recording_id, + ) + + assert artifact is not None + assert artifact.path == artifact_path + assert artifact.sha256 == member.sha256 + + def test_e10_profile_pins_integrated_realtime_budget() -> None: _fusion, runner = _worker_modules() profile_path = ( diff --git a/tests/test_session_recording.py b/tests/test_session_recording.py index 6d65940..a77c736 100644 --- a/tests/test_session_recording.py +++ b/tests/test_session_recording.py @@ -13,6 +13,11 @@ from types import SimpleNamespace import pytest import k1link.sessions.recording as recording_module +from k1link.artifact_gateway import ( + ArtifactGateway, + CentralArtifactStore, + LocalArtifactCache, +) from k1link.sessions.models import ReplayArtifact, ReplayCommand from k1link.sessions.recording import ( CACHE_SCHEMA, @@ -131,6 +136,54 @@ def test_materializer_reuses_only_a_digest_validated_private_cache(tmp_path: Pat assert exporter.calls == 1 +def test_materializer_restores_exact_central_recording_without_export( + tmp_path: Path, +) -> None: + command = _command(tmp_path / "source") + central_recording = tmp_path / "central.rrd" + central_recording.write_bytes(b"RRF2centrally-prepared-recording") + gateway = ArtifactGateway( + CentralArtifactStore(tmp_path / "central-store", create=True), + LocalArtifactCache( + tmp_path / "artifact-cache", + max_bytes=1024 * 1024, + free_space_reserve_bytes=0, + ), + ) + gateway.publish( + namespace="sessions", + key=command.session_id, + artifact_type="recorded-session", + subject_id=command.session_id, + sources=( + ("base-rrd", RERUN_RECORDING_MEDIA_TYPE, central_recording), + ), + metadata={ + "base-source-sha256": _sha256(command.primary_artifact.path), + "base-timeline": "session_time", + "base-timeline-start-ns": "0", + "base-timeline-end-ns": "2500000000", + }, + ) + exporter = FakeExporter() + materializer = SessionRecordingMaterializer( + tmp_path / "private", + exporter=exporter, + artifact_gateway=gateway, + ) + + restored = materializer.materialize(command) + + assert exporter.calls == 0 + assert restored.path.is_relative_to(materializer.recordings_root) + assert restored.path.read_bytes() == central_recording.read_bytes() + assert restored.sha256 == _sha256(central_recording) + assert restored.source_sha256 == _sha256(command.primary_artifact.path) + assert restored.timeline_end_ns == 2_500_000_000 + assert materializer.materialize(command) == restored + assert exporter.calls == 0 + + def test_default_recording_cache_has_no_application_byte_quota( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,