OXYGENHUB

Prune build caches before they are kept 077a65242359

A kept cache took every build's files into the next run (each crate
under every hash a lockfile or feature set gave it) until the job's disk
was full. Over cache_prune_percent of its mount's size (runner config,
default 60), a passing job's cache now loses the files the job did not
use, oldest first; one still over is not kept. Used: read, written or
changed since the job started, sharing a cargo unit hash with such a
file, another link of one, or under one of the job's outputs.

A microVM's flow-init mounts cache disks strictatime, prunes them before
powering off and reports \x1eC frames; without word from it the runner
measures the image with dumpe2fs. The runner prunes the syscall tier's
directories itself, having set their atimes to the epoch at take time.
The job's log says what was kept and pruned, or why not.

Closes #72
Author
Andy McCall @andymccall <[email protected]>
via
agent agent:claude@andymccall <[email protected]> · keyless, under your grant for changes,refs,seal · key 6bb2facb6e21
When
· 2026-10-20 04:29 UTC
Signature
◈ sealed by agent:claude@andymccall
Hash
077a65242359898d8967ed32f416ba4678d3ed58947c49052f7b195ae0a4d186
Depends on
2594278f49b2 280ef951d193 4cf0c99a1486 59078d2b8d46 82cf6e91d398 8f211e20bea5 a083cc802892 c3db750ed96e d60aac6c3f8a da72c3b192c3 e0918b887850 ea56c56eaf90 ea81796bb757 eb5c4db673e5 fc10d9e80383
Depended on by
010b9a79de79 788f57e4d266 3554afefb6aa 8bf49b95b09e
In bookmarks
main
In tags
v0.3.0 v0.3.1 v0.4.0 v0.5.0 v0.5.1 v0.5.2 v0.5.3
Edits
55 atoms, 2 tree operations, 15 files

Browse the tree at this change

added crates/flow-init/src/lib.rs +5 −0
1 +
//! The parts of the guest init the runner shares: pruning a build cache
2 +
//! before it is kept (oxygenflow#72), done in the guest for a microVM's
3 +
//! disks and on the runner for the syscall tier's directories.
4 +
5 +
pub mod prune;
added crates/flow-init/src/prune.rs +365 −0
1 +
//! Pruning a build cache before it is kept (oxygenflow#72, spec §5.4).
2 +
//!
3 +
//! A kept disk starts the next run, so without pruning it carries every
4 +
//! build's files: each crate's artifacts under every hash a lockfile or a
5 +
//! feature set ever gave it. Before a passing job's cache disk goes back
6 +
//! into the pool, the files this run did not use are removed, oldest
7 +
//! first, until what is on the disk is at most a share of its size (the
8 +
//! cap). The microVM's guest init prunes its disks before it powers off,
9 +
//! since only it can see inside them; the runner prunes the syscall
10 +
//! tier's directories itself. Both use this module.
11 +
//!
12 +
//! What the run used is told by the files' times against when the job
13 +
//! started: written (mtime), changed (ctime) or read (atime). Reads are
14 +
//! seen only where the file system records them: the guest mounts a cache
15 +
//! disk `strictatime`; on the runner, where a directory sits on whatever
16 +
//! file system the runner has, each file's atime is set to the epoch
17 +
//! before the job ([`forget_reads`]), so even `relatime` records its
18 +
//! first read. A file system mounted `noatime` records none, and then
19 +
//! nothing is pruned.
20 +
//!
21 +
//! A build tool does not read everything it relies on. Cargo, finding a
22 +
//! crate fresh, reads its fingerprint and only looks at its rlib and
23 +
//! binaries, so their times say nothing. Every name cargo gives a unit's
24 +
//! files carries the unit's hash (`.fingerprint/foo-<hash>/`,
25 +
//! `deps/libfoo-<hash>.rlib`, `build/foo-<hash>/`), so a file stays when
26 +
//! a file this run used carries one of the hashes in its path: a fresh
27 +
//! crate's fingerprint keeps its artifacts. A file with another link
28 +
//! that stays, or under a path the job's outputs name, stays too. What
29 +
//! goes is what this run's build did not touch, which a later build needs
30 +
//! only if an older lockfile comes back, and then cargo rebuilds it.
31 +
32 +
use serde::{Deserialize, Serialize};
33 +
use std::collections::{HashMap, HashSet};
34 +
use std::os::unix::fs::MetadataExt;
35 +
use std::path::{Path, PathBuf};
36 +
use std::time::{Duration, SystemTime};
37 +
38 +
/// The share of a cache disk a kept cache may fill when no runner says
39 +
/// otherwise: room for this run's build and as much again.
40 +
pub const DEFAULT_PERCENT: u8 = 60;
41 +
42 +
/// What pruning a cache found and did, in bytes of blocks on its disk.
43 +
#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq)]
44 +
pub struct Pruned {
45 +
    /// The disk's size, and the share of it the cache may fill.
46 +
    pub size: u64,
47 +
    pub percent: u8,
48 +
    /// What was on the disk before pruning, and after.
49 +
    pub before: u64,
50 +
    pub after: u64,
51 +
    /// What went: bytes and files.
52 +
    pub freed: u64,
53 +
    pub files: u64,
54 +
    /// Why nothing could be pruned, when so (no access times, or an
55 +
    /// error): only the size then decides.
56 +
    #[serde(default, skip_serializing_if = "Option::is_none")]
57 +
    pub unpruned: Option<String>,
58 +
}
59 +
60 +
impl Pruned {
61 +
    /// The most the cache may keep: `percent` of `size`.
62 +
    pub fn cap(&self) -> u64 {
63 +
        cap(self.size, self.percent)
64 +
    }
65 +
66 +
    /// Whether what is left may be kept.
67 +
    pub fn fits(&self) -> bool {
68 +
        self.after <= self.cap()
69 +
    }
70 +
}
71 +
72 +
fn cap(size: u64, percent: u8) -> u64 {
73 +
    (size as u128 * percent.min(100) as u128 / 100) as u64
74 +
}
75 +
76 +
/// Now, as a job's start to compare file times against. File systems
77 +
/// stamp times from a clock that may lag this one by a tick, so the
78 +
/// moments either side of it are left to pass: nothing done before the
79 +
/// call (such as [`forget_reads`]) can look as though it came after.
80 +
pub fn start_clock() -> SystemTime {
81 +
    std::thread::sleep(Duration::from_millis(20));
82 +
    let t = SystemTime::now();
83 +
    std::thread::sleep(Duration::from_millis(20));
84 +
    t
85 +
}
86 +
87 +
/// Whether the file system holding `path` records reads (not `noatime`).
88 +
pub fn records_reads(path: &Path) -> bool {
89 +
    let Ok(c) = std::ffi::CString::new(path.as_os_str().as_encoded_bytes()) else { return false };
90 +
    let mut st: libc::statvfs = unsafe { std::mem::zeroed() };
91 +
    if unsafe { libc::statvfs(c.as_ptr(), &mut st) } != 0 {
92 +
        return false;
93 +
    }
94 +
    st.f_flag & libc::ST_NOATIME == 0
95 +
}
96 +
97 +
/// The blocks in use on the file system mounted at `path`, and its size
98 +
/// (as `df` counts them).
99 +
pub fn usage(path: &Path) -> std::io::Result<(u64, u64)> {
100 +
    let c = std::ffi::CString::new(path.as_os_str().as_encoded_bytes())?;
101 +
    let mut st: libc::statvfs = unsafe { std::mem::zeroed() };
102 +
    if unsafe { libc::statvfs(c.as_ptr(), &mut st) } != 0 {
103 +
        return Err(std::io::Error::last_os_error());
104 +
    }
105 +
    let unit = st.f_frsize as u64;
106 +
    Ok(((st.f_blocks as u64).saturating_sub(st.f_bfree as u64) * unit, st.f_blocks as u64 * unit))
107 +
}
108 +
109 +
/// Set every regular file's atime under `root` to the epoch, its mtime
110 +
/// kept, so the job's first read of each is recorded even on `relatime`
111 +
/// (which records a read only of a file read no later than it changed).
112 +
/// Links are not followed. The count of files.
113 +
pub fn forget_reads(root: &Path) -> std::io::Result<u64> {
114 +
    let mut n = 0;
115 +
    let mut err = None;
116 +
    walk(root, &mut |p, md| {
117 +
        if !md.file_type().is_file() {
118 +
            return;
119 +
        }
120 +
        let Ok(c) = std::ffi::CString::new(p.as_os_str().as_encoded_bytes()) else { return };
121 +
        let times = [libc::timespec { tv_sec: 0, tv_nsec: 0 }, libc::timespec { tv_sec: 0, tv_nsec: libc::UTIME_OMIT }];
122 +
        if unsafe { libc::utimensat(libc::AT_FDCWD, c.as_ptr(), times.as_ptr(), libc::AT_SYMLINK_NOFOLLOW) } == 0 {
123 +
            n += 1;
124 +
        } else if err.is_none() {
125 +
            err = Some(std::io::Error::last_os_error());
126 +
        }
127 +
    })?;
128 +
    match err {
129 +
        Some(e) => Err(e),
130 +
        None => Ok(n),
131 +
    }
132 +
}
133 +
134 +
/// Each entry under `root` (not `root` itself) with its metadata, parents
135 +
/// before children; links are not followed, nor other file systems
136 +
/// entered.
137 +
fn walk(root: &Path, f: &mut dyn FnMut(&Path, &std::fs::Metadata)) -> std::io::Result<()> {
138 +
    let dev = std::fs::symlink_metadata(root)?.dev();
139 +
    let mut dirs = vec![root.to_path_buf()];
140 +
    while let Some(d) = dirs.pop() {
141 +
        let Ok(entries) = std::fs::read_dir(&d) else { continue };
142 +
        for e in entries.flatten() {
143 +
            let p = e.path();
144 +
            let Ok(md) = std::fs::symlink_metadata(&p) else { continue };
145 +
            if md.dev() != dev {
146 +
                continue;
147 +
            }
148 +
            f(&p, &md);
149 +
            if md.is_dir() {
150 +
                dirs.push(p);
151 +
            }
152 +
        }
153 +
    }
154 +
    Ok(())
155 +
}
156 +
157 +
/// The unit hashes in `rel`: a 16-digit lowercase hex word after a `-`
158 +
/// in any of its names, as cargo names a unit's files.
159 +
fn hashes(rel: &Path) -> impl Iterator<Item = String> + '_ {
160 +
    rel.components().filter_map(|c| c.as_os_str().to_str()).flat_map(|name| {
161 +
        name.split('-').skip(1).filter_map(|w| {
162 +
            let w = w.split('.').next().unwrap_or("");
163 +
            (w.len() == 16 && w.bytes().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))).then(|| w.to_string())
164 +
        })
