Prune build caches before they are kept 077a65242359

//! The parts of the guest init the runner shares: pruning a build cache

//! before it is kept (oxygenflow#72), done in the guest for a microVM's

//! disks and on the runner for the syscall tier's directories.

pub mod prune;

//! Pruning a build cache before it is kept (oxygenflow#72, spec §5.4).

//!

//! A kept disk starts the next run, so without pruning it carries every

//! build's files: each crate's artifacts under every hash a lockfile or a

//! feature set ever gave it. Before a passing job's cache disk goes back

//! into the pool, the files this run did not use are removed, oldest

//! first, until what is on the disk is at most a share of its size (the

//! cap). The microVM's guest init prunes its disks before it powers off,

//! since only it can see inside them; the runner prunes the syscall

//! tier's directories itself. Both use this module.

//!

//! What the run used is told by the files' times against when the job

//! started: written (mtime), changed (ctime) or read (atime). Reads are

//! seen only where the file system records them: the guest mounts a cache

//! disk `strictatime`; on the runner, where a directory sits on whatever

//! file system the runner has, each file's atime is set to the epoch

//! before the job ([`forget_reads`]), so even `relatime` records its

//! first read. A file system mounted `noatime` records none, and then

//! nothing is pruned.

//!

//! A build tool does not read everything it relies on. Cargo, finding a

//! crate fresh, reads its fingerprint and only looks at its rlib and

//! binaries, so their times say nothing. Every name cargo gives a unit's

//! files carries the unit's hash (`.fingerprint/foo-<hash>/`,

//! `deps/libfoo-<hash>.rlib`, `build/foo-<hash>/`), so a file stays when

//! a file this run used carries one of the hashes in its path: a fresh

//! crate's fingerprint keeps its artifacts. A file with another link

//! that stays, or under a path the job's outputs name, stays too. What

//! goes is what this run's build did not touch, which a later build needs

//! only if an older lockfile comes back, and then cargo rebuilds it.

use serde::{Deserialize, Serialize};

use std::collections::{HashMap, HashSet};

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

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

use std::time::{Duration, SystemTime};

/// The share of a cache disk a kept cache may fill when no runner says

/// otherwise: room for this run's build and as much again.

pub const DEFAULT_PERCENT: u8 = 60;

/// What pruning a cache found and did, in bytes of blocks on its disk.

#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq)]

pub struct Pruned {

    /// The disk's size, and the share of it the cache may fill.

    pub size: u64,

    pub percent: u8,

    /// What was on the disk before pruning, and after.

    pub before: u64,

    pub after: u64,

    /// What went: bytes and files.

    pub freed: u64,

    pub files: u64,

    /// Why nothing could be pruned, when so (no access times, or an

    /// error): only the size then decides.

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

    pub unpruned: Option<String>,

}

impl Pruned {

    /// The most the cache may keep: `percent` of `size`.

    pub fn cap(&self) -> u64 {

        cap(self.size, self.percent)

    }

    /// Whether what is left may be kept.

    pub fn fits(&self) -> bool {

        self.after <= self.cap()

    }

}

fn cap(size: u64, percent: u8) -> u64 {

    (size as u128 * percent.min(100) as u128 / 100) as u64

}

/// Now, as a job's start to compare file times against. File systems

/// stamp times from a clock that may lag this one by a tick, so the

/// moments either side of it are left to pass: nothing done before the

/// call (such as [`forget_reads`]) can look as though it came after.

pub fn start_clock() -> SystemTime {

    std::thread::sleep(Duration::from_millis(20));

    let t = SystemTime::now();

    std::thread::sleep(Duration::from_millis(20));

    t

}

/// Whether the file system holding `path` records reads (not `noatime`).

pub fn records_reads(path: &Path) -> bool {

    let Ok(c) = std::ffi::CString::new(path.as_os_str().as_encoded_bytes()) else { return false };

    let mut st: libc::statvfs = unsafe { std::mem::zeroed() };

    if unsafe { libc::statvfs(c.as_ptr(), &mut st) } != 0 {

        return false;

    }

    st.f_flag & libc::ST_NOATIME == 0

}

/// The blocks in use on the file system mounted at `path`, and its size

/// (as `df` counts them).

pub fn usage(path: &Path) -> std::io::Result<(u64, u64)> {

    let c = std::ffi::CString::new(path.as_os_str().as_encoded_bytes())?;

    let mut st: libc::statvfs = unsafe { std::mem::zeroed() };

    if unsafe { libc::statvfs(c.as_ptr(), &mut st) } != 0 {

        return Err(std::io::Error::last_os_error());

    }

    let unit = st.f_frsize as u64;

    Ok(((st.f_blocks as u64).saturating_sub(st.f_bfree as u64) * unit, st.f_blocks as u64 * unit))

}

/// Set every regular file's atime under `root` to the epoch, its mtime

/// kept, so the job's first read of each is recorded even on `relatime`

/// (which records a read only of a file read no later than it changed).

/// Links are not followed. The count of files.

