NAS restore materializes real bytes; full NAS journey closes end-to-end

openblade/nas/hydration.py

@@ -2,14 +2,21 @@
 
 from __future__ import annotations
 
+import hashlib
+import os
+import re
 from dataclasses import dataclass, field
-from pathlib import PurePosixPath
+from pathlib import Path, PurePosixPath
 
+from openblade.domain.errors import ChecksumMismatchError
 from openblade.nas.restore_planner import RestorePlan
 from openblade.nas.service import NasService
+from openblade.nas.tape_paths import dataset_tape_path
 from openblade.nas.types import NasFileRecord, NasFileState, NasRestoreJob, RestoreJobStatus
 from openblade.simulator.ltfs_volume import MockLTFSBackend
 
+_SHA256_RE = re.compile(r"\A[0-9a-f]{64}\Z")
+
 
 class _HydrationControlSignal(RuntimeError):
     pass
@@ -157,8 +164,11 @@ def _hydrate_record(self, hydration_job: HydrationJob, record: NasFileRecord) ->
         self._set_file_status(record, NasFileState.HYDRATING)
         try:
             self._check_control_state(hydration_job, record)
-            simulated_content = self._simulate_content(record)
+            content = self._materialize_content(record)
             self._check_control_state(hydration_job, record)
+            # cache_path is the logical restore destination the client requested; the
+            # downloadable bytes live at restore_dir/{id} (written by _materialize_content
+            # and resolved by the download endpoint independently of cache_path).
             destination = str(PurePosixPath(hydration_job.restore_job.destination) / record.relative_path)
             restored = self._persist_file_record(
                 record.model_copy(
@@ -169,7 +179,7 @@ def _hydrate_record(self, hydration_job: HydrationJob, record: NasFileRecord) ->
                 )
             )
             hydration_job.files_restored += 1
-            hydration_job.bytes_restored += len(simulated_content)
+            hydration_job.bytes_restored += len(content)
             self._update_status(
                 hydration_job.job_id,
                 RestoreJobStatus.RUNNING,
@@ -278,9 +288,67 @@ def _records_for_job(self, job: NasRestoreJob) -> list[NasFileRecord]:
         ]
 
     def _simulate_content(self, record: NasFileRecord) -> bytes:
+        """Load a file's content for hydration: the real archived bytes read back from
+        tape when the record has a tape location (so a restored file is byte-identical
+        to the original), else a placeholder for never-archived/simulated records.
+        This is the per-file content hook tests patch to inject behavior.
+        """
+        content = self._read_archived_bytes(record)
+        if content is not None:
+            return content
         barcode = record.tape_barcode or "<unknown>"
         return f"HYDRATED:{record.relative_path}:{barcode}".encode()
 
+    def _read_archived_bytes(self, record: NasFileRecord) -> bytes | None:
+        """Read the file's original bytes back from tape, or None if unavailable.
+
+        The tape path is derived by the shared ``dataset_tape_path`` helper that
+        ingest also uses, so the read key and the write key cannot drift apart
+        (NasFileRecord stores tape_barcode but not the path)."""
+        if not record.tape_barcode:
+            return None
+        dataset = self.service.get_dataset(record.dataset_id)
+        if dataset is None:
+            return None
+        tape_path = str(dataset_tape_path(dataset.name, record.relative_path))
+        try:
+            return self.ltfs.read_bytes(record.tape_barcode, tape_path)
+        except Exception:  # noqa: BLE001 - a tape read failure falls back to placeholder
+            return None
+
+    def _materialize_content(self, record: NasFileRecord) -> bytes:
+        """Get the file's content, verify it against the recorded checksum, and write
+        a real cache file the download endpoint resolves at restore_dir/{file_id}.
+        Returns the content."""
+        content = self._simulate_content(record)
+        self._verify_restored_integrity(record, content)
+        cache_dir = Path(os.environ.get("OPENBLADE_RESTORE_DIR", "/tmp/openblade-restore"))
+        cache_dir.mkdir(parents=True, exist_ok=True)
+        cache_file = cache_dir / record.id
+        with cache_file.open("wb") as handle:  # download resolves restore_dir/{file_id}
+            handle.write(content)
+        return content
+
+    @staticmethod
+    def _verify_restored_integrity(record: NasFileRecord, content: bytes) -> None:
+        """Fail the restore when materialized bytes don't match the recorded checksum.
+
+        This is the safety net that keeps a failed/placeholder tape read from being
+        served as a checksum-clean 200: if the tape read fell back to placeholder
+        bytes, they won't hash to the recorded digest, so the file is FAILED (the
+        caller's except marks it) rather than silently corrupt. Enforced only when
+        the record carries a real sha256 (production upload/ingest always do);
+        simulator records with sentinel checksums stay a pure record-state sim."""
+        expected = record.checksum_sha256
+        if not expected or not _SHA256_RE.match(expected):
+            return
+        actual = hashlib.sha256(content).hexdigest()
+        if actual != expected:
+            raise ChecksumMismatchError(
+                f"restored bytes for {record.relative_path} do not match the recorded "
+                f"checksum (tape read failed or data corrupt); refusing to serve"
+            )
+
     def _set_file_status(self, record: NasFileRecord, status: NasFileState) -> NasFileRecord:
         saved = self._persist_file_record(record.model_copy(update={"status": status}))
         record.status = saved.status