165 +
    })
166 +
}
167 +
168 +
fn nanos(s: i64, ns: i64) -> i128 {
169 +
    s as i128 * 1_000_000_000 + ns as i128
170 +
}
171 +
172 +
/// One file, however many names it has.
173 +
struct File {
174 +
    paths: Vec<PathBuf>,
175 +
    links: u64,
176 +
    bytes: u64,
177 +
    /// When it was last used as far as can be told: the later of its
178 +
    /// atime and mtime.
179 +
    last: i128,
180 +
    used: bool,
181 +
}
182 +
183 +
/// Prune the cache at `root`, a disk of `size` bytes with `before` in
184 +
/// use, down to `percent` of its size, removing only regular files this
185 +
/// run did not use (see the module's notes), those used longest ago
186 +
/// first. `started` is when the job started ([`start_clock`]); `keep` are
187 +
/// paths relative to `root` (the job's outputs) that stay whatever their
188 +
/// times. Under the cap nothing is touched. `after` is `before` less what
189 +
/// was freed.
190 +
pub fn prune(root: &Path, started: SystemTime, keep: &[String], size: u64, before: u64, percent: u8) -> std::io::Result<Pruned> {
191 +
    let mut out = Pruned { size, percent, before, after: before, ..Pruned::default() };
192 +
    if before <= out.cap() {
193 +
        return Ok(out);
194 +
    }
195 +
    let since = started.duration_since(SystemTime::UNIX_EPOCH).map(|d| d.as_nanos() as i128).unwrap_or(0);
196 +
    let keep: Vec<PathBuf> = keep.iter().map(|k| PathBuf::from(k.trim_start_matches('/'))).filter(|k| !k.as_os_str().is_empty()).collect();
197 +
    let mut files: HashMap<(u64, u64), File> = HashMap::new();
198 +
    walk(root, &mut |p, md| {
199 +
        if !md.file_type().is_file() {
200 +
            return;
201 +
        }
202 +
        let rel = p.strip_prefix(root).unwrap_or(p);
203 +
        let used = [nanos(md.atime(), md.atime_nsec()), nanos(md.mtime(), md.mtime_nsec()), nanos(md.ctime(), md.ctime_nsec())]
204 +
            .into_iter()
205 +
            .any(|t| t >= since)
206 +
            || keep.iter().any(|k| rel.starts_with(k));
207 +
        let f = files.entry((md.dev(), md.ino())).or_insert_with(|| File {
208 +
            paths: Vec::new(),
209 +
            links: md.nlink(),
210 +
            bytes: md.blocks() * 512,
211 +
            last: nanos(md.atime(), md.atime_nsec()).max(nanos(md.mtime(), md.mtime_nsec())),
212 +
            used: false,
213 +
        });
214 +
        f.paths.push(p.to_path_buf());
215 +
        f.used |= used;
216 +
    })?;
217 +
    // A hash any used file carries keeps every file carrying it.
218 +
    let live: HashSet<String> =
219 +
        files.values().filter(|f| f.used).flat_map(|f| f.paths.iter().flat_map(|p| hashes(p.strip_prefix(root).unwrap_or(p)))).collect();
220 +
    let mut gone: Vec<File> = files
221 +
        .into_values()
222 +
        .filter(|f| {
223 +
            !f.used
224 +
                // A link from outside the cache: removing it frees nothing.
225 +
                && f.links == f.paths.len() as u64
226 +
                && !f.paths.iter().any(|p| hashes(p.strip_prefix(root).unwrap_or(p)).any(|h| live.contains(&h)))
227 +
        })
228 +
        .collect();
229 +
    gone.sort_by_key(|f| f.last);
230 +
    let mut emptied: HashSet<PathBuf> = HashSet::new();
231 +
    for f in gone {
232 +
        if out.after <= out.cap() {
233 +
            break;
234 +
        }
235 +
        let mut removed = true;
236 +
        for p in &f.paths {
237 +
            if std::fs::remove_file(p).is_err() {
238 +
                removed = false;
239 +
            } else if let Some(parent) = p.parent() {
240 +
                emptied.insert(parent.to_path_buf());
241 +
            }
242 +
        }
243 +
        if removed {
244 +
            out.after = out.after.saturating_sub(f.bytes);
245 +
            out.freed += f.bytes;
246 +
            out.files += 1;
247 +
        }
248 +
    }
249 +
    // Directories pruning emptied go too, deepest first; never `root`.
250 +
    let mut dirs: Vec<PathBuf> = emptied.into_iter().collect();
251 +
    dirs.sort_by_key(|d| std::cmp::Reverse(d.components().count()));
252 +
    for d in dirs {
253 +
        let mut d = d.as_path();
254 +
        while d != root && d.starts_with(root) && std::fs::remove_dir(d).is_ok() {
255 +
            match d.parent() {
256 +
                Some(p) => d = p,
257 +
                None => break,
258 +
            }
259 +
        }
260 +
    }
261 +
    Ok(out)
262 +
}
263 +
264 +
#[cfg(test)]
265 +
mod tests {
266 +
    use super::*;
267 +
268 +
    /// Times around a job that started at `T`, ahead of the clock: a file
269 +
    /// whose times are set is changed (ctime) now, before it.
270 +
    const T: u64 = 4_000_000_000;
271 +
272 +
    fn at(secs: u64) -> SystemTime {
273 +
        SystemTime::UNIX_EPOCH + Duration::from_secs(secs)
274 +
    }
275 +
276 +
    /// A file of `kib` KiB at `rel`, last read and written at `secs`.
277 +
    fn file(root: &Path, rel: &str, kib: usize, secs: u64) -> PathBuf {
278 +
        let p = root.join(rel);
279 +
        std::fs::create_dir_all(p.parent().unwrap()).unwrap();
280 +
        let bytes: Vec<u8> = (0..kib << 10).map(|i: usize| (i.wrapping_mul(2_654_435_761) >> 13) as u8).collect();
281 +
        std::fs::write(&p, bytes).unwrap();
282 +
        let f = std::fs::File::options().write(true).open(&p).unwrap();
283 +
        f.set_times(std::fs::FileTimes::new().set_accessed(at(secs)).set_modified(at(secs))).unwrap();
284 +
        p
285 +
    }
286 +
287 +
    fn bytes(p: &Path) -> u64 {
288 +
        std::fs::metadata(p).unwrap().blocks() * 512
289 +
    }
290 +
291 +
    #[test]
292 +
    fn hashes_are_cargos_unit_hashes() {
293 +
        let h = |p: &str| hashes(Path::new(p)).collect::<Vec<_>>();
294 +
        assert_eq!(h("debug/deps/libdep-8f2b19826c505967.rlib"), ["8f2b19826c505967"]);
295 +
        assert_eq!(h("debug/.fingerprint/dep-8f2b19826c505967/lib-dep"), ["8f2b19826c505967"]);
296 +
        assert_eq!(h("debug/build/my-crate-0123456789abcdef/out/x.o"), ["0123456789abcdef"]);
297 +
        assert!(h("debug/incremental/app-2dadnjnzlrswc/s-hndl5uwdpw-1uaho4m.lock").is_empty());
298 +
        assert!(h("debug/app").is_empty() && h("debug/deps/x-0123456789ABCDEF").is_empty() && h("x-0123456789abcde").is_empty());
299 +
    }
300 +
301 +
    #[test]
302 +
    fn under_the_cap_nothing_is_touched() {
303 +
        let dir = tempfile::tempdir().unwrap();
304 +
        let old = file(dir.path(), "debug/deps/libold-1111111111111111.rlib", 64, 1_000);
305 +
        let p = prune(dir.path(), at(T), &[], 1 << 30, 64 << 10, 60).unwrap();
306 +
        assert!(old.exists() && p.fits() && p.freed == 0 && p.after == p.before, "{p:?}");
307 +
    }
308 +
309 +
    /// A cargo target as a run leaves it: this lockfile's units used this
310 +
    /// run (one rebuilt, one fresh: only its fingerprint read), an older
311 +
    /// lockfile's not, and an output the run did not touch.
312 +
    #[test]
313 +
    fn what_this_run_did_not_use_goes_oldest_first_and_its_units_stay_whole() {
314 +
        let dir = tempfile::tempdir().unwrap();
315 +
        let root = dir.path();
316 +
        let app = file(root, "debug/deps/app-aaaaaaaaaaaaaaaa", 256, T + 100);
317 +
        let app_fp = file(root, "debug/.fingerprint/app-aaaaaaaaaaaaaaaa/bin-app", 1, T + 100);
318 +
        let dep_rlib = file(root, "debug/deps/libdep-bbbbbbbbbbbbbbbb.rlib", 256, T - 5_000);
319 +
        let dep_fp = file(root, "debug/.fingerprint/dep-bbbbbbbbbbbbbbbb/lib-dep", 1, T - 5_000);
320 +
        std::fs::File::open(&dep_fp).unwrap().set_times(std::fs::FileTimes::new().set_accessed(at(T + 50))).unwrap();
321 +
        let script = file(root, "debug/build/dep-bbbbbbbbbbbbbbbb/out/gen.rs", 64, T - 5_000);
322 +
        // Uplifted: another name of the rebuilt binary.
323 +
        std::fs::hard_link(&app, root.join("debug/app")).unwrap();
324 +
        // An older lockfile's units, untouched, the oldest first to go.
325 +
        let oldest = file(root, "debug/deps/libdep-cccccccccccccccc.rlib", 256, T - 9_000);
326 +
        let oldest_fp = file(root, "debug/.fingerprint/dep-cccccccccccccccc/lib-dep", 1, T - 8_999);
327 +
        let older = file(root, "debug/deps/libdep-dddddddddddddddd.rlib", 256, T - 8_000);
328 +
        // A declared output, older still, that the run did not touch.
329 +
        let output = file(root, "release/tool", 128, T - 10_000);
330 +
        let all = [&app, &app_fp, &dep_rlib, &dep_fp, &script, &oldest, &oldest_fp, &older, &output];
331 +
        let before: u64 = all.iter().map(|p| bytes(p)).sum();
332 +
        // Room for all but the oldest unit.
333 +
        let want = before - bytes(&oldest) - bytes(&oldest_fp);
334 +
        let size = (want * 100).div_ceil(60);
335 +
        let p = prune(root, at(T), &["/release/tool".into()], size, before, 60).unwrap();
336 +
        assert!(p.fits(), "{p:?}");
337 +
        assert!(
338 +
            !oldest.exists() && !root.join("debug/.fingerprint/dep-cccccccccccccccc").exists(),
339 +
            "the oldest unit went, its emptied directory too"
340 +
        );
341 +
        assert!(older.exists(), "and pruning stopped under the cap: {p:?}");
342 +
        for kept in [&app, &app_fp, &dep_rlib, &dep_fp, &script, &output, &root.join("debug/app")] {
343 +
            assert!(kept.exists(), "{} stays", kept.display());
344 +
        }
345 +
        assert_eq!((p.files, p.freed, p.after), (2, p.before - want, want));
346 +
        // With no room at all: still only what the run did not use goes.
347 +
        let p = prune(root, at(T), &["release/tool".into()], 1, p.after, 60).unwrap();
348 +
        assert!(!older.exists() && !p.fits() && p.files == 1, "{p:?}");
349 +
        for kept in [&app, &dep_rlib, &dep_fp, &script, &output] {
350 +
            assert!(kept.exists(), "{} stays", kept.display());
351 +
        }
352 +
        assert!(root.join("debug/deps").is_dir(), "a directory still holding files stays");
353 +
    }
354 +
355 +
    #[test]
356 +
    fn reads_are_forgotten_so_the_next_is_seen() {
357 +
        let dir = tempfile::tempdir().unwrap();
358 +
        let p = file(dir.path(), "a/b", 1, 5_000);
359 +
        std::os::unix::fs::symlink("/etc/passwd", dir.path().join("link")).unwrap();
360 +
        assert_eq!(forget_reads(dir.path()).unwrap(), 1);
361 +
        let md = std::fs::metadata(&p).unwrap();
362 +
        assert_eq!((md.atime(), md.mtime()), (0, 5_000), "the mtime is kept");
363 +
        assert!(std::fs::symlink_metadata(dir.path().join("link")).unwrap().atime() > 0, "a link is not followed");
364 +
    }
365 +
}
modified CHANGELOG.md +2 −0
…
3 unchanged lines
4 4
5 5
## Unreleased
6 6
7 +
- **Build caches are pruned before they are kept (#72, spec §5.4).** A kept cache no longer grows until the job's disk is full. When a passing job's cache holds more than `cache_prune_percent` of its mount's size (runner config, default 60), the files the job did not read, write or share a cargo unit hash with are removed, oldest first, until it is under; one still over is not kept, and the next run starts empty. A microVM's guest init prunes its disk before it powers off; the runner prunes the syscall tier's directories. The job's log says how much was kept and pruned, or why the cache was not kept.
8 +
7 9
## 0.2.0 (2026-10-20)
8 10
9 11
- **Approval gates and environments (#57, #59, oxygenhub#235, spec §5.3.2, §5.3.3).** `approval { by "maintain"; others-only; expires "24h" }` on a job makes the hub hold it until a person with that role (write, maintain by default, or admin) approves it, and with `others-only` not the person who pushed. `approval { sealed #true }` asks for an approval signed with the approver's own key (`oxy seal --flow`), the default for a job that names an `environment "prod"`; the hub names the environment in the job's identity tokens only once it is approved. Both are part of the job's hash.
…
74 unchanged lines
modified Cargo.lock +2 −0
…
506 unchanged lines
507 507
 "blake3",
508 508
 "clap",
509 509
 "flow-core",
510 +
 "flow-init",
510 511
 "hex",
511 512
 "ignore",
512 513
 "jet",
…
41 unchanged lines
554 555
 "libc",
555 556
 "serde",
556 557
 "serde_json",
558 +
 "tempfile",
557 559
]
558 560
559 561
[[package]]
…
1482 unchanged lines
modified Cargo.toml +1 −0
…
12 unchanged lines
13 13
oxy-core = { path = "deps/oxygen/crates/oxy-core" }
14 14
flow-core = { path = "crates/flow-core" }
15 15
jet = { path = "crates/jet" }
16 +
flow-init = { path = "crates/flow-init" }
16 17
serde = { version = "1", features = ["derive"] }
17 18
serde_json = "1"
18 19
blake3 = "1"
…
7 unchanged lines
modified crates/flow-init/Cargo.toml +3 −0
…
8 unchanged lines
9 9
libc = "0.2"
10 10
serde.workspace = true
11 11
serde_json.workspace = true
12 +
13 +
[dev-dependencies]
14 +
tempfile.workspace = true
modified crates/flow-init/src/main.rs +62 −3
…
6 unchanged lines
7 7
//!
8 8
//! Frames, one per line on the console: `\x1eO ` stdout, `\x1eE ` stderr,
9 9
//! `\x1eX <code>` the exit code (128 + signal when killed), `\x1eF <text>`
10 -
//! a failure of init itself.
10 +
//! a failure of init itself, `\x1eC <json>` a build cache pruned before
11 +
//! the guest powers off (oxygenflow#72): `{"target": ..., "pruned":
12 +
//! flow_init::prune::Pruned}`, sent before the exit code.
11 13
14 +
use flow_init::prune;
12 15
use serde::Deserialize;
13 16
use std::ffi::CString;
14 17
use std::io::{Read, Write};
…
38 unchanged lines
53 56
    size_mib: u32,
54 57
    #[serde(default)]
55 58
    device: Option<String>,
59 +
    /// A build cache's disk (oxygenflow#72): mounted `strictatime`, so
60 +
    /// what the job reads is seen, and after a passing job pruned to this
61 +
    /// percentage of its size before the guest powers off.
62 +
    #[serde(default)]
63 +
    prune: Option<u8>,
64 +
    /// Paths on the disk, relative to it, that pruning leaves: the job's
65 +
    /// outputs, which the runner reads after the guest stops.
66 +
    #[serde(default)]
67 +
    keep: Vec<String>,
56 68
}
57 69
58 70
fn main() {
…
102 unchanged lines
161 173
        match (m.kind.as_str(), &m.device) {
162 174
            ("tmpfs", _) => mount("tmpfs", &target, "tmpfs", nosuid, &format!("size={}m,uid={uid},gid={gid},mode=755", m.size_mib))?,
163 175
            ("disk", Some(dev)) => {
164 -
                mount(dev, &target, "ext4", nosuid, "")?;
176 +
                let atime = if m.prune.is_some() { libc::MS_STRICTATIME | libc::MS_LAZYTIME } else { 0 };
177 +
                mount(dev, &target, "ext4", nosuid | atime, "")?;
165 178
                chown(&target, uid, gid)?;
166 179
            }
167 180
            (kind, _) => return Err(format!("unknown mount kind {kind}")),
…
33 unchanged lines
201 214
        return Err(format!("enter the rootfs: {}", std::io::Error::last_os_error()));
202 215
    }
203 216
217 +
    // What the job reads or writes on a cache disk from now on is what
218 +
    // this run used.
219 +
    let started = job.mounts.iter().any(|m| m.prune.is_some()).then(prune::start_clock);
204 220
    let argv: Vec<CString> = job.argv.iter().map(|a| cstr(a)).collect::<Result<_, _>>()?;
205 221
    // glibc's execvpe searches the caller's PATH, and PID 1 has none of its
206 222
    // own, so the program is found here, in the job's PATH, before forking.
…
52 unchanged lines
259 275
    }
