A build cache: disk mounts that carry a build forward d60aac6c3f8a

//! The build cache (oxygenflow#28, spec §5.4): a disk mount that names a

//! `cache` is started from a disk kept by an earlier passing job, rather

//! than empty, so a build carries its compiled dependencies forward.

//!

//! Each runner keeps a pool under `<home>/cache`, by repository, job,

//! cache name and bookmark, each disk named by its key: the BLAKE3 of the

//! image, the job, the cache name and the content of its key files. A job

//! takes, in order:

//!

//! 1. its bookmark's disk with the same key (moved: no copying);

//! 2. its bookmark's newest disk (moved; cargo rebuilds what changed);

//! 3. another bookmark's disk with the same key, then its newest (copied,

//!    so a proposal never takes main's cache away);

//! 4. else an empty disk, as without a cache.

//!

//! A passing job's disk goes back into its bookmark's place; a failing

//! job's is thrown away, so nothing it wrote reaches a later job. The

//! receipt names what the job started from. The pool is kept under its

//! size limit by dropping the disks used longest ago.

use anyhow::{Context, Result};

use flow_core::pipeline::Mount;

use std::os::unix::fs::MetadataExt;

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

use std::time::SystemTime;

/// The pool's default size limit.

pub const DEFAULT_GB: u64 = 64;

/// Where a job's cached disks come from and go back to.

pub struct Pool {

    root: PathBuf,

    limit: u64,

}

/// Which job: the pool's coordinates other than the key.

pub struct Ctx<'a> {

    pub repo: &'a str,

    pub bookmark: &'a str,

    pub job: &'a str,

    pub image: &'a str,

}

/// A name safe as one path segment: kept readable, with a short hash so

/// two names that clean up alike stay apart.

fn segment(s: &str) -> String {

    let clean: String = s.chars().map(|c| if c.is_ascii_alphanumeric() || "._-".contains(c) { c } else { '_' }).take(48).collect();

    format!("{clean}-{}", &blake3::hash(s.as_bytes()).to_hex()[..8])

}

impl Pool {

    pub fn new(home_root: &Path, gb: Option<u64>) -> Pool {

        Pool { root: home_root.join("cache"), limit: gb.unwrap_or(DEFAULT_GB).saturating_mul(1 << 30) }

    }

    fn cache_dir(&self, ctx: &Ctx, name: &str) -> PathBuf {

        self.root.join(segment(ctx.repo)).join(segment(ctx.job)).join(segment(name))

    }

    /// A mount's key: the image, job, cache name and its key files'

    /// content hashes (`hash_of` gives a file's, `None` when absent).

    pub fn key(ctx: &Ctx, m: &Mount, hash_of: impl Fn(&str) -> Option<String>) -> String {

        let mut h = blake3::Hasher::new();

        for part in [ctx.image, ctx.job, m.cache.as_deref().unwrap_or("")] {

            h.update(part.as_bytes());

            h.update(b"\n");

        }

        for k in &m.key {

            h.update(k.as_bytes());

            h.update(b"=");

            h.update(hash_of(k).unwrap_or_else(|| "absent".into()).as_bytes());

            h.update(b"\n");

        }

        h.finalize().to_hex()[..32].to_string()

    }

    /// Start `dest` from the pool for mount `m`, a disk of `m.size_mib`.

    /// What it started from, or `None` (the caller makes an empty disk).

