feat(cache): unchanged sources keep their times; claims list the caches held (closes #52, refs #51) 82cf6e91d398

- **The build cache pays off (#51, #52, spec §5.4).** A cached disk keeps the content hashes and times of the sources it was built from, and the next job gives every unchanged file its old time back, so cargo rebuilds only the crates that changed instead of every path crate. Claims list the caches a runner holds, and the hub sends a job to the runner holding its cache, as it does for images.

    /// The build caches this runner holds a disk of (oxygenflow#51), each

    /// [`cache_slot`] of a repository and job, so the hub can send a job

    /// where its cache is. Absent: an older runner.

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

    pub caches: Option<Vec<String>>,

}

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

/// two names that clean up alike stay apart. The build cache's pool is laid

/// out by these (oxygenflow#28).

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

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

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

}

/// Where a repository's job keeps its build caches, as runners report

/// them in [`ClaimRequest::caches`] (oxygenflow#51): `repo/job`, each part

/// a [`segment`].

pub fn cache_slot(repo: &str, job: &str) -> String {

    format!("{}/{}", segment(repo), segment(job))

//!

//! Beside each disk, `<key>.sources` 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.

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

}

fn sources_of(disk: &Path) -> PathBuf {

    disk.with_extension("sources")

}

fn read_sources(disk: &Path) -> Option<Sources> {

    serde_json::from_slice(&std::fs::read(sources_of(disk)).ok()?).ok()

}

use flow_core::api::segment;

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

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

            }

            let sources = read_sources(&p);

                let _ = std::fs::remove_file(sources_of(&p));

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

                return Some(Taken {

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

                    sources: read_sources(&p),

                });

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

        let tmp = sources_of(&dest).with_extension("sources-part");

        std::fs::write(&tmp, serde_json::to_vec(sources)?)?;

        std::fs::rename(&tmp, sources_of(&dest))?;

    }

    /// The repository and job slots holding at least one disk, as

    /// [`flow_core::api::cache_slot`] names them, for the claim

    /// (oxygenflow#51).

    pub fn slots(&self) -> Vec<String> {

        let mut out = Vec::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 any = false;

                walk(&job.path(), &mut |p| any |= p.extension().is_some_and(|e| e == "ext4"));

                if any {

                    out.push(format!("{}/{}", repo.file_name().to_string_lossy(), job.file_name().to_string_lossy()));

                }

            }

        }

        out.sort();

        out

                let _ = std::fs::remove_file(sources_of(&p));

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

        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();

        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)");

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

        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.slots(), vec![flow_core::api::cache_slot("andy/hub", "test")], "what the claim reports (#51)");

        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");

        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");

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

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

        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());

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

                caches: Some(crate::cache::Pool::new(&home.root, cfg.cache_gb).slots()),

        let job = claim.job.in_sandbox(&claim.sandbox).with_flow_env(&claim.flow_env());

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

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

        // workspace is imaged: the sources the disk was built from give

        // unchanged files their old times back (#52).

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

        let cache_ctx =

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

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

        let mut built_from: Option<crate::cache::Sources> = None;

        if !syscall_tier {

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

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

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

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

                    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));

            }

        }

        let sources = if cached.is_empty() {

            crate::cache::Sources::new()

        } else {

            let (sources, restored) = workspace::restore_times(&scratch.join("tree"), &files, built_from.as_ref());

            if built_from.is_some() {

                note(&mut sink, format!("cache: {restored} of {} files unchanged since the cached build, their times kept", sources.len()));

            }

            sources

        };

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

/// Give the files of `dir` whose content a cached disk was built from

/// the mtime they had then (oxygenflow#52), and describe this checkout's

/// sources for the disk to keep: unchanged files are fresh to a build

/// tool that judges by time, changed ones newer than anything it built.

/// Whole seconds, as the workspace image keeps them. Returns the sources

/// and how many files got their time back.

