OXYGENHUB
# 33d8ada65f6a oxygenflow / crates / flow / src / cache.rs
33.0 KiB · Rust · last changed Build caches for the syscall tier 16 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
//! A disk is in the [`Form`] its tier mounts (oxygenflow#70): a microVM's
22
//! is an ext4 image, `<key>.ext4`; the syscall tier's is the host
23
//! directory it binds at the mount's target, `<key>.tree`, taken and kept
24
//! the same way. A directory goes into the pool with nothing in it
25
//! writable by others or set-id, and the pool itself is the runner's
26
//! alone.
27
//!
28
//! Beside each disk, `<key>.sources` (`<key>.tree-sources` for a
29
//! directory) lists the sources it was built from (oxygenflow#52): each
30
//! file's content hash and the mtime it had. A job that takes the disk
31
//! gives every file whose content is unchanged that mtime back, so a build
32
//! tool that judges freshness by time (cargo) rebuilds only what changed;
33
//! a changed file keeps the time of its checkout, newer than anything on
34
//! the disk, so it is always rebuilt.
35
36
use anyhow::{Context, Result};
37
use flow_core::pipeline::Mount;
38
use std::os::unix::fs::{MetadataExt, PermissionsExt};
39
use std::path::{Path, PathBuf};
40
use std::time::SystemTime;
41
42
/// The pool's default size limit.
43
pub const DEFAULT_GB: u64 = 64;
44
45
/// The sources a cached disk was built from: path to content hash and
46
/// mtime, in whole seconds (what the workspace image keeps).
47
pub type Sources = std::collections::BTreeMap<String, (String, i64)>;
48
49
/// What a job started a cached disk from.
50
pub struct Taken {
51
    /// For the log and the receipt.
52
    pub how: String,
53
    /// The sources it was built from, when they were kept with it.
54
    pub sources: Option<Sources>,
55
}
56
57
/// How a tier holds a disk mount's files (oxygenflow#70).
58
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
59
pub enum Form {
60
    /// An ext4 image of the mount's size, attached to a microVM.
61
    Image,
62
    /// A host directory, bound into the syscall sandbox.
63
    Dir,
64
}
65
66
impl Form {
67
    fn ext(self) -> &'static str {
68
        match self {
69
            Form::Image => "ext4",
70
            Form::Dir => "tree",
71
        }
72
    }
73
}
74
75
/// Which form a kept entry of the pool is in, by its name.
76
fn form_of(p: &Path) -> Option<Form> {
77
    match p.extension()?.to_str()? {
78
        "ext4" => Some(Form::Image),
79
        "tree" => Some(Form::Dir),
80
        _ => None,
81
    }
82
}
83
84
fn sources_of(disk: &Path) -> PathBuf {
85
    match form_of(disk) {
86
        Some(Form::Dir) => disk.with_extension("tree-sources"),
87
        _ => disk.with_extension("sources"),
88
    }
89
}
90
91
fn read_sources(disk: &Path) -> Option<Sources> {
92
    serde_json::from_slice(&std::fs::read(sources_of(disk)).ok()?).ok()
93
}
94
95
/// Where a job's cached disks come from and go back to.
96
pub struct Pool {
97
    root: PathBuf,
98
    limit: u64,
99
}
100
101
/// Which job: the pool's coordinates other than the key, and the form its
102
/// tier keeps a disk in.
103
pub struct Ctx<'a> {
104
    pub repo: &'a str,
105
    pub bookmark: &'a str,
106
    pub job: &'a str,
107
    pub image: &'a str,
108
    pub form: Form,
109
}
110
111
use flow_core::api::segment;
112
113
/// Held while the pool changes: jobs run side by side, and a directory,
114
/// unlike an image, can be half moved or half removed while another job
115
/// copies it.
116
static CHANGING: std::sync::Mutex<()> = std::sync::Mutex::new(());
117
118
fn changing() -> std::sync::MutexGuard<'static, ()> {
119
    CHANGING.lock().unwrap_or_else(|p| p.into_inner())
