OXYGENHUB
# ed4aef8b70ca oxygenflow / crates / flow / src / cache.rs
14.9 KiB · Rust · last changed feat(cache): unchanged sources keep their times; claims list the caches held (closes #52, refs #51) 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
//! Beside each disk, `<key>.sources` lists the sources it was built from
22
//! (oxygenflow#52): each file's content hash and the mtime it had. A job
23
//! that takes the disk gives every file whose content is unchanged that
24
//! mtime back, so a build tool that judges freshness by time (cargo)
25
//! rebuilds only what changed; a changed file keeps the time of its
26
//! checkout, newer than anything on the disk, so it is always rebuilt.
27
28
use anyhow::{Context, Result};
29
use flow_core::pipeline::Mount;
30
use std::os::unix::fs::MetadataExt;
31
use std::path::{Path, PathBuf};
32
use std::time::SystemTime;
33
34
/// The pool's default size limit.
35
pub const DEFAULT_GB: u64 = 64;
36
37
/// The sources a cached disk was built from: path to content hash and
38
/// mtime, in whole seconds (what the workspace image keeps).
39
pub type Sources = std::collections::BTreeMap<String, (String, i64)>;
40
41
/// What a job started a cached disk from.
42
pub struct Taken {
43
    /// For the log and the receipt.
44
    pub how: String,
45
    /// The sources it was built from, when they were kept with it.
46
    pub sources: Option<Sources>,
47
}
48
49
fn sources_of(disk: &Path) -> PathBuf {
50
    disk.with_extension("sources")
51
}
52
53
fn read_sources(disk: &Path) -> Option<Sources> {
54
    serde_json::from_slice(&std::fs::read(sources_of(disk)).ok()?).ok()
55
}
56
57
/// Where a job's cached disks come from and go back to.
58
pub struct Pool {
59
    root: PathBuf,
60
    limit: u64,
61
}
62
63
/// Which job: the pool's coordinates other than the key.
64
pub struct Ctx<'a> {
65
    pub repo: &'a str,
66
    pub bookmark: &'a str,
67
    pub job: &'a str,
68
    pub image: &'a str,
69
}
70
71
use flow_core::api::segment;
72
73
impl Pool {
74
    pub fn new(home_root: &Path, gb: Option<u64>) -> Pool {
75
        Pool { root: home_root.join("cache"), limit: gb.unwrap_or(DEFAULT_GB).saturating_mul(1 << 30) }
76
    }
77
78
    fn cache_dir(&self, ctx: &Ctx, name: &str) -> PathBuf {
79
        self.root.join(segment(ctx.repo)).join(segment(ctx.job)).join(segment(name))
80
    }
81
82
    /// A mount's key: the image, job, cache name and its key files'
83
    /// content hashes (`hash_of` gives a file's, `None` when absent).
84
    pub fn key(ctx: &Ctx, m: &Mount, hash_of: impl Fn(&str) -> Option<String>) -> String {
85
        let mut h = blake3::Hasher::new();
86
        for part in [ctx.image, ctx.job, m.cache.as_deref().unwrap_or("")] {
87
            h.update(part.as_bytes());
88
            h.update(b"\n");
89
        }
90
        for k in &m.key {
91
            h.update(k.as_bytes());
92
            h.update(b"=");
93
            h.update(hash_of(k).unwrap_or_else(|| "absent".into()).as_bytes());
94
            h.update(b"\n");
95
        }
96
        h.finalize().to_hex()[..32].to_string()
97
    }
98
99
    /// Start `dest` from the pool for mount `m`, a disk of `m.size_mib`.
100
    /// What it started from, or `None` (the caller makes an empty disk).
101
    pub fn take(&self, ctx: &Ctx, m: &Mount, key: &str, dest: &Path) -> Option<Taken> {
102
        let name = m.cache.as_deref()?;
103
        let dir = self.cache_dir(ctx, name);
104
        let size = m.size_mib as u64 * (1 << 20);
105
        let fits = |p: &Path| std::fs::metadata(p).is_ok_and(|md| md.len() == size);
106
        let mine = dir.join(segment(ctx.bookmark));
107
        let exact = mine.join(format!("{key}.ext4"));
108
        // A moved disk takes its sources with it.
109
        if fits(&exact) {
110
            let sources = read_sources(&exact);
111
            if std::fs::rename(&exact, dest).is_ok() {
112
                let _ = std::fs::remove_file(sources_of(&exact));
113
                return Some(Taken { how: format!("cache {name}: this bookmark's, same key"), sources });
114
            }
115
        }
116
        if let Some(p) = newest(&mine, |p| fits(p)) {
117
            let sources = read_sources(&p);
118
            if std::fs::rename(&p, dest).is_ok() {
119
                let _ = std::fs::remove_file(sources_of(&p));
120
                return Some(Taken { how: format!("cache {name}: this bookmark's, an older key"), sources });
121
            }
122
        }
123
        let others: Vec<PathBuf> =
124
            std::fs::read_dir(&dir).into_iter().flatten().flatten().map(|e| e.path()).filter(|p| p.is_dir() && *p != mine).collect();
125
        let pick = others
126
            .iter()
127
            .map(|d| d.join(format!("{key}.ext4")))
128
            .filter(|p| fits(p))
129
            .max_by_key(|p| modified(p))
130
            .map(|p| (p, "same key"))
131
            .or_else(|| others.iter().filter_map(|d| newest(d, |p| fits(p))).max_by_key(|p| modified(p)).map(|p| (p, "an older key")));
132
        if let Some((p, which)) = pick {
133
            let from = p.parent().and_then(|d| d.file_name()).map(|n| n.to_string_lossy().into_owned()).unwrap_or_default();
134
            if sparse_copy(&p, dest).is_ok() {
135
                return Some(Taken {
136
                    how: format!("cache {name}: copied from another bookmark's ({from}), {which}"),
137
                    sources: read_sources(&p),
138
                });
139
            }
140
        }
141
        None
142
    }
143
144
    /// Keep a passing job's disk for mount `m` under its key, then keep
145
    /// the pool under its limit.
146
    pub fn keep(&self, ctx: &Ctx, m: &Mount, key: &str, disk: &Path, sources: &Sources) -> Result<()> {
147
        let Some(name) = m.cache.as_deref() else { return Ok(()) };
148
        let mine = self.cache_dir(ctx, name).join(segment(ctx.bookmark));
149
        std::fs::create_dir_all(&mine)?;
150
        let dest = mine.join(format!("{key}.ext4"));
151
        if std::fs::rename(disk, &dest).is_err() {
152
            sparse_copy(disk, &dest).with_context(|| format!("keep {}", disk.display()))?;
153
        }
154
        // Moved disks keep their old time: mark it used now.
155
        std::fs::File::options().write(true).open(&dest)?.set_modified(SystemTime::now())?;
156
        let tmp = sources_of(&dest).with_extension("sources-part");
157
        std::fs::write(&tmp, serde_json::to_vec(sources)?)?;
158
        std::fs::rename(&tmp, sources_of(&dest))?;
159
        // One disk per key is enough; older keys of this bookmark go once
160
        // the pool is over its limit.
161
        self.evict();
162
        Ok(())
163
    }
164
165
    /// The repository and job slots holding at least one disk, as
166
    /// [`flow_core::api::cache_slot`] names them, for the claim
167
    /// (oxygenflow#51).
168
    pub fn slots(&self) -> Vec<String> {
169
        let mut out = Vec::new();
170
        for repo in std::fs::read_dir(&self.root).into_iter().flatten().flatten() {
171
            for job in std::fs::read_dir(repo.path()).into_iter().flatten().flatten() {
172
                let mut any = false;
173
                walk(&job.path(), &mut |p| any |= p.extension().is_some_and(|e| e == "ext4"));
174
                if any {
175
                    out.push(format!("{}/{}", repo.file_name().to_string_lossy(), job.file_name().to_string_lossy()));
176
                }
177
            }
178
        }
179
        out.sort();
180
        out
181
    }
182
183
    /// Drop the disks used longest ago until the pool's blocks fit.
184
    pub fn evict(&self) -> u64 {
185
        let mut disks: Vec<(SystemTime, u64, PathBuf)> = Vec::new();
186
        walk(&self.root, &mut |p| {
187
            if p.extension().is_some_and(|e| e == "ext4") {
188
                if let Ok(md) = std::fs::metadata(p) {
189
                    disks.push((md.modified().unwrap_or(SystemTime::UNIX_EPOCH), md.blocks() * 512, p.to_path_buf()));
190
                }
191
            }
192
        });
193
        let mut total: u64 = disks.iter().map(|d| d.1).sum();
194
        disks.sort();
195
        let mut freed = 0;
196
        for (_, bytes, p) in disks {
197
            if total <= self.limit {
198
                break;
199
            }
200
            if std::fs::remove_file(&p).is_ok() {
201
                let _ = std::fs::remove_file(sources_of(&p));
202
                total -= bytes;
203
                freed += bytes;
204
            }
205
        }
206
        freed
207
    }
208
}
209
210
fn modified(p: &Path) -> SystemTime {
211
    std::fs::metadata(p).and_then(|m| m.modified()).unwrap_or(SystemTime::UNIX_EPOCH)
212
}
213
214
fn newest(dir: &Path, ok: impl Fn(&Path) -> bool) -> Option<PathBuf> {
215
    std::fs::read_dir(dir)
216
        .ok()?
217
        .flatten()
218
        .map(|e| e.path())
219
        .filter(|p| p.extension().is_some_and(|e| e == "ext4") && ok(p))
220
        .max_by_key(|p| modified(p))
221
}
222
223
fn walk(dir: &Path, f: &mut dyn FnMut(&Path)) {
224
    for e in std::fs::read_dir(dir).into_iter().flatten().flatten() {
225
        let p = e.path();
226
        if p.is_dir() {
227
            walk(&p, f);
228
        } else {
229
            f(&p);
230
        }
231
    }
232
}
233
234
/// Copy a sparse disk without writing its holes.
235
fn sparse_copy(from: &Path, to: &Path) -> Result<()> {
236
    let tmp = to.with_extension("part");
237
    let st = std::process::Command::new("cp").arg("--sparse=always").arg(from).arg(&tmp).status().context("run cp")?;
238
    if !st.success() {
239
        let _ = std::fs::remove_file(&tmp);
240
        anyhow::bail!("cp {} failed", from.display());
241
    }
242
    std::fs::rename(&tmp, to)?;
243
    Ok(())
244
}
245
246
#[cfg(test)]
247
mod tests {
248
    use super::*;
249
    use flow_core::pipeline::MountSource;
250
251
    fn mount(key: &[&str]) -> Mount {
252
        Mount {
253
            source: MountSource::Disk,
254
            target: "/src/target".into(),
255
            size_mib: 1,
256
            cache: Some("target".into()),
257
            key: key.iter().map(|s| s.to_string()).collect(),
258
        }
259
    }
260
261
    fn disk(path: &Path, fill: &[u8]) {
262
        let f = std::fs::File::create(path).unwrap();
263
        f.set_len(1 << 20).unwrap();
264
        std::os::unix::fs::FileExt::write_at(&f, fill, 0).unwrap();
265
    }
266
267
    fn first(path: &Path) -> Vec<u8> {
268
        std::fs::read(path).unwrap()[..4].to_vec()
269
    }
270
271
    #[test]
272
    fn a_kept_disk_starts_the_next_job_of_its_bookmark_then_others_copy_it() {
273
        let home = tempfile::tempdir().unwrap();
274
        let work = tempfile::tempdir().unwrap();
275
        let pool = Pool::new(home.path(), None);
276
        let m = mount(&["Cargo.lock"]);
277
        let main = Ctx { repo: "andy/hub", bookmark: "main", job: "test", image: "img" };
278
        let key = Pool::key(&main, &m, |_| Some("lock-1".into()));
279
        // Nothing yet: the caller makes an empty disk.
280
        assert!(pool.take(&main, &m, &key, &work.path().join("a.ext4")).is_none());
281
        // A passing job's disk is kept and moved into the next job's place.
282
        disk(&work.path().join("a.ext4"), b"MAIN");
283
        let built: Sources = [("src/lib.rs".to_string(), ("h1".to_string(), 1_000))].into();
284
        pool.keep(&main, &m, &key, &work.path().join("a.ext4"), &built).unwrap();
285
        assert!(!work.path().join("a.ext4").exists(), "moved, not copied");
286
        let taken = pool.take(&main, &m, &key, &work.path().join("b.ext4")).unwrap();
287
        assert!(taken.how.contains("same key"), "{}", taken.how);
288
        assert_eq!(taken.sources.as_ref(), Some(&built), "its sources come with it (#52)");
289
        assert_eq!(first(&work.path().join("b.ext4")), b"MAIN");
290
        pool.keep(&main, &m, &key, &work.path().join("b.ext4"), &built).unwrap();
291
        // A changed key takes the bookmark's newest disk all the same.
292
        let key2 = Pool::key(&main, &m, |_| Some("lock-2".into()));
293
        assert_ne!(key, key2);
294
        let taken = pool.take(&main, &m, &key2, &work.path().join("c.ext4")).unwrap();
295
        assert!(taken.how.contains("older key"), "{}", taken.how);
296
        assert_eq!(taken.sources.as_ref(), Some(&built));
297
        pool.keep(&main, &m, &key2, &work.path().join("c.ext4"), &built).unwrap();
298
        assert_eq!(pool.slots(), vec![flow_core::api::cache_slot("andy/hub", "test")], "what the claim reports (#51)");
299
        // Another bookmark copies it, leaving main's in place.
300
        let feature = Ctx { bookmark: "feature", ..main };
301
        let taken = pool.take(&feature, &m, &key2, &work.path().join("d.ext4")).unwrap();
302
        assert!(taken.how.contains("copied from another bookmark"), "{}", taken.how);
303
        assert_eq!(taken.sources.as_ref(), Some(&built), "a copy reads them and leaves them");
304
        assert_eq!(first(&work.path().join("d.ext4")), b"MAIN");
305
        let taken = pool.take(&main, &m, &key2, &work.path().join("e.ext4")).unwrap();
306
        assert!(taken.how.contains("this bookmark's") && taken.sources.is_some(), "main's is still there, with its sources");
307
        // Another job or a different size never shares it.
308
        let other_job = Ctx { job: "build", ..main };
309
        assert!(pool.take(&other_job, &m, &key2, &work.path().join("f.ext4")).is_none());
310
        let mut bigger = mount(&["Cargo.lock"]);
311
        bigger.size_mib = 2;
312
        assert!(pool.take(&feature, &bigger, &key2, &work.path().join("g.ext4")).is_none());
313
    }
314
315
    #[test]
316
    fn the_key_follows_the_image_the_job_and_the_key_files() {
317
        let ctx = Ctx { repo: "r", bookmark: "main", job: "test", image: "img-1" };
318
        let m = mount(&["Cargo.lock", "rust-toolchain.toml"]);
319
        let k = Pool::key(&ctx, &m, |p| Some(format!("{p}-v1")));
320
        assert_eq!(k, Pool::key(&ctx, &m, |p| Some(format!("{p}-v1"))));
321
        assert_ne!(k, Pool::key(&ctx, &m, |p| Some(format!("{p}-v2"))), "a key file changed");
322
        assert_ne!(k, Pool::key(&Ctx { image: "img-2", ..ctx }, &m, |p| Some(format!("{p}-v1"))), "another image");
323
        assert_ne!(k, Pool::key(&ctx, &m, |_| None), "a key file missing");
324
    }
325
326
    #[test]
327
    fn the_pool_drops_the_disks_used_longest_ago() {
328
        let home = tempfile::tempdir().unwrap();
329
        let work = tempfile::tempdir().unwrap();
330
        let pool = Pool { root: home.path().join("cache"), limit: 0 };
331
        let m = mount(&[]);
332
        let ctx = Ctx { repo: "r", bookmark: "main", job: "j", image: "i" };
333
        let full = |p: &Path| {
334
            let f = std::fs::File::create(p).unwrap();
335
            std::os::unix::fs::FileExt::write_at(&f, &vec![7u8; 1 << 20], 0).unwrap();
336
        };
337
        full(&work.path().join("a.ext4"));
338
        // Over a limit of nothing: kept, then dropped at once.
339
        pool.keep(&ctx, &m, "k1", &work.path().join("a.ext4"), &Sources::new()).unwrap();
340
        assert!(pool.take(&ctx, &m, "k1", &work.path().join("b.ext4")).is_none());
341
        let roomy = Pool { root: home.path().join("cache"), limit: 1 << 30 };
342
        full(&work.path().join("c.ext4"));
343
        roomy.keep(&ctx, &m, "k2", &work.path().join("c.ext4"), &Sources::new()).unwrap();
344
        assert!(roomy.take(&ctx, &m, "k2", &work.path().join("d.ext4")).is_some(), "under the limit it stays");
345
    }
346
}