haz_cache/

restore.rs

//! [`Cache::restore`] per `CACHE-019` and `CACHE-020`.
//!
//! Restoration materialises the recorded outputs of a cache
//! entry at their workspace-absolute paths and returns the
//! captured `stdout`/`stderr` byte streams for the caller to
//! emit. The caller drives stream emission (subject to the
//! task's `output.mode`); the cache layer is only responsible
//! for producing the bytes and publishing the files.
//!
//! Two-phase publish per `CACHE-020`:
//!
//! 1. **Stage.** A per-restore directory under [`crate::layout::cache_root`]
//!    named `.restore-<hex-key>-<random>/` collects every output
//!    blob, byte-identical to what the entry directory records.
//!    Each staged file is written with the recorded mode and
//!    `fsync`-ed. The target's parent directory is created in
//!    this phase too, so phase 2 can be pure renames.
//! 2. **Publish.** Each staged file is renamed onto its target
//!    workspace-absolute path. Renames are atomic on the host
//!    filesystem (same FS, since both source and target sit
//!    under `<workspace_root>`).
//!
//! The cache holds the workspace root so that
//! `workspace_absolute_path` strings recorded in the manifest
//! (rooted at `/`) can be mapped to real filesystem paths.
//!
//! Failure handling matches the spec's "all or nothing" intent
//! best-effort: every failure inside `restore` returns an error,
//! and the staging directory is wiped on the way out (success or
//! failure) so transient publishing state is not leaked. If a
//! failure occurs partway through phase 2, some targets are
//! published and others are not; the caller is expected to treat
//! the result as a miss and re-run the task fresh, per
//! `CACHE-020` second paragraph.

use std::path::{Path, PathBuf};

use haz_domain::path::CanonicalPath;
use haz_vfs::{FsError, WritableFilesystem};
use snafu::{ResultExt, Snafu};

use crate::cache::Cache;
use crate::layout;
use crate::manifest::Manifest;

/// Captured streams returned to the caller on a successful
/// restore. The caller decides how to emit them per the task's
/// configured `output.mode` (`CACHE-019` steps 2 and 3); the
/// cache is opinion-free on emission.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RestoredStreams {
    /// Bytes that were captured on the run's `stdout`.
    pub stdout: Vec<u8>,
    /// Bytes that were captured on the run's `stderr`.
    pub stderr: Vec<u8>,
}

/// Failure modes for [`Cache::restore`].
#[derive(Debug, Snafu)]
pub enum RestoreError {
    /// Underlying filesystem error during one of the restore
    /// phases (reading a cached blob, staging the copy, creating
    /// the target's parent directory, renaming onto the target).
    /// The wrapped [`FsError`] carries the specific path.
    #[snafu(display("filesystem error during cache restore: {source}"))]
    Io {
        /// The originating filesystem error.
        source: FsError,
    },
}

impl<Fs: WritableFilesystem> Cache<Fs> {
    /// Restore the cache entry described by `manifest` per
    /// `CACHE-019`.
    ///
    /// Materialises every output declared in the manifest at its
    /// workspace-absolute path with the recorded mode and
    /// returns the captured `stdout`/`stderr` bytes. The caller
    /// MUST have just obtained `manifest` from
    /// [`Cache::lookup`]; the cache trusts the manifest content
    /// (paths, content hashes, sizes, modes) as truth and does
    /// not re-verify it.
    ///
    /// On error, the staging directory under the cache root is
    /// removed regardless of which phase failed, so no transient
    /// scratch space leaks. If the error occurs after one or
    /// more targets have already been renamed onto, those
    /// targets remain published; the caller MUST treat the error
    /// as a miss and re-run the task fresh
    /// (`CACHE-020` second paragraph).
    ///
    /// # Errors
    ///
    /// Returns [`RestoreError::Io`] wrapping the underlying
    /// [`FsError`] if any filesystem operation along the phases
    /// fails.
    pub fn restore(&self, manifest: &Manifest) -> Result<RestoredStreams, RestoreError> {
        let suffix = random_suffix_hex();
        let stage_dir = layout::restore_staging_dir(self.cache_root(), &manifest.key, &suffix);
        let result = self.restore_inner(manifest, &stage_dir);
        // Best-effort cleanup of the staging directory. On
        // success it is already empty after every rename; on
        // failure it may hold staged files that never made it to
        // their targets. Either way we drop it. We deliberately
        // ignore errors here; the caller is informed of the
        // primary failure via `result`.
        let _ = self.fs().remove_dir_all(&stage_dir);
        result
    }