260 276
    for r in readers {
261 277
        let _ = r.join();
278 +
    }
279 +
    let code = if libc::WIFEXITED(status) { libc::WEXITSTATUS(status) } else { 128 + libc::WTERMSIG(status) };
280 +
    // A failing job's disks are thrown away: only a passing one's are
281 +
    // worth pruning.
282 +
    if let (Some(started), 0) = (started, code) {
283 +
        prune_caches(console, &job.mounts, started);
284 +
    }
285 +
    Ok(code)
286 +
}
287 +
288 +
/// Prune each cache disk and say so to the host. Whatever the job left
289 +
/// running stops first, so nothing changes the disks meanwhile.
290 +
fn prune_caches(console: &Mutex<std::io::Stdout>, mounts: &[MountSpec], started: std::time::SystemTime) {
291 +
    unsafe {
292 +
        libc::kill(-1, libc::SIGKILL);
293 +
        let mut status = 0;
294 +
        while libc::waitpid(-1, &mut status, 0) > 0 {}
295 +
    }
296 +
    for m in mounts {
297 +
        let Some(percent) = m.prune else { continue };
298 +
        let root = std::path::Path::new(&m.target);
299 +
        let pruned = match prune::usage(root) {
300 +
            Ok((used, size)) => {
301 +
                let mut p = prune::prune(root, started, &m.keep, size, used, percent).unwrap_or_else(|e| prune::Pruned {
302 +
                    size,
303 +
                    percent,
304 +
                    before: used,
305 +
                    after: used,
306 +
                    unpruned: Some(format!("pruning failed: {e}")),
307 +
                    ..Default::default()
308 +
                });
309 +
                // What is left as the file system counts it.
310 +
                if p.freed > 0 {
311 +
                    unsafe { libc::sync() };
312 +
                    if let Ok((now, _)) = prune::usage(root) {
313 +
                        p.after = now;
314 +
                    }
315 +
                }
316 +
                p
317 +
            }
318 +
            Err(e) => prune::Pruned { percent, unpruned: Some(format!("statvfs {}: {e}", m.target)), ..Default::default() },
319 +
        };
320 +
        let msg = serde_json::json!({ "target": m.target, "pruned": pruned });
321 +
        frame(console, b'C', msg.to_string().as_bytes());
262 322
    }
263 -
    Ok(if libc::WIFEXITED(status) { libc::WEXITSTATUS(status) } else { 128 + libc::WTERMSIG(status) })
264 323
}
265 324
266 325
/// Frame a stream line by line; a long line is cut at 16 KiB.
…
23 unchanged lines
modified crates/flow/Cargo.toml +1 −0
…
11 unchanged lines
12 12
[dependencies]
13 13
flow-core.workspace = true
14 14
jet.workspace = true
15 +
flow-init.workspace = true
15 16
oxy-core.workspace = true
16 17
serde.workspace = true
17 18
serde_json.workspace = true
…
13 unchanged lines
modified crates/flow/src/cache.rs +259 −16
…
31 unchanged lines
32 32
//! tool that judges freshness by time (cargo) rebuilds only what changed;
33 33
//! a changed file keeps the time of its checkout, newer than anything on
34 34
//! the disk, so it is always rebuilt.
35 +
//!
36 +
//! A passing job's disk is pruned before it is kept (oxygenflow#72; see
37 +
//! [`flow_init::prune`]): over [`flow_init::prune::DEFAULT_PERCENT`] of
38 +
//! its size (the runner's `cache_prune_percent`), the files the job did
39 +
//! not use go, oldest first, and a cache still over it is not kept, so a
40 +
//! kept target never grows until the job's disk is full. A microVM's
41 +
//! guest prunes its image before it stops (the runner cannot look inside
42 +
//! an ext4 image without mounting it); the runner prunes a directory
43 +
//! itself, having set its files' atimes to the epoch before the job
44 +
//! ([`forget_reads`]) so that every read is seen. The job's log says what
45 +
//! was kept and pruned.
35 46
36 47
use anyhow::{Context, Result};
37 48
use flow_core::pipeline::Mount;
49 +
use flow_init::prune::Pruned;
38 50
use std::os::unix::fs::{MetadataExt, PermissionsExt};
39 51
use std::path::{Path, PathBuf};
40 52
use std::time::SystemTime;
…
49 unchanged lines
90 102
91 103
fn read_sources(disk: &Path) -> Option<Sources> {
92 104
    serde_json::from_slice(&std::fs::read(sources_of(disk)).ok()?).ok()
105 +
}
106 +
107 +
/// How a passing job's cache disk is pruned before it is kept
108 +
/// (oxygenflow#72).
109 +
pub struct Prune<'a> {
110 +
    /// When the job started ([`flow_init::prune::start_clock`]): what it