120
}
121
122
impl Pool {
123
    pub fn new(home_root: &Path, gb: Option<u64>) -> Pool {
124
        Pool { root: home_root.join("cache"), limit: gb.unwrap_or(DEFAULT_GB).saturating_mul(1 << 30) }
125
    }
126
127
    fn cache_dir(&self, ctx: &Ctx, name: &str) -> PathBuf {
128
        self.root.join(segment(ctx.repo)).join(segment(ctx.job)).join(segment(name))
129
    }
130
131
    /// A mount's key: the image, job, cache name and its key files'
132
    /// content hashes (`hash_of` gives a file's, `None` when absent).
133
    pub fn key(ctx: &Ctx, m: &Mount, hash_of: impl Fn(&str) -> Option<String>) -> String {
134
        let mut h = blake3::Hasher::new();
135
        for part in [ctx.image, ctx.job, m.cache.as_deref().unwrap_or("")] {
136
            h.update(part.as_bytes());
137
            h.update(b"\n");
138
        }
139
        for k in &m.key {
140
            h.update(k.as_bytes());
141
            h.update(b"=");
142
            h.update(hash_of(k).unwrap_or_else(|| "absent".into()).as_bytes());
143
            h.update(b"\n");
144
        }
145
        h.finalize().to_hex()[..32].to_string()
146
    }
147
148
    /// Start `dest` from the pool for mount `m`: an image of `m.size_mib`,
149
    /// or a directory, as `ctx.form` says; `dest` must not exist yet. What
150
    /// it started from, or `None` (the caller makes an empty disk).
151
    ///
152
    /// A disk this job's bookmark kept is moved out of the pool, so a job
153
    /// of the same slot running beside this one can never take it too.
154
    pub fn take(&self, ctx: &Ctx, m: &Mount, key: &str, dest: &Path) -> Option<Taken> {
155
        let name = m.cache.as_deref()?;
156
        let _changing = changing();
157
        let dir = self.cache_dir(ctx, name);
158
        let size = m.size_mib as u64 * (1 << 20);
159
        let form = ctx.form;
160
        let fits = |p: &Path| match form {
161
            Form::Image => std::fs::metadata(p).is_ok_and(|md| md.len() == size),
162
            Form::Dir => std::fs::symlink_metadata(p).is_ok_and(|md| md.is_dir()),
163
        };
164
        let ext = form.ext();
165
        let mine = dir.join(segment(ctx.bookmark));
166
        let exact = mine.join(format!("{key}.{ext}"));
167
        // A moved disk takes its sources with it.
168
        if fits(&exact) {
169
            let sources = read_sources(&exact);
170
            if std::fs::rename(&exact, dest).is_ok() {
171
                let _ = std::fs::remove_file(sources_of(&exact));
172
                return Some(Taken { how: format!("cache {name}: this bookmark's, same key"), sources });
173
            }
174
        }
175
        if let Some(p) = newest(&mine, form, |p| fits(p)) {
176
            let sources = read_sources(&p);
177
            if std::fs::rename(&p, dest).is_ok() {
178
                let _ = std::fs::remove_file(sources_of(&p));
179
                return Some(Taken { how: format!("cache {name}: this bookmark's, an older key"), sources });
180
            }
181
        }
182
        let others: Vec<PathBuf> =
183
            std::fs::read_dir(&dir).into_iter().flatten().flatten().map(|e| e.path()).filter(|p| p.is_dir() && *p != mine).collect();
184
        let pick = others
185
            .iter()
186
            .map(|d| d.join(format!("{key}.{ext}")))
187
            .filter(|p| fits(p))
188
            .max_by_key(|p| modified(p))
189
            .map(|p| (p, "same key"))
190
            .or_else(|| {
191
                others.iter().filter_map(|d| newest(d, form, |p| fits(p))).max_by_key(|p| modified(p)).map(|p| (p, "an older key"))
192
            });
193
        if let Some((p, which)) = pick {
194
            let from = p.parent().and_then(|d| d.file_name()).map(|n| n.to_string_lossy().into_owned()).unwrap_or_default();
195
            if copy(form, &p, dest).is_ok() {
196
                return Some(Taken {
197
                    how: format!("cache {name}: copied from another bookmark's ({from}), {which}"),
198
                    sources: read_sources(&p),
199
                });
200
            }
201
        }
202
        None
203
    }
204
205
    /// Keep a passing job's disk for mount `m` under its key, then keep
206
    /// the pool under its limit.
207
    pub fn keep(&self, ctx: &Ctx, m: &Mount, key: &str, disk: &Path, sources: &Sources) -> Result<()> {
208
        let Some(name) = m.cache.as_deref() else { return Ok(()) };
209
        let _changing = changing();
210
        // Whatever the job left, nothing in the pool is anyone else's to
211
        // write or runs as the runner (oxygenflow#70).
212
        if ctx.form == Form::Dir {
213
            seal(disk).with_context(|| format!("seal {}", disk.display()))?;
214
        }
215
        let mine = self.cache_dir(ctx, name).join(segment(ctx.bookmark));
216
        std::os::unix::fs::DirBuilderExt::mode(std::fs::DirBuilder::new().recursive(true), 0o700).create(&mine)?;
217
        std::fs::set_permissions(&self.root, std::fs::Permissions::from_mode(0o700))?;
218
        let dest = mine.join(format!("{key}.{}", ctx.form.ext()));
219
        if ctx.form == Form::Dir {
220
            // Another job of this slot kept one since: this one is newer.
221
            crate::workspace::remove_tree(&dest)?;
222
        }
223
        if std::fs::rename(disk, &dest).is_err() {
224
            copy(ctx.form, disk, &dest).with_context(|| format!("keep {}", disk.display()))?;
225
        }
226
        // Moved disks keep their old time: mark it used now.
227
        std::fs::File::open(&dest)?.set_modified(SystemTime::now())?;
228
        let sidecar = sources_of(&dest);
229
        let tmp = sidecar.with_extension("sources-part");
230
        let _ = std::fs::remove_file(&tmp);
231
        {
232
            use std::io::Write;
233
            use std::os::unix::fs::OpenOptionsExt;
234
            std::fs::OpenOptions::new().write(true).create_new(true).mode(0o600).open(&tmp)?.write_all(&serde_json::to_vec(sources)?)?;
235
        }
236
        std::fs::rename(&tmp, sidecar)?;
237
        // One disk per key is enough; older keys of this bookmark go once
238
        // the pool is over its limit.
239
        self.evict();
240
        Ok(())
241
    }
242
243
    /// After the job: a passing job's disk is kept; a failing one's is
244
    /// thrown away, so nothing it wrote reaches a later job.
245
    pub fn settle(&self, ctx: &Ctx, m: &Mount, key: &str, disk: &Path, sources: &Sources, passed: bool) -> Result<()> {
246
        if passed {
247
            return self.keep(ctx, m, key, disk, sources);
248
        }
249
        let gone = match ctx.form {
250
            Form::Dir => crate::workspace::remove_tree(disk),
251
            Form::Image => std::fs::remove_file(disk).or_else(|e| if e.kind() == std::io::ErrorKind::NotFound { Ok(()) } else { Err(e) }),
252
        };
253
        gone.with_context(|| format!("discard {}", disk.display()))
254
    }
255
256
    /// The repository and job slots holding at least one disk, as
257
    /// [`flow_core::api::cache_slot`] names them, for the claim
258
    /// (oxygenflow#51), each with when its newest disk was kept (Unix ms):
259
    /// a disk's time is set when it is kept, so the newest is the freshest
260
    /// build (oxygenflow#54). Images and directories alike, so a job goes
261
    /// where its cache is whichever tier keeps it.
262
    pub fn slot_times(&self) -> std::collections::BTreeMap<String, u64> {
263
        let mut out = std::collections::BTreeMap::new();
264
        for repo in std::fs::read_dir(&self.root).into_iter().flatten().flatten() {
265
            for job in std::fs::read_dir(repo.path()).into_iter().flatten().flatten() {
266
                let mut newest: Option<SystemTime> = None;
267
                // `<job>/<cache>/<bookmark>/<disk>`
268
                kept(&job.path(), 2, &mut |p, _| {
269
                    let t = modified(p);
270
                    newest = Some(newest.map_or(t, |n| n.max(t)));
271
                });
272
                if let Some(t) = newest {
273
                    let ms = t.duration_since(SystemTime::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
274
                    out.insert(format!("{}/{}", repo.file_name().to_string_lossy(), job.file_name().to_string_lossy()), ms);
275
                }
276
            }
277
        }
278
        out
279
    }
280
281
    /// Drop the disks used longest ago until the pool's blocks fit: an
282
    /// image's blocks, or everything in a directory. The caller holds
283
    /// [`changing`].
284
    fn evict(&self) -> u64 {
285
        let mut disks: Vec<(SystemTime, u64, PathBuf)> = Vec::new();
286
        // `<repo>/<job>/<cache>/<bookmark>/<disk>`
287
        kept(&self.root, 4, &mut |p, form| {
288
            if let Ok(md) = std::fs::symlink_metadata(p) {
289
                let bytes = match form {
290
                    Form::Image => md.blocks() * 512,
291
                    Form::Dir => tree_bytes(p),
292
                };
293
                disks.push((md.modified().unwrap_or(SystemTime::UNIX_EPOCH), bytes, p.to_path_buf()));
294
            }
295
        });
296
        let mut total: u64 = disks.iter().map(|d| d.1).sum();
297
        disks.sort();
298
        let mut freed = 0;
299
        for (_, bytes, p) in disks {
300
            if total <= self.limit {
301
                break;
302
            }
303
            let gone = match form_of(&p) {
304
                Some(Form::Dir) => crate::workspace::remove_tree(&p),
305
                _ => std::fs::remove_file(&p),
306
            };
307
            if gone.is_ok() {
308
                let _ = std::fs::remove_file(sources_of(&p));
309
                total -= bytes;
310
                freed += bytes;
311
            }
312
        }
313
        freed
314
    }
315
}
316
317
/// Where a job's `n`th disk mount is in its scratch directory, in the form
318
/// its tier mounts it: `microvm::run` attaches `mount-N.ext4`, and
319
/// `syscall::run` binds the directory `mount-N`.
320
pub fn disk_path(form: Form, scratch: &Path, n: usize) -> PathBuf {
321
    match form {
322
        Form::Image => scratch.join(format!("mount-{n}.ext4")),
323
        Form::Dir => crate::syscall::disk_dir(scratch, n),
324
    }
325
}
326
327
fn modified(p: &Path) -> SystemTime {
328
    std::fs::symlink_metadata(p).and_then(|m| m.modified()).unwrap_or(SystemTime::UNIX_EPOCH)
329
}
330
331
fn newest(dir: &Path, form: Form, ok: impl Fn(&Path) -> bool) -> Option<PathBuf> {
332
    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))