    fn restore_inner(
        &self,
        manifest: &Manifest,
        stage_dir: &Path,
    ) -> Result<RestoredStreams, RestoreError> {
        self.fs().create_dir_all(stage_dir).context(IoSnafu)?;

        let stdout = self
            .fs()
            .read(&layout::stdout_path(self.cache_root(), &manifest.key))
            .context(IoSnafu)?;
        let stderr = self
            .fs()
            .read(&layout::stderr_path(self.cache_root(), &manifest.key))
            .context(IoSnafu)?;

        // Stage outputs into the sibling restore directory and
        // prepare each target's parent.
        let mut planned: Vec<(PathBuf, PathBuf)> = Vec::with_capacity(manifest.outputs.len());
        for (i, blob) in manifest.outputs.iter().enumerate() {
            let src =
                layout::output_blob_path(self.cache_root(), &manifest.key, &blob.content_hash);
            let bytes = self.fs().read(&src).context(IoSnafu)?;
            let staged = stage_dir.join(format!("{i:08}"));
            self.fs().write_file(&staged, &bytes).context(IoSnafu)?;
            self.fs()
                .set_permissions(&staged, blob.mode)
                .context(IoSnafu)?;
            self.fs().fsync_file(&staged).context(IoSnafu)?;

            let target =
                workspace_path_from_canonical(self.workspace_root(), &blob.workspace_absolute_path);
            if let Some(parent) = target.parent() {
                self.fs().create_dir_all(parent).context(IoSnafu)?;
            }
            planned.push((staged, target));
        }

        // Atomically publish each staged file into place.
        for (staged, target) in &planned {
            self.fs().rename(staged, target).context(IoSnafu)?;
        }

        Ok(RestoredStreams { stdout, stderr })
    }
}

/// Map a workspace-anchored [`CanonicalPath`] onto a real
/// filesystem path by joining each validated segment onto
/// `workspace_root`.
///
/// Walks segments rather than concatenating the rendered string,
/// so a host-OS separator inside a single segment cannot be
/// reinterpreted as a separator. This is belt-and-braces: the
/// [`CanonicalPath`] type's construction already rejects
/// segments that contain `/` (`PATH-002`) or that resolve to `.`
/// or `..`. The segment walk preserves that invariant across the
/// boundary into [`std::path::PathBuf`].
fn workspace_path_from_canonical(workspace_root: &Path, canonical: &CanonicalPath) -> PathBuf {
    let mut p = workspace_root.to_path_buf();
    for segment in canonical.segments() {
        p.push(segment.as_str());
    }
    p
}

/// 16 lowercase hex characters of randomness, same shape as the
/// store-time tmp suffix. Kept module-local rather than shared
/// with `store.rs` to avoid a thin shared module for a four-line
/// helper; the two call sites are independent and divergence is
/// not a concern.
fn random_suffix_hex() -> String {
    let r: u64 = rand::random();
    format!("{r:016x}")
}

#[cfg(test)]
mod tests {
    use std::path::{Path, PathBuf};

    use haz_domain::settings::cache::HashAlgo;
    use haz_vfs::{Filesystem, MemFilesystem, WritableFilesystem};

    use crate::cache::Cache;
    use crate::key::CacheKey;
    use crate::store::{StoreInputs, StoredOutput};

    const WORKSPACE_ROOT: &str = "/ws";