111 +
    /// read or wrote since is what it used.
112 +
    pub started: SystemTime,
113 +
    /// The share of the disk the kept cache may fill.
114 +
    pub percent: u8,
115 +
    /// What the guest did to a microVM's image, when it said.
116 +
    pub guest: Option<&'a Pruned>,
93 117
}
94 118
95 119
/// Where a job's cached disks come from and go back to.
…
144 unchanged lines
240 264
        Ok(())
241 265
    }
242 266
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<()> {
267 +
    /// After the job: a passing job's disk is pruned and kept, unless it
268 +
    /// is still over its cap; a failing one's is thrown away, so nothing
269 +
    /// it wrote reaches a later job. For a kept or refused cache, what to
270 +
    /// say in the job's log.
271 +
    #[allow(clippy::too_many_arguments)]
272 +
    pub fn settle(
273 +
        &self,
274 +
        ctx: &Ctx,
275 +
        m: &Mount,
276 +
        key: &str,
277 +
        disk: &Path,
278 +
        sources: &Sources,
279 +
        passed: bool,
280 +
        prune: &Prune,
281 +
    ) -> Result<Option<String>> {
246 282
        if passed {
247 -
            return self.keep(ctx, m, key, disk, sources);
283 +
            let Some(name) = m.cache.as_deref() else { return Ok(None) };
284 +
            let pruned = pruned(ctx.form, m, disk, prune)?;
285 +
            let line = describe(name, &pruned);
286 +
            if pruned.fits() {
287 +
                self.keep(ctx, m, key, disk, sources)?;
288 +
            } else {
289 +
                discard(ctx.form, disk)?;
290 +
            }
291 +
            return Ok(Some(line));
248 292
        }
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()))
293 +
        discard(ctx.form, disk)?;
294 +
        Ok(None)
254 295
    }