    pub fn take(&self, ctx: &Ctx, m: &Mount, key: &str, dest: &Path) -> Option<String> {

        let name = m.cache.as_deref()?;

        let dir = self.cache_dir(ctx, name);

        let size = m.size_mib as u64 * (1 << 20);

        let fits = |p: &Path| std::fs::metadata(p).is_ok_and(|md| md.len() == size);

        let mine = dir.join(segment(ctx.bookmark));

        let exact = mine.join(format!("{key}.ext4"));

        if fits(&exact) && std::fs::rename(&exact, dest).is_ok() {

            return Some(format!("cache {name}: this bookmark's, same key"));

        }

        if let Some(p) = newest(&mine, |p| fits(p)) {

            if std::fs::rename(&p, dest).is_ok() {

                return Some(format!("cache {name}: this bookmark's, an older key"));

            }

        }

        let others: Vec<PathBuf> =

            std::fs::read_dir(&dir).into_iter().flatten().flatten().map(|e| e.path()).filter(|p| p.is_dir() && *p != mine).collect();

        let pick = others

            .iter()

            .map(|d| d.join(format!("{key}.ext4")))

            .filter(|p| fits(p))

            .max_by_key(|p| modified(p))

            .map(|p| (p, "same key"))

            .or_else(|| others.iter().filter_map(|d| newest(d, |p| fits(p))).max_by_key(|p| modified(p)).map(|p| (p, "an older key")));

        if let Some((p, which)) = pick {

            let from = p.parent().and_then(|d| d.file_name()).map(|n| n.to_string_lossy().into_owned()).unwrap_or_default();

            if sparse_copy(&p, dest).is_ok() {

                return Some(format!("cache {name}: copied from another bookmark's ({from}), {which}"));

            }

        }

        None

    }

    /// Keep a passing job's disk for mount `m` under its key, then keep

    /// the pool under its limit.

    pub fn keep(&self, ctx: &Ctx, m: &Mount, key: &str, disk: &Path) -> Result<()> {

        let Some(name) = m.cache.as_deref() else { return Ok(()) };

        let mine = self.cache_dir(ctx, name).join(segment(ctx.bookmark));

        std::fs::create_dir_all(&mine)?;

        let dest = mine.join(format!("{key}.ext4"));

        if std::fs::rename(disk, &dest).is_err() {

            sparse_copy(disk, &dest).with_context(|| format!("keep {}", disk.display()))?;

        }

        // Moved disks keep their old time: mark it used now.

        std::fs::File::options().write(true).open(&dest)?.set_modified(SystemTime::now())?;

        // One disk per key is enough; older keys of this bookmark go once

        // the pool is over its limit.

        self.evict();

        Ok(())

    }

    /// Drop the disks used longest ago until the pool's blocks fit.

    pub fn evict(&self) -> u64 {

        let mut disks: Vec<(SystemTime, u64, PathBuf)> = Vec::new();

        walk(&self.root, &mut |p| {

            if p.extension().is_some_and(|e| e == "ext4") {

                if let Ok(md) = std::fs::metadata(p) {

                    disks.push((md.modified().unwrap_or(SystemTime::UNIX_EPOCH), md.blocks() * 512, p.to_path_buf()));

                }

            }

        });

        let mut total: u64 = disks.iter().map(|d| d.1).sum();

        disks.sort();

        let mut freed = 0;

        for (_, bytes, p) in disks {

            if total <= self.limit {

                break;

            }

            if std::fs::remove_file(&p).is_ok() {

                total -= bytes;

                freed += bytes;

            }

        }

        freed

    }

}

fn modified(p: &Path) -> SystemTime {

    std::fs::metadata(p).and_then(|m| m.modified()).unwrap_or(SystemTime::UNIX_EPOCH)

}

fn newest(dir: &Path, ok: impl Fn(&Path) -> bool) -> Option<PathBuf> {

    std::fs::read_dir(dir)

        .ok()?

        .flatten()

        .map(|e| e.path())

        .filter(|p| p.extension().is_some_and(|e| e == "ext4") && ok(p))

        .max_by_key(|p| modified(p))

}

fn walk(dir: &Path, f: &mut dyn FnMut(&Path)) {

    for e in std::fs::read_dir(dir).into_iter().flatten().flatten() {

        let p = e.path();

        if p.is_dir() {

            walk(&p, f);

        } else {

            f(&p);

        }

    }

}

/// Copy a sparse disk without writing its holes.

