Build caches for the syscall tier 2594278f49b2

- **The syscall tier keeps its build caches too (#70, spec §5.4).** A cached `ephemeral://disk` mount in a `runtime "syscall"` job now starts from the directory a passing job of the same slot kept, and goes back into the runner's pool when the job passes, with the same order, bookmark rules, `cache_gb` eviction and claim reporting as the microVM tier's disks, so jobs go where their cache is on boards without KVM as well. Nothing kept is writable by others or set-id, and the pool is the runner's alone.

//! A disk is in the [`Form`] its tier mounts (oxygenflow#70): a microVM's

//! is an ext4 image, `<key>.ext4`; the syscall tier's is the host

//! directory it binds at the mount's target, `<key>.tree`, taken and kept

//! the same way. A directory goes into the pool with nothing in it

//! writable by others or set-id, and the pool itself is the runner's

//! alone.

//!

//! Beside each disk, `<key>.sources` (`<key>.tree-sources` for a

//! directory) lists the sources it was built from (oxygenflow#52): each

//! file's content hash and the mtime it had. A job that takes the disk

//! gives every file whose content is unchanged that mtime back, so a build

//! tool that judges freshness by time (cargo) rebuilds only what changed;

//! a changed file keeps the time of its checkout, newer than anything on

//! the disk, so it is always rebuilt.

use std::os::unix::fs::{MetadataExt, PermissionsExt};

}

/// How a tier holds a disk mount's files (oxygenflow#70).

#[derive(Clone, Copy, Debug, PartialEq, Eq)]

pub enum Form {

    /// An ext4 image of the mount's size, attached to a microVM.

    Image,

    /// A host directory, bound into the syscall sandbox.

    Dir,

}

impl Form {

    fn ext(self) -> &'static str {

        match self {

            Form::Image => "ext4",

            Form::Dir => "tree",

        }

    }

}

/// Which form a kept entry of the pool is in, by its name.

