OXYGENHUB
# 8302cca90c95 oxygenflow / crates / flow / src / checkpoint.rs
9.9 KiB · Rust · last changed flow builds for Windows, offering the WASM tier 10 days ago source blame history raw
1
//! Checkpoints (oxygenflow#77, docs/handover.md §7): the directories a job
2
//! declares with `checkpoint "/src/…"`, saved when its runner hands it back
3
//! and restored before the next attempt runs the same exec, so a tool that
4
//! skips finished work (cargo, make) carries on where it stopped.
5
//!
6
//! The format is the runner's own, so CI builds need no new crate: a magic
7
//! line, then entries (a kind byte, the path under /src, mode, modification
8
//! time and, for a file, its bytes). Files keep their times, since cargo's
9
//! fingerprints read them. Links are left out. Unpacking accepts only paths
10
//! under the job's own checkpoints, without `.` or `..`.
11
12
use anyhow::{bail, Context, Result};
13
use flow_core::pipeline::{Job, MountSource};
14
use std::path::{Path, PathBuf};
15
use std::time::{Duration, SystemTime, UNIX_EPOCH};
16
17
/// At most this much is saved: the hub keeps it as it keeps an artifact.
18
pub const MAX_BYTES: usize = 512 << 20;
19
20
const MAGIC: &[u8] = b"OXYFLOW-CHECKPOINT 1\n";
21
const DIR: u8 = 0;
22
const FILE: u8 = 1;
23
24
/// What was saved or restored.
25
#[derive(Debug, Default, PartialEq, Eq)]
26
pub struct Counts {
27
    pub files: usize,
28
    pub dirs: usize,
29
    pub bytes: u64,
30
}
31
32
/// Where checkpoint `path` (under /src) is on this host, for a job of a
33
/// directory tier (WASM, syscall): in a disk mount's directory when a mount
34
/// holds it, else in the job's tree. `None` for one under a tmpfs, which
35
/// the syscall tier keeps in memory.
36
pub fn host_dir(path: &str, job: &Job, tree: &Path, scratch: &Path, wasm: bool) -> Option<PathBuf> {
37
    let under = |t: &str| path == t || path.starts_with(&format!("{}/", t.trim_end_matches('/')));
38
    let mut disks = 0;
39
    let mut best: Option<(usize, PathBuf)> = None;
40
    for (i, m) in job.mounts.iter().enumerate() {
41
        let dir = match m.source {
42
            MountSource::Disk => {
43
                disks += 1;
44
                Some(crate::syscall::disk_dir(scratch, disks - 1))
45
            }
46
            MountSource::Tmpfs if wasm => Some(scratch.join(format!("tmpfs-{i}"))),
47
            MountSource::Tmpfs => None,
48
        };
49
        if under(&m.target) && best.as_ref().is_none_or(|(len, _)| m.target.len() > *len) {
50
            let rest = path[m.target.trim_end_matches('/').len()..].trim_start_matches('/');
51
            best = Some((m.target.len(), dir.map(|d| d.join(rest)).unwrap_or_default()));
52
        }
53
    }
54
    match best {
55
        Some((_, d)) if d.as_os_str().is_empty() => None,
56
        Some((_, d)) => Some(d),
57
        None => Some(tree.join(path.trim_start_matches("/src").trim_start_matches('/'))),
58
    }
59
}
60
61
/// Pack `dirs` (each a checkpoint path and where it is on this host).
62
pub fn pack(dirs: &[(String, PathBuf)]) -> Result<(Vec<u8>, Counts)> {
63
    let mut out = MAGIC.to_vec();
64
    let mut n = Counts::default();
65
    for (path, host) in dirs {
66
        if !host.is_dir() {
67
            continue;
68
        }
69
        let mut stack = vec![(host.clone(), path.trim_end_matches('/').to_string())];
70
        while let Some((at, name)) = stack.pop() {
71
            let meta = std::fs::symlink_metadata(&at)?;
72
            if meta.is_dir() {
73
                entry(&mut out, DIR, &name, &meta, &[]);
74
                n.dirs += 1;
75
                let mut children: Vec<_> = std::fs::read_dir(&at)?.collect::<std::io::Result<_>>()?;
76
                children.sort_by_key(|e| e.file_name());
77
                for c in children.into_iter().rev() {
78
                    let Some(child) = c.file_name().to_str().map(String::from) else { continue };
79
                    stack.push((c.path(), format!("{name}/{child}")));
80
                }
81
            } else if meta.is_file() {
82
                let bytes = std::fs::read(&at).with_context(|| format!("read {}", at.display()))?;
83
                n.files += 1;
84
                n.bytes += bytes.len() as u64;
85
                entry(&mut out, FILE, &name, &meta, &bytes);
86
            }
87
            if out.len() > MAX_BYTES {
88
                bail!("the checkpoint is over {} MiB", MAX_BYTES >> 20);
89
            }
90
        }
91
    }
92
    Ok((out, n))
93
}
94
95
fn entry(out: &mut Vec<u8>, kind: u8, path: &str, meta: &std::fs::Metadata, bytes: &[u8]) {
96
    let since = meta.modified().ok().and_then(|t| t.duration_since(UNIX_EPOCH).ok()).unwrap_or_default();
97
    out.push(kind);
98
    out.extend((path.len() as u32).to_le_bytes());
99
    out.extend(path.as_bytes());
100
    out.extend(crate::portable::mode(meta).to_le_bytes());
101
    out.extend(since.as_secs().to_le_bytes());
102
    out.extend(since.subsec_nanos().to_le_bytes());
103
    out.extend((bytes.len() as u64).to_le_bytes());
104
    out.extend(bytes);
105
}
106
107
/// Restore a checkpoint: each entry at `place(checkpoint)` joined with the
108
/// rest of its path, for the job's own `checkpoints` only.
109
pub fn unpack(bytes: &[u8], checkpoints: &[String], place: impl Fn(&str) -> Option<PathBuf>) -> Result<Counts> {
110
    let Some(mut rest) = bytes.strip_prefix(MAGIC) else { bail!("not a checkpoint") };
111
    let mut n = Counts::default();
112
    let take = |rest: &mut &[u8], k: usize| -> Result<Vec<u8>> {
113
        if rest.len() < k {
114
            bail!("the checkpoint is cut short");
115
        }
116
        let (a, b) = rest.split_at(k);
117
        *rest = b;
118
        Ok(a.to_vec())
119
    };
120
    while !rest.is_empty() {
121
        let kind = take(&mut rest, 1)?[0];
122
        let len = u32::from_le_bytes(take(&mut rest, 4)?.try_into().unwrap()) as usize;
123
        let path = String::from_utf8(take(&mut rest, len)?).context("a path that is not UTF-8")?;
124
        let mode = u32::from_le_bytes(take(&mut rest, 4)?.try_into().unwrap());
125
        let secs = u64::from_le_bytes(take(&mut rest, 8)?.try_into().unwrap());
126
        let nanos = u32::from_le_bytes(take(&mut rest, 4)?.try_into().unwrap());
127
        let size = u64::from_le_bytes(take(&mut rest, 8)?.try_into().unwrap()) as usize;
128
        let data = take(&mut rest, size)?;
129
        // Under one of the job's checkpoints, without . or .. on the way.
130
        let Some(root) = checkpoints.iter().map(|c| c.trim_end_matches('/')).find(|c| path == *c || path.starts_with(&format!("{c}/")))
131
        else {
132
            bail!("{path} is not under the job's checkpoints");
133
        };
134
        let below = path[root.len()..].trim_start_matches('/');
135
        if below.split('/').any(|p| p == "." || p == ".." || (p.is_empty() && !below.is_empty())) {
136
            bail!("{path}: . or .. in a checkpoint path");
137
        }
138
        let Some(base) = place(root) else { continue };
139
        let at = if below.is_empty() { base } else { base.join(below) };
140
        match kind {
141
            DIR => {
142
                std::fs::create_dir_all(&at)?;
143
                n.dirs += 1;
144
            }
145
            FILE => {
146
                if let Some(parent) = at.parent() {
147
                    std::fs::create_dir_all(parent)?;
148
                }
149
                // Never through a link already there.
150
                if std::fs::symlink_metadata(&at).is_ok_and(|m| !m.is_file()) {
151
                    bail!("{path}: something other than a file is in the way");
152
                }
153
                std::fs::write(&at, &data)?;
154
                let f = std::fs::File::options().write(true).open(&at)?;
155
                crate::portable::set_mode(&at, mode & 0o777)?;
156
                f.set_modified(SystemTime::UNIX_EPOCH + Duration::new(secs, nanos))?;
157
                n.files += 1;
158
                n.bytes += data.len() as u64;
159
            }
160
            other => bail!("{path}: unknown entry kind {other}"),
161
        }
162
    }
163
    Ok(n)
164
}
165
166
#[cfg(test)]
167
mod tests {
168
    use super::*;
169
170
    #[test]
171
    fn a_checkpoint_comes_back_with_its_times_and_only_where_it_belongs() {
172
        let from = tempfile::tempdir().unwrap();
173
        let target = from.path().join("target");
174
        std::fs::create_dir_all(target.join("debug/deps")).unwrap();
175
        std::fs::write(target.join("debug/deps/libx.rlib"), b"built").unwrap();
176
        std::fs::write(target.join("debug/.fingerprint"), b"fp").unwrap();
177
        let old = SystemTime::UNIX_EPOCH + Duration::new(1_700_000_000, 123_000_000);
178
        std::fs::File::options().write(true).open(target.join("debug/deps/libx.rlib")).unwrap().set_modified(old).unwrap();
179
        let (bytes, n) = pack(&[("/src/target".into(), target.clone())]).unwrap();
180
        assert_eq!((n.files, n.dirs, n.bytes), (2, 3, 7));
181
182
        let to = tempfile::tempdir().unwrap();
183
        let place = |root: &str| (root == "/src/target").then(|| to.path().join("t"));
184
        let back = unpack(&bytes, &["/src/target".into()], place).unwrap();
185
        assert_eq!(back, n);
186
        let restored = to.path().join("t/debug/deps/libx.rlib");
187
        assert_eq!(std::fs::read(&restored).unwrap(), b"built");
188
        assert_eq!(std::fs::metadata(&restored).unwrap().modified().unwrap(), old, "cargo reads the times");
189
190
        // Not the job's checkpoint, or a path that climbs out: refused.
191
        assert!(unpack(&bytes, &["/src/other".into()], place).is_err());
192
        let mut evil = MAGIC.to_vec();
193
        let meta = std::fs::metadata(target.join("debug/.fingerprint")).unwrap();
194
        entry(&mut evil, FILE, "/src/target/../escape", &meta, b"x");
195
        assert!(unpack(&evil, &["/src/target".into()], place).is_err());
196
        assert!(!to.path().join("escape").exists());
197
        assert!(unpack(b"junk", &["/src/target".into()], place).is_err());
198
    }
199
200
    #[test]
201
    fn a_checkpoint_under_a_disk_mount_is_in_the_mount_s_directory() {
202
        let p = flow_core::Pipeline::parse(
203
            "schema-version 1\nsandbox \"w\" {\n    runtime \"wasm\"\n}\njob \"j\" {\n    sandbox \"w\"\n    mount {\n        source \"ephemeral://disk\"\n        target \"/src/target\"\n        size \"1GiB\"\n    }\n    checkpoint \"/src/target\" \"/src/state\"\n    exec \"x.wasm\"\n}\n",
204
        )
205
        .unwrap();
206
        let job = &p.jobs["j"];
207
        assert_eq!(job.checkpoints, ["/src/state", "/src/target"]);
208
        let (tree, scratch) = (Path::new("/w/tree"), Path::new("/w"));
209
        assert_eq!(host_dir("/src/target", job, tree, scratch, true), Some(crate::syscall::disk_dir(scratch, 0)));
210
        assert_eq!(host_dir("/src/state", job, tree, scratch, true), Some(PathBuf::from("/w/tree/state")));
211
    }
212
}