fn sparse_copy(from: &Path, to: &Path) -> Result<()> {

    let tmp = to.with_extension("part");

    let st = std::process::Command::new("cp").arg("--sparse=always").arg(from).arg(&tmp).status().context("run cp")?;

    if !st.success() {

        let _ = std::fs::remove_file(&tmp);

        anyhow::bail!("cp {} failed", from.display());

    }

    std::fs::rename(&tmp, to)?;

    Ok(())

}

#[cfg(test)]

mod tests {

    use super::*;

    use flow_core::pipeline::MountSource;

    fn mount(key: &[&str]) -> Mount {

        Mount {

            source: MountSource::Disk,

            target: "/src/target".into(),

            size_mib: 1,

            cache: Some("target".into()),

            key: key.iter().map(|s| s.to_string()).collect(),

        }

    }

    fn disk(path: &Path, fill: &[u8]) {

        let f = std::fs::File::create(path).unwrap();

        f.set_len(1 << 20).unwrap();

        std::os::unix::fs::FileExt::write_at(&f, fill, 0).unwrap();

    }

    fn first(path: &Path) -> Vec<u8> {

        std::fs::read(path).unwrap()[..4].to_vec()

    }

    #[test]

    fn a_kept_disk_starts_the_next_job_of_its_bookmark_then_others_copy_it() {

        let home = tempfile::tempdir().unwrap();

        let work = tempfile::tempdir().unwrap();

        let pool = Pool::new(home.path(), None);

        let m = mount(&["Cargo.lock"]);

        let main = Ctx { repo: "andy/hub", bookmark: "main", job: "test", image: "img" };

        let key = Pool::key(&main, &m, |_| Some("lock-1".into()));

        // Nothing yet: the caller makes an empty disk.

        assert_eq!(pool.take(&main, &m, &key, &work.path().join("a.ext4")), None);

        // A passing job's disk is kept and moved into the next job's place.

        disk(&work.path().join("a.ext4"), b"MAIN");

        pool.keep(&main, &m, &key, &work.path().join("a.ext4")).unwrap();

        assert!(!work.path().join("a.ext4").exists(), "moved, not copied");

        let how = pool.take(&main, &m, &key, &work.path().join("b.ext4")).unwrap();

        assert!(how.contains("same key"), "{how}");

        assert_eq!(first(&work.path().join("b.ext4")), b"MAIN");

        pool.keep(&main, &m, &key, &work.path().join("b.ext4")).unwrap();

        // A changed key takes the bookmark's newest disk all the same.

        let key2 = Pool::key(&main, &m, |_| Some("lock-2".into()));

        assert_ne!(key, key2);

        let how = pool.take(&main, &m, &key2, &work.path().join("c.ext4")).unwrap();

        assert!(how.contains("older key"), "{how}");

        pool.keep(&main, &m, &key2, &work.path().join("c.ext4")).unwrap();

        // Another bookmark copies it, leaving main's in place.

        let feature = Ctx { bookmark: "feature", ..main };

        let how = pool.take(&feature, &m, &key2, &work.path().join("d.ext4")).unwrap();

        assert!(how.contains("copied from another bookmark"), "{how}");

        assert_eq!(first(&work.path().join("d.ext4")), b"MAIN");

        assert!(pool.take(&main, &m, &key2, &work.path().join("e.ext4")).unwrap().contains("this bookmark's"), "main's is still there");

        // Another job or a different size never shares it.

        let other_job = Ctx { job: "build", ..main };

        assert_eq!(pool.take(&other_job, &m, &key2, &work.path().join("f.ext4")), None);

        let mut bigger = mount(&["Cargo.lock"]);

        bigger.size_mib = 2;

        assert_eq!(pool.take(&feature, &bigger, &key2, &work.path().join("g.ext4")), None);

    }

    #[test]

