OXYGENHUB
# 32f54b393396 oxygenflow / crates / flow / src / workspace.rs
9.6 KiB · Rust · last changed flow builds for Windows, offering the WASM tier 10 days ago source blame history raw
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`.
16
pub fn export_state(repo: &Repo, heads: &[ChangeHash], dir: &Path) -> Result<Files> {
17
    let snapshot = repo.load_state(heads).context("load the state")?.materialize();
18
    let mut files = Vec::new();
19
    for (path, f) in &snapshot.files {
20
        if path.split('/').any(|p| p.is_empty() || p == "." || p == ".." || p == ".oxy") {
21
            continue;
22
        }
23
        let target = dir.join(path);
24
        match (&f.kind, &f.content) {
25
            (NodeKind::Dir, _) | (_, FileContent::Dir) => std::fs::create_dir_all(&target)?,
26
            (NodeKind::Subrepo, _) => {}
27
            (kind, content) => {
28
                let bytes = match content {
29
                    FileContent::Text { bytes, .. } => bytes.clone(),
30
                    FileContent::Opaque { blobs } => match blobs.first() {
31
                        Some(b) => repo.read_blob(b)?,
32
                        None => Vec::new(),
33
                    },
34
                    FileContent::Dir => unreachable!(),
35
                };
36
                if let Some(parent) = target.parent() {
37
                    std::fs::create_dir_all(parent)?;
38
                }
39
                if *kind == NodeKind::Symlink {
40
                    let to = String::from_utf8_lossy(&bytes).into_owned();
41
                    crate::portable::symlink(Path::new(&to), &target).with_context(|| format!("symlink {path}"))?;
42
                } else {
43
                    std::fs::write(&target, &bytes).with_context(|| format!("write {path}"))?;
44
                    if f.mode.exec {
45
                        crate::portable::set_mode(&target, 0o755)?;
46
                    }
47
                }
48
                files.push((path.clone(), *blake3::hash(&bytes).as_bytes()));
49
            }
50
        }
51
    }
52
    Ok(files)
53
}
54
55
/// Copy a working directory (without `.oxy` and what `.oxyignore` and
56
/// `.gitignore` exclude) under `dir`, for `flow run`.
57
pub fn copy_tree(src: &Path, dir: &Path) -> Result<Files> {
58
    let mut files = Vec::new();
59
    // A file at the top would otherwise find no directory to go in.
60
    std::fs::create_dir_all(dir)?;
61
    let walk = ignore::WalkBuilder::new(src)
62
        .hidden(false)
63
        .git_ignore(true)
64
        .require_git(false)
65
        .add_custom_ignore_filename(".oxyignore")
66
        .filter_entry(|e| e.file_name() != ".oxy" && e.file_name() != ".git")
67
        .build();
68
    for entry in walk {
69
        let entry = entry?;
70
        let rel = entry.path().strip_prefix(src)?;
71
        if rel.as_os_str().is_empty() {
72
            continue;
73
        }
74
        let rel_s = rel.to_string_lossy().replace('\\', "/");
75
        let target = dir.join(rel);
76
        let ft = entry.file_type().context("file type")?;
77
        if ft.is_dir() {
78
            std::fs::create_dir_all(&target)?;
79
        } else if ft.is_symlink() {
80
            let to = std::fs::read_link(entry.path())?;
81
            crate::portable::symlink(&to, &target)?;
82
            files.push((rel_s, *blake3::hash(to.to_string_lossy().as_bytes()).as_bytes()));
83
        } else if ft.is_file() {
84
            std::fs::copy(entry.path(), &target)?;
85
            files.push((rel_s, *blake3::hash(&std::fs::read(entry.path())?).as_bytes()));
86
        }
87
    }
88
    Ok(files)
89
}
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.
93
pub fn image(dir: &Path, out: &Path) -> Result<()> {
94
    let used: u64 = ignore::WalkBuilder::new(dir)
95
        .hidden(false)
96
        .ignore(false)
97
        .git_ignore(false)
98
        .build()
99
        .flatten()
100
        .filter_map(|e| e.metadata().ok())
101
        .map(|m| m.len().max(4096))
102
        .sum();
103
    let mib = used * 12 / 10 / (1 << 20) + 512;
104
    let (uid, gid) = owner();
105
    let status = std::process::Command::new("mkfs.ext4")
106
        .args(["-q", "-F", "-L", "workspace", "-E"])
107
        .arg(format!("root_owner={uid}:{gid}"))
108
        .arg("-d")
109
        .arg(dir)
110
        .arg(out)
111
        .arg(format!("{mib}M"))
112
        .status()
113
        .context("run mkfs.ext4 (e2fsprogs 1.43 or later)")?;
114
    if !status.success() {
115
        bail!("mkfs.ext4 failed for the workspace image");
116
    }
117
    Ok(())
118
}
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.
126
pub fn restore_times(dir: &Path, files: &Files, before: Option<&crate::cache::Sources>) -> (crate::cache::Sources, usize) {
127
    let mut out = crate::cache::Sources::new();
128
    let mut restored = 0;
129
    for (path, hash) in files {
130
        let p = dir.join(path);
131
        let Ok(md) = std::fs::symlink_metadata(&p) else { continue };
132
        if !md.is_file() {
133
            continue;
134
        }
135
        let hex = hex::encode(hash);
136
        let kept = before.and_then(|b| b.get(path)).filter(|(h, _)| *h == hex).map(|(_, t)| *t);
137
        let t = match kept {
138
            Some(t) => {
139
                let at = std::time::UNIX_EPOCH + std::time::Duration::from_secs(t.max(0) as u64);
140
                if std::fs::File::options().write(true).open(&p).and_then(|f| f.set_modified(at)).is_err() {
141
                    continue;
142
                }
143
                restored += 1;
144
                t
145
            }
146
            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
        };
148
        out.insert(path.clone(), (hex, t));
149
    }
150
    (out, restored)
151
}
152
153
/// An empty ext4 filesystem of `mib` MiB at `out`, owned by this user like
154
/// the workspace. mkfs.ext4 creates the file sparse, so only what the job
155
/// writes takes space on the runner.
156
pub fn empty_ext4(out: &Path, mib: u32, label: &str) -> Result<()> {
157
    let (uid, gid) = owner();
158
    let status = std::process::Command::new("mkfs.ext4")
159
        .args(["-q", "-F", "-L", label, "-E"])
160
        .arg(format!("root_owner={uid}:{gid}"))
161
        .arg(out)
162
        .arg(format!("{mib}M"))
163
        .status()
164
        .context("run mkfs.ext4 (e2fsprogs 1.43 or later)")?;
165
    if !status.success() {
166
        bail!("mkfs.ext4 failed for a scratch disk");
167
    }
168
    Ok(())
169
}
170
171
/// Remove `dir` and everything in it, even where a job took away its own
172
/// write or search permission on a directory; links are never followed.
173
pub fn remove_tree(dir: &Path) -> std::io::Result<()> {
174
    match std::fs::remove_dir_all(dir) {
175
        Ok(()) => Ok(()),
176
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
177
        Err(_) => {
178
            open_up(dir);
179
            std::fs::remove_dir_all(dir)
180
        }
181
    }
182
}
183
184
/// The runner's user and group, which own a workspace image's root.
185
fn owner() -> (u32, u32) {
186
    #[cfg(unix)]
187
    {
188
        unsafe { (libc::getuid(), libc::getgid()) }
189
    }
190
    #[cfg(not(unix))]
191
    {
192
        (0, 0)
193
    }
194
}
195
196
fn open_up(dir: &Path) {
197
    let Ok(meta) = std::fs::symlink_metadata(dir) else { return };
198
    if !meta.is_dir() {
199
        return;
200
    }
201
    let _ = crate::portable::set_mode(dir, 0o700);
202
    if let Ok(entries) = std::fs::read_dir(dir) {
203
        for e in entries.flatten() {
204
            open_up(&e.path());
205
        }
206
    }
207
}
208
209
#[cfg(test)]
210
mod tests {
211
    use super::*;
212
    use std::time::UNIX_EPOCH;
213
214
    fn secs(p: &Path) -> i64 {
215
        std::fs::metadata(p).unwrap().modified().unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs() as i64
216
    }
217
218
    /// oxygenflow#52: an unchanged file gets the time it had when the cached
219
    /// disk was built; a changed or new one keeps its checkout's, newer.
220
    #[test]
221
    fn unchanged_sources_get_their_cached_times_back() {
222
        let dir = tempfile::tempdir().unwrap();
223
        for (name, text) in [("same.rs", "a"), ("changed.rs", "b2"), ("new.rs", "c")] {
224
            std::fs::write(dir.path().join(name), text).unwrap();
225
        }
226
        std::os::unix::fs::symlink("same.rs", dir.path().join("link.rs")).unwrap();
227
        let h = |t: &str| *blake3::hash(t.as_bytes()).as_bytes();
228
        let files: Files =
229
            vec![("same.rs".into(), h("a")), ("changed.rs".into(), h("b2")), ("new.rs".into(), h("c")), ("link.rs".into(), h("same.rs"))];
230
        let now = secs(&dir.path().join("same.rs"));
231
        let before: crate::cache::Sources =
232
            [("same.rs".to_string(), (hex::encode(h("a")), 1_000_000)), ("changed.rs".to_string(), (hex::encode(h("b1")), 1_000_000))]
233
                .into();
234
235
        let (sources, restored) = restore_times(dir.path(), &files, Some(&before));
236
        assert_eq!(restored, 1);
237
        assert_eq!(secs(&dir.path().join("same.rs")), 1_000_000, "unchanged: fresh to cargo");
238
        assert!(secs(&dir.path().join("changed.rs")) >= now, "changed: newer than the cached build");
239
        assert_eq!(sources["same.rs"], (hex::encode(h("a")), 1_000_000));
240
        assert_eq!(sources["changed.rs"].0, hex::encode(h("b2")), "kept with its new content");
241
        assert!(sources.contains_key("new.rs") && !sources.contains_key("link.rs"), "files only");
242
243
        // Without a cached build nothing moves, and the sources are listed.
244
        let (again, restored) = restore_times(dir.path(), &files, None);
245
        assert_eq!((restored, again.len()), (0, 3));
246
    }
247
}