pub fn forget_reads(root: &Path) -> std::io::Result<u64> {

    let mut n = 0;

    let mut err = None;

    walk(root, &mut |p, md| {

        if !md.file_type().is_file() {

            return;

        }

        let Ok(c) = std::ffi::CString::new(p.as_os_str().as_encoded_bytes()) else { return };

        let times = [libc::timespec { tv_sec: 0, tv_nsec: 0 }, libc::timespec { tv_sec: 0, tv_nsec: libc::UTIME_OMIT }];

        if unsafe { libc::utimensat(libc::AT_FDCWD, c.as_ptr(), times.as_ptr(), libc::AT_SYMLINK_NOFOLLOW) } == 0 {

            n += 1;

        } else if err.is_none() {

            err = Some(std::io::Error::last_os_error());

        }

    })?;

    match err {

        Some(e) => Err(e),

        None => Ok(n),

    }

}

/// Each entry under `root` (not `root` itself) with its metadata, parents

/// before children; links are not followed, nor other file systems

/// entered.

fn walk(root: &Path, f: &mut dyn FnMut(&Path, &std::fs::Metadata)) -> std::io::Result<()> {

    let dev = std::fs::symlink_metadata(root)?.dev();

    let mut dirs = vec![root.to_path_buf()];

    while let Some(d) = dirs.pop() {

        let Ok(entries) = std::fs::read_dir(&d) else { continue };

        for e in entries.flatten() {

            let p = e.path();

            let Ok(md) = std::fs::symlink_metadata(&p) else { continue };

            if md.dev() != dev {

                continue;

            }

            f(&p, &md);

            if md.is_dir() {

                dirs.push(p);

            }

        }

    }

    Ok(())

}

/// The unit hashes in `rel`: a 16-digit lowercase hex word after a `-`

/// in any of its names, as cargo names a unit's files.

fn hashes(rel: &Path) -> impl Iterator<Item = String> + '_ {

    rel.components().filter_map(|c| c.as_os_str().to_str()).flat_map(|name| {

        name.split('-').skip(1).filter_map(|w| {

            let w = w.split('.').next().unwrap_or("");

            (w.len() == 16 && w.bytes().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))).then(|| w.to_string())

        })

    })

}

fn nanos(s: i64, ns: i64) -> i128 {

    s as i128 * 1_000_000_000 + ns as i128

}

/// One file, however many names it has.

struct File {

    paths: Vec<PathBuf>,

    links: u64,

    bytes: u64,

    /// When it was last used as far as can be told: the later of its

    /// atime and mtime.

    last: i128,

    used: bool,

}

/// Prune the cache at `root`, a disk of `size` bytes with `before` in

/// use, down to `percent` of its size, removing only regular files this

/// run did not use (see the module's notes), those used longest ago

/// first. `started` is when the job started ([`start_clock`]); `keep` are

/// paths relative to `root` (the job's outputs) that stay whatever their

/// times. Under the cap nothing is touched. `after` is `before` less what

/// was freed.

pub fn prune(root: &Path, started: SystemTime, keep: &[String], size: u64, before: u64, percent: u8) -> std::io::Result<Pruned> {

    let mut out = Pruned { size, percent, before, after: before, ..Pruned::default() };

    if before <= out.cap() {

        return Ok(out);

    }

    let since = started.duration_since(SystemTime::UNIX_EPOCH).map(|d| d.as_nanos() as i128).unwrap_or(0);

    let keep: Vec<PathBuf> = keep.iter().map(|k| PathBuf::from(k.trim_start_matches('/'))).filter(|k| !k.as_os_str().is_empty()).collect();

    let mut files: HashMap<(u64, u64), File> = HashMap::new();

    walk(root, &mut |p, md| {

        if !md.file_type().is_file() {

            return;

        }

        let rel = p.strip_prefix(root).unwrap_or(p);

        let used = [nanos(md.atime(), md.atime_nsec()), nanos(md.mtime(), md.mtime_nsec()), nanos(md.ctime(), md.ctime_nsec())]

            .into_iter()

            .any(|t| t >= since)

            || keep.iter().any(|k| rel.starts_with(k));

        let f = files.entry((md.dev(), md.ino())).or_insert_with(|| File {

            paths: Vec::new(),

            links: md.nlink(),

            bytes: md.blocks() * 512,

            last: nanos(md.atime(), md.atime_nsec()).max(nanos(md.mtime(), md.mtime_nsec())),

            used: false,

        });

        f.paths.push(p.to_path_buf());

        f.used |= used;

    })?;

    // A hash any used file carries keeps every file carrying it.

    let live: HashSet<String> =

        files.values().filter(|f| f.used).flat_map(|f| f.paths.iter().flat_map(|p| hashes(p.strip_prefix(root).unwrap_or(p)))).collect();

    let mut gone: Vec<File> = files

        .into_values()

        .filter(|f| {

            !f.used

                // A link from outside the cache: removing it frees nothing.

                && f.links == f.paths.len() as u64

                && !f.paths.iter().any(|p| hashes(p.strip_prefix(root).unwrap_or(p)).any(|h| live.contains(&h)))

        })

        .collect();

    gone.sort_by_key(|f| f.last);

    let mut emptied: HashSet<PathBuf> = HashSet::new();

    for f in gone {

        if out.after <= out.cap() {

            break;

        }

        let mut removed = true;

        for p in &f.paths {

            if std::fs::remove_file(p).is_err() {

                removed = false;

            } else if let Some(parent) = p.parent() {

                emptied.insert(parent.to_path_buf());

            }

        }

        if removed {

            out.after = out.after.saturating_sub(f.bytes);

            out.freed += f.bytes;

            out.files += 1;

        }

    }

    // Directories pruning emptied go too, deepest first; never `root`.

    let mut dirs: Vec<PathBuf> = emptied.into_iter().collect();

    dirs.sort_by_key(|d| std::cmp::Reverse(d.components().count()));

    for d in dirs {

        let mut d = d.as_path();

        while d != root && d.starts_with(root) && std::fs::remove_dir(d).is_ok() {

            match d.parent() {

                Some(p) => d = p,

                None => break,

            }

        }

    }

    Ok(out)

}

