OXYGENHUB
# 92587de58cef oxygenflow / crates / flow / src / cache.rs
12.1 KiB · Rust · last changed A build cache: disk mounts that carry a build forward 22 days ago source blame history raw
1
//! The build cache (oxygenflow#28, spec §5.4): a disk mount that names a
2
//! `cache` is started from a disk kept by an earlier passing job, rather
3
//! than empty, so a build carries its compiled dependencies forward.
4
//!
5
//! Each runner keeps a pool under `<home>/cache`, by repository, job,
6
//! cache name and bookmark, each disk named by its key: the BLAKE3 of the
7
//! image, the job, the cache name and the content of its key files. A job
8
//! takes, in order:
9
//!
10
//! 1. its bookmark's disk with the same key (moved: no copying);
11
//! 2. its bookmark's newest disk (moved; cargo rebuilds what changed);
12
//! 3. another bookmark's disk with the same key, then its newest (copied,
13
//!    so a proposal never takes main's cache away);
14
//! 4. else an empty disk, as without a cache.
15
//!
16
//! A passing job's disk goes back into its bookmark's place; a failing
17
//! job's is thrown away, so nothing it wrote reaches a later job. The
18
//! receipt names what the job started from. The pool is kept under its
19
//! size limit by dropping the disks used longest ago.
20
21
use anyhow::{Context, Result};
22
use flow_core::pipeline::Mount;
23
use std::os::unix::fs::MetadataExt;
24
use std::path::{Path, PathBuf};
25
use std::time::SystemTime;
26
27
/// The pool's default size limit.
28
pub const DEFAULT_GB: u64 = 64;
29
30
/// Where a job's cached disks come from and go back to.
31
pub struct Pool {
32
    root: PathBuf,
33
    limit: u64,
34
}
35
36
/// Which job: the pool's coordinates other than the key.
37
pub struct Ctx<'a> {
38
    pub repo: &'a str,
39
    pub bookmark: &'a str,
40
    pub job: &'a str,
41
    pub image: &'a str,