    fn the_key_follows_the_image_the_job_and_the_key_files() {

        let ctx = Ctx { repo: "r", bookmark: "main", job: "test", image: "img-1" };

        let m = mount(&["Cargo.lock", "rust-toolchain.toml"]);

        let k = Pool::key(&ctx, &m, |p| Some(format!("{p}-v1")));

        assert_eq!(k, Pool::key(&ctx, &m, |p| Some(format!("{p}-v1"))));

        assert_ne!(k, Pool::key(&ctx, &m, |p| Some(format!("{p}-v2"))), "a key file changed");

        assert_ne!(k, Pool::key(&Ctx { image: "img-2", ..ctx }, &m, |p| Some(format!("{p}-v1"))), "another image");

        assert_ne!(k, Pool::key(&ctx, &m, |_| None), "a key file missing");

    }

    #[test]

    fn the_pool_drops_the_disks_used_longest_ago() {

        let home = tempfile::tempdir().unwrap();

        let work = tempfile::tempdir().unwrap();

        let pool = Pool { root: home.path().join("cache"), limit: 0 };

        let m = mount(&[]);

        let ctx = Ctx { repo: "r", bookmark: "main", job: "j", image: "i" };

        let full = |p: &Path| {

            let f = std::fs::File::create(p).unwrap();

            std::os::unix::fs::FileExt::write_at(&f, &vec![7u8; 1 << 20], 0).unwrap();

        };

        full(&work.path().join("a.ext4"));

        // Over a limit of nothing: kept, then dropped at once.

        pool.keep(&ctx, &m, "k1", &work.path().join("a.ext4")).unwrap();

        assert_eq!(pool.take(&ctx, &m, "k1", &work.path().join("b.ext4")), None);

        let roomy = Pool { root: home.path().join("cache"), limit: 1 << 30 };

        full(&work.path().join("c.ext4"));

        roomy.keep(&ctx, &m, "k2", &work.path().join("c.ext4")).unwrap();

        assert!(roomy.take(&ctx, &m, "k2", &work.path().join("d.ext4")).is_some(), "under the limit it stays");

    }

}