333
}
334
335
/// Each kept image or directory `depth` levels of directories below `dir`,
336
/// with its form; never inside a kept directory.
337
fn kept(dir: &Path, depth: usize, f: &mut dyn FnMut(&Path, Form)) {
338
    for e in std::fs::read_dir(dir).into_iter().flatten().flatten() {
339
        let p = e.path();
340
        let Ok(t) = e.file_type() else { continue };
341
        if depth == 0 {
342
            match form_of(&p) {
343
                Some(Form::Image) if t.is_file() => f(&p, Form::Image),
344
                Some(Form::Dir) if t.is_dir() => f(&p, Form::Dir),
345
                _ => {}
346
            }
347
        } else if t.is_dir() {
348
            kept(&p, depth - 1, f);
349
        }
350
    }
351
}
352
353
/// The blocks a directory's files take, each file once however many links
354
/// it has; links are not followed.
355
fn tree_bytes(dir: &Path) -> u64 {
356
    fn walk(p: &Path, seen: &mut std::collections::HashSet<(u64, u64)>) -> u64 {
357
        let Ok(md) = std::fs::symlink_metadata(p) else { return 0 };
358
        if md.nlink() > 1 && !md.is_dir() && !seen.insert((md.dev(), md.ino())) {
359
            return 0;
360
        }
361
        let mut bytes = md.blocks() * 512;
362
        if md.is_dir() {
363
            for e in std::fs::read_dir(p).into_iter().flatten().flatten() {
364
                bytes += walk(&e.path(), seen);
365
            }
366
        }
367
        bytes
368
    }
369
    walk(dir, &mut Default::default())
370
}
371
372
/// Make a directory a job left fit for the pool: nothing in it writable by
373
/// group or others, nothing set-id or sticky, and everything readable by
374
/// the runner (directories searchable and writable too, so it can be
375
/// copied and removed). Links are not followed. The job ran as the
376
/// runner's user, so all of it is the runner's to change; times are kept.
377
fn seal(p: &Path) -> std::io::Result<()> {
378
    let md = std::fs::symlink_metadata(p)?;
379
    if md.file_type().is_symlink() {
380
        return Ok(());
381
    }
382
    let mode = md.mode() & 0o7777;
383
    let want = (mode & !0o7022) | if md.is_dir() { 0o700 } else { 0o400 };
384
    if want != mode {
385
        std::fs::set_permissions(p, std::fs::Permissions::from_mode(want))?;
386
    }
387
    if md.is_dir() {
388
        for e in std::fs::read_dir(p)? {
389
            seal(&e?.path())?;
390
        }
391
    }
392
    Ok(())
393
}
394
395
fn copy(form: Form, from: &Path, to: &Path) -> Result<()> {
396
    match form {
397
        Form::Image => sparse_copy(from, to),
398
        Form::Dir => tree_copy(from, to),
399
    }
400
}
401
402
/// Copy a sparse disk without writing its holes.
403
fn sparse_copy(from: &Path, to: &Path) -> Result<()> {
404
    let tmp = to.with_extension("part");
405
    let st = std::process::Command::new("cp").arg("--sparse=always").arg(from).arg(&tmp).status().context("run cp")?;
406
    if !st.success() {
407
        let _ = std::fs::remove_file(&tmp);
408
        anyhow::bail!("cp {} failed", from.display());
409
    }
410
    std::fs::rename(&tmp, to)?;
411
    Ok(())
412
}
413
414
/// Copy a directory with its modes and times (a build tool judges by
415
/// them), sharing blocks where the filesystem can.
416
fn tree_copy(from: &Path, to: &Path) -> Result<()> {
417
    let tmp = to.with_extension("tree-part");
418
    crate::workspace::remove_tree(&tmp)?;
419
    let st = std::process::Command::new("cp")
420
        .args(["-a", "--reflink=auto", "--no-target-directory"])
421
        .arg(from)
422
        .arg(&tmp)
423
        .status()
424
        .context("run cp")?;
425
    if !st.success() {
426
        let _ = crate::workspace::remove_tree(&tmp);
427
        anyhow::bail!("cp {} failed", from.display());
428
    }
429
    std::fs::rename(&tmp, to).inspect_err(|_| {
430
        let _ = crate::workspace::remove_tree(&tmp);
431
    })?;
432
    Ok(())
433
}
434
435
#[cfg(test)]
436
mod tests {
437
    use super::*;
438
    use flow_core::pipeline::MountSource;
439
440
    fn mount(key: &[&str]) -> Mount {
441
        Mount {
442
            source: MountSource::Disk,
443
            target: "/src/target".into(),
444
            size_mib: 1,
445
            cache: Some("target".into()),
446
            key: key.iter().map(|s| s.to_string()).collect(),
447
        }
448
    }
449
450
    fn disk(path: &Path, fill: &[u8]) {
451
        let f = std::fs::File::create(path).unwrap();
452
        f.set_len(1 << 20).unwrap();
453
        std::os::unix::fs::FileExt::write_at(&f, fill, 0).unwrap();
454
    }
455
456
    fn first(path: &Path) -> Vec<u8> {
457
        std::fs::read(path).unwrap()[..4].to_vec()
458
    }
459
460
    #[test]
461
    fn a_kept_disk_starts_the_next_job_of_its_bookmark_then_others_copy_it() {
462
        let home = tempfile::tempdir().unwrap();
463
        let work = tempfile::tempdir().unwrap();
464
        let pool = Pool::new(home.path(), None);
465
        let m = mount(&["Cargo.lock"]);
466
        let main = Ctx { repo: "andy/hub", bookmark: "main", job: "test", image: "img", form: Form::Image };
467
        let key = Pool::key(&main, &m, |_| Some("lock-1".into()));
468
        // Nothing yet: the caller makes an empty disk.
469
        assert!(pool.take(&main, &m, &key, &work.path().join("a.ext4")).is_none());
470
        // A passing job's disk is kept and moved into the next job's place.
471
        disk(&work.path().join("a.ext4"), b"MAIN");
472
        let built: Sources = [("src/lib.rs".to_string(), ("h1".to_string(), 1_000))].into();
473
        pool.keep(&main, &m, &key, &work.path().join("a.ext4"), &built).unwrap();
474
        assert!(!work.path().join("a.ext4").exists(), "moved, not copied");
475
        let taken = pool.take(&main, &m, &key, &work.path().join("b.ext4")).unwrap();
476
        assert!(taken.how.contains("same key"), "{}", taken.how);
477
        assert_eq!(taken.sources.as_ref(), Some(&built), "its sources come with it (#52)");
478
        assert_eq!(first(&work.path().join("b.ext4")), b"MAIN");
479
        pool.keep(&main, &m, &key, &work.path().join("b.ext4"), &built).unwrap();
480
        // A changed key takes the bookmark's newest disk all the same.
481
        let key2 = Pool::key(&main, &m, |_| Some("lock-2".into()));
482
        assert_ne!(key, key2);
483
        let taken = pool.take(&main, &m, &key2, &work.path().join("c.ext4")).unwrap();
484
        assert!(taken.how.contains("older key"), "{}", taken.how);
485
        assert_eq!(taken.sources.as_ref(), Some(&built));
486
        pool.keep(&main, &m, &key2, &work.path().join("c.ext4"), &built).unwrap();
487
        assert_eq!(
488
            pool.slot_times().into_keys().collect::<Vec<_>>(),
489
            vec![flow_core::api::cache_slot("andy/hub", "test")],
490
            "what the claim reports (#51)"
491
        );
492
        // With when it was last kept (#54): just now.
493
        let now = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_millis() as u64;
494
        let kept = pool.slot_times()[&flow_core::api::cache_slot("andy/hub", "test")];
495
        assert!(now.abs_diff(kept) < 60_000, "kept {kept}, now {now}");
496
        // Another bookmark copies it, leaving main's in place.
497
        let feature = Ctx { bookmark: "feature", ..main };
498
        let taken = pool.take(&feature, &m, &key2, &work.path().join("d.ext4")).unwrap();
499
        assert!(taken.how.contains("copied from another bookmark"), "{}", taken.how);
500
        assert_eq!(taken.sources.as_ref(), Some(&built), "a copy reads them and leaves them");
501
        assert_eq!(first(&work.path().join("d.ext4")), b"MAIN");
502
        let taken = pool.take(&main, &m, &key2, &work.path().join("e.ext4")).unwrap();
503
        assert!(taken.how.contains("this bookmark's") && taken.sources.is_some(), "main's is still there, with its sources");
504
        // Another job or a different size never shares it.
505
        let other_job = Ctx { job: "build", ..main };
506
        assert!(pool.take(&other_job, &m, &key2, &work.path().join("f.ext4")).is_none());
507
        let mut bigger = mount(&["Cargo.lock"]);
508
        bigger.size_mib = 2;
509
        assert!(pool.take(&feature, &bigger, &key2, &work.path().join("g.ext4")).is_none());
510
    }
511
512
    #[test]
513
    fn the_key_follows_the_image_the_job_and_the_key_files() {
514
        let ctx = Ctx { repo: "r", bookmark: "main", job: "test", image: "img-1", form: Form::Image };
515
        let m = mount(&["Cargo.lock", "rust-toolchain.toml"]);
516
        let k = Pool::key(&ctx, &m, |p| Some(format!("{p}-v1")));
517
        assert_eq!(k, Pool::key(&ctx, &m, |p| Some(format!("{p}-v1"))));
518
        assert_ne!(k, Pool::key(&ctx, &m, |p| Some(format!("{p}-v2"))), "a key file changed");
519
        assert_ne!(k, Pool::key(&Ctx { image: "img-2", ..ctx }, &m, |p| Some(format!("{p}-v1"))), "another image");
520
        assert_ne!(k, Pool::key(&ctx, &m, |_| None), "a key file missing");
521
    }
522
523
    #[test]
524
    fn the_pool_drops_the_disks_used_longest_ago() {
525
        let home = tempfile::tempdir().unwrap();
526
        let work = tempfile::tempdir().unwrap();
527
        let pool = Pool { root: home.path().join("cache"), limit: 0 };
528
        let m = mount(&[]);
529
        let ctx = Ctx { repo: "r", bookmark: "main", job: "j", image: "i", form: Form::Image };
530
        let full = |p: &Path| {
531
            let f = std::fs::File::create(p).unwrap();
532
            std::os::unix::fs::FileExt::write_at(&f, &vec![7u8; 1 << 20], 0).unwrap();
533
        };
534
        full(&work.path().join("a.ext4"));
535
        // Over a limit of nothing: kept, then dropped at once.
536
        pool.keep(&ctx, &m, "k1", &work.path().join("a.ext4"), &Sources::new()).unwrap();
537
        assert!(pool.take(&ctx, &m, "k1", &work.path().join("b.ext4")).is_none());
538
        let roomy = Pool { root: home.path().join("cache"), limit: 1 << 30 };
539
        full(&work.path().join("c.ext4"));
540
        roomy.keep(&ctx, &m, "k2", &work.path().join("c.ext4"), &Sources::new()).unwrap();
541
        assert!(roomy.take(&ctx, &m, "k2", &work.path().join("d.ext4")).is_some(), "under the limit it stays");
542
    }
543
544
    /// A directory as the syscall tier leaves one: `marker` in
545
    /// `debug/marker`, and a file of `kib` KiB.
546
    fn tree(path: &Path, marker: &[u8], kib: usize) {
547
        std::fs::create_dir_all(path.join("debug")).unwrap();
548
        std::fs::write(path.join("debug/marker"), marker).unwrap();
549
        // Not all one byte, so a compressing filesystem stores it whole.
550
        let bytes: Vec<u8> = (0..kib << 10).map(|i: usize| (i.wrapping_mul(2_654_435_761) >> 13) as u8).collect();
551
        std::fs::write(path.join("debug/blob"), bytes).unwrap();
552
    }
553
554
    fn dir_ctx<'a>(bookmark: &'a str) -> Ctx<'a> {