255 296
256 297
    /// The repository and job slots holding at least one disk, as
…
54 unchanged lines
311 352
            }
312 353
        }
313 354
        freed
355 +
    }
356 +
}
357 +
358 +
/// Throw a disk away.
359 +
fn discard(form: Form, disk: &Path) -> Result<()> {
360 +
    let gone = match form {
361 +
        Form::Dir => crate::workspace::remove_tree(disk),
362 +
        Form::Image => std::fs::remove_file(disk).or_else(|e| if e.kind() == std::io::ErrorKind::NotFound { Ok(()) } else { Err(e) }),
363 +
    };
364 +
    gone.with_context(|| format!("discard {}", disk.display()))
365 +
}
366 +
367 +
/// Before a syscall job: set the atime of every file of a directory it
368 +
/// took from the pool to the epoch, so that whatever the file system's
369 +
/// atime rule, the job's first read of each is seen when the directory is
370 +
/// pruned (oxygenflow#72). A microVM's guest mounts its disk
371 +
/// `strictatime` instead.
372 +
pub fn forget_reads(form: Form, disk: &Path) -> Result<()> {
373 +
    if form == Form::Dir && disk.is_dir() {
374 +
        flow_init::prune::forget_reads(disk).with_context(|| format!("set the atimes in {}", disk.display()))?;
375 +
    }
376 +
    Ok(())
377 +
}
378 +
379 +
/// Prune a passing job's disk, or say why it was not: a directory here,
380 +
/// an image by what its guest said (else only its size is measured).
381 +
fn pruned(form: Form, m: &Mount, disk: &Path, prune: &Prune) -> Result<Pruned> {
382 +
    let size = m.size_mib as u64 * (1 << 20);
383 +
    let percent = prune.percent;
384 +
    let unpruned = |before: u64, why: String| Pruned { size, percent, before, after: before, unpruned: Some(why), ..Pruned::default() };
385 +
    Ok(match form {
386 +
        Form::Dir => {
387 +
            let before = tree_bytes(disk);
388 +
            if !flow_init::prune::records_reads(disk) {
389 +
                unpruned(before, "this runner's file system records no reads (noatime), so what the job used cannot be told".into())
390 +
            } else {
391 +
                flow_init::prune::prune(disk, prune.started, &[], size, before, percent)
392 +
                    .unwrap_or_else(|e| unpruned(tree_bytes(disk), format!("pruning failed: {e}")))
393 +
            }
394 +
        }
395 +
        Form::Image => match prune.guest.filter(|g| g.size > 0) {
396 +
            Some(g) => g.clone(),
397 +
            None => {
398 +
                let (before, size) = image_usage(disk)?;
399 +
                let why = prune.guest.and_then(|g| g.unpruned.clone()).unwrap_or_else(|| "the guest did not prune it".into());
400 +
                Pruned { size, ..unpruned(before, why) }
401 +
            }
402 +
        },
403 +
    })
404 +
}
405 +
406 +
/// What an ext4 image holds and its size, as its superblock counts them.
407 +
fn image_usage(image: &Path) -> Result<(u64, u64)> {
408 +
    let out = std::process::Command::new("dumpe2fs").arg("-h").arg(image).output().context("run dumpe2fs (e2fsprogs)")?;
409 +
    if !out.status.success() {
410 +
        anyhow::bail!("dumpe2fs {}: {}", image.display(), String::from_utf8_lossy(&out.stderr).trim());
411 +
    }
412 +
    superblock_usage(&String::from_utf8_lossy(&out.stdout)).with_context(|| format!("dumpe2fs {}: no block counts", image.display()))
413 +
}
414 +
415 +
fn superblock_usage(header: &str) -> Option<(u64, u64)> {
416 +
    let field = |name: &str| -> Option<u64> {
417 +
        header.lines().find_map(|l| l.strip_prefix(name)?.strip_prefix(':')).and_then(|v| v.trim().parse().ok())
418 +
    };
419 +
    let (count, free, block) = (field("Block count")?, field("Free blocks")?, field("Block size")?);
420 +
    Some((count.saturating_sub(free) * block, count * block))
421 +
}
422 +
423 +
/// The job log's line for a passing job's cache: what was kept, or why
424 +
/// not (oxygenflow#72).
425 +
pub fn describe(name: &str, p: &Pruned) -> String {
426 +
    use crate::images::human;
427 +
    let cap = format!("{}% of its {} disk", p.percent, human(p.size));
428 +
    let pruned = format!("{} in {} file{} this run did not use", human(p.freed), p.files, if p.files == 1 { "" } else { "s" });
429 +
    if p.fits() && p.files > 0 {
430 +
        format!("cache {name}: kept {} of {} (pruned {pruned}; the cap is {cap})", human(p.after), human(p.before))
431 +
    } else if p.fits() {
432 +
        format!("cache {name}: kept {}, under its cap of {cap}", human(p.after))
433 +
    } else {
434 +
        let after = if p.files > 0 { format!("{} after pruning {pruned}", human(p.after)) } else { human(p.after) };
435 +
        let why = p.unpruned.as_ref().map(|u| format!(" ({u})")).unwrap_or_default();
436 +
        format!(
437 +
            "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",
438 +
            human(p.cap())
439 +
        )
314 440
    }
315 441
}
316 442
…
120 unchanged lines
437 563
    use super::*;
438 564
    use flow_core::pipeline::MountSource;
439 565
566 +
    /// Pruning that never prunes: the cap is the whole disk.
567 +
    fn keep_all() -> Prune<'static> {
568 +
        Prune { started: SystemTime::now(), percent: 100, guest: None }
569 +
    }
570 +
440 571
    fn mount(key: &[&str]) -> Mount {
441 572
        Mount {
442 573
            source: MountSource::Disk,
…
131 unchanged lines
574 705
        let old = SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_000_000);
575 706
        std::fs::File::open(w("a/debug/marker")).unwrap().set_modified(old).unwrap();
576 707
        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();
708 +
        pool.settle(&main, &m, &key, &w("a"), &built, true, &keep_all()).unwrap();
578 709
        assert!(!w("a").exists(), "moved, not copied");
579 710
        assert_eq!(
580 711
            pool.slot_times().into_keys().collect::<Vec<_>>(),
…
12 unchanged lines
593 724
        assert!(pool.slot_times().is_empty());
594 725
        // Kept again, and under a changed key the newest is taken all the same.
595 726
        std::fs::write(w("b/debug/marker"), b"SECOND").unwrap();
596 -
        pool.settle(&main, &m, &key, &w("b"), &built, true).unwrap();
727 +
        pool.settle(&main, &m, &key, &w("b"), &built, true, &keep_all()).unwrap();
597 728
        let key2 = Pool::key(&main, &m, |_| Some("lock-2".into()));
598 729
        let taken = pool.take(&main, &m, &key2, &w("d")).unwrap();
599 730
        assert!(taken.how.contains("older key"), "{}", taken.how);
600 731
        assert_eq!(std::fs::read(w("d/debug/marker")).unwrap(), b"SECOND");
601 -
        pool.settle(&main, &m, &key2, &w("d"), &built, true).unwrap();
732 +
        pool.settle(&main, &m, &key2, &w("d"), &built, true, &keep_all()).unwrap();
602 733
        // Another bookmark copies it, with its times, and main keeps its own.
603 734
        let feature = dir_ctx("feature");
604 735
        let taken = pool.take(&feature, &m, &key2, &w("e")).unwrap();
…
2 unchanged lines
607 738
        assert_eq!(std::fs::read(w("e/debug/marker")).unwrap(), b"SECOND");
608 739
        assert!(pool.take(&main, &m, &key2, &w("f")).is_some(), "main's is still there");
609 740
        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();
741 +
        pool.settle(&main, &m, &key2, &w("f"), &built, true, &keep_all()).unwrap();
611 742
        // A microVM job of the same slot never takes a directory, nor a
612 743
        // syscall job an image.
613 744
        let image = Ctx { form: Form::Image, ..main };
…
14 unchanged lines
628 759
        let m = mount(&[]);
629 760
        let main = dir_ctx("main");
630 761
        tree(&work.path().join("a"), b"GOOD", 1);
631 -
        pool.settle(&main, &m, "k", &work.path().join("a"), &Sources::new(), true).unwrap();
762 +
        pool.settle(&main, &m, "k", &work.path().join("a"), &Sources::new(), true, &keep_all()).unwrap();
632 763
        let taken = pool.take(&main, &m, "k", &work.path().join("b"));
633 764
        assert!(taken.is_some());
634 765
        // The job that took it fails after writing to it: not kept, gone.
635 766
        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();
767 +
        pool.settle(&main, &m, "k", &work.path().join("b"), &Sources::new(), false, &keep_all()).unwrap();
637 768
        assert!(!work.path().join("b").exists());
638 769
        assert!(pool.take(&main, &m, "k", &work.path().join("c")).is_none(), "nothing it wrote reaches a later job");
639 770
        assert!(pool.slot_times().is_empty());
640 771
        // A failed image is thrown away the same way.
641 772
        let image = Ctx { form: Form::Image, ..main };
642 773
        disk(&work.path().join("d.ext4"), b"BAD!");
643 -
        pool.settle(&image, &m, "k", &work.path().join("d.ext4"), &Sources::new(), false).unwrap();
774 +
        pool.settle(&image, &m, "k", &work.path().join("d.ext4"), &Sources::new(), false, &keep_all()).unwrap();
644 775
        assert!(!work.path().join("d.ext4").exists() && pool.slot_times().is_empty());
645 776
    }
646 777
…
70 unchanged lines
717 848
        assert_eq!(std::fs::metadata(b.join("open")).unwrap().mode() & 0o777, 0o755);
718 849
        assert_eq!(std::fs::read_link(b.join("link")).unwrap(), Path::new("/etc/passwd"), "a link is kept, not followed");
719 850
        assert_eq!(modified(&b.join("debug/marker")), when);
851 +
    }
