//! 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.
//!
//! 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.
//!
//! 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 anyhow::{Context, Result};
use flow_core::pipeline::Mount;
use flow_init::prune::Pruned;
use std::os::unix::fs::{MetadataExt, PermissionsExt};
use std::path::{Path, PathBuf};
use std::time::SystemTime;

/// The pool's default size limit.
pub const DEFAULT_GB: u64 = 64;

/// The sources a cached disk was built from: path to content hash and
/// mtime, in whole seconds (what the workspace image keeps).
pub type Sources = std::collections::BTreeMap<String, (String, i64)>;

/// What a job started a cached disk from.
pub struct Taken {
    /// For the log and the receipt.
    pub how: String,
    /// The sources it was built from, when they were kept with it.
    pub sources: Option<Sources>,
}

/// 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,
    }
}

fn sources_of(disk: &Path) -> PathBuf {
    match form_of(disk) {
        Some(Form::Dir) => disk.with_extension("tree-sources"),
        _ => disk.with_extension("sources"),
    }
}

fn read_sources(disk: &Path) -> Option<Sources> {
    serde_json::from_slice(&std::fs::read(sources_of(disk)).ok()?).ok()
}

/// 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>,
}

/// 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, and the form its
/// tier keeps a disk in.
pub struct Ctx<'a> {
    pub repo: &'a str,
    pub bookmark: &'a str,
    pub job: &'a str,
    pub image: &'a str,
    pub form: Form,
}

use flow_core::api::segment;

/// 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())
}

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`: 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.
    pub fn take(&self, ctx: &Ctx, m: &Mount, key: &str, dest: &Path) -> Option<Taken> {
        let name = m.cache.as_deref()?;
        let _changing = changing();
        let dir = self.cache_dir(ctx, name);
        let size = m.size_mib as u64 * (1 << 20);
        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 mine = dir.join(segment(ctx.bookmark));
        let exact = mine.join(format!("{key}.{ext}"));
        // A moved disk takes its sources with it.
        if fits(&exact) {
            let sources = read_sources(&exact);
            if std::fs::rename(&exact, dest).is_ok() {
                let _ = std::fs::remove_file(sources_of(&exact));
                return Some(Taken { how: format!("cache {name}: this bookmark's, same key"), sources });
            }
        }
        if let Some(p) = newest(&mine, form, |p| fits(p)) {
            let sources = read_sources(&p);
            if std::fs::rename(&p, dest).is_ok() {
                let _ = std::fs::remove_file(sources_of(&p));
                return Some(Taken { how: format!("cache {name}: this bookmark's, an older key"), sources });
            }
        }
        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}.{ext}")))
            .filter(|p| fits(p))
            .max_by_key(|p| modified(p))
            .map(|p| (p, "same key"))
            .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 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 copy(form, &p, dest).is_ok() {
                return Some(Taken {
                    how: format!("cache {name}: copied from another bookmark's ({from}), {which}"),
                    sources: read_sources(&p),
                });
            }
        }
        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, sources: &Sources) -> Result<()> {
        let Some(name) = m.cache.as_deref() else { return 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()))?;
        }
        let mine = self.cache_dir(ctx, name).join(segment(ctx.bookmark));
        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)?;
        }
        if std::fs::rename(disk, &dest).is_err() {
            copy(ctx.form, disk, &dest).with_context(|| format!("keep {}", disk.display()))?;
        }
        // Moved disks keep their old time: mark it used now.
        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)?;
        // One disk per key is enough; older keys of this bookmark go once
        // the pool is over its limit.
        self.evict();
        Ok(())
    }

    /// 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>> {
        if passed {
            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)
    }

    /// The repository and job slots holding at least one disk, as
    /// [`flow_core::api::cache_slot`] names them, for the claim
    /// (oxygenflow#51), each with when its newest disk was kept (Unix ms):
    /// a disk's time is set when it is kept, so the newest is the freshest
    /// build (oxygenflow#54). Images and directories alike, so a job goes
    /// where its cache is whichever tier keeps it.
    pub fn slot_times(&self) -> std::collections::BTreeMap<String, u64> {
        let mut out = std::collections::BTreeMap::new();
        for repo in std::fs::read_dir(&self.root).into_iter().flatten().flatten() {
            for job in std::fs::read_dir(repo.path()).into_iter().flatten().flatten() {
                let mut newest: Option<SystemTime> = None;
                // `<job>/<cache>/<bookmark>/<disk>`
                kept(&job.path(), 2, &mut |p, _| {
                    let t = modified(p);
                    newest = Some(newest.map_or(t, |n| n.max(t)));
                });
                if let Some(t) = newest {
                    let ms = t.duration_since(SystemTime::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
                    out.insert(format!("{}/{}", repo.file_name().to_string_lossy(), job.file_name().to_string_lossy()), ms);
                }
            }
        }
        out
    }

    /// 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 {
        let mut disks: Vec<(SystemTime, u64, PathBuf)> = Vec::new();
        // `<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 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;
            }
            let gone = match form_of(&p) {
                Some(Form::Dir) => crate::workspace::remove_tree(&p),
                _ => std::fs::remove_file(&p),
            };
            if gone.is_ok() {
                let _ = std::fs::remove_file(sources_of(&p));
                total -= bytes;
                freed += bytes;
            }
        }
        freed
    }
}