    fn sample_key() -> CacheKey {
        let mut bytes = [0u8; 32];
        bytes[0] = 0xAB;
        bytes[1] = 0xCD;
        CacheKey::from_bytes(bytes)
    }

    fn make_cache(fs: MemFilesystem, algo: HashAlgo) -> Cache<MemFilesystem> {
        Cache::new(fs, Path::new(WORKSPACE_ROOT), algo)
    }

    /// Build a [`MemFilesystem`] preloaded with a workspace and
    /// a single output file on disk.
    fn fs_with_one_output(target: &Path, bytes: &[u8], mode: u32) -> MemFilesystem {
        let mut fs = MemFilesystem::new();
        fs.add_dir(target.parent().unwrap()).unwrap();
        fs.add_file_with_mode(target, bytes.to_vec(), mode).unwrap();
        fs
    }

    /// Drive store then restore as the executor would. Returns
    /// the cache (so tests can inspect on-disk state after) and
    /// the [`RestoredStreams`] returned by `restore`.
    fn store_then_restore(
        fs: MemFilesystem,
        algo: HashAlgo,
        outputs: &[StoredOutput<'_>],
        stdout: &[u8],
        stderr: &[u8],
    ) -> (
        Cache<MemFilesystem>,
        crate::restore::RestoredStreams,
        crate::manifest::Manifest,
    ) {
        let cache = make_cache(fs, algo);
        let key = sample_key();
        let inputs = StoreInputs {
            outputs,
            stdout,
            stderr,
            created_at_unix: 1_715_700_000,
        };
        cache.store(&key, &inputs).unwrap();
        let manifest = cache.lookup(&key).expect("store should produce a hit");
        let restored = cache.restore(&manifest).expect("restore should succeed");
        (cache, restored, manifest)
    }

    // ---- happy path: round-trip ----

    #[test]
    fn cache_019_restore_after_store_round_trips_outputs() {
        let blob = b"hello-world";
        let target = PathBuf::from("/ws/proj/out");
        let fs = fs_with_one_output(&target, blob, 0o644);

        let outs = [StoredOutput {
            workspace_absolute_path: "/proj/out",
            on_disk_path: &target,
            mode: 0o644,
        }];
        let (cache, _restored, _manifest) = store_then_restore(
            fs,
            HashAlgo::Blake3,
            &outs,
            b"stdout-bytes",
            b"stderr-bytes",
        );

        // Target on disk holds the restored bytes.
        let got = cache.fs().read(&target).unwrap();
        assert_eq!(got, blob);
        let mode = cache.fs().mode_of(&target).unwrap();
        assert_eq!(mode, 0o644);
    }

    #[test]
    fn cache_019_restore_returns_captured_stdout_and_stderr_bytes() {
        let blob = b"";
        let target = PathBuf::from("/ws/proj/out");
        let fs = fs_with_one_output(&target, blob, 0o644);
        let outs = [StoredOutput {
            workspace_absolute_path: "/proj/out",
            on_disk_path: &target,
            mode: 0o644,
        }];
        let (_cache, restored, _manifest) =
            store_then_restore(fs, HashAlgo::Blake3, &outs, b"out-bytes\n", b"err-bytes\n");
        assert_eq!(restored.stdout, b"out-bytes\n");
        assert_eq!(restored.stderr, b"err-bytes\n");
    }

    // ---- degenerate input shapes ----

    #[test]
    fn cache_019_restore_with_no_outputs_returns_empty_streams_when_streams_are_empty() {
        let mut fs = MemFilesystem::new();
        fs.add_dir("/ws").unwrap();
        let (_cache, restored, manifest) = store_then_restore(fs, HashAlgo::Blake3, &[], b"", b"");
        assert!(restored.stdout.is_empty());
        assert!(restored.stderr.is_empty());
        assert_eq!(manifest.outputs.len(), 0);
    }

    #[test]
    fn cache_019_restore_with_multiple_outputs_materialises_each_at_its_path() {
        let mut fs = MemFilesystem::new();
        fs.add_dir("/ws/proj").unwrap();
        fs.add_file_with_mode("/ws/proj/a", b"alpha".to_vec(), 0o644)
            .unwrap();
        fs.add_file_with_mode("/ws/proj/b", b"beta-bytes".to_vec(), 0o755)
            .unwrap();
        let on_a = PathBuf::from("/ws/proj/a");
        let on_b = PathBuf::from("/ws/proj/b");
        let outs = [
            StoredOutput {
                workspace_absolute_path: "/proj/a",
                on_disk_path: &on_a,
                mode: 0o644,
            },
            StoredOutput {
                workspace_absolute_path: "/proj/b",
                on_disk_path: &on_b,
                mode: 0o755,
            },
        ];
        let (cache, _restored, _manifest) =
            store_then_restore(fs, HashAlgo::Blake3, &outs, b"", b"");
        assert_eq!(cache.fs().read(&on_a).unwrap(), b"alpha");
        assert_eq!(cache.fs().read(&on_b).unwrap(), b"beta-bytes");
        assert_eq!(cache.fs().mode_of(&on_a).unwrap(), 0o644);
        assert_eq!(cache.fs().mode_of(&on_b).unwrap(), 0o755);
    }

    // ---- intermediate parent directories ----

    #[test]
    fn cache_019_restore_creates_missing_intermediate_directories_for_target() {
        let blob = b"deep-output";
        let target = PathBuf::from("/ws/proj/nested/deep/out");
        // Build fs with the deep file present (so store can read
        // it), then drop the nested chain BEFORE restore to model
        // the target's parent vanishing between store and
        // restore. We do this by issuing a fresh store on a
        // brand-new filesystem.
        let mut fs = MemFilesystem::new();
        fs.add_dir("/ws/proj/nested/deep").unwrap();
        fs.add_file_with_mode(&target, blob.to_vec(), 0o644)
            .unwrap();

        let cache = make_cache(fs, HashAlgo::Blake3);
        let key = sample_key();
        let outs = [StoredOutput {
            workspace_absolute_path: "/proj/nested/deep/out",
            on_disk_path: &target,
            mode: 0o644,
        }];
        let inputs = StoreInputs {
            outputs: &outs,
            stdout: b"",
            stderr: b"",
            created_at_unix: 0,
        };
        cache.store(&key, &inputs).unwrap();

        // Now wipe the workspace's proj/ tree to simulate
        // "outputs are gone between store and restore".
        cache.fs().remove_dir_all(Path::new("/ws/proj")).unwrap();

        let manifest = cache.lookup(&key).expect("entry still hits");
        cache
            .restore(&manifest)
            .expect("restore must re-create the path");
        assert_eq!(cache.fs().read(&target).unwrap(), blob);
    }

    // ---- overwrite of existing target ----

    #[test]
    fn cache_020_cache_019_restore_overwrites_an_existing_target_file() {
        let target = PathBuf::from("/ws/proj/out");
        let fs = fs_with_one_output(&target, b"original", 0o644);
        let outs = [StoredOutput {
            workspace_absolute_path: "/proj/out",
            on_disk_path: &target,
            mode: 0o644,
        }];
        let cache = make_cache(fs, HashAlgo::Blake3);
        let key = sample_key();
        cache
            .store(
                &key,
                &StoreInputs {
                    outputs: &outs,
                    stdout: b"",
                    stderr: b"",
                    created_at_unix: 0,
                },
            )
            .unwrap();

        // Mutate the file in place to model a divergent run.
        cache.fs().write_file(&target, b"divergent").unwrap();

        let manifest = cache.lookup(&key).unwrap();
        cache.restore(&manifest).unwrap();
        assert_eq!(cache.fs().read(&target).unwrap(), b"original");
    }

    // ---- I/O errors propagate ----

    #[test]
    fn cache_019_restore_propagates_missing_cached_blob_as_io_error() {
        let target = PathBuf::from("/ws/proj/out");
        let fs = fs_with_one_output(&target, b"x", 0o644);
        let cache = make_cache(fs, HashAlgo::Blake3);
        let key = sample_key();
        let outs = [StoredOutput {
            workspace_absolute_path: "/proj/out",
            on_disk_path: &target,
            mode: 0o644,
        }];
        cache
            .store(
                &key,
                &StoreInputs {
                    outputs: &outs,
                    stdout: b"",
                    stderr: b"",
                    created_at_unix: 0,
                },
            )
            .unwrap();

        let manifest = cache.lookup(&key).unwrap();

        // Tamper: delete the cache entry directory after the
        // lookup but before the restore. Lookup observed the
        // entry; restore must surface the missing-blob failure.
        let entry = crate::layout::entry_dir(cache.cache_root(), &key);
        cache.fs().remove_dir_all(&entry).unwrap();

        let err = cache.restore(&manifest).unwrap_err();
        let msg = format!("{err}");
        assert!(msg.contains("filesystem error"), "got: {msg}");
    }

    // ---- staging cleanup ----

    #[test]
    fn cache_019_restore_leaves_no_staging_directory_after_success() {
        let target = PathBuf::from("/ws/proj/out");
        let fs = fs_with_one_output(&target, b"x", 0o644);
        let outs = [StoredOutput {
            workspace_absolute_path: "/proj/out",
            on_disk_path: &target,
            mode: 0o644,
        }];
        let (cache, _restored, _manifest) =
            store_then_restore(fs, HashAlgo::Blake3, &outs, b"", b"");

        for entry in cache.fs().read_dir(cache.cache_root()).unwrap() {
            let name = entry
                .path
                .file_name()
                .unwrap()
                .to_string_lossy()
                .into_owned();
            assert!(
                !name.starts_with(".restore-"),
                "staging directory must not persist after a successful restore, found: {name}"
            );
        }
    }

    #[test]
    fn cache_019_restore_leaves_no_staging_directory_after_failure() {
        let target = PathBuf::from("/ws/proj/out");
        let fs = fs_with_one_output(&target, b"x", 0o644);
        let cache = make_cache(fs, HashAlgo::Blake3);
        let key = sample_key();
        let outs = [StoredOutput {
            workspace_absolute_path: "/proj/out",
            on_disk_path: &target,
            mode: 0o644,
        }];
        cache
            .store(
                &key,
                &StoreInputs {
                    outputs: &outs,
                    stdout: b"",
                    stderr: b"",
                    created_at_unix: 0,
                },
            )
            .unwrap();
        let manifest = cache.lookup(&key).unwrap();

        // Force a phase-1 failure by deleting the cache entry
        // (so reading the cached blob fails).
        let entry = crate::layout::entry_dir(cache.cache_root(), &key);
        cache.fs().remove_dir_all(&entry).unwrap();

        let _ = cache.restore(&manifest).unwrap_err();
        for entry in cache.fs().read_dir(cache.cache_root()).unwrap() {
            let name = entry
                .path
                .file_name()
                .unwrap()
                .to_string_lossy()
                .into_owned();
            assert!(
                !name.starts_with(".restore-"),
                "staging directory must be cleaned up after a failed restore, found: {name}"
            );
        }
    }

    // ---- different hash algo ----

    #[test]
    fn cache_019_restore_works_under_sha256() {
        let target = PathBuf::from("/ws/proj/out");
        let fs = fs_with_one_output(&target, b"sha-bytes", 0o600);
        let outs = [StoredOutput {
            workspace_absolute_path: "/proj/out",
            on_disk_path: &target,
            mode: 0o600,
        }];
        let (cache, restored, _manifest) =
            store_then_restore(fs, HashAlgo::Sha256, &outs, b"sha-stdout", b"sha-stderr");
        assert_eq!(cache.fs().read(&target).unwrap(), b"sha-bytes");
        assert_eq!(cache.fs().mode_of(&target).unwrap(), 0o600);
        assert_eq!(restored.stdout, b"sha-stdout");
        assert_eq!(restored.stderr, b"sha-stderr");
    }
}