555
        Ctx { repo: "andy/flow", bookmark, job: "test", image: "img", form: Form::Dir }
556
    }
557
558
    /// oxygenflow#70: the syscall tier's directories are taken and kept as
559
    /// a microVM's images are, and never mixed with them.
560
    #[test]
561
    fn a_kept_directory_starts_the_next_syscall_job_and_others_copy_it() {
562
        let home = tempfile::tempdir().unwrap();
563
        let work = tempfile::tempdir().unwrap();
564
        let pool = Pool::new(home.path(), None);
565
        let m = mount(&["Cargo.lock"]);
566
        let main = dir_ctx("main");
567
        let key = Pool::key(&main, &m, |_| Some("lock-1".into()));
568
        let w = |n: &str| work.path().join(n);
569
        assert!(pool.take(&main, &m, &key, &w("a")).is_none(), "nothing kept yet");
570
        // A passing job's directory goes into the pool, times and all.
571
        tree(&w("a"), b"FIRST", 4);
572
        // Named like an image: the pool never looks inside a directory.
573
        std::fs::write(w("a/debug/inner.ext4"), b"not a disk").unwrap();
574
        let old = SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_000_000);
575
        std::fs::File::open(w("a/debug/marker")).unwrap().set_modified(old).unwrap();
576
        let built: Sources = [("src/lib.rs".to_string(), ("h1".to_string(), 1_000))].into();