852 +
    /// A file of `kib` KiB at `rel` under `root`, last read and written
853 +
    /// at `secs` (ahead of the clock, so the time it was changed, now,
854 +
    /// is before a job starting at [`T`]).
855 +
    fn aged(root: &Path, rel: &str, kib: usize, secs: u64) -> PathBuf {
856 +
        let p = root.join(rel);
857 +
        std::fs::create_dir_all(p.parent().unwrap()).unwrap();
858 +
        let bytes: Vec<u8> = (0..kib << 10).map(|i: usize| (i.wrapping_mul(2_654_435_761) >> 13) as u8).collect();
859 +
        std::fs::write(&p, bytes).unwrap();
860 +
        let t = SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(secs);
861 +
        std::fs::File::options()
862 +
            .write(true)
863 +
            .open(&p)
864 +
            .unwrap()
865 +
            .set_times(std::fs::FileTimes::new().set_accessed(t).set_modified(t))
866 +
            .unwrap();
867 +
        p
868 +
    }
869 +
870 +
    const T: u64 = 4_000_000_000;
871 +
872 +
    fn at_t() -> SystemTime {
873 +
        SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(T)
874 +
    }
875 +
876 +
    /// oxygenflow#72: a syscall job's directory is pruned of what the job
877 +
    /// did not use before it is kept, and the log says so; one still over
878 +
    /// its cap is not kept at all.
879 +
    #[test]
880 +
    fn a_directory_is_pruned_of_what_the_job_did_not_use_before_it_is_kept() {
881 +
        let home = tempfile::tempdir().unwrap();
882 +
        let work = tempfile::tempdir().unwrap();
883 +
        let pool = Pool::new(home.path(), None);
884 +
        let main = dir_ctx("main");
885 +
        let mut m = mount(&[]);
886 +
        m.size_mib = 8;
887 +
        let a = work.path().join("a");
888 +
        // This run's build, and an older lockfile's that it did not touch.
889 +
        aged(&a, "debug/deps/libnew-aaaaaaaaaaaaaaaa.rlib", 2 << 10, T + 10);
890 +
        aged(&a, "debug/deps/libold-bbbbbbbbbbbbbbbb.rlib", 3 << 10, T - 1_000);
891 +
        let prune = Prune { started: at_t(), percent: 50, guest: None };
892 +
        let line = pool.settle(&main, &m, "k", &a, &Sources::new(), true, &prune).unwrap().unwrap();
893 +
        assert_eq!(
894 +
            line,
895 +
            "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)"
896 +
        );
897 +
        let b = work.path().join("b");
898 +
        assert!(pool.take(&main, &m, "k", &b).unwrap().how.contains("same key"));
899 +
        assert!(b.join("debug/deps/libnew-aaaaaaaaaaaaaaaa.rlib").is_file() && !b.join("debug/deps/libold-bbbbbbbbbbbbbbbb.rlib").exists());
900 +
        // Before the next job its reads are forgotten, its writes not.
901 +
        forget_reads(Form::Dir, &b).unwrap();
902 +
        let md = std::fs::metadata(b.join("debug/deps/libnew-aaaaaaaaaaaaaaaa.rlib")).unwrap();
903 +
        assert_eq!((md.atime(), md.mtime()), (0, (T + 10) as i64));
904 +
        // This run's own files over the cap: nothing it used goes, and the
905 +
        // cache is not kept, so the next run starts empty.
906 +
        aged(&b, "debug/deps/libbig-cccccccccccccccc.rlib", 3 << 10, T + 20);
907 +
        let line = pool.settle(&main, &m, "k", &b, &Sources::new(), true, &prune).unwrap().unwrap();
908 +
        assert!(
909 +
            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"),
910 +
            "{line}"
911 +
        );
912 +
        assert!(!b.exists() && pool.take(&main, &m, "k", &work.path().join("c")).is_none());
913 +
    }
914 +
915 +
    /// oxygenflow#72: a microVM's image is kept or not by what its guest
916 +
    /// said; without word from it, by what its superblock says is in use.
917 +
    #[test]
918 +
    fn an_image_is_kept_by_what_its_guest_pruned_or_else_by_its_size() {
919 +
        let home = tempfile::tempdir().unwrap();
920 +
        let work = tempfile::tempdir().unwrap();
921 +
        let pool = Pool::new(home.path(), None);
922 +
        let ctx = Ctx { repo: "r", bookmark: "main", job: "j", image: "i", form: Form::Image };
923 +
        let mut m = mount(&[]);
924 +
        m.size_mib = 16;
925 +
        let fits = Pruned { size: 16 << 20, percent: 60, before: 12 << 20, after: 6 << 20, freed: 6 << 20, files: 40, unpruned: None };
926 +
        let over = Pruned { after: 11 << 20, ..fits.clone() };
927 +
        disk(&work.path().join("a.ext4"), b"KEEP");
928 +
        let line = pool
929 +
            .settle(&ctx, &m, "k", &work.path().join("a.ext4"), &Sources::new(), true, &Prune { guest: Some(&fits), ..keep_all() })
930 +
            .unwrap()
931 +
            .unwrap();
932 +
        assert_eq!(
933 +
            line,
934 +
            "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)"
935 +
        );
936 +
        assert!(pool.take(&ctx, &m, "k", &work.path().join("b.ext4")).is_none(), "a 1 MiB disk is not the mount's 16 MiB");
937 +
        disk(&work.path().join("c.ext4"), b"OVER");
938 +
        let line = pool
939 +
            .settle(&ctx, &m, "k", &work.path().join("c.ext4"), &Sources::new(), true, &Prune { guest: Some(&over), ..keep_all() })
940 +
            .unwrap()
941 +
            .unwrap();
942 +
        assert!(line.starts_with("cache target: not kept: 11 MiB after pruning 6 MiB in 40 files"), "{line}");
943 +
        assert!(!work.path().join("c.ext4").exists());
944 +
        // No word from the guest: an empty file system is well under.
945 +
        let img = work.path().join("d.ext4");
946 +
        crate::workspace::empty_ext4(&img, 16, "scratch").unwrap();
947 +
        let line = pool.settle(&ctx, &m, "k", &img, &Sources::new(), true, &Prune { percent: 60, ..keep_all() }).unwrap().unwrap();
948 +
        assert!(line.starts_with("cache target: kept ") && line.contains("under its cap of 60% of its 16 MiB disk"), "{line}");
949 +
        assert!(pool.take(&ctx, &m, "k", &work.path().join("e.ext4")).is_some());
950 +
        // ... and one nearly all in use is not kept, saying why.
951 +
        let line = pool
952 +
            .settle(&ctx, &m, "k", &work.path().join("e.ext4"), &Sources::new(), true, &Prune { percent: 1, ..keep_all() })
953 +
            .unwrap()
954 +
            .unwrap();
955 +
        assert!(line.contains("not kept") && line.contains("(the guest did not prune it)"), "{line}");
956 +
    }
957 +
958 +
    #[test]
959 +
    fn the_superblock_says_what_is_in_use() {
960 +
        let h = "Filesystem volume name:   scratch\nBlock count:              4096\nFree blocks:              1024\nBlock size:               4096\n";
961 +
        assert_eq!(superblock_usage(h), Some((3072 * 4096, 4096 * 4096)));
962 +
        assert_eq!(superblock_usage("Block count: 1\n"), None);
720 963
    }
721 964
}
modified crates/flow/src/home.rs +5 −0
…
63 unchanged lines
64 64
    /// The build cache's size limit in GiB (#28; default 64).
65 65
    #[serde(default)]
66 66
    pub cache_gb: Option<u64>,