pub fn restore_times(dir: &Path, files: &Files, before: Option<&crate::cache::Sources>) -> (crate::cache::Sources, usize) {

    let mut out = crate::cache::Sources::new();

    let mut restored = 0;

    for (path, hash) in files {

        let p = dir.join(path);

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

        if !md.is_file() {

            continue;

        }

        let hex = hex::encode(hash);

        let kept = before.and_then(|b| b.get(path)).filter(|(h, _)| *h == hex).map(|(_, t)| *t);

        let t = match kept {

            Some(t) => {

                let at = std::time::UNIX_EPOCH + std::time::Duration::from_secs(t.max(0) as u64);

                if std::fs::File::options().write(true).open(&p).and_then(|f| f.set_modified(at)).is_err() {

                    continue;

                }

                restored += 1;

                t

            }

            None => md.modified().ok().and_then(|m| m.duration_since(std::time::UNIX_EPOCH).ok()).map(|d| d.as_secs() as i64).unwrap_or(0),

        };

        out.insert(path.clone(), (hex, t));

    }

    (out, restored)

}

        }

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    use std::time::UNIX_EPOCH;

    fn secs(p: &Path) -> i64 {

        std::fs::metadata(p).unwrap().modified().unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs() as i64

    }

    /// oxygenflow#52: an unchanged file gets the time it had when the cached

    /// disk was built; a changed or new one keeps its checkout's, newer.

    #[test]

    fn unchanged_sources_get_their_cached_times_back() {

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

        for (name, text) in [("same.rs", "a"), ("changed.rs", "b2"), ("new.rs", "c")] {

            std::fs::write(dir.path().join(name), text).unwrap();

        std::os::unix::fs::symlink("same.rs", dir.path().join("link.rs")).unwrap();

        let h = |t: &str| *blake3::hash(t.as_bytes()).as_bytes();

        let files: Files =

            vec![("same.rs".into(), h("a")), ("changed.rs".into(), h("b2")), ("new.rs".into(), h("c")), ("link.rs".into(), h("same.rs"))];

        let now = secs(&dir.path().join("same.rs"));

        let before: crate::cache::Sources =

            [("same.rs".to_string(), (hex::encode(h("a")), 1_000_000)), ("changed.rs".to_string(), (hex::encode(h("b1")), 1_000_000))]

                .into();

        let (sources, restored) = restore_times(dir.path(), &files, Some(&before));

        assert_eq!(restored, 1);

        assert_eq!(secs(&dir.path().join("same.rs")), 1_000_000, "unchanged: fresh to cargo");

        assert!(secs(&dir.path().join("changed.rs")) >= now, "changed: newer than the cached build");

        assert_eq!(sources["same.rs"], (hex::encode(h("a")), 1_000_000));

        assert_eq!(sources["changed.rs"].0, hex::encode(h("b2")), "kept with its new content");

        assert!(sources.contains_key("new.rs") && !sources.contains_key("link.rs"), "files only");

        // Without a cached build nothing moves, and the sources are listed.

        let (again, restored) = restore_times(dir.path(), &files, None);

        assert_eq!((restored, again.len()), (0, 3));

- **Unchanged sources keep their times** (oxygenflow#52). Beside each kept

  disk the pool keeps its sources: each workspace file's content hash and

  the mtime it had, in whole seconds (what the workspace image keeps). A

  job that takes the disk gives each file whose content is unchanged that

  mtime back before the workspace is imaged; a changed or new file keeps

  the time of its checkout, newer than anything on the disk. A build tool

  that judges freshness by time, such as cargo, then rebuilds exactly the

  crates whose sources changed, where a fresh checkout would make every

  path crate look changed. The log says how many files kept their times.

- **Jobs go where their cache is** (oxygenflow#51). Each claim lists the

  caches the runner holds a disk of (`caches`, each

  `flow_core::api::cache_slot(repo, job)`), and the hub ranks a runner

  holding a job's cache above one holding its image (§11.4): a job another

  free runner holds the cache for is held for it for up to 15 seconds, and

  `placed` says "holds its build cache".