577
        pool.settle(&main, &m, &key, &w("a"), &built, true).unwrap();
578
        assert!(!w("a").exists(), "moved, not copied");
579
        assert_eq!(
580
            pool.slot_times().into_keys().collect::<Vec<_>>(),
581
            vec![flow_core::api::cache_slot("andy/flow", "test")],
582
            "the claim reports it like an image (#51)"
583
        );
584
        // The next job of its bookmark moves it out: a job of the same
585
        // slot beside it finds nothing to take.
586
        let taken = pool.take(&main, &m, &key, &w("b")).unwrap();
587
        assert!(taken.how.contains("this bookmark's, same key"), "{}", taken.how);
588
        assert_eq!(taken.sources.as_ref(), Some(&built), "its sources come with it (#52)");
589
        assert_eq!(std::fs::read(w("b/debug/marker")).unwrap(), b"FIRST");
590
        assert_eq!(modified(&w("b/debug/marker")), old, "its files keep their times");
591
        assert_eq!(std::fs::read(w("b/debug/inner.ext4")).unwrap(), b"not a disk");
592
        assert!(pool.take(&main, &m, &key, &w("c")).is_none(), "taken once");
593
        assert!(pool.slot_times().is_empty());
594
        // Kept again, and under a changed key the newest is taken all the same.