fn form_of(p: &Path) -> Option<Form> {

    match p.extension()?.to_str()? {

        "ext4" => Some(Form::Image),

        "tree" => Some(Form::Dir),

        _ => None,

    }

    match form_of(disk) {

        Some(Form::Dir) => disk.with_extension("tree-sources"),

        _ => disk.with_extension("sources"),

    }

/// Which job: the pool's coordinates other than the key, and the form its

/// tier keeps a disk in.

    pub form: Form,

/// Held while the pool changes: jobs run side by side, and a directory,

/// unlike an image, can be half moved or half removed while another job

/// copies it.

static CHANGING: std::sync::Mutex<()> = std::sync::Mutex::new(());

fn changing() -> std::sync::MutexGuard<'static, ()> {

    CHANGING.lock().unwrap_or_else(|p| p.into_inner())

}

    /// Start `dest` from the pool for mount `m`: an image of `m.size_mib`,

    /// or a directory, as `ctx.form` says; `dest` must not exist yet. What

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

    ///

    /// A disk this job's bookmark kept is moved out of the pool, so a job

    /// of the same slot running beside this one can never take it too.

        let _changing = changing();

        let form = ctx.form;

        let fits = |p: &Path| match form {

            Form::Image => std::fs::metadata(p).is_ok_and(|md| md.len() == size),

            Form::Dir => std::fs::symlink_metadata(p).is_ok_and(|md| md.is_dir()),

        };

        let ext = form.ext();

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

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

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

            .or_else(|| {

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

            });

            if copy(form, &p, dest).is_ok() {

        let _changing = changing();

        // Whatever the job left, nothing in the pool is anyone else's to

        // write or runs as the runner (oxygenflow#70).

        if ctx.form == Form::Dir {

            seal(disk).with_context(|| format!("seal {}", disk.display()))?;

        }

        std::os::unix::fs::DirBuilderExt::mode(std::fs::DirBuilder::new().recursive(true), 0o700).create(&mine)?;

        std::fs::set_permissions(&self.root, std::fs::Permissions::from_mode(0o700))?;

        let dest = mine.join(format!("{key}.{}", ctx.form.ext()));

        if ctx.form == Form::Dir {

            // Another job of this slot kept one since: this one is newer.

            crate::workspace::remove_tree(&dest)?;

        }

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

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

        let sidecar = sources_of(&dest);

        let tmp = sidecar.with_extension("sources-part");

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

        {

            use std::io::Write;

            use std::os::unix::fs::OpenOptionsExt;

            std::fs::OpenOptions::new().write(true).create_new(true).mode(0o600).open(&tmp)?.write_all(&serde_json::to_vec(sources)?)?;

        }

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

    }

    /// After the job: a passing job's disk is kept; a failing one's is

    /// thrown away, so nothing it wrote reaches a later job.

    pub fn settle(&self, ctx: &Ctx, m: &Mount, key: &str, disk: &Path, sources: &Sources, passed: bool) -> Result<()> {

        if passed {

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

        }

        let gone = match ctx.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()))

    /// build (oxygenflow#54). Images and directories alike, so a job goes

    /// where its cache is whichever tier keeps it.

                // `<job>/<cache>/<bookmark>/<disk>`

                kept(&job.path(), 2, &mut |p, _| {

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

    /// image's blocks, or everything in a directory. The caller holds

    /// [`changing`].

    fn evict(&self) -> u64 {

        // `<repo>/<job>/<cache>/<bookmark>/<disk>`

        kept(&self.root, 4, &mut |p, form| {

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

                let bytes = match form {

                    Form::Image => md.blocks() * 512,

                    Form::Dir => tree_bytes(p),

                };

                disks.push((md.modified().unwrap_or(SystemTime::UNIX_EPOCH), bytes, p.to_path_buf()));

            let gone = match form_of(&p) {

                Some(Form::Dir) => crate::workspace::remove_tree(&p),

                _ => std::fs::remove_file(&p),

            };

            if gone.is_ok() {

/// Where a job's `n`th disk mount is in its scratch directory, in the form

/// its tier mounts it: `microvm::run` attaches `mount-N.ext4`, and

/// `syscall::run` binds the directory `mount-N`.

pub fn disk_path(form: Form, scratch: &Path, n: usize) -> PathBuf {

    match form {

        Form::Image => scratch.join(format!("mount-{n}.ext4")),

        Form::Dir => crate::syscall::disk_dir(scratch, n),

    }

}

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

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

    std::fs::read_dir(dir).ok()?.flatten().map(|e| e.path()).filter(|p| form_of(p) == Some(form) && ok(p)).max_by_key(|p| modified(p))

/// Each kept image or directory `depth` levels of directories below `dir`,

/// with its form; never inside a kept directory.

fn kept(dir: &Path, depth: usize, f: &mut dyn FnMut(&Path, Form)) {

        let Ok(t) = e.file_type() else { continue };

        if depth == 0 {

            match form_of(&p) {

                Some(Form::Image) if t.is_file() => f(&p, Form::Image),

                Some(Form::Dir) if t.is_dir() => f(&p, Form::Dir),

                _ => {}

            }

        } else if t.is_dir() {

            kept(&p, depth - 1, f);

        }

    }

}

/// The blocks a directory's files take, each file once however many links

/// it has; links are not followed.

fn tree_bytes(dir: &Path) -> u64 {

    fn walk(p: &Path, seen: &mut std::collections::HashSet<(u64, u64)>) -> u64 {

        let Ok(md) = std::fs::symlink_metadata(p) else { return 0 };

        if md.nlink() > 1 && !md.is_dir() && !seen.insert((md.dev(), md.ino())) {

            return 0;

        }

        let mut bytes = md.blocks() * 512;

        if md.is_dir() {

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

                bytes += walk(&e.path(), seen);

            }

        }

        bytes

    }

    walk(dir, &mut Default::default())

}

/// Make a directory a job left fit for the pool: nothing in it writable by

/// group or others, nothing set-id or sticky, and everything readable by

/// the runner (directories searchable and writable too, so it can be

/// copied and removed). Links are not followed. The job ran as the

/// runner's user, so all of it is the runner's to change; times are kept.

fn seal(p: &Path) -> std::io::Result<()> {

    let md = std::fs::symlink_metadata(p)?;

    if md.file_type().is_symlink() {

        return Ok(());

    }

    let mode = md.mode() & 0o7777;

    let want = (mode & !0o7022) | if md.is_dir() { 0o700 } else { 0o400 };

    if want != mode {

        std::fs::set_permissions(p, std::fs::Permissions::from_mode(want))?;

    }

    if md.is_dir() {

        for e in std::fs::read_dir(p)? {

            seal(&e?.path())?;

    }

    Ok(())

}

fn copy(form: Form, from: &Path, to: &Path) -> Result<()> {

    match form {

        Form::Image => sparse_copy(from, to),

        Form::Dir => tree_copy(from, to),

/// Copy a directory with its modes and times (a build tool judges by

/// them), sharing blocks where the filesystem can.

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

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

    crate::workspace::remove_tree(&tmp)?;

    let st = std::process::Command::new("cp")

        .args(["-a", "--reflink=auto", "--no-target-directory"])

        .arg(from)

        .arg(&tmp)

        .status()

        .context("run cp")?;

    if !st.success() {

        let _ = crate::workspace::remove_tree(&tmp);

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

    }

    std::fs::rename(&tmp, to).inspect_err(|_| {

        let _ = crate::workspace::remove_tree(&tmp);

    })?;

    Ok(())

}

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

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

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

    }

    /// A directory as the syscall tier leaves one: `marker` in

    /// `debug/marker`, and a file of `kib` KiB.

    fn tree(path: &Path, marker: &[u8], kib: usize) {

        std::fs::create_dir_all(path.join("debug")).unwrap();

        std::fs::write(path.join("debug/marker"), marker).unwrap();

        // Not all one byte, so a compressing filesystem stores it whole.

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

        std::fs::write(path.join("debug/blob"), bytes).unwrap();

    }

    fn dir_ctx<'a>(bookmark: &'a str) -> Ctx<'a> {

        Ctx { repo: "andy/flow", bookmark, job: "test", image: "img", form: Form::Dir }

    }

    /// oxygenflow#70: the syscall tier's directories are taken and kept as

    /// a microVM's images are, and never mixed with them.

    #[test]

    fn a_kept_directory_starts_the_next_syscall_job_and_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 = dir_ctx("main");

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

        let w = |n: &str| work.path().join(n);

        assert!(pool.take(&main, &m, &key, &w("a")).is_none(), "nothing kept yet");

        // A passing job's directory goes into the pool, times and all.

        tree(&w("a"), b"FIRST", 4);

        // Named like an image: the pool never looks inside a directory.

        std::fs::write(w("a/debug/inner.ext4"), b"not a disk").unwrap();

        let old = SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_000_000);

        std::fs::File::open(w("a/debug/marker")).unwrap().set_modified(old).unwrap();

        let built: Sources = [("src/lib.rs".to_string(), ("h1".to_string(), 1_000))].into();

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

        assert!(!w("a").exists(), "moved, not copied");

        assert_eq!(

            pool.slot_times().into_keys().collect::<Vec<_>>(),

            vec![flow_core::api::cache_slot("andy/flow", "test")],

            "the claim reports it like an image (#51)"

        );

        // The next job of its bookmark moves it out: a job of the same

        // slot beside it finds nothing to take.

        let taken = pool.take(&main, &m, &key, &w("b")).unwrap();

        assert!(taken.how.contains("this bookmark's, same key"), "{}", taken.how);

        assert_eq!(taken.sources.as_ref(), Some(&built), "its sources come with it (#52)");

        assert_eq!(std::fs::read(w("b/debug/marker")).unwrap(), b"FIRST");

        assert_eq!(modified(&w("b/debug/marker")), old, "its files keep their times");

        assert_eq!(std::fs::read(w("b/debug/inner.ext4")).unwrap(), b"not a disk");

        assert!(pool.take(&main, &m, &key, &w("c")).is_none(), "taken once");

        assert!(pool.slot_times().is_empty());

        // Kept again, and under a changed key the newest is taken all the same.

        std::fs::write(w("b/debug/marker"), b"SECOND").unwrap();

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

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

        let taken = pool.take(&main, &m, &key2, &w("d")).unwrap();

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

        assert_eq!(std::fs::read(w("d/debug/marker")).unwrap(), b"SECOND");

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

        // Another bookmark copies it, with its times, and main keeps its own.

        let feature = dir_ctx("feature");

        let taken = pool.take(&feature, &m, &key2, &w("e")).unwrap();

        assert!(taken.how.contains("copied from another bookmark's (main-"), "{}", taken.how);

        assert_eq!(taken.sources.as_ref(), Some(&built));

        assert_eq!(std::fs::read(w("e/debug/marker")).unwrap(), b"SECOND");

        assert!(pool.take(&main, &m, &key2, &w("f")).is_some(), "main's is still there");

        assert_eq!(modified(&w("e/debug/marker")), modified(&w("f/debug/marker")), "a copy keeps the times");

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

        // A microVM job of the same slot never takes a directory, nor a

        // syscall job an image.

        let image = Ctx { form: Form::Image, ..main };

        assert!(pool.take(&image, &m, &key2, &w("g.ext4")).is_none());

        disk(&w("h.ext4"), b"DISK");

        pool.keep(&image, &m, &key2, &w("h.ext4"), &built).unwrap();

        let taken = pool.take(&main, &m, &key2, &w("i")).unwrap();

        assert!(w("i/debug/marker").is_file(), "{}", taken.how);

        assert!(pool.take(&main, &m, &key2, &w("j")).is_none(), "the image is not a directory");

        assert!(pool.take(&image, &m, &key2, &w("k.ext4")).is_some(), "and is still there for a microVM");

    }

    #[test]

    fn a_failing_jobs_directory_is_thrown_away() {

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

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

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

        let m = mount(&[]);

        let main = dir_ctx("main");

        tree(&work.path().join("a"), b"GOOD", 1);

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

        let taken = pool.take(&main, &m, "k", &work.path().join("b"));

        assert!(taken.is_some());

        // The job that took it fails after writing to it: not kept, gone.

        std::fs::write(work.path().join("b/debug/marker"), b"POISON").unwrap();

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

        assert!(!work.path().join("b").exists());

        assert!(pool.take(&main, &m, "k", &work.path().join("c")).is_none(), "nothing it wrote reaches a later job");

        assert!(pool.slot_times().is_empty());

        // A failed image is thrown away the same way.

        let image = Ctx { form: Form::Image, ..main };

        disk(&work.path().join("d.ext4"), b"BAD!");

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

        assert!(!work.path().join("d.ext4").exists() && pool.slot_times().is_empty());

    }

    #[test]

    fn directories_are_evicted_by_their_size_oldest_first() {

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

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

        // Room for one 512 KiB directory, not two.

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

        let m = mount(&[]);

        let (main, feature) = (dir_ctx("main"), dir_ctx("feature"));

        tree(&work.path().join("a"), b"OLD", 512);

        pool.keep(&main, &m, "k1", &work.path().join("a"), &Sources::new()).unwrap();

        let kept = pool.cache_dir(&main, "target").join(segment("main")).join("k1.tree");

        assert!(kept.is_dir(), "under the limit it stays");

        assert!(tree_bytes(&kept) >= 512 << 10, "a directory counts what is in it");

        // Older than the next one, however fast this test runs.

        std::fs::File::open(&kept).unwrap().set_modified(SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1)).unwrap();

        tree(&work.path().join("b"), b"NEW", 512);

        pool.keep(&feature, &m, "k1", &work.path().join("b"), &Sources::new()).unwrap();

        assert!(!kept.exists() && !kept.with_extension("tree-sources").exists(), "the one used longest ago goes, sources and all");

        let taken = pool.take(&main, &m, "k1", &work.path().join("c")).unwrap();

        assert!(taken.how.contains("bookmark's (feature-"), "{}", taken.how);

        assert_eq!(std::fs::read(work.path().join("c/debug/marker")).unwrap(), b"NEW");

    }

    /// oxygenflow#70: the job wrote what it liked; the pool keeps nothing

    /// others may write or that runs as the runner, and is the runner's.

    #[test]

    fn nothing_kept_is_writable_by_others_or_set_id() {

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

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

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

        let m = mount(&[]);

        let main = dir_ctx("main");

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

        tree(&a, b"X", 1);

        let mode = |p: &Path, m: u32| std::fs::set_permissions(p, std::fs::Permissions::from_mode(m)).unwrap();

        std::fs::write(a.join("open"), b"").unwrap();

        mode(&a.join("open"), 0o777);

        std::fs::write(a.join("suid"), b"").unwrap();

        mode(&a.join("suid"), 0o4755);

        std::fs::write(a.join("debug/sealed"), b"secret").unwrap();

        mode(&a.join("debug/sealed"), 0o000);

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

        let when = modified(&a.join("debug/marker"));

        mode(&a.join("debug"), 0o1777);

        mode(&a, 0o500);

        pool.keep(&main, &m, "k", &a, &Sources::new()).unwrap();

        let root = home.path().join("cache");

        assert_eq!(std::fs::metadata(&root).unwrap().mode() & 0o777, 0o700, "the pool is the runner's alone");

        let mut seen = 0;

        fn each(p: &Path, f: &mut dyn FnMut(&Path, &std::fs::Metadata)) {

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

            f(p, &md);

            if md.is_dir() {

                for e in std::fs::read_dir(p).unwrap() {

                    each(&e.unwrap().path(), f);

                }

            }

        }

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

            seen += 1;

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

                assert_eq!(md.mode() & 0o7022, 0, "{} is {:o}", p.display(), md.mode());

            }

        });

        assert!(seen >= 10, "{seen}");

        // Still readable, with its times, so the next job builds on it.

        let taken = pool.take(&main, &m, "k", &work.path().join("b"));

        assert!(taken.is_some());

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

        assert_eq!(std::fs::read(b.join("debug/sealed")).unwrap(), b"secret");

        assert_eq!(std::fs::metadata(b.join("open")).unwrap().mode() & 0o777, 0o755);

        assert_eq!(std::fs::read_link(b.join("link")).unwrap(), Path::new("/etc/passwd"), "a link is kept, not followed");

        assert_eq!(modified(&b.join("debug/marker")), when);

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

        // microVM takes an image already in place as it is, the syscall tier

        // a directory (#70). Taken before the workspace is imaged: the

        // sources the disk was built from give unchanged files their old

        // times back (#52).

        let form = if syscall_tier { crate::cache::Form::Dir } else { crate::cache::Form::Image };

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

        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 = match pool.take(&cache_ctx, m, &key, &crate::cache::disk_path(form, &scratch, i)) {

                Some(t) => {

                    if built_from.is_none() {

                        built_from = t.sources;

                    t.how

                }

                None => format!("cache {name}: none kept yet, an empty disk"),

            };

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

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

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

        // thrown away, so nothing it wrote reaches a later job.

        let kept = passed && output_error.is_none();

            let disk = crate::cache::disk_path(form, &scratch, *i);

            if let Err(e) = pool.settle(&cache_ctx, disk_mounts[*i], key, &disk, &sources, kept) {

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

- **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).

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

size the mount asks for.

- **The syscall tier keeps directories** (oxygenflow#70). Its

  `ephemeral://disk` mount is a host directory bound at the target, so

  its pool holds directories (`<key>.tree`) beside the microVM's images

  (`<key>.ext4`), taken and kept the same way: moved out of the pool for a

  job of the same bookmark (so two jobs of one slot running at once never

  take the same one), copied with their modes and times from another

  bookmark's, kept only when the job and its outputs pass and otherwise

  thrown away, and counted by the blocks of everything in them toward

  `cache_gb`. A directory has no size, so `size` does not choose between

  them. Each tier takes only its own form: a syscall job never starts from

  a microVM's image, nor a microVM from a directory. The job runs as the

  runner's user, so what it wrote is the runner's; before a directory goes

  into the pool nothing in it is left writable by group or others, or

  set-id, and the pool is readable by the runner alone. Claims report the

  slots of both forms in `caches` and `cache_times`, so jobs go where

  their cache is whichever tier they run in.