#[cfg(test)]

mod tests {

    use super::*;

    /// Times around a job that started at `T`, ahead of the clock: a file

    /// whose times are set is changed (ctime) now, before it.

    const T: u64 = 4_000_000_000;

    fn at(secs: u64) -> SystemTime {

        SystemTime::UNIX_EPOCH + Duration::from_secs(secs)

    }

    /// A file of `kib` KiB at `rel`, last read and written at `secs`.

    fn file(root: &Path, rel: &str, kib: usize, secs: u64) -> PathBuf {

        let p = root.join(rel);

        std::fs::create_dir_all(p.parent().unwrap()).unwrap();

        let bytes: Vec<u8> = (0..kib << 10).map(|i: usize| (i.wrapping_mul(2_654_435_761) >> 13) as u8).collect();

        std::fs::write(&p, bytes).unwrap();

        let f = std::fs::File::options().write(true).open(&p).unwrap();

        f.set_times(std::fs::FileTimes::new().set_accessed(at(secs)).set_modified(at(secs))).unwrap();

        p

    }

    fn bytes(p: &Path) -> u64 {

        std::fs::metadata(p).unwrap().blocks() * 512

    }

    #[test]

    fn hashes_are_cargos_unit_hashes() {

        let h = |p: &str| hashes(Path::new(p)).collect::<Vec<_>>();

        assert_eq!(h("debug/deps/libdep-8f2b19826c505967.rlib"), ["8f2b19826c505967"]);

        assert_eq!(h("debug/.fingerprint/dep-8f2b19826c505967/lib-dep"), ["8f2b19826c505967"]);

        assert_eq!(h("debug/build/my-crate-0123456789abcdef/out/x.o"), ["0123456789abcdef"]);

        assert!(h("debug/incremental/app-2dadnjnzlrswc/s-hndl5uwdpw-1uaho4m.lock").is_empty());

        assert!(h("debug/app").is_empty() && h("debug/deps/x-0123456789ABCDEF").is_empty() && h("x-0123456789abcde").is_empty());

    }

    #[test]

    fn under_the_cap_nothing_is_touched() {

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

        let old = file(dir.path(), "debug/deps/libold-1111111111111111.rlib", 64, 1_000);

        let p = prune(dir.path(), at(T), &[], 1 << 30, 64 << 10, 60).unwrap();

        assert!(old.exists() && p.fits() && p.freed == 0 && p.after == p.before, "{p:?}");

    }

    /// A cargo target as a run leaves it: this lockfile's units used this

    /// run (one rebuilt, one fresh: only its fingerprint read), an older

    /// lockfile's not, and an output the run did not touch.

    #[test]

    fn what_this_run_did_not_use_goes_oldest_first_and_its_units_stay_whole() {

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

        let root = dir.path();

        let app = file(root, "debug/deps/app-aaaaaaaaaaaaaaaa", 256, T + 100);

        let app_fp = file(root, "debug/.fingerprint/app-aaaaaaaaaaaaaaaa/bin-app", 1, T + 100);

        let dep_rlib = file(root, "debug/deps/libdep-bbbbbbbbbbbbbbbb.rlib", 256, T - 5_000);

        let dep_fp = file(root, "debug/.fingerprint/dep-bbbbbbbbbbbbbbbb/lib-dep", 1, T - 5_000);

        std::fs::File::open(&dep_fp).unwrap().set_times(std::fs::FileTimes::new().set_accessed(at(T + 50))).unwrap();

        let script = file(root, "debug/build/dep-bbbbbbbbbbbbbbbb/out/gen.rs", 64, T - 5_000);

        // Uplifted: another name of the rebuilt binary.

        std::fs::hard_link(&app, root.join("debug/app")).unwrap();

        // An older lockfile's units, untouched, the oldest first to go.

        let oldest = file(root, "debug/deps/libdep-cccccccccccccccc.rlib", 256, T - 9_000);

        let oldest_fp = file(root, "debug/.fingerprint/dep-cccccccccccccccc/lib-dep", 1, T - 8_999);

        let older = file(root, "debug/deps/libdep-dddddddddddddddd.rlib", 256, T - 8_000);

        // A declared output, older still, that the run did not touch.

        let output = file(root, "release/tool", 128, T - 10_000);

        let all = [&app, &app_fp, &dep_rlib, &dep_fp, &script, &oldest, &oldest_fp, &older, &output];

        let before: u64 = all.iter().map(|p| bytes(p)).sum();

        // Room for all but the oldest unit.

        let want = before - bytes(&oldest) - bytes(&oldest_fp);

        let size = (want * 100).div_ceil(60);

        let p = prune(root, at(T), &["/release/tool".into()], size, before, 60).unwrap();

        assert!(p.fits(), "{p:?}");

        assert!(

            !oldest.exists() && !root.join("debug/.fingerprint/dep-cccccccccccccccc").exists(),

            "the oldest unit went, its emptied directory too"

        );

        assert!(older.exists(), "and pruning stopped under the cap: {p:?}");

        for kept in [&app, &app_fp, &dep_rlib, &dep_fp, &script, &output, &root.join("debug/app")] {

            assert!(kept.exists(), "{} stays", kept.display());

        }

        assert_eq!((p.files, p.freed, p.after), (2, p.before - want, want));

        // With no room at all: still only what the run did not use goes.

        let p = prune(root, at(T), &["release/tool".into()], 1, p.after, 60).unwrap();

        assert!(!older.exists() && !p.fits() && p.files == 1, "{p:?}");

        for kept in [&app, &dep_rlib, &dep_fp, &script, &output] {

            assert!(kept.exists(), "{} stays", kept.display());

        }

        assert!(root.join("debug/deps").is_dir(), "a directory still holding files stays");

    }