595
        std::fs::write(w("b/debug/marker"), b"SECOND").unwrap();
596
        pool.settle(&main, &m, &key, &w("b"), &built, true).unwrap();
597
        let key2 = Pool::key(&main, &m, |_| Some("lock-2".into()));
598
        let taken = pool.take(&main, &m, &key2, &w("d")).unwrap();
599
        assert!(taken.how.contains("older key"), "{}", taken.how);
600
        assert_eq!(std::fs::read(w("d/debug/marker")).unwrap(), b"SECOND");
601
        pool.settle(&main, &m, &key2, &w("d"), &built, true).unwrap();
602
        // Another bookmark copies it, with its times, and main keeps its own.
603
        let feature = dir_ctx("feature");
604
        let taken = pool.take(&feature, &m, &key2, &w("e")).unwrap();
605
        assert!(taken.how.contains("copied from another bookmark's (main-"), "{}", taken.how);
606
        assert_eq!(taken.sources.as_ref(), Some(&built));
607
        assert_eq!(std::fs::read(w("e/debug/marker")).unwrap(), b"SECOND");
608
        assert!(pool.take(&main, &m, &key2, &w("f")).is_some(), "main's is still there");
609
        assert_eq!(modified(&w("e/debug/marker")), modified(&w("f/debug/marker")), "a copy keeps the times");