- **A build cache (#28, spec §5.4).** A disk mount may name a `cache` and the `key` files it depends on, such as `Cargo.lock`. The runner keeps passing jobs' disks and starts the next job of the same repository, job and bookmark from one, so builds carry their compiled dependencies forward; another bookmark copies it rather than taking it. Failing jobs' disks are never kept, the receipt names what a disk started from, and the pool stays under `cache_gb` (default 64 GiB).

                    .map(|m| {

                        // A cache mount adds its name and key files; a plain

                        // one encodes as before caches existed.

                        let mut v = vec![text(m.source.as_str()), text(&m.target), Value::Uint(m.size_mib as u64)];

                        if let Some(c) = &m.cache {

                            v.push(text(c));

                            v.push(Value::Array(m.key.iter().map(|k| text(k)).collect()));

                        }

                        Value::Array(v)

                    })

/// after it, so it cannot carry anything between jobs (spec §2.2 I4) --

/// unless it names a `cache` (spec §5.4, #28): then a disk kept from an

/// earlier passing job may start it, and the receipt says which.

    /// `cache "name"`: keep this disk after a passing job and start later

    /// jobs from it (disk mounts only).

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub cache: Option<String>,

    /// `key "Cargo.lock" ...`: files whose content picks the cache entry.

    #[serde(default, skip_serializing_if = "Vec::is_empty")]

    pub key: Vec<String>,

    let (mut cache, mut keys) = (None, Vec::new());

            "cache" => {

                cache = match ctx.one_string(c) {

                    Some(n) if !n.is_empty() && n.len() <= 64 && n.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b)) => {

                        Some(n)

                    }

                    Some(n) => {

                        ctx.err(c.span(), format!("cache name {n:?}: letters, digits, '.', '_' and '-', at most 64"));

                        None

                    }

                    None => None,

                }

            }

            "key" => {

                keys = ctx.strings(c);

                if keys.is_empty() {

                    ctx.err(c.span(), "`key` names one or more files, as \"Cargo.lock\"");

                }

                for k in &keys {

                    if k.starts_with('/') || k.split('/').any(|p| p.is_empty() || p == "." || p == "..") {

                        ctx.err(c.span(), format!("key file {k:?} must be a path in the repository, without . or .."));

                    }

                }

            }

    }

    if source == Some(MountSource::Tmpfs) && cache.is_some() {

        ctx.err(node.span(), "only an ephemeral://disk mount can be a cache");

    }

    if !keys.is_empty() && cache.is_none() {

        ctx.err(node.span(), "`key` goes with `cache`");

    Some(Mount { source: source?, target: target?, size_mib: size?, cache, key: keys })

            Mount { source: MountSource::Disk, target: "/src/target".into(), size_mib: 8192, cache: None, key: vec![] },

            Mount { source: MountSource::Tmpfs, target: "/tmp/cache".into(), size_mib: 256, cache: None, key: vec![] },

fn a_disk_mount_can_be_a_cache_keyed_by_files() {

    let src = |extra: &str| {

        doc(&format!(

            "job \"b\" {{\n    sandbox \"alpine\"\n    mount {{\n        source \"ephemeral://disk\"\n        target \"/src/target\"\n        size \"8GiB\"\n{extra}    }}\n    exec \"cargo\" \"build\"\n}}\n"

        ))

    };

    let plain = Pipeline::parse(&src("")).unwrap();

    let cached = Pipeline::parse(&src("        cache \"target\"\n        key \"Cargo.lock\" \"rust-toolchain.toml\"\n")).unwrap();

    let m = &cached.jobs["b"].mounts[0];

    assert_eq!((m.cache.as_deref(), m.key.clone()), (Some("target"), vec!["Cargo.lock".to_string(), "rust-toolchain.toml".to_string()]));

    assert_ne!(cached.hash(), plain.hash(), "a cache is part of the job");

    let keyless = Pipeline::parse(&src("        cache \"target\"\n")).unwrap();

    assert_ne!(keyless.hash(), cached.hash());

    let ps = problems(&src("        cache \"no spaces\"\n"));

    assert!(ps.iter().any(|(_, m)| m.contains("cache name")), "{ps:?}");

    let ps = problems(&src("        key \"Cargo.lock\"\n"));

    assert!(ps.iter().any(|(_, m)| m.contains("goes with `cache`")), "{ps:?}");

    let ps = problems(&src("        cache \"t\"\n        key \"../x\"\n"));

    assert!(ps.iter().any(|(_, m)| m.contains("path in the repository")), "{ps:?}");

    let tmpfs = doc("job \"b\" {\n    sandbox \"alpine\"\n    mount {\n        source \"ephemeral://tmpfs\"\n        target \"/tmp/c\"\n        size \"64MiB\"\n        cache \"c\"\n    }\n    exec \"true\"\n}\n");

    assert!(problems(&tmpfs).iter().any(|(_, m)| m.contains("only an ephemeral://disk")));

}

#[test]

    /// The build cache's size limit in GiB (#28; default 64).

    #[serde(default)]

    pub cache_gb: Option<u64>,

mod cache;

        // A disk already in place came from the build cache (#28).

        if !path.exists() {

            crate::workspace::empty_ext4(&path, m.size_mib, "scratch")?;

        }

            Mount { source: MountSource::Disk, target: "/src/target".into(), size_mib: 8192, cache: None, key: vec![] },

            Mount { source: MountSource::Tmpfs, target: "/tmp/cache".into(), size_mib: 256, cache: None, key: vec![] },

            Mount { source: MountSource::Disk, target: "/tmp/z".into(), size_mib: 64, cache: None, key: vec![] },

                Mount { source: MountSource::Tmpfs, target: "/src/cache".into(), size_mib: 64, cache: None, key: vec![] },

                Mount { source: MountSource::Disk, target: "/src/target".into(), size_mib: 1024, cache: None, key: vec![] },

                Mount { source: MountSource::Disk, target: "/src/work".into(), size_mib: 1024, cache: None, key: vec![] },

        // Disk mounts that name a cache start from a kept disk (#28); the

        // microVM takes a disk already in place as it is.

        let pool = crate::cache::Pool::new(&home.root, cfg.cache_gb);

        let cache_ctx =

            crate::cache::Ctx { repo: &claim.repo, bookmark: &claim.bookmark, job: &claim.job.name, image: &claim.sandbox.rootfs };

        let mut cached: Vec<(usize, String, String)> = Vec::new();

        if !syscall_tier {

            for (i, m) in job.mounts.iter().filter(|m| m.source == flow_core::pipeline::MountSource::Disk).enumerate() {

                let Some(name) = &m.cache else { continue };

                let key = crate::cache::Pool::key(&cache_ctx, m, |p| files.iter().find(|(f, _)| f == p).map(|(_, h)| hex::encode(h)));

                let how = pool.take(&cache_ctx, m, &key, &scratch.join(format!("mount-{i}.ext4")));

                let how = how.unwrap_or_else(|| format!("cache {name}: none kept yet, an empty disk"));

                note(&mut sink, format!("{} ({})", how, m.target));

                cached.push((i, key, how));

            }

        }

        let mut outcome = if syscall_tier {

        }

        // The receipt says what a cached disk started from.

        let disk_mounts: Vec<&flow_core::pipeline::Mount> =

            job.mounts.iter().filter(|m| m.source == flow_core::pipeline::MountSource::Disk).collect();

        for (i, key, how) in &cached {

            outcome.environment.parts.push((format!("cache {}", disk_mounts[*i].target), format!("{key}: {how}")));

        // Measured before a cached disk leaves the scratch directory.

        let peak_disk_mib = crate::facts::allocated_mib(&scratch);

        // A passing job's cached disks are kept; a failing job's are not,

        // so nothing it wrote reaches a later job.

        if passed && output_error.is_none() {

            for (i, key, _) in &cached {

                if let Err(e) = pool.keep(&cache_ctx, disk_mounts[*i], key, &scratch.join(format!("mount-{i}.ext4"))) {

                    note(&mut sink, format!("cache: not kept: {e:#}"));

                }

            }

        }

            peak_disk_mib,

### 5.4 The Build Cache (oxygenflow#28)

An `ephemeral://disk` mount may name a `cache`, and the files that key it:

```kdl

mount {

    source "ephemeral://disk"

    target "/src/target"

    size "24GiB"

    cache "target"

    key "Cargo.lock" "rust-toolchain.toml"

}

```

Such a disk is not always empty at the start. Each runner keeps a pool of

the disks of passing jobs, by repository, job, cache name and bookmark,

named by the **key**: the BLAKE3 of the sandbox image, the job's name, the

cache name, and each key file's content in the job's state. A job's disk

starts as, in order: its bookmark's disk with the same key (moved), its

bookmark's newest disk (moved), another bookmark's with the same key and

then another's newest (copied, so a proposal never takes the default

bookmark's), and otherwise an empty disk. A disk must be the size the

mount asks for.

- **Only passing jobs keep a disk.** When the job and its outputs pass,

  its cached disks go back into the pool under their key; otherwise they

  are discarded with the scratch directory, so nothing a failing job wrote

  reaches a later one.

- **The receipt says so.** The job's environment gains `cache TARGET` with

  the key and what the disk started from, so the receipt's environment

  hash differs from a cold build's.

- **Bounded.** The pool keeps under `cache_gb` in the runner's config

  (default 64 GiB) by dropping the disks used longest ago.

- The syscall tier keeps its disks empty for now.

- A cache is a speed-up, never a source of truth: jobs must build

  correctly from an empty disk, which a key change or an eviction gives

  them at any time.