    #[test]

    fn reads_are_forgotten_so_the_next_is_seen() {

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

        let p = file(dir.path(), "a/b", 1, 5_000);

        std::os::unix::fs::symlink("/etc/passwd", dir.path().join("link")).unwrap();

        assert_eq!(forget_reads(dir.path()).unwrap(), 1);

        let md = std::fs::metadata(&p).unwrap();

        assert_eq!((md.atime(), md.mtime()), (0, 5_000), "the mtime is kept");

        assert!(std::fs::symlink_metadata(dir.path().join("link")).unwrap().atime() > 0, "a link is not followed");

    }

}

- **Build caches are pruned before they are kept (#72, spec §5.4).** A kept cache no longer grows until the job's disk is full. When a passing job's cache holds more than `cache_prune_percent` of its mount's size (runner config, default 60), the files the job did not read, write or share a cargo unit hash with are removed, oldest first, until it is under; one still over is not kept, and the next run starts empty. A microVM's guest init prunes its disk before it powers off; the runner prunes the syscall tier's directories. The job's log says how much was kept and pruned, or why the cache was not kept.

 "flow-init",

 "tempfile",

flow-init = { path = "crates/flow-init" }

[dev-dependencies]

tempfile.workspace = true

//! a failure of init itself, `\x1eC <json>` a build cache pruned before

//! the guest powers off (oxygenflow#72): `{"target": ..., "pruned":

//! flow_init::prune::Pruned}`, sent before the exit code.

use flow_init::prune;

    /// A build cache's disk (oxygenflow#72): mounted `strictatime`, so

    /// what the job reads is seen, and after a passing job pruned to this

    /// percentage of its size before the guest powers off.

    #[serde(default)]

    prune: Option<u8>,

    /// Paths on the disk, relative to it, that pruning leaves: the job's

    /// outputs, which the runner reads after the guest stops.

    #[serde(default)]

    keep: Vec<String>,

                let atime = if m.prune.is_some() { libc::MS_STRICTATIME | libc::MS_LAZYTIME } else { 0 };

                mount(dev, &target, "ext4", nosuid | atime, "")?;

    // What the job reads or writes on a cache disk from now on is what

    // this run used.

    let started = job.mounts.iter().any(|m| m.prune.is_some()).then(prune::start_clock);

    }

    let code = if libc::WIFEXITED(status) { libc::WEXITSTATUS(status) } else { 128 + libc::WTERMSIG(status) };

    // A failing job's disks are thrown away: only a passing one's are

    // worth pruning.

    if let (Some(started), 0) = (started, code) {

        prune_caches(console, &job.mounts, started);

    }

    Ok(code)

}

/// Prune each cache disk and say so to the host. Whatever the job left

/// running stops first, so nothing changes the disks meanwhile.