610
        pool.settle(&main, &m, &key2, &w("f"), &built, true).unwrap();
611
        // A microVM job of the same slot never takes a directory, nor a
612
        // syscall job an image.
613
        let image = Ctx { form: Form::Image, ..main };
614
        assert!(pool.take(&image, &m, &key2, &w("g.ext4")).is_none());
615
        disk(&w("h.ext4"), b"DISK");
616
        pool.keep(&image, &m, &key2, &w("h.ext4"), &built).unwrap();
617
        let taken = pool.take(&main, &m, &key2, &w("i")).unwrap();
618
        assert!(w("i/debug/marker").is_file(), "{}", taken.how);
619
        assert!(pool.take(&main, &m, &key2, &w("j")).is_none(), "the image is not a directory");
620
        assert!(pool.take(&image, &m, &key2, &w("k.ext4")).is_some(), "and is still there for a microVM");
621
    }
622
623
    #[test]
624
    fn a_failing_jobs_directory_is_thrown_away() {
625
        let home = tempfile::tempdir().unwrap();
626
        let work = tempfile::tempdir().unwrap();
627
        let pool = Pool::new(home.path(), None);
628
        let m = mount(&[]);
629
        let main = dir_ctx("main");
630
        tree(&work.path().join("a"), b"GOOD", 1);
631
        pool.settle(&main, &m, "k", &work.path().join("a"), &Sources::new(), true).unwrap();
632
        let taken = pool.take(&main, &m, "k", &work.path().join("b"));
633
        assert!(taken.is_some());
634
        // The job that took it fails after writing to it: not kept, gone.
635
        std::fs::write(work.path().join("b/debug/marker"), b"POISON").unwrap();
636
        pool.settle(&main, &m, "k", &work.path().join("b"), &Sources::new(), false).unwrap();
637
        assert!(!work.path().join("b").exists());
638
        assert!(pool.take(&main, &m, "k", &work.path().join("c")).is_none(), "nothing it wrote reaches a later job");
639
        assert!(pool.slot_times().is_empty());
640
        // A failed image is thrown away the same way.
641
        let image = Ctx { form: Form::Image, ..main };
642
        disk(&work.path().join("d.ext4"), b"BAD!");
643
        pool.settle(&image, &m, "k", &work.path().join("d.ext4"), &Sources::new(), false).unwrap();
644
        assert!(!work.path().join("d.ext4").exists() && pool.slot_times().is_empty());
645
    }