42
}
43
44
/// A name safe as one path segment: kept readable, with a short hash so
45
/// two names that clean up alike stay apart.
46
fn segment(s: &str) -> String {
47
    let clean: String = s.chars().map(|c| if c.is_ascii_alphanumeric() || "._-".contains(c) { c } else { '_' }).take(48).collect();
48
    format!("{clean}-{}", &blake3::hash(s.as_bytes()).to_hex()[..8])
49
}
50
51
impl Pool {
52
    pub fn new(home_root: &Path, gb: Option<u64>) -> Pool {
53
        Pool { root: home_root.join("cache"), limit: gb.unwrap_or(DEFAULT_GB).saturating_mul(1 << 30) }
54
    }
55
56
    fn cache_dir(&self, ctx: &Ctx, name: &str) -> PathBuf {
57
        self.root.join(segment(ctx.repo)).join(segment(ctx.job)).join(segment(name))
58
    }
59
60
    /// A mount's key: the image, job, cache name and its key files'
61
    /// content hashes (`hash_of` gives a file's, `None` when absent).
62
    pub fn key(ctx: &Ctx, m: &Mount, hash_of: impl Fn(&str) -> Option<String>) -> String {
63
        let mut h = blake3::Hasher::new();
64
        for part in [ctx.image, ctx.job, m.cache.as_deref().unwrap_or("")] {
65
            h.update(part.as_bytes());
66
            h.update(b"\n");
67
        }
68
        for k in &m.key {
69
            h.update(k.as_bytes());
70
            h.update(b"=");
71
            h.update(hash_of(k).unwrap_or_else(|| "absent".into()).as_bytes());
72
            h.update(b"\n");
73
        }
74
        h.finalize().to_hex()[..32].to_string()
75
    }
76
77
    /// Start `dest` from the pool for mount `m`, a disk of `m.size_mib`.
78
    /// What it started from, or `None` (the caller makes an empty disk).
79
    pub fn take(&self, ctx: &Ctx, m: &Mount, key: &str, dest: &Path) -> Option<String> {
80
        let name = m.cache.as_deref()?;
81
        let dir = self.cache_dir(ctx, name);
82
        let size = m.size_mib as u64 * (1 << 20);
83
        let fits = |p: &Path| std::fs::metadata(p).is_ok_and(|md| md.len() == size);
84
        let mine = dir.join(segment(ctx.bookmark));
85
        let exact = mine.join(format!("{key}.ext4"));
86
        if fits(&exact) && std::fs::rename(&exact, dest).is_ok() {
87
            return Some(format!("cache {name}: this bookmark's, same key"));
88
        }
89
        if let Some(p) = newest(&mine, |p| fits(p)) {
90
            if std::fs::rename(&p, dest).is_ok() {
91
                return Some(format!("cache {name}: this bookmark's, an older key"));
92
            }
93
        }
94
        let others: Vec<PathBuf> =
95
            std::fs::read_dir(&dir).into_iter().flatten().flatten().map(|e| e.path()).filter(|p| p.is_dir() && *p != mine).collect();
96
        let pick = others
97
            .iter()
98
            .map(|d| d.join(format!("{key}.ext4")))
99
            .filter(|p| fits(p))
100
            .max_by_key(|p| modified(p))
101
            .map(|p| (p, "same key"))
102
            .or_else(|| others.iter().filter_map(|d| newest(d, |p| fits(p))).max_by_key(|p| modified(p)).map(|p| (p, "an older key")));
103
        if let Some((p, which)) = pick {
104
            let from = p.parent().and_then(|d| d.file_name()).map(|n| n.to_string_lossy().into_owned()).unwrap_or_default();
105
            if sparse_copy(&p, dest).is_ok() {
106
                return Some(format!("cache {name}: copied from another bookmark's ({from}), {which}"));
107
            }
108
        }
109
        None
110
    }
111
112
    /// Keep a passing job's disk for mount `m` under its key, then keep
113
    /// the pool under its limit.
114
    pub fn keep(&self, ctx: &Ctx, m: &Mount, key: &str, disk: &Path) -> Result<()> {
115
        let Some(name) = m.cache.as_deref() else { return Ok(()) };
116
        let mine = self.cache_dir(ctx, name).join(segment(ctx.bookmark));
117
        std::fs::create_dir_all(&mine)?;
118
        let dest = mine.join(format!("{key}.ext4"));
119
        if std::fs::rename(disk, &dest).is_err() {
120
            sparse_copy(disk, &dest).with_context(|| format!("keep {}", disk.display()))?;
121
        }
122
        // Moved disks keep their old time: mark it used now.
123
        std::fs::File::options().write(true).open(&dest)?.set_modified(SystemTime::now())?;
124
        // One disk per key is enough; older keys of this bookmark go once
125
        // the pool is over its limit.
126
        self.evict();
127
        Ok(())
128
    }
129
130
    /// Drop the disks used longest ago until the pool's blocks fit.
131
    pub fn evict(&self) -> u64 {
132
        let mut disks: Vec<(SystemTime, u64, PathBuf)> = Vec::new();
133
        walk(&self.root, &mut |p| {
134
            if p.extension().is_some_and(|e| e == "ext4") {
135
                if let Ok(md) = std::fs::metadata(p) {
136
                    disks.push((md.modified().unwrap_or(SystemTime::UNIX_EPOCH), md.blocks() * 512, p.to_path_buf()));
137
                }
138
            }
139
        });
140
        let mut total: u64 = disks.iter().map(|d| d.1).sum();
141
        disks.sort();
142
        let mut freed = 0;
143
        for (_, bytes, p) in disks {
144
            if total <= self.limit {
145
                break;
146
            }
147
            if std::fs::remove_file(&p).is_ok() {
148
                total -= bytes;
149
                freed += bytes;
150
            }
151
        }
152
        freed
153
    }
154
}
155
156
fn modified(p: &Path) -> SystemTime {
157
    std::fs::metadata(p).and_then(|m| m.modified()).unwrap_or(SystemTime::UNIX_EPOCH)
158
}
159
160
fn newest(dir: &Path, ok: impl Fn(&Path) -> bool) -> Option<PathBuf> {
161
    std::fs::read_dir(dir)
162
        .ok()?
163
        .flatten()
164
        .map(|e| e.path())
165
        .filter(|p| p.extension().is_some_and(|e| e == "ext4") && ok(p))
166
        .max_by_key(|p| modified(p))
167
}
168
169
fn walk(dir: &Path, f: &mut dyn FnMut(&Path)) {
170
    for e in std::fs::read_dir(dir).into_iter().flatten().flatten() {
171
        let p = e.path();
172
        if p.is_dir() {
173
            walk(&p, f);
174
        } else {
175
            f(&p);
176
        }
177
    }
178
}
179
180
/// Copy a sparse disk without writing its holes.
181
fn sparse_copy(from: &Path, to: &Path) -> Result<()> {
182
    let tmp = to.with_extension("part");
183
    let st = std::process::Command::new("cp").arg("--sparse=always").arg(from).arg(&tmp).status().context("run cp")?;
184
    if !st.success() {
185
        let _ = std::fs::remove_file(&tmp);
186
        anyhow::bail!("cp {} failed", from.display());
187
    }
188
    std::fs::rename(&tmp, to)?;
189
    Ok(())
190
}
191
192
#[cfg(test)]
193
mod tests {
194
    use super::*;
195
    use flow_core::pipeline::MountSource;
196
197
    fn mount(key: &[&str]) -> Mount {
198
        Mount {
199
            source: MountSource::Disk,
200
            target: "/src/target".into(),
201
            size_mib: 1,
202
            cache: Some("target".into()),
203
            key: key.iter().map(|s| s.to_string()).collect(),
204
        }
205
    }
206
207
    fn disk(path: &Path, fill: &[u8]) {
208
        let f = std::fs::File::create(path).unwrap();
209
        f.set_len(1 << 20).unwrap();
210
        std::os::unix::fs::FileExt::write_at(&f, fill, 0).unwrap();
211
    }
212
213
    fn first(path: &Path) -> Vec<u8> {
214
        std::fs::read(path).unwrap()[..4].to_vec()
215
    }
216
217
    #[test]
218
    fn a_kept_disk_starts_the_next_job_of_its_bookmark_then_others_copy_it() {
219
        let home = tempfile::tempdir().unwrap();
220
        let work = tempfile::tempdir().unwrap();
221
        let pool = Pool::new(home.path(), None);
222
        let m = mount(&["Cargo.lock"]);
223
        let main = Ctx { repo: "andy/hub", bookmark: "main", job: "test", image: "img" };
224
        let key = Pool::key(&main, &m, |_| Some("lock-1".into()));
225
        // Nothing yet: the caller makes an empty disk.
226
        assert_eq!(pool.take(&main, &m, &key, &work.path().join("a.ext4")), None);
227
        // A passing job's disk is kept and moved into the next job's place.
228
        disk(&work.path().join("a.ext4"), b"MAIN");
229
        pool.keep(&main, &m, &key, &work.path().join("a.ext4")).unwrap();
230
        assert!(!work.path().join("a.ext4").exists(), "moved, not copied");
231
        let how = pool.take(&main, &m, &key, &work.path().join("b.ext4")).unwrap();
232
        assert!(how.contains("same key"), "{how}");
233
        assert_eq!(first(&work.path().join("b.ext4")), b"MAIN");
234
        pool.keep(&main, &m, &key, &work.path().join("b.ext4")).unwrap();
235
        // A changed key takes the bookmark's newest disk all the same.
236
        let key2 = Pool::key(&main, &m, |_| Some("lock-2".into()));
237
        assert_ne!(key, key2);
238
        let how = pool.take(&main, &m, &key2, &work.path().join("c.ext4")).unwrap();
239
        assert!(how.contains("older key"), "{how}");
240
        pool.keep(&main, &m, &key2, &work.path().join("c.ext4")).unwrap();
241
        // Another bookmark copies it, leaving main's in place.
242
        let feature = Ctx { bookmark: "feature", ..main };
243
        let how = pool.take(&feature, &m, &key2, &work.path().join("d.ext4")).unwrap();
244
        assert!(how.contains("copied from another bookmark"), "{how}");
245
        assert_eq!(first(&work.path().join("d.ext4")), b"MAIN");
246
        assert!(pool.take(&main, &m, &key2, &work.path().join("e.ext4")).unwrap().contains("this bookmark's"), "main's is still there");
247
        // Another job or a different size never shares it.
248
        let other_job = Ctx { job: "build", ..main };
249
        assert_eq!(pool.take(&other_job, &m, &key2, &work.path().join("f.ext4")), None);
250
        let mut bigger = mount(&["Cargo.lock"]);
251
        bigger.size_mib = 2;
252
        assert_eq!(pool.take(&feature, &bigger, &key2, &work.path().join("g.ext4")), None);
253
    }
254
255
    #[test]
256
    fn the_key_follows_the_image_the_job_and_the_key_files() {
257
        let ctx = Ctx { repo: "r", bookmark: "main", job: "test", image: "img-1" };
258
        let m = mount(&["Cargo.lock", "rust-toolchain.toml"]);
259
        let k = Pool::key(&ctx, &m, |p| Some(format!("{p}-v1")));
260
        assert_eq!(k, Pool::key(&ctx, &m, |p| Some(format!("{p}-v1"))));
261
        assert_ne!(k, Pool::key(&ctx, &m, |p| Some(format!("{p}-v2"))), "a key file changed");
262
        assert_ne!(k, Pool::key(&Ctx { image: "img-2", ..ctx }, &m, |p| Some(format!("{p}-v1"))), "another image");
263
        assert_ne!(k, Pool::key(&ctx, &m, |_| None), "a key file missing");
264
    }
265
266
    #[test]
267
    fn the_pool_drops_the_disks_used_longest_ago() {
268
        let home = tempfile::tempdir().unwrap();
269
        let work = tempfile::tempdir().unwrap();
270
        let pool = Pool { root: home.path().join("cache"), limit: 0 };
271
        let m = mount(&[]);
272
        let ctx = Ctx { repo: "r", bookmark: "main", job: "j", image: "i" };
273
        let full = |p: &Path| {
274
            let f = std::fs::File::create(p).unwrap();
275
            std::os::unix::fs::FileExt::write_at(&f, &vec![7u8; 1 << 20], 0).unwrap();
276
        };
277
        full(&work.path().join("a.ext4"));
278
        // Over a limit of nothing: kept, then dropped at once.
279
        pool.keep(&ctx, &m, "k1", &work.path().join("a.ext4")).unwrap();
280
        assert_eq!(pool.take(&ctx, &m, "k1", &work.path().join("b.ext4")), None);
281
        let roomy = Pool { root: home.path().join("cache"), limit: 1 << 30 };
282
        full(&work.path().join("c.ext4"));
283
        roomy.keep(&ctx, &m, "k2", &work.path().join("c.ext4")).unwrap();
284
        assert!(roomy.take(&ctx, &m, "k2", &work.path().join("d.ext4")).is_some(), "under the limit it stays");
285
    }
286
}