67 +
    /// The share of its disk, in percent, a kept build cache may fill
68 +
    /// (oxygenflow#72; default 60): over it, the files the job did not use
69 +
    /// are pruned, oldest first, and a cache still over it is not kept.
70 +
    #[serde(default)]
71 +
    pub cache_prune_percent: Option<u8>,
67 72
    /// Let microVM jobs reach the hosts their sandbox allows, through this
68 73
    /// runner (oxygenflow#55); off unless set. The runner then offers the
69 74
    /// `egress` label, which such jobs need.
…
207 unchanged lines
modified crates/flow/src/microvm.rs +60 −6
…
21 unchanged lines
22 22
    pub cancelled: bool,
23 23
    /// The most memory the sandbox held, in MiB (oxygenflow#41).
24 24
    pub peak_memory_mib: Option<u64>,
25 +
    /// Each cache disk the guest pruned before it stopped, by its target
26 +
    /// (oxygenflow#72); the syscall tier's are pruned by the runner.
27 +
    pub pruned: Vec<(String, flow_init::prune::Pruned)>,
25 28
}
26 29
27 30
/// Where a job's records go, and whether it is still wanted: asked a few
…
55 unchanged lines
83 86
    let rootfs = home.verified(&rootfs, &sandbox.rootfs)?;
84 87
    let environment = environment(home, sandbox, job)?;
85 88
86 -
    let mut spec = guest_spec(job);
89 +
    let percent = home.config.runner.as_ref().and_then(|r| r.cache_prune_percent).unwrap_or(flow_init::prune::DEFAULT_PERCENT);
90 +
    let mut spec = guest_spec(job, percent);
87 91
    spec["loopback"] = serde_json::Value::Bool(sandbox.network == flow_core::Network::Loopback);
88 92
    // Allow-listed egress (oxygenflow#55): a vsock device and the runner's
89 93
    // proxy behind it; the guest relays 127.0.0.1:3128 to it.
…
93 unchanged lines
183 187
    // The VMM's peak resident memory is the guest memory it touched: read
184 188
    // while it runs, since it is gone once it exits.
185 189
    let mut peak_memory_mib = None;
190 +
    let mut pruned = Vec::new();
186 191
    loop {
187 192
        if let Some(m) = crate::facts::peak_rss_mib(child.id()) {
188 193
            peak_memory_mib = Some(m);
…
15 unchanged lines
204 209
                        "E" => sink.record(rec("err", body)),
205 210
                        "X" => exit_code = body.trim().parse().ok(),
206 211
                        "F" => sink.record(rec("flow", &format!("guest init failed: {body}"))),
212 +
                        "C" => match pruned_frame(body) {
213 +
                            Some(p) => pruned.push(p),
214 +
                            None => sink.record(rec("console", &text)),
215 +
                        },
207 216
                        _ => sink.record(rec("console", &text)),
208 217
                    }
209 218
                } else if !text.is_empty() {
…
36 unchanged lines
246 255
        timed_out,
247 256
        cancelled,
248 257
        peak_memory_mib,
258 +
        pruned,
249 259
    })
250 260
}
251 261
262 +
/// A `\x1eC` frame: a cache disk the guest pruned (oxygenflow#72).
263 +
fn pruned_frame(body: &str) -> Option<(String, flow_init::prune::Pruned)> {
264 +
    #[derive(serde::Deserialize)]
265 +
    struct Frame {
266 +
        target: String,
267 +
        pruned: flow_init::prune::Pruned,
268 +
    }
269 +
    let f: Frame = serde_json::from_str(body).ok()?;
270 +
    Some((f.target, f.pruned))
271 +
}
272 +
252 273
/// The proxy variables a job with egress gets, in both spellings tools
253 274
/// read, unless the job sets them itself.
254 275
fn add_proxy_env(spec: &mut serde_json::Value) {
…
14 unchanged lines
269 290
}
270 291
271 292
/// What `flow-init` reads from `/job.json`: argv, env, and the mounts
272 -
/// with the device each disk mount is attached as (vdc onwards).
273 -
pub fn guest_spec(job: &Job) -> serde_json::Value {
293 +
/// with the device each disk mount is attached as (vdc onwards). A cache
294 +
/// disk is pruned to `percent` of its size after a passing job
295 +
/// (oxygenflow#72), leaving the job's outputs on it.
296 +
pub fn guest_spec(job: &Job, percent: u8) -> serde_json::Value {
274 297
    let mut next_disk = b'c';
275 298
    let mounts: Vec<serde_json::Value> = job
276 299
        .mounts
…
3 unchanged lines
280 303
            MountSource::Disk => {
281 304
                let dev = format!("/dev/vd{}", next_disk as char);
282 305
                next_disk += 1;
283 -
                serde_json::json!({ "kind": "disk", "target": m.target, "size_mib": m.size_mib, "device": dev })
306 +
                let mut disk = serde_json::json!({ "kind": "disk", "target": m.target, "size_mib": m.size_mib, "device": dev });
307 +
                if m.cache.is_some() {
308 +
                    let prefix = format!("{}/", m.target);
309 +
                    let keep: Vec<&str> = job.outputs.iter().filter_map(|o| o.path.strip_prefix(&prefix)).collect();
310 +
                    disk["prune"] = percent.into();
311 +
                    disk["keep"] = keep.into();
312 +
                }
313 +
                disk
284 314
            }
285 315
        })
286 316
        .collect();
…
41 unchanged lines
328 358
            "name": "t", "sandbox": "s", "scope": [], "depends_on": [], "env": {"A": "1"}, "exec": ["true"]
329 359
        }))
330 360
        .unwrap();
331 -
        assert_eq!(guest_spec(&job)["mounts"], serde_json::json!([]), "no mounts");
361 +
        assert_eq!(guest_spec(&job, 60)["mounts"], serde_json::json!([]), "no mounts");
332 362
        job.mounts = vec![
333 363
            Mount { source: MountSource::Disk, target: "/src/target".into(), size_mib: 8192, cache: None, key: vec![] },
334 364
            Mount { source: MountSource::Tmpfs, target: "/tmp/cache".into(), size_mib: 256, cache: None, key: vec![] },
335 365
            Mount { source: MountSource::Disk, target: "/tmp/z".into(), size_mib: 64, cache: None, key: vec![] },
336 366
        ];
337 -
        let spec = guest_spec(&job);
367 +
        let spec = guest_spec(&job, 60);
338 368
        assert_eq!(spec["mounts"][0], serde_json::json!({"kind": "disk", "target": "/src/target", "size_mib": 8192, "device": "/dev/vdc"}));
339 369
        assert_eq!(spec["mounts"][1], serde_json::json!({"kind": "tmpfs", "target": "/tmp/cache", "size_mib": 256}));
340 370
        assert_eq!(spec["mounts"][2]["device"], "/dev/vdd");
341 371
        assert_eq!(spec["env"], serde_json::json!([["A", "1"]]));
372 +
    }
373 +
374 +
    /// oxygenflow#72: a cache disk is pruned in the guest, leaving the
375 +
    /// job's outputs on it; a plain disk is not, and the guest's report
376 +
    /// comes back as a frame.
377 +
    #[test]
378 +
    fn a_cache_disk_is_pruned_in_the_guest_but_its_outputs_stay() {
379 +
        let mut job: Job = serde_json::from_value(serde_json::json!({
380 +
            "name": "t", "sandbox": "s", "scope": [], "depends_on": [], "env": {}, "exec": ["true"],
381 +
            "outputs": [{"name": "bin", "path": "/src/target/release/tool"}, {"name": "w", "path": "/src/w"}]
382 +
        }))
383 +
        .unwrap();
384 +
        job.mounts = vec![
385 +
            Mount { source: MountSource::Disk, target: "/src/target".into(), size_mib: 8192, cache: Some("t".into()), key: vec![] },
386 +
            Mount { source: MountSource::Disk, target: "/tmp/z".into(), size_mib: 64, cache: None, key: vec![] },
387 +
        ];
388 +
        let spec = guest_spec(&job, 45);
389 +
        assert_eq!((spec["mounts"][0]["prune"].as_u64(), &spec["mounts"][0]["keep"]), (Some(45), &serde_json::json!(["release/tool"])));
390 +
        assert!(spec["mounts"][1].get("prune").is_none() && spec["mounts"][1].get("keep").is_none());
391 +
        let (target, p) =
392 +
            pruned_frame(r#"{"target":"/src/target","pruned":{"size":100,"percent":60,"before":90,"after":50,"freed":40,"files":3}}"#)
393 +
                .unwrap();
394 +
        assert_eq!((target.as_str(), p.after, p.cap(), p.fits(), p.unpruned), ("/src/target", 50, 60, true, None));
395 +
        assert!(pruned_frame("not json").is_none());
342 396
    }
343 397
}
modified crates/flow/src/runner.rs +20 −4
…
1001 unchanged lines
1002 1002
                None => format!("cache {name}: none kept yet, an empty disk"),