fn prune_caches(console: &Mutex<std::io::Stdout>, mounts: &[MountSpec], started: std::time::SystemTime) {

    unsafe {

        libc::kill(-1, libc::SIGKILL);

        let mut status = 0;

        while libc::waitpid(-1, &mut status, 0) > 0 {}

    }

    for m in mounts {

        let Some(percent) = m.prune else { continue };

        let root = std::path::Path::new(&m.target);

        let pruned = match prune::usage(root) {

            Ok((used, size)) => {

                let mut p = prune::prune(root, started, &m.keep, size, used, percent).unwrap_or_else(|e| prune::Pruned {

                    size,

                    percent,

                    before: used,

                    after: used,

                    unpruned: Some(format!("pruning failed: {e}")),

                    ..Default::default()

                });

                // What is left as the file system counts it.

                if p.freed > 0 {

                    unsafe { libc::sync() };

                    if let Ok((now, _)) = prune::usage(root) {

                        p.after = now;

                    }

                }

                p

            }

            Err(e) => prune::Pruned { percent, unpruned: Some(format!("statvfs {}: {e}", m.target)), ..Default::default() },

        };

        let msg = serde_json::json!({ "target": m.target, "pruned": pruned });

        frame(console, b'C', msg.to_string().as_bytes());

flow-init.workspace = true

//!

//! A passing job's disk is pruned before it is kept (oxygenflow#72; see

//! [`flow_init::prune`]): over [`flow_init::prune::DEFAULT_PERCENT`] of

//! its size (the runner's `cache_prune_percent`), the files the job did

//! not use go, oldest first, and a cache still over it is not kept, so a

//! kept target never grows until the job's disk is full. A microVM's

//! guest prunes its image before it stops (the runner cannot look inside

//! an ext4 image without mounting it); the runner prunes a directory

//! itself, having set its files' atimes to the epoch before the job

//! ([`forget_reads`]) so that every read is seen. The job's log says what

//! was kept and pruned.

use flow_init::prune::Pruned;

}

/// How a passing job's cache disk is pruned before it is kept

/// (oxygenflow#72).

pub struct Prune<'a> {

    /// When the job started ([`flow_init::prune::start_clock`]): what it

    /// read or wrote since is what it used.

    pub started: SystemTime,

    /// The share of the disk the kept cache may fill.

    pub percent: u8,

    /// What the guest did to a microVM's image, when it said.

    pub guest: Option<&'a Pruned>,

    /// After the job: a passing job's disk is pruned and kept, unless it

    /// is still over its cap; a failing one's is thrown away, so nothing

    /// it wrote reaches a later job. For a kept or refused cache, what to

    /// say in the job's log.

    #[allow(clippy::too_many_arguments)]

    pub fn settle(

        &self,

        ctx: &Ctx,

        m: &Mount,

        key: &str,

        disk: &Path,

        sources: &Sources,

        passed: bool,

        prune: &Prune,

    ) -> Result<Option<String>> {

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

            let pruned = pruned(ctx.form, m, disk, prune)?;

            let line = describe(name, &pruned);

            if pruned.fits() {

                self.keep(ctx, m, key, disk, sources)?;

            } else {

                discard(ctx.form, disk)?;

            }

            return Ok(Some(line));

        discard(ctx.form, disk)?;

        Ok(None)

    }

}

/// Throw a disk away.

fn discard(form: Form, disk: &Path) -> Result<()> {

    let gone = match form {

        Form::Dir => crate::workspace::remove_tree(disk),

        Form::Image => std::fs::remove_file(disk).or_else(|e| if e.kind() == std::io::ErrorKind::NotFound { Ok(()) } else { Err(e) }),

    };

    gone.with_context(|| format!("discard {}", disk.display()))

}

/// Before a syscall job: set the atime of every file of a directory it

/// took from the pool to the epoch, so that whatever the file system's

/// atime rule, the job's first read of each is seen when the directory is

/// pruned (oxygenflow#72). A microVM's guest mounts its disk

/// `strictatime` instead.

pub fn forget_reads(form: Form, disk: &Path) -> Result<()> {

    if form == Form::Dir && disk.is_dir() {

        flow_init::prune::forget_reads(disk).with_context(|| format!("set the atimes in {}", disk.display()))?;

    }

    Ok(())

}

/// Prune a passing job's disk, or say why it was not: a directory here,

/// an image by what its guest said (else only its size is measured).

fn pruned(form: Form, m: &Mount, disk: &Path, prune: &Prune) -> Result<Pruned> {

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

    let percent = prune.percent;

    let unpruned = |before: u64, why: String| Pruned { size, percent, before, after: before, unpruned: Some(why), ..Pruned::default() };

    Ok(match form {

        Form::Dir => {

            let before = tree_bytes(disk);

            if !flow_init::prune::records_reads(disk) {

                unpruned(before, "this runner's file system records no reads (noatime), so what the job used cannot be told".into())

            } else {

                flow_init::prune::prune(disk, prune.started, &[], size, before, percent)

                    .unwrap_or_else(|e| unpruned(tree_bytes(disk), format!("pruning failed: {e}")))

            }

        }

        Form::Image => match prune.guest.filter(|g| g.size > 0) {

            Some(g) => g.clone(),

            None => {

                let (before, size) = image_usage(disk)?;

                let why = prune.guest.and_then(|g| g.unpruned.clone()).unwrap_or_else(|| "the guest did not prune it".into());

                Pruned { size, ..unpruned(before, why) }

            }

        },

    })

}

/// What an ext4 image holds and its size, as its superblock counts them.

fn image_usage(image: &Path) -> Result<(u64, u64)> {

    let out = std::process::Command::new("dumpe2fs").arg("-h").arg(image).output().context("run dumpe2fs (e2fsprogs)")?;

    if !out.status.success() {

        anyhow::bail!("dumpe2fs {}: {}", image.display(), String::from_utf8_lossy(&out.stderr).trim());

    }

    superblock_usage(&String::from_utf8_lossy(&out.stdout)).with_context(|| format!("dumpe2fs {}: no block counts", image.display()))

}

fn superblock_usage(header: &str) -> Option<(u64, u64)> {

    let field = |name: &str| -> Option<u64> {

        header.lines().find_map(|l| l.strip_prefix(name)?.strip_prefix(':')).and_then(|v| v.trim().parse().ok())

    };

    let (count, free, block) = (field("Block count")?, field("Free blocks")?, field("Block size")?);

    Some((count.saturating_sub(free) * block, count * block))

}