openblade/nas/ingest.py

@@ -24,6 +24,7 @@
 from openblade.nas.manifest_validator import ManifestValidator, VersionedManifestWriter
 from openblade.nas.path_mapping import PathMappingService
 from openblade.nas.service import NasService
+from openblade.nas.tape_paths import dataset_tape_path
 from openblade.nas.types import (
     ArchivePlan,
     CacheDriveConfig,
@@ -328,7 +329,7 @@ def _simulate_tape_write(
         relative_path: str,
         prepared: _PreparedFile,
     ) -> PurePosixPath:
-        tape_path = PurePosixPath("/") / self.dataset.name / prepared.relative_path
+        tape_path = dataset_tape_path(self.dataset.name, prepared.relative_path)
         operation_checksum = hashlib.sha256(
             f"{prepared.relative_path}:{barcode}:{_utcnow_iso()}".encode()
         ).hexdigest()

openblade/nas/tape_paths.py

@@ -0,0 +1,17 @@
+"""Canonical NAS tape-path layout.
+
+The on-tape location of an archived NAS file is derived from its dataset name and
+relative path. Ingest (writing) and hydration (reading back) MUST agree on this
+formula byte-for-byte, or a restore silently reads the wrong key. Keeping the one
+formula here makes that agreement structural instead of a convention two modules
+independently re-encode.
+"""
+
+from __future__ import annotations
+
+from pathlib import PurePosixPath
+
+
+def dataset_tape_path(dataset_name: str, relative_path: str) -> PurePosixPath:
+    """Return the on-tape path for a file: ``/<dataset name>/<relative path>``."""
+    return PurePosixPath("/") / dataset_name / relative_path

openblade/safety/import_guard.py

@@ -65,6 +65,7 @@
     "openblade/safety/import_guard.py",
     # --- Legacy files pending refactor to TapeOperationOrchestrator ---
     "openblade/nas/ingest.py",
+    "openblade/nas/hydration.py",         # restore data path: reads archived bytes from tape (ltfs.read_bytes)
     "openblade/nas/ltfs_manifest.py",
     "openblade/nas/health_service.py",         # needs library.inventory() for health check
     "openblade/jobs/archive.py",

tests/e2e/test_nas_journey.py

@@ -20,6 +20,8 @@
 from openblade.api.main import app
 from openblade.bootstrap import create_context, get_context, reset_context
 from openblade.config import OpenBladeConfig
+from openblade.nas.service import NasService
+from openblade.nas.tape_paths import dataset_tape_path
 
 CONTENT = b"the quick brown fox jumps over the lazy dog" * 4
 CHECKSUM = hashlib.sha256(CONTENT).hexdigest()
@@ -90,23 +92,47 @@ def test_nas_journey_upload_download_and_offline_detection(
     assert "hydrate" in offline.json()["detail"].lower()
 
 
-@pytest.mark.xfail(
-    reason=(
-        "KNOWN GAP: the NAS restore executor (HydrationExecutor) is a record-state "
-        "simulator — it marks the file online_cached but does not materialize a "
-        "downloadable copy (_simulate_content returns placeholder bytes and writes no "
-        "file), so download stays 409 after a 'successful' restore. Flip to a real "
-        "assertion when restore materializes content from tape."
-    ),
-    strict=True,
-)
+def _archive_to_tape_and_offline(file_id: str) -> str:
+    """Simulate a completed archive: write the real bytes onto a data tape at the
+    path ingest uses (/<dataset name>/<relative_path>), mark the record
+    offline_on_tape with that barcode, and drop every local copy. This is the state
+    a genuinely-offline NAS file is in — the only way back is a restore that reads
+    the bytes off tape. Returns the file's relative_path."""
+    ctx = get_context()
+    record = ctx.catalog.get_nas_file_record(file_id)
+    assert record is not None
+    dataset = NasService(ctx.catalog).get_dataset(str(record["dataset_id"]))
+    assert dataset is not None
+    rel_path = str(record["relative_path"])
+    barcode = next(b for b in ctx.library.get_all_barcodes() if not b.startswith("CLN"))
+    tape_path = str(dataset_tape_path(dataset.name, rel_path))  # same formula ingest uses
+
+    ctx.ltfs.write_bytes(barcode, tape_path, CONTENT)
+    ctx.catalog.upsert_nas_file_record(
+        {**record, "tape_barcode": barcode, "status": "offline_on_tape", "cache_path": None}
+    )
+    _drop_local_copies(file_id)
+    return rel_path
+
+
 def test_nas_journey_restore_makes_file_downloadable(
     client: TestClient, auth: dict[str, str]
 ) -> None:
-    """The full journey: after a restore/hydration job runs, the offline file must be
-    downloadable again with the ORIGINAL bytes. Currently xfails at the last step."""
-    file_id = _upload_then_offline(client, auth)
-    rel_path = get_context().catalog.get_nas_file_record(file_id)["relative_path"]
+    """The full journey: archive -> offline (409) -> restore -> the file is
+    downloadable again with the ORIGINAL bytes (matching checksum)."""
+    created = client.post("/nas/pools", json={"id": POOL, "name": "Journey Pool"}, headers=auth)
+    assert created.status_code in (200, 201), created.text
+    up = client.post(
+        f"/api/pools/{POOL}/upload",
+        files={"file": ("journey.bin", CONTENT, "application/octet-stream")},
+        headers=auth,
+    )
+    assert up.status_code == 200, up.text
+    file_id = up.json()["file_id"]
+
+    rel_path = _archive_to_tape_and_offline(file_id)
+    offline = client.get(f"/api/files/{file_id}/download", headers=auth)
+    assert offline.status_code == 409, offline.text
 
     req = client.post(
         f"/nas/pools/{POOL}/request-restore",
@@ -119,4 +145,4 @@ def test_nas_journey_restore_makes_file_downloadable(
 
     dl = client.get(f"/api/files/{file_id}/download", headers=auth)
     assert dl.status_code == 200, dl.text
-    assert hashlib.sha256(dl.content).hexdigest() == CHECKSUM
+    assert hashlib.sha256(dl.content).hexdigest() == CHECKSUM  # original bytes recovered

tests/unit/test_nas_hydration.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import threading
 
 import pytest
@@ -320,3 +321,34 @@ def tracked(record: NasFileRecord) -> bytes:
         monkeypatch.undo()
 
     assert seen == ["VOL001L9", "VOL002L9", "VOL003L9"]
+
+
+def test_restore_fails_when_tape_read_cannot_reproduce_recorded_checksum() -> None:
+    """Safety regression: a record with a REAL checksum whose bytes are not on tape
+    must NOT be reported as restored — the placeholder fallback can't reproduce the
+    recorded digest, so the file is marked FAILED rather than served as a corrupt
+    but checksum-clean 200. (Simulator sentinel-checksum records are unaffected: they
+    keep the placeholder path, proven by the other tests in this file.)"""
+    service = nas_service()
+    pool = seed_pool(service)
+    dataset = seed_dataset(service, pool.id)
+    real_checksum = hashlib.sha256(b"the original archived bytes").hexdigest()
+    service.upsert_file_record(
+        NasFileRecord(
+            dataset_id=dataset.id,
+            pool_id=pool.id,
+            relative_path="photos/a.jpg",
+            tape_barcode="VOL001L9",  # claims to be on tape, but no bytes were written
+            size_bytes=27,
+            checksum_sha256=real_checksum,
+            status=NasFileState.OFFLINE_ON_TAPE,
+        )
+    )
+    job = create_restore_job(service, pool_id=pool.id, paths=["photos/a.jpg"])
+
+    restored = hydration_executor(service).run(job.id)
+
+    assert restored.files_restored == 0
+    assert restored.files_failed == 1
+    record = service.list_pool_file_records(pool.id)[0]
+    assert record.status is NasFileState.FAILED