1003 1003
            };
1004 1004
            note(&mut sink, format!("{} ({})", how, m.target));
1005 +
            // What the job reads is seen when the disk is pruned (#72).
1006 +
            if let Err(e) = crate::cache::forget_reads(form, &crate::cache::disk_path(form, &scratch, i)) {
1007 +
                note(&mut sink, format!("cache {name}: {e:#}"));
1008 +
            }
1005 1009
            cached.push((i, key, how));
1006 1010
        }
1011 +
        // What the job reads or writes on its cache from now on is what it
1012 +
        // used; the rest may be pruned before the cache is kept (#72).
1013 +
        let cache_started = if cached.is_empty() { std::time::SystemTime::now() } else { flow_init::prune::start_clock() };
1007 1014
        let sources = if cached.is_empty() {
1008 1015
            crate::cache::Sources::new()
1009 1016
        } else {
…
131 unchanged lines
1141 1148
        };
1142 1149
        // Measured before a cached disk leaves the scratch directory.
1143 1150
        let peak_disk_mib = crate::facts::allocated_mib(&scratch);
1144 -
        // A passing job's cached disks are kept; a failing job's are
1145 -
        // thrown away, so nothing it wrote reaches a later job.
1151 +
        // A passing job's cached disks are pruned and kept (#72); a
1152 +
        // failing job's are thrown away, so nothing it wrote reaches a
1153 +
        // later job.
1146 1154
        let kept = passed && output_error.is_none();
1147 1155
        for (i, key, _) in &cached {
1148 1156
            let disk = crate::cache::disk_path(form, &scratch, *i);
1149 -
            if let Err(e) = pool.settle(&cache_ctx, disk_mounts[*i], key, &disk, &sources, kept) {
1150 -
                note(&mut sink, format!("cache: not {}: {e:#}", if kept { "kept" } else { "discarded" }));
1157 +
            let m = disk_mounts[*i];
1158 +
            let prune = crate::cache::Prune {
1159 +
                started: cache_started,
1160 +
                percent: cfg.cache_prune_percent.unwrap_or(flow_init::prune::DEFAULT_PERCENT),
1161 +
                guest: outcome.pruned.iter().find(|(t, _)| *t == m.target).map(|(_, p)| p),
1162 +
            };
1163 +
            match pool.settle(&cache_ctx, m, key, &disk, &sources, kept, &prune) {
1164 +
                Ok(Some(line)) => note(&mut sink, line),
1165 +
                Ok(None) => {}
1166 +
                Err(e) => note(&mut sink, format!("cache: not {}: {e:#}", if kept { "kept" } else { "discarded" })),
1151 1167
            }
1152 1168
        }
1153 1169
        Ok(Finish {
…
586 unchanged lines
modified crates/flow/src/syscall.rs +1 −0
…
285 unchanged lines
286 286
        timed_out,
287 287
        cancelled,
288 288
        peak_memory_mib,
289 +
        pruned: Vec::new(),
289 290
    })
290 291
}
291 292
…
573 unchanged lines
modified docs/m0.md +7 −3
…
146 unchanged lines
147 147
   variable the pipeline sets keeps its value. They are part of the job's
148 148
   environment, so of its receipt.
149 149
5. `flow-init` frames the child's output on the serial console as
150 -
   `\x1eO <line>` (stdout) and `\x1eE <line>` (stderr), then
151 -
   `\x1eX <code>`, and powers off. The kernel runs `quiet` so the host sees
150 +
   `\x1eO <line>` (stdout) and `\x1eE <line>` (stderr). After a passing
151 +
   job it prunes each build cache disk (mounted `strictatime`) of the
152 +
   files the job did not use when it is over the runner's
153 +
   `cache_prune_percent` of its size, and reports each as
154 +
   `\x1eC <json>` (spec §5.4, oxygenflow#72); then `\x1eX <code>`, and
155 +
   it powers off. The kernel runs `quiet` so the host sees
152 156
   only frames; anything else is kept as `console` records.
153 157
6. The host reads frames, streams them as log records, and kills the VM
154 158
   at the timeout (the guest cannot extend it). Scratch files are deleted
…
20 unchanged lines
175 179
- **Root:** PID 1 binds the unpacked tree read-only as the new root, the workspace at `/src`, tmpfs at `/tmp`, `/run` and `/run/secrets`, a fresh `/proc`, and a `/dev` of `null`, `zero`, `full`, `random` and `urandom` with `/dev/shm`. It brings `lo` up for `network "loopback"` and `"egress"` (with the vsock relay of spec §4.3.1 for the latter), pivots into the root and detaches the old one, so the host's files, the spec and the scratch directory are out of reach.
176 180
- **Job:** the job gets no new privileges, then Landlock (read and execute everywhere, write beneath `/src`, `/tmp`, `/dev` and its mounts), then the seccomp filter, then `execve` with exactly the job's environment, found the way `flow-init` finds it (the two share that code). The filter also refuses io_uring (which could open sockets past it), the new mount calls, kexec, swap and reboot.
177 181
- **Unpacking:** `debugfs rdump` unpacks the image into `trees/<hash>` in the home once, without mount rights. Set-id bits and device nodes are dropped.
178 -
- **Ephemeral mounts:** a `tmpfs` mount is a tmpfs of its size. An `ephemeral://disk` mount is a scratch directory, so its size is not enforced in this tier. One that names a `cache` starts from a directory a passing job of the slot kept, and goes back into the runner's cache pool when the job passes (spec §5.4, oxygenflow#70).
182 +
- **Ephemeral mounts:** a `tmpfs` mount is a tmpfs of its size. An `ephemeral://disk` mount is a scratch directory, so its size is not enforced in this tier. One that names a `cache` starts from a directory a passing job of the slot kept, and goes back into the runner's cache pool when the job passes (spec §5.4, oxygenflow#70), pruned first by the runner of the files the job did not use when over the cap (oxygenflow#72).
179 183
- **Artifacts:** they are read from the workspace and mount directories without following links.
180 184
- **Receipts:** the environment's runtime is `syscall`. Its parts are the `flow` binary's BLAKE3, the host kernel's release, memory, CPUs and mounts.
181 185
- **Limits:** the job's cgroup gets `memory.max`, `memory.swap.max 0` and `cpu.max`. A job killed for memory is told so in its log. At the end the group is killed and removed.
…
74 unchanged lines
modified docs/spec.md +23 −0
…
415 unchanged lines
416 416
  hash differs from a cold build's.
417 417
- **Bounded.** The pool keeps under `cache_gb` in the runner's config
418 418
  (default 64 GiB) by dropping the disks used longest ago.
419 +
- **Pruned before it is kept** (oxygenflow#72). A kept disk would
420 +
  otherwise carry every build's files (each crate under every hash a
421 +
  lockfile or feature set gave it) until the job's disk is full. When a
422 +
  passing job's cache holds more than `cache_prune_percent` of its
423 +
  mount's `size` (runner config, default 60), the files the job did not
424 +
  use are removed, used longest ago first, until it is under that cap; one
425 +
  still over it is not kept, so the next run starts empty. A file was
426 +
  used if it was read, written or changed since the job started, or it
427 +
  shares a unit hash with one that was (a 16-digit hex word after a `-`
428 +
  in its path, as cargo names a unit's fingerprint, rlibs, binaries and
429 +
  build-script output: a crate cargo found fresh is only looked at, but
430 +
  its fingerprint is read), or is another link of one, or lies under one
431 +
  of the job's outputs. A microVM's guest (`flow-init`) prunes its disks
432 +
  before it powers off, having mounted them `strictatime` so every read
433 +
  is seen, and tells the runner what it kept (`\x1eC` frames); without
434 +
  word from it the runner only measures the image (`dumpe2fs`) and keeps
435 +
  it if it is under the cap. The syscall tier's runner prunes a directory
436 +
  itself, having set each file's atime to the epoch when the job took it,
437 +
  so even a `relatime` file system records the job's first read of each;
438 +
  on a `noatime` one nothing is pruned and only the size decides. The
439 +
  job's log says what happened: `cache target: kept 12.3 GiB of 23.1 GiB
440 +
  (pruned 10.8 GiB in 4113 files this run did not use; the cap is 60% of
441 +
  its 24.0 GiB disk)`, or `not kept:` and why.
419 442
- **Unchanged sources keep their times** (oxygenflow#52). Beside each kept
420 443
  disk the pool keeps its sources: each workspace file's content hash and
421 444
  the mtime it had, in whole seconds (what the workspace image keeps). A
…
352 unchanged lines