646
647
    #[test]
648
    fn directories_are_evicted_by_their_size_oldest_first() {
649
        let home = tempfile::tempdir().unwrap();
650
        let work = tempfile::tempdir().unwrap();
651
        // Room for one 512 KiB directory, not two.
652
        let pool = Pool { root: home.path().join("cache"), limit: 800 << 10 };
653
        let m = mount(&[]);
654
        let (main, feature) = (dir_ctx("main"), dir_ctx("feature"));
655
        tree(&work.path().join("a"), b"OLD", 512);
656
        pool.keep(&main, &m, "k1", &work.path().join("a"), &Sources::new()).unwrap();
657
        let kept = pool.cache_dir(&main, "target").join(segment("main")).join("k1.tree");
658
        assert!(kept.is_dir(), "under the limit it stays");
659
        assert!(tree_bytes(&kept) >= 512 << 10, "a directory counts what is in it");
660
        // Older than the next one, however fast this test runs.
661
        std::fs::File::open(&kept).unwrap().set_modified(SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1)).unwrap();
662
        tree(&work.path().join("b"), b"NEW", 512);
663
        pool.keep(&feature, &m, "k1", &work.path().join("b"), &Sources::new()).unwrap();
664
        assert!(!kept.exists() && !kept.with_extension("tree-sources").exists(), "the one used longest ago goes, sources and all");
665
        let taken = pool.take(&main, &m, "k1", &work.path().join("c")).unwrap();
666
        assert!(taken.how.contains("bookmark's (feature-"), "{}", taken.how);
667
        assert_eq!(std::fs::read(work.path().join("c/debug/marker")).unwrap(), b"NEW");
668
    }
669
670
    /// oxygenflow#70: the job wrote what it liked; the pool keeps nothing
671
    /// others may write or that runs as the runner, and is the runner's.
672
    #[test]
673
    fn nothing_kept_is_writable_by_others_or_set_id() {
674
        let home = tempfile::tempdir().unwrap();
675
        let work = tempfile::tempdir().unwrap();
676
        let pool = Pool::new(home.path(), None);
677
        let m = mount(&[]);
678
        let main = dir_ctx("main");
679
        let a = work.path().join("a");
680
        tree(&a, b"X", 1);
681
        let mode = |p: &Path, m: u32| std::fs::set_permissions(p, std::fs::Permissions::from_mode(m)).unwrap();
682
        std::fs::write(a.join("open"), b"").unwrap();
683
        mode(&a.join("open"), 0o777);
684
        std::fs::write(a.join("suid"), b"").unwrap();
685
        mode(&a.join("suid"), 0o4755);
686
        std::fs::write(a.join("debug/sealed"), b"secret").unwrap();
687
        mode(&a.join("debug/sealed"), 0o000);
688
        std::os::unix::fs::symlink("/etc/passwd", a.join("link")).unwrap();
689
        let when = modified(&a.join("debug/marker"));
690
        mode(&a.join("debug"), 0o1777);
691
        mode(&a, 0o500);
692
        pool.keep(&main, &m, "k", &a, &Sources::new()).unwrap();
693
        let root = home.path().join("cache");
694
        assert_eq!(std::fs::metadata(&root).unwrap().mode() & 0o777, 0o700, "the pool is the runner's alone");
695
        let mut seen = 0;
696
        fn each(p: &Path, f: &mut dyn FnMut(&Path, &std::fs::Metadata)) {
697
            let md = std::fs::symlink_metadata(p).unwrap();
698
            f(p, &md);
699
            if md.is_dir() {
700
                for e in std::fs::read_dir(p).unwrap() {
701
                    each(&e.unwrap().path(), f);
702
                }
703
            }
704
        }
705
        each(&root, &mut |p, md| {
706
            seen += 1;
707
            if !md.file_type().is_symlink() {
708
                assert_eq!(md.mode() & 0o7022, 0, "{} is {:o}", p.display(), md.mode());
709
            }
710
        });
711
        assert!(seen >= 10, "{seen}");
712
        // Still readable, with its times, so the next job builds on it.
713
        let taken = pool.take(&main, &m, "k", &work.path().join("b"));
714
        assert!(taken.is_some());
715
        let b = work.path().join("b");
716
        assert_eq!(std::fs::read(b.join("debug/sealed")).unwrap(), b"secret");
717
        assert_eq!(std::fs::metadata(b.join("open")).unwrap().mode() & 0o777, 0o755);
718
        assert_eq!(std::fs::read_link(b.join("link")).unwrap(), Path::new("/etc/passwd"), "a link is kept, not followed");
719
        assert_eq!(modified(&b.join("debug/marker")), when);
720
    }
721
}