/// The job log's line for a passing job's cache: what was kept, or why

/// not (oxygenflow#72).

pub fn describe(name: &str, p: &Pruned) -> String {

    use crate::images::human;

    let cap = format!("{}% of its {} disk", p.percent, human(p.size));

    let pruned = format!("{} in {} file{} this run did not use", human(p.freed), p.files, if p.files == 1 { "" } else { "s" });

    if p.fits() && p.files > 0 {

        format!("cache {name}: kept {} of {} (pruned {pruned}; the cap is {cap})", human(p.after), human(p.before))

    } else if p.fits() {

        format!("cache {name}: kept {}, under its cap of {cap}", human(p.after))

    } else {

        let after = if p.files > 0 { format!("{} after pruning {pruned}", human(p.after)) } else { human(p.after) };

        let why = p.unpruned.as_ref().map(|u| format!(" ({u})")).unwrap_or_default();

        format!(

            "cache {name}: not kept: {after} is over its cap of {} ({cap}){why}; the next run starts empty, and a larger `size` on the mount gives the cache room",

            human(p.cap())

        )

    /// Pruning that never prunes: the cap is the whole disk.

    fn keep_all() -> Prune<'static> {

        Prune { started: SystemTime::now(), percent: 100, guest: None }

    }

        pool.settle(&main, &m, &key, &w("a"), &built, true, &keep_all()).unwrap();

        pool.settle(&main, &m, &key, &w("b"), &built, true, &keep_all()).unwrap();

        pool.settle(&main, &m, &key2, &w("d"), &built, true, &keep_all()).unwrap();

        pool.settle(&main, &m, &key2, &w("f"), &built, true, &keep_all()).unwrap();

        pool.settle(&main, &m, "k", &work.path().join("a"), &Sources::new(), true, &keep_all()).unwrap();

        pool.settle(&main, &m, "k", &work.path().join("b"), &Sources::new(), false, &keep_all()).unwrap();

        pool.settle(&image, &m, "k", &work.path().join("d.ext4"), &Sources::new(), false, &keep_all()).unwrap();

    }

    /// A file of `kib` KiB at `rel` under `root`, last read and written

    /// at `secs` (ahead of the clock, so the time it was changed, now,

    /// is before a job starting at [`T`]).

    fn aged(root: &Path, rel: &str, kib: usize, secs: u64) -> PathBuf {

        let p = root.join(rel);

        std::fs::create_dir_all(p.parent().unwrap()).unwrap();

        let bytes: Vec<u8> = (0..kib << 10).map(|i: usize| (i.wrapping_mul(2_654_435_761) >> 13) as u8).collect();

        std::fs::write(&p, bytes).unwrap();

        let t = SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(secs);

        std::fs::File::options()

            .write(true)

            .open(&p)

            .unwrap()

            .set_times(std::fs::FileTimes::new().set_accessed(t).set_modified(t))

            .unwrap();

        p

    }

    const T: u64 = 4_000_000_000;

    fn at_t() -> SystemTime {

        SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(T)

    }

    /// oxygenflow#72: a syscall job's directory is pruned of what the job

    /// did not use before it is kept, and the log says so; one still over

    /// its cap is not kept at all.

    #[test]

    fn a_directory_is_pruned_of_what_the_job_did_not_use_before_it_is_kept() {

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

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

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

        let main = dir_ctx("main");

        let mut m = mount(&[]);

        m.size_mib = 8;

        let a = work.path().join("a");

        // This run's build, and an older lockfile's that it did not touch.

        aged(&a, "debug/deps/libnew-aaaaaaaaaaaaaaaa.rlib", 2 << 10, T + 10);

        aged(&a, "debug/deps/libold-bbbbbbbbbbbbbbbb.rlib", 3 << 10, T - 1_000);

        let prune = Prune { started: at_t(), percent: 50, guest: None };

        let line = pool.settle(&main, &m, "k", &a, &Sources::new(), true, &prune).unwrap().unwrap();

        assert_eq!(

            line,

            "cache target: kept 2 MiB of 5 MiB (pruned 3 MiB in 1 file this run did not use; the cap is 50% of its 8 MiB disk)"

        );

        let b = work.path().join("b");

        assert!(pool.take(&main, &m, "k", &b).unwrap().how.contains("same key"));

        assert!(b.join("debug/deps/libnew-aaaaaaaaaaaaaaaa.rlib").is_file() && !b.join("debug/deps/libold-bbbbbbbbbbbbbbbb.rlib").exists());

        // Before the next job its reads are forgotten, its writes not.

        forget_reads(Form::Dir, &b).unwrap();

        let md = std::fs::metadata(b.join("debug/deps/libnew-aaaaaaaaaaaaaaaa.rlib")).unwrap();

        assert_eq!((md.atime(), md.mtime()), (0, (T + 10) as i64));

        // This run's own files over the cap: nothing it used goes, and the

        // cache is not kept, so the next run starts empty.

        aged(&b, "debug/deps/libbig-cccccccccccccccc.rlib", 3 << 10, T + 20);

        let line = pool.settle(&main, &m, "k", &b, &Sources::new(), true, &prune).unwrap().unwrap();

        assert!(

            line.starts_with("cache target: not kept: 5 MiB is over its cap of 4 MiB (50% of its 8 MiB disk); the next run starts empty"),

            "{line}"

        );

        assert!(!b.exists() && pool.take(&main, &m, "k", &work.path().join("c")).is_none());

    }

    /// oxygenflow#72: a microVM's image is kept or not by what its guest

    /// said; without word from it, by what its superblock says is in use.

    #[test]

    fn an_image_is_kept_by_what_its_guest_pruned_or_else_by_its_size() {

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

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

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

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

        let mut m = mount(&[]);

        m.size_mib = 16;

        let fits = Pruned { size: 16 << 20, percent: 60, before: 12 << 20, after: 6 << 20, freed: 6 << 20, files: 40, unpruned: None };

        let over = Pruned { after: 11 << 20, ..fits.clone() };

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

        let line = pool

            .settle(&ctx, &m, "k", &work.path().join("a.ext4"), &Sources::new(), true, &Prune { guest: Some(&fits), ..keep_all() })

            .unwrap()

            .unwrap();

        assert_eq!(

            line,

            "cache target: kept 6 MiB of 12 MiB (pruned 6 MiB in 40 files this run did not use; the cap is 60% of its 16 MiB disk)"

        );

        assert!(pool.take(&ctx, &m, "k", &work.path().join("b.ext4")).is_none(), "a 1 MiB disk is not the mount's 16 MiB");

        disk(&work.path().join("c.ext4"), b"OVER");

        let line = pool

            .settle(&ctx, &m, "k", &work.path().join("c.ext4"), &Sources::new(), true, &Prune { guest: Some(&over), ..keep_all() })

            .unwrap()

            .unwrap();

        assert!(line.starts_with("cache target: not kept: 11 MiB after pruning 6 MiB in 40 files"), "{line}");

        assert!(!work.path().join("c.ext4").exists());

        // No word from the guest: an empty file system is well under.

        let img = work.path().join("d.ext4");

        crate::workspace::empty_ext4(&img, 16, "scratch").unwrap();

        let line = pool.settle(&ctx, &m, "k", &img, &Sources::new(), true, &Prune { percent: 60, ..keep_all() }).unwrap().unwrap();

        assert!(line.starts_with("cache target: kept ") && line.contains("under its cap of 60% of its 16 MiB disk"), "{line}");

        assert!(pool.take(&ctx, &m, "k", &work.path().join("e.ext4")).is_some());

        // ... and one nearly all in use is not kept, saying why.

        let line = pool

            .settle(&ctx, &m, "k", &work.path().join("e.ext4"), &Sources::new(), true, &Prune { percent: 1, ..keep_all() })

            .unwrap()

            .unwrap();

        assert!(line.contains("not kept") && line.contains("(the guest did not prune it)"), "{line}");

    }

    #[test]

    fn the_superblock_says_what_is_in_use() {

        let h = "Filesystem volume name:   scratch\nBlock count:              4096\nFree blocks:              1024\nBlock size:               4096\n";

        assert_eq!(superblock_usage(h), Some((3072 * 4096, 4096 * 4096)));

        assert_eq!(superblock_usage("Block count: 1\n"), None);

    /// The share of its disk, in percent, a kept build cache may fill

    /// (oxygenflow#72; default 60): over it, the files the job did not use

    /// are pruned, oldest first, and a cache still over it is not kept.

    #[serde(default)]

    pub cache_prune_percent: Option<u8>,

    /// Each cache disk the guest pruned before it stopped, by its target

    /// (oxygenflow#72); the syscall tier's are pruned by the runner.

    pub pruned: Vec<(String, flow_init::prune::Pruned)>,

    let percent = home.config.runner.as_ref().and_then(|r| r.cache_prune_percent).unwrap_or(flow_init::prune::DEFAULT_PERCENT);

    let mut spec = guest_spec(job, percent);

    let mut pruned = Vec::new();

                        "C" => match pruned_frame(body) {

                            Some(p) => pruned.push(p),

                            None => sink.record(rec("console", &text)),

                        },

        pruned,

/// A `\x1eC` frame: a cache disk the guest pruned (oxygenflow#72).

fn pruned_frame(body: &str) -> Option<(String, flow_init::prune::Pruned)> {

    #[derive(serde::Deserialize)]

    struct Frame {

        target: String,

        pruned: flow_init::prune::Pruned,

    }

    let f: Frame = serde_json::from_str(body).ok()?;

    Some((f.target, f.pruned))

}

/// with the device each disk mount is attached as (vdc onwards). A cache

/// disk is pruned to `percent` of its size after a passing job

/// (oxygenflow#72), leaving the job's outputs on it.

pub fn guest_spec(job: &Job, percent: u8) -> serde_json::Value {

                let mut disk = serde_json::json!({ "kind": "disk", "target": m.target, "size_mib": m.size_mib, "device": dev });

                if m.cache.is_some() {

                    let prefix = format!("{}/", m.target);

                    let keep: Vec<&str> = job.outputs.iter().filter_map(|o| o.path.strip_prefix(&prefix)).collect();

                    disk["prune"] = percent.into();

                    disk["keep"] = keep.into();

                }

                disk

        assert_eq!(guest_spec(&job, 60)["mounts"], serde_json::json!([]), "no mounts");

        let spec = guest_spec(&job, 60);

    }

    /// oxygenflow#72: a cache disk is pruned in the guest, leaving the

    /// job's outputs on it; a plain disk is not, and the guest's report

    /// comes back as a frame.

    #[test]

    fn a_cache_disk_is_pruned_in_the_guest_but_its_outputs_stay() {

        let mut job: Job = serde_json::from_value(serde_json::json!({

            "name": "t", "sandbox": "s", "scope": [], "depends_on": [], "env": {}, "exec": ["true"],

            "outputs": [{"name": "bin", "path": "/src/target/release/tool"}, {"name": "w", "path": "/src/w"}]

        }))

        .unwrap();

        job.mounts = vec![

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

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

        ];

        let spec = guest_spec(&job, 45);

        assert_eq!((spec["mounts"][0]["prune"].as_u64(), &spec["mounts"][0]["keep"]), (Some(45), &serde_json::json!(["release/tool"])));

        assert!(spec["mounts"][1].get("prune").is_none() && spec["mounts"][1].get("keep").is_none());

        let (target, p) =

            pruned_frame(r#"{"target":"/src/target","pruned":{"size":100,"percent":60,"before":90,"after":50,"freed":40,"files":3}}"#)

                .unwrap();

        assert_eq!((target.as_str(), p.after, p.cap(), p.fits(), p.unpruned), ("/src/target", 50, 60, true, None));

        assert!(pruned_frame("not json").is_none());

            // What the job reads is seen when the disk is pruned (#72).

            if let Err(e) = crate::cache::forget_reads(form, &crate::cache::disk_path(form, &scratch, i)) {

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

            }

        // What the job reads or writes on its cache from now on is what it

        // used; the rest may be pruned before the cache is kept (#72).

        let cache_started = if cached.is_empty() { std::time::SystemTime::now() } else { flow_init::prune::start_clock() };

        // A passing job's cached disks are pruned and kept (#72); a

        // failing job's are thrown away, so nothing it wrote reaches a

        // later job.

            let m = disk_mounts[*i];

            let prune = crate::cache::Prune {

                started: cache_started,

                percent: cfg.cache_prune_percent.unwrap_or(flow_init::prune::DEFAULT_PERCENT),

                guest: outcome.pruned.iter().find(|(t, _)| *t == m.target).map(|(_, p)| p),

            };

            match pool.settle(&cache_ctx, m, key, &disk, &sources, kept, &prune) {

                Ok(Some(line)) => note(&mut sink, line),

                Ok(None) => {}

                Err(e) => note(&mut sink, format!("cache: not {}: {e:#}", if kept { "kept" } else { "discarded" })),

        pruned: Vec::new(),

   `\x1eO <line>` (stdout) and `\x1eE <line>` (stderr). After a passing

   job it prunes each build cache disk (mounted `strictatime`) of the

   files the job did not use when it is over the runner's

   `cache_prune_percent` of its size, and reports each as

   `\x1eC <json>` (spec §5.4, oxygenflow#72); then `\x1eX <code>`, and

   it powers off. The kernel runs `quiet` so the host sees

- **Ephemeral mounts:** a `tmpfs` mount is a tmpfs of its size. An `ephemeral://disk` mount is a scratch directory, so its size is not enforced in this tier. One that names a `cache` starts from a directory a passing job of the slot kept, and goes back into the runner's cache pool when the job passes (spec §5.4, oxygenflow#70), pruned first by the runner of the files the job did not use when over the cap (oxygenflow#72).

- **Pruned before it is kept** (oxygenflow#72). A kept disk would

  otherwise carry every build's files (each crate under every hash a

  lockfile or feature set gave it) until the job's disk is full. When a

  passing job's cache holds more than `cache_prune_percent` of its

  mount's `size` (runner config, default 60), the files the job did not

  use are removed, used longest ago first, until it is under that cap; one

  still over it is not kept, so the next run starts empty. A file was

  used if it was read, written or changed since the job started, or it

  shares a unit hash with one that was (a 16-digit hex word after a `-`

  in its path, as cargo names a unit's fingerprint, rlibs, binaries and

  build-script output: a crate cargo found fresh is only looked at, but

  its fingerprint is read), or is another link of one, or lies under one

  of the job's outputs. A microVM's guest (`flow-init`) prunes its disks

  before it powers off, having mounted them `strictatime` so every read

  is seen, and tells the runner what it kept (`\x1eC` frames); without

  word from it the runner only measures the image (`dumpe2fs`) and keeps

  it if it is under the cap. The syscall tier's runner prunes a directory

  itself, having set each file's atime to the epoch when the job took it,

  so even a `relatime` file system records the job's first read of each;

  on a `noatime` one nothing is pruned and only the size decides. The

  job's log says what happened: `cache target: kept 12.3 GiB of 23.1 GiB

  (pruned 10.8 GiB in 4113 files this run did not use; the cap is 60% of

  its 24.0 GiB disk)`, or `not kept:` and why.