OXYGENHUB

Coverage / crates/flow/src/workspace.rs 77.9%

197 of 253 lines run, 27 of 59 functions · run #208 on main · run missed

1
//! A job's workspace: the files of the state being built, written to a
2
//! scratch directory and packed into an ext4 image for the microVM
3
//! (docs/m0.md §2). Returns each file's content hash for the scope hash.
4
5
use anyhow::{bail, Context, Result};
6
use flow_core::Hash;
7
use oxy_core::hash::ChangeHash;
8
use oxy_core::objects::NodeKind;
9
use oxy_core::repo::Repo;
10
use oxy_core::state::FileContent;
11
use std::path::Path;
12
13
pub type Files = Vec<(String, Hash)>;
14
15
/// Write the state `heads` of `repo` under `dir`.
160×
pub fn export_state(repo: &Repo, heads: &[ChangeHash], dir: &Path) -> Result<Files> {
170×
    let snapshot = repo.load_state(heads).context("load the state")?.materialize();
180×
    let mut files = Vec::new();
190×
    for (path, f) in &snapshot.files {
200×
        if path.split('/').any(|p| p.is_empty() || p == "." || p == ".." || p == ".oxy") {
210×
            continue;
220×
        }
230×
        let target = dir.join(path);
240×
        match (&f.kind, &f.content) {
250×
            (NodeKind::Dir, _) | (_, FileContent::Dir) => std::fs::create_dir_all(&target)?,
260×
            (NodeKind::Subrepo, _) => {}
270×
            (kind, content) => {
280×
                let bytes = match content {
290×
                    FileContent::Text { bytes, .. } => bytes.clone(),
300×
                    FileContent::Opaque { blobs } => match blobs.first() {
310×
                        Some(b) => repo.read_blob(b)?,
320×
                        None => Vec::new(),
33
                    },
340×
                    FileContent::Dir => unreachable!(),
35
                };
360×
                if let Some(parent) = target.parent() {
370×
                    std::fs::create_dir_all(parent)?;
380×
                }
390×
                if *kind == NodeKind::Symlink {
400×
                    let to = String::from_utf8_lossy(&bytes).into_owned();
410×
                    crate::portable::symlink(Path::new(&to), &target).with_context(|| format!("symlink {path}"))?;
42
                } else {
430×
                    std::fs::write(&target, &bytes).with_context(|| format!("write {path}"))?;
440×
                    if f.mode.exec {
450×
                        crate::portable::set_mode(&target, 0o755)?;
460×
                    }
47
                }
480×
                files.push((path.clone(), *blake3::hash(&bytes).as_bytes()));
49
            }
50
        }
51
    }
520×
    Ok(files)
530×
}
54
55
/// Copy a working directory (without `.oxy` and what `.oxyignore` and
56
/// `.gitignore` exclude) under `dir`, for `flow run`.
5714×
pub fn copy_tree(src: &Path, dir: &Path) -> Result<Files> {
5814×
    let mut files = Vec::new();
59
    // A file at the top would otherwise find no directory to go in.
6014×
    std::fs::create_dir_all(dir)?;
6114×
    let walk = ignore::WalkBuilder::new(src)
6214×
        .hidden(false)
6314×
        .git_ignore(true)
6414×
        .require_git(false)
6514×
        .add_custom_ignore_filename(".oxyignore")
66168×
        .filter_entry(|e| e.file_name() != ".oxy" && e.file_name() != ".git")
6714×
        .build();
68182×
    for entry in walk {
69182×
        let entry = entry?;
70182×
        let rel = entry.path().strip_prefix(src)?;
71182×
        if rel.as_os_str().is_empty() {
7214×
            continue;
73168×
        }
74168×
        let rel_s = rel.to_string_lossy().replace('\\', "/");
75168×
        let target = dir.join(rel);
76168×
        let ft = entry.file_type().context("file type")?;
77168×
        if ft.is_dir() {
7842×
            std::fs::create_dir_all(&target)?;
79126×
        } else if ft.is_symlink() {
800×
            let to = std::fs::read_link(entry.path())?;
810×
            crate::portable::symlink(&to, &target)?;
820×
            files.push((rel_s, *blake3::hash(to.to_string_lossy().as_bytes()).as_bytes()));
83126×
        } else if ft.is_file() {
84126×
            std::fs::copy(entry.path(), &target)?;
85126×
            files.push((rel_s, *blake3::hash(&std::fs::read(entry.path())?).as_bytes()));
860×
        }
87
    }
8814×
    Ok(files)
8914×
}
90
91
/// Pack `dir` into an ext4 image whose root belongs to this user, so the
92
/// job (which runs as the owner of `/src`) can write to it.
932×
pub fn image(dir: &Path, out: &Path) -> Result<()> {
942×
    let used: u64 = ignore::WalkBuilder::new(dir)
952×
        .hidden(false)
962×
        .ignore(false)
972×
        .git_ignore(false)
982×
        .build()
992×
        .flatten()
10013×
        .filter_map(|e| e.metadata().ok())
10113×
        .map(|m| m.len().max(4096))
1022×
        .sum();
1032×
    let mib = used * 12 / 10 / (1 << 20) + 512;
1042×
    let (uid, gid) = owner();
1052×
    let status = std::process::Command::new("mkfs.ext4")
1062×
        .args(["-q", "-F", "-L", "workspace", "-E"])
1072×
        .arg(format!("root_owner={uid}:{gid}"))
1082×
        .arg("-d")
1092×
        .arg(dir)
1102×
        .arg(out)
1112×
        .arg(format!("{mib}M"))
1122×
        .status()
1132×
        .context("run mkfs.ext4 (e2fsprogs 1.43 or later)")?;
1142×
    if !status.success() {
1150×
        bail!("mkfs.ext4 failed for the workspace image");
1162×
    }
1172×
    Ok(())
1182×
}
119
120
/// Give the files of `dir` whose content a cached disk was built from
121
/// the mtime they had then (oxygenflow#52), and describe this checkout's
122
/// sources for the disk to keep: unchanged files are fresh to a build
123
/// tool that judges by time, changed ones newer than anything it built.
124
/// Whole seconds, as the workspace image keeps them. Returns the sources
125
/// and how many files got their time back.
1265×
pub fn restore_times(dir: &Path, files: &Files, before: Option<&crate::cache::Sources>) -> (crate::cache::Sources, usize) {
1275×
    let mut out = crate::cache::Sources::new();
1285×
    let mut restored = 0;
12924×
    for (path, hash) in files {
13024×
        let p = dir.join(path);
13124×
        let Ok(md) = std::fs::symlink_metadata(&p) else { continue };
13224×
        if !md.is_file() {
1335×
            continue;
13419×
        }
13519×
        let hex = hex::encode(hash);
13619×
        let kept = before.and_then(|b| b.get(path)).filter(|(h, _)| *h == hex).map(|(_, t)| *t);
13719×
        let t = match kept {
1389×
            Some(t) => {
1399×
                let at = std::time::UNIX_EPOCH + std::time::Duration::from_secs(t.max(0) as u64);
1409×
                if std::fs::File::options().write(true).open(&p).and_then(|f| f.set_modified(at)).is_err() {
1410×
                    continue;
1429×
                }
1439×
                restored += 1;
1449×
                t
145
            }
14610×
            None => md.modified().ok().and_then(|m| m.duration_since(std::time::UNIX_EPOCH).ok()).map(|d| d.as_secs() as i64).unwrap_or(0),
147
        };
14819×
        out.insert(path.clone(), (hex, t));
149
    }
150
    // Directories too, after their files (#52): a build script's
151
    // `rerun-if-changed` on a directory looks at its time as well. A
152
    // directory whose listing is unchanged gets its old time back; one that
153
    // gained or lost an entry keeps its checkout's, so the change shows.
1545×
    restore_dir_times(dir, Path::new(""), before, &mut out, &mut restored);
1555×
    (out, restored)
1565×
}
157
158
/// The key a directory's entry has in the sources: its path and a slash.
1598×
fn dir_key(rel: &Path) -> String {
1608×
    format!("{}/", rel.display())
1618×
}
162
1638×
fn restore_dir_times(
1648×
    root: &Path,
1658×
    rel: &Path,
1668×
    before: Option<&crate::cache::Sources>,
1678×
    out: &mut crate::cache::Sources,
1688×
    restored: &mut usize,
1698×
) {
1708×
    let here = root.join(rel);
1718×
    let Ok(entries) = std::fs::read_dir(&here) else { return };
1728×
    let mut names: Vec<(String, bool)> = Vec::new();
17327×
    for e in entries.flatten() {
17427×
        let is_dir = e.file_type().is_ok_and(|t| t.is_dir());
17527×
        names.push((e.file_name().to_string_lossy().into_owned(), is_dir));
176
    }
1778×
    names.sort();
17827×
    for (name, is_dir) in &names {
17927×
        if *is_dir {
1803×
            restore_dir_times(root, &rel.join(name), before, out, restored);
18124×
        }
182
    }
1838×
    let mut h = blake3::Hasher::new();
18427×
    for (name, is_dir) in &names {
18527×
        h.update(name.as_bytes());
18627×
        h.update(if *is_dir { b"/\n".as_slice() } else { b"\n".as_slice() });
187
    }
1888×
    let hex = h.finalize().to_hex().to_string();
1898×
    let key = dir_key(rel);
1908×
    let kept = before.and_then(|b| b.get(&key)).filter(|(h, _)| *h == hex).map(|(_, t)| *t);
1918×
    let t = match kept {
1923×
        Some(t) => {
1933×
            let at = std::time::UNIX_EPOCH + std::time::Duration::from_secs(t.max(0) as u64);
1943×
            if std::fs::File::open(&here).and_then(|f| f.set_modified(at)).is_err() {
1950×
                return;
1963×
            }
1973×
            *restored += 1;
1983×
            t
199
        }
2005×
        None => std::fs::metadata(&here)
2015×
            .ok()
2025×
            .and_then(|m| m.modified().ok())
2035×
            .and_then(|m| m.duration_since(std::time::UNIX_EPOCH).ok())
2045×
            .map(|d| d.as_secs() as i64)
2055×
            .unwrap_or(0),
206
    };
2078×
    out.insert(key, (hex, t));
2088×
}
209
210
/// An empty ext4 filesystem of `mib` MiB at `out`, owned by this user like
211
/// the workspace. mkfs.ext4 creates the file sparse, so only what the job
212
/// writes takes space on the runner.
2131×
pub fn empty_ext4(out: &Path, mib: u32, label: &str) -> Result<()> {
2141×
    let (uid, gid) = owner();
2151×
    let status = std::process::Command::new("mkfs.ext4")
2161×
        .args(["-q", "-F", "-L", label, "-E"])
2171×
        .arg(format!("root_owner={uid}:{gid}"))
2181×
        .arg(out)
2191×
        .arg(format!("{mib}M"))
2201×
        .status()
2211×
        .context("run mkfs.ext4 (e2fsprogs 1.43 or later)")?;
2221×
    if !status.success() {
2230×
        bail!("mkfs.ext4 failed for a scratch disk");
2241×
    }
2251×
    Ok(())
2261×
}
227
228
/// Remove `dir` and everything in it, even where a job took away its own
229
/// write or search permission on a directory; links are never followed.
23028×
pub fn remove_tree(dir: &Path) -> std::io::Result<()> {
23128×
    match std::fs::remove_dir_all(dir) {
23217×
        Ok(()) => Ok(()),
23311×
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
234
        Err(_) => {
2350×
            open_up(dir);
2360×
            std::fs::remove_dir_all(dir)
237
        }
238
    }
23928×
}
240
241
/// The runner's user and group, which own a workspace image's root.
2423×
fn owner() -> (u32, u32) {
243
    #[cfg(unix)]
244
    {
2453×
        unsafe { (libc::getuid(), libc::getgid()) }
246
    }
247
    #[cfg(not(unix))]
248
    {
249
        (0, 0)
250
    }
2513×
}
252
2530×
fn open_up(dir: &Path) {
2540×
    let Ok(meta) = std::fs::symlink_metadata(dir) else { return };
2550×
    if !meta.is_dir() {
2560×
        return;
2570×
    }
2580×
    let _ = crate::portable::set_mode(dir, 0o700);
2590×
    if let Ok(entries) = std::fs::read_dir(dir) {
2600×
        for e in entries.flatten() {
2610×
            open_up(&e.path());
2620×
        }
2630×
    }
2640×
}
265
266
#[cfg(test)]
267
mod tests {
268
    use super::*;
269
    use std::time::UNIX_EPOCH;
270
2715×
    fn secs(p: &Path) -> i64 {
2725×
        std::fs::metadata(p).unwrap().modified().unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs() as i64
2735×
    }
274
275
    /// oxygenflow#52: an unchanged file gets the time it had when the cached
276
    /// disk was built; a changed or new one keeps its checkout's, newer.
277
    #[test]
2781×
    fn unchanged_sources_get_their_cached_times_back() {
2791×
        let dir = tempfile::tempdir().unwrap();
2803×
        for (name, text) in [("same.rs", "a"), ("changed.rs", "b2"), ("new.rs", "c")] {
2813×
            std::fs::write(dir.path().join(name), text).unwrap();
2823×
        }
2831×
        std::os::unix::fs::symlink("same.rs", dir.path().join("link.rs")).unwrap();
28410×
        let h = |t: &str| *blake3::hash(t.as_bytes()).as_bytes();
2851×
        let files: Files =
2861×
            vec![("same.rs".into(), h("a")), ("changed.rs".into(), h("b2")), ("new.rs".into(), h("c")), ("link.rs".into(), h("same.rs"))];
2871×
        let now = secs(&dir.path().join("same.rs"));
2881×
        let before: crate::cache::Sources =
2891×
            [("same.rs".to_string(), (hex::encode(h("a")), 1_000_000)), ("changed.rs".to_string(), (hex::encode(h("b1")), 1_000_000))]
2901×
                .into();
291
2921×
        let (sources, restored) = restore_times(dir.path(), &files, Some(&before));
2931×
        assert_eq!(restored, 1);
2941×
        assert_eq!(secs(&dir.path().join("same.rs")), 1_000_000, "unchanged: fresh to cargo");
2951×
        assert!(secs(&dir.path().join("changed.rs")) >= now, "changed: newer than the cached build");
2961×
        assert_eq!(sources["same.rs"], (hex::encode(h("a")), 1_000_000));
2971×
        assert_eq!(sources["changed.rs"].0, hex::encode(h("b2")), "kept with its new content");
2981×
        assert!(sources.contains_key("new.rs") && !sources.contains_key("link.rs"), "files only");
299
300
        // Without a cached build nothing moves, and the sources are listed:
301
        // three files and the directory holding them.
3021×
        let (again, restored) = restore_times(dir.path(), &files, None);
3031×
        assert_eq!((restored, again.len()), (0, 4));
3041×
        assert!(again.contains_key("/"));
305
306
        // A directory whose listing is unchanged gets its time back; one
307
        // that gained an entry keeps its new time, so a build script's
308
        // `rerun-if-changed` on it sees the change.
3091×
        std::fs::create_dir(dir.path().join("migrations")).unwrap();
3101×
        std::fs::write(dir.path().join("migrations/0001.sql"), "x").unwrap();
3111×
        let mut files = files;
3121×
        files.push(("migrations/0001.sql".into(), h("x")));
3131×
        let (mut kept, _) = restore_times(dir.path(), &files, None);
3141×
        kept.get_mut("migrations/").unwrap().1 = 2_000_000;
3151×
        kept.get_mut("migrations/0001.sql").unwrap().1 = 2_000_000;
3161×
        let (_, restored) = restore_times(dir.path(), &files, Some(&kept));
3171×
        assert_eq!(secs(&dir.path().join("migrations")), 2_000_000, "unchanged listing: its old time");
3181×
        assert!(restored >= 2);
3191×
        std::fs::write(dir.path().join("migrations/0002.sql"), "y").unwrap();
3201×
        files.push(("migrations/0002.sql".into(), h("y")));
3211×
        let (now_sources, _) = restore_times(dir.path(), &files, Some(&kept));
3221×
        assert!(secs(&dir.path().join("migrations")) > 2_000_000, "a new entry: it looks changed");
3231×
        assert_ne!(now_sources["migrations/"].0, kept["migrations/"].0);
3241×
    }
325
}