/// 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())
        )
    }
}

/// 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),
    }
}

fn modified(p: &Path) -> SystemTime {
    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)) {
    for e in std::fs::read_dir(dir).into_iter().flatten().flatten() {
        let p = e.path();
        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 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(())
}

/// 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(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use flow_core::pipeline::MountSource;

    /// Pruning that never prunes: the cap is the whole disk.
    fn keep_all() -> Prune<'static> {
        Prune { started: SystemTime::now(), percent: 100, guest: None }
    }

    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", form: Form::Image };
        let key = Pool::key(&main, &m, |_| Some("lock-1".into()));
        // Nothing yet: the caller makes an empty disk.
        assert!(pool.take(&main, &m, &key, &work.path().join("a.ext4")).is_none());
        // A passing job's disk is kept and moved into the next job's place.
        disk(&work.path().join("a.ext4"), b"MAIN");
        let built: Sources = [("src/lib.rs".to_string(), ("h1".to_string(), 1_000))].into();
        pool.keep(&main, &m, &key, &work.path().join("a.ext4"), &built).unwrap();
        assert!(!work.path().join("a.ext4").exists(), "moved, not copied");
        let taken = pool.take(&main, &m, &key, &work.path().join("b.ext4")).unwrap();
        assert!(taken.how.contains("same key"), "{}", taken.how);
        assert_eq!(taken.sources.as_ref(), Some(&built), "its sources come with it (#52)");
        assert_eq!(first(&work.path().join("b.ext4")), b"MAIN");
        pool.keep(&main, &m, &key, &work.path().join("b.ext4"), &built).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 taken = pool.take(&main, &m, &key2, &work.path().join("c.ext4")).unwrap();
        assert!(taken.how.contains("older key"), "{}", taken.how);
        assert_eq!(taken.sources.as_ref(), Some(&built));
        pool.keep(&main, &m, &key2, &work.path().join("c.ext4"), &built).unwrap();
        assert_eq!(
            pool.slot_times().into_keys().collect::<Vec<_>>(),
            vec![flow_core::api::cache_slot("andy/hub", "test")],
            "what the claim reports (#51)"
        );
        // With when it was last kept (#54): just now.
        let now = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_millis() as u64;
        let kept = pool.slot_times()[&flow_core::api::cache_slot("andy/hub", "test")];
        assert!(now.abs_diff(kept) < 60_000, "kept {kept}, now {now}");
        // Another bookmark copies it, leaving main's in place.
        let feature = Ctx { bookmark: "feature", ..main };
        let taken = pool.take(&feature, &m, &key2, &work.path().join("d.ext4")).unwrap();
        assert!(taken.how.contains("copied from another bookmark"), "{}", taken.how);
        assert_eq!(taken.sources.as_ref(), Some(&built), "a copy reads them and leaves them");
        assert_eq!(first(&work.path().join("d.ext4")), b"MAIN");
        let taken = pool.take(&main, &m, &key2, &work.path().join("e.ext4")).unwrap();
        assert!(taken.how.contains("this bookmark's") && taken.sources.is_some(), "main's is still there, with its sources");
        // Another job or a different size never shares it.
        let other_job = Ctx { job: "build", ..main };
        assert!(pool.take(&other_job, &m, &key2, &work.path().join("f.ext4")).is_none());
        let mut bigger = mount(&["Cargo.lock"]);
        bigger.size_mib = 2;
        assert!(pool.take(&feature, &bigger, &key2, &work.path().join("g.ext4")).is_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", form: Form::Image };
        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", form: Form::Image };
        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"), &Sources::new()).unwrap();
        assert!(pool.take(&ctx, &m, "k1", &work.path().join("b.ext4")).is_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"), &Sources::new()).unwrap();
        assert!(roomy.take(&ctx, &m, "k2", &work.path().join("d.ext4")).is_some(), "under the limit it stays");
    }

    /// 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, &keep_all()).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, &keep_all()).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, &keep_all()).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, &keep_all()).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, &keep_all()).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, &keep_all()).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, &keep_all()).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);
    }
    /// 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);
    }
}