OXYGENHUB
# 42c8e2f4219f oxygenflow / crates / jet / src / images.rs
10.4 KiB · Rust · last changed jet: the Jetstream crate, starting with images by hash 25 days ago source blame history raw
59078d2b Andy McCall ↶ 1
//! Images by hash (oxygenflow#39, docs/jetstream.md §8).
2
//!
3
//! A rootfs image is named by the BLAKE3 of its bytes, as pipelines pin it
4
//! (`cas://blob/HASH`). To move it, it is cut into chunks with Oxygen's
5
//! FastCDC (32 KiB to 256 KiB, `oxy_core::chunk`), each named by its own
6
//! BLAKE3, and described by a [`Manifest`]: the image's hash, size and
7
//! architecture, and its chunks in order. A sender sends only the chunks a
8
//! receiver lacks; a receiver writes the image back, leaving runs of zeros
9
//! as holes without fetching them, and checks the whole hash before using
10
//! it. Chunk boundaries depend only on content, so an image rebuilt with a
11
//! small change shares nearly all its chunks with the one before.
12
13
use oxy_core::chunk::{cut_point, MAX_SIZE};
14
use serde::{Deserialize, Serialize};
15
use std::collections::HashSet;
16
use std::io::{Read, Seek, SeekFrom, Write};
17
use std::path::Path;
18
19
/// The manifest format this crate writes and reads.
20
pub const VERSION: u32 = 1;
21
/// Largest image accepted: 64 GiB.
22
pub const MAX_IMAGE: u64 = 64 << 30;
23
24
#[derive(Debug, thiserror::Error)]
25
pub enum Error {
26
    #[error("{0}")]
27
    Io(#[from] std::io::Error),
28
    #[error("{0}")]
29
    Bad(String),
30
}
31
32
pub type Result<T> = std::result::Result<T, Error>;
33
34
/// One chunk: its BLAKE3 (hex) and length.
35
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
36
pub struct Chunk(pub String, pub u32);
37
38
impl Chunk {
39
    pub fn hash(&self) -> &str {
40
        &self.0
41
    }
42
43
    pub fn len(&self) -> usize {
44
        self.1 as usize
45
    }
46
47
    pub fn is_empty(&self) -> bool {
48
        self.1 == 0
49
    }
50
51
    /// Whether this is a chunk of zeros, which a receiver writes as a hole
52
    /// without fetching it.
53
    pub fn is_zeros(&self) -> bool {
54
        self.0 == zeros_hash(self.len())
55
    }
56
}
57
58
/// An image: its hash, size, architecture and chunks.
59
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
60
pub struct Manifest {
61
    pub version: u32,
62
    /// BLAKE3 of the whole image, hex: its name.
63
    pub hash: String,
64
    pub size: u64,
65
    /// `x86_64`, `riscv64`, …, as its import recorded (OxygenFlow spec §11.2).
66
    #[serde(default, skip_serializing_if = "Option::is_none")]
67
    pub arch: Option<String>,
68
    pub chunks: Vec<Chunk>,
69
}
70
71
impl Manifest {
72
    /// The distinct chunks a receiver must fetch: all but zeros.
73
    pub fn needed(&self) -> Vec<&str> {
74
        let mut seen = HashSet::new();
75
        self.chunks.iter().filter(|c| !c.is_zeros() && seen.insert(c.hash())).map(|c| c.hash()).collect()
76
    }
77
78
    /// Check its shape: a known version, hex names, chunks in the
79
    /// chunker's bounds that add up to the size.
80
    pub fn check(&self) -> Result<()> {
81
        if self.version != VERSION {
82
            return Err(Error::Bad(format!("image manifest version {} is not known; this reads {VERSION}", self.version)));
83
        }
84
        if !is_hash(&self.hash) {
85
            return Err(Error::Bad(format!("{:?} is not an image hash", self.hash)));
86
        }
87
        if self.size > MAX_IMAGE {
88
            return Err(Error::Bad(format!("the image is {} GiB; at most {} GiB", self.size >> 30, MAX_IMAGE >> 30)));
89
        }
90
        let mut total = 0u64;
91
        for c in &self.chunks {
92
            if !is_hash(c.hash()) || c.is_empty() || c.len() > MAX_SIZE {
93
                return Err(Error::Bad(format!("chunk {:?} of {} bytes is not a chunk", c.hash(), c.len())));
94
            }
95
            total += c.1 as u64;
96
        }
97
        if total != self.size {
98
            return Err(Error::Bad(format!("the chunks hold {total} bytes; the image is {}", self.size)));
99
        }
100
        Ok(())
101
    }
102
}
103
104
/// Whether `s` is 64 lower-case hex digits.
105
pub fn is_hash(s: &str) -> bool {
106
    s.len() == 64 && s.bytes().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
107
}
108
109
/// The BLAKE3 of `len` zero bytes, hex; remembered per length.
110
pub fn zeros_hash(len: usize) -> String {
111
    use std::sync::Mutex;
112
    static MEMO: Mutex<Vec<(usize, String)>> = Mutex::new(Vec::new());
113
    let mut memo = MEMO.lock().unwrap_or_else(|p| p.into_inner());
114
    if let Some((_, h)) = memo.iter().find(|(l, _)| *l == len) {
115
        return h.clone();
116
    }
117
    let h = blake3::hash(&vec![0u8; len]).to_hex().to_string();
118
    if memo.len() < 64 {
119
        memo.push((len, h.clone()));
120
    }
121
    h
122
}
123
124
/// Cut `r` into chunks as Oxygen's chunker would cut it whole, calling
125
/// `each` with every chunk's bytes, and answer the manifest. Reads in
126
/// bounded memory: the chunker never looks past [`MAX_SIZE`].
127
pub fn scan<R: Read>(mut r: R, arch: Option<String>, mut each: impl FnMut(&Chunk, &[u8]) -> Result<()>) -> Result<Manifest> {
128
    const FILL: usize = 8 << 20;
129
    let mut buf: Vec<u8> = Vec::with_capacity(FILL + MAX_SIZE);
130
    let (mut pos, mut eof, mut size) = (0usize, false, 0u64);
131
    let mut whole = blake3::Hasher::new();
132
    let mut chunks = Vec::new();
133
    loop {
134
        if !eof && buf.len() - pos < MAX_SIZE {
135
            buf.drain(..pos);
136
            pos = 0;
137
            let start = buf.len();
138
            buf.resize(start + FILL, 0);
139
            let mut got = 0;
140
            while got < FILL {
141
                match r.read(&mut buf[start + got..])? {
142
                    0 => {
143
                        eof = true;
144
                        break;
145
                    }
146
                    n => got += n,
147
                }
148
            }
149
            buf.truncate(start + got);
150
        }
151
        if pos == buf.len() {
152
            break;
153
        }
154
        let n = cut_point(&buf[pos..]);
155
        let bytes = &buf[pos..pos + n];
156
        whole.update(bytes);
157
        let c = Chunk(blake3::hash(bytes).to_hex().to_string(), n as u32);
158
        each(&c, bytes)?;
159
        chunks.push(c);
160
        size += n as u64;
161
        pos += n;
162
    }
163
    Ok(Manifest { version: VERSION, hash: whole.finalize().to_hex().to_string(), size, arch, chunks })
164
}
165
166
/// [`scan`] of a file.
167
pub fn scan_file(path: &Path, arch: Option<String>, each: impl FnMut(&Chunk, &[u8]) -> Result<()>) -> Result<Manifest> {
168
    scan(std::io::BufReader::with_capacity(1 << 20, std::fs::File::open(path)?), arch, each)
169
}
170
171
/// Write image `m` to `out`, getting each chunk's bytes from `fetch`
172
/// (zeros are holes, never fetched; a chunk seen before is fetched once
173
/// per run of repeats), checking every chunk and then the whole hash. The
174
/// image is written beside `out` and renamed into place only when whole.
175
pub fn assemble(m: &Manifest, out: &Path, mut fetch: impl FnMut(&Chunk) -> Result<Vec<u8>>) -> Result<()> {
176
    m.check()?;
177
    let tmp = out.with_extension("part");
178
    let mut f = std::fs::File::create(&tmp)?;
179
    let mut whole = blake3::Hasher::new();
180
    let mut last: Option<(String, Vec<u8>)> = None;
181
    for c in &m.chunks {
182
        if c.is_zeros() {
183
            whole.update(&vec![0u8; c.len()]);
184
            f.seek(SeekFrom::Current(c.len() as i64))?;
185
            continue;
186
        }
187
        let bytes = match &last {
188
            Some((h, b)) if h == c.hash() => b.clone(),
189
            _ => {
190
                let b = fetch(c)?;
191
                if b.len() != c.len() || blake3::hash(&b).to_hex().as_str() != c.hash() {
192
                    let _ = std::fs::remove_file(&tmp);
193
                    return Err(Error::Bad(format!("chunk {} came back with other bytes", &c.hash()[..12])));
194
                }
195
                last = Some((c.0.clone(), b.clone()));
196
                b
197
            }
198
        };
199
        whole.update(&bytes);
200
        f.write_all(&bytes)?;
201
    }
202
    f.set_len(m.size)?;
203
    f.sync_all()?;
204
    drop(f);
205
    let got = whole.finalize().to_hex().to_string();
206
    if got != m.hash {
207
        let _ = std::fs::remove_file(&tmp);
208
        return Err(Error::Bad(format!("the image came out as {}, not {}", &got[..12], &m.hash[..12])));
209
    }
210
    std::fs::rename(&tmp, out)?;
211
    Ok(())
212
}
213
214
#[cfg(test)]
215
mod tests {
216
    use super::*;
217
218
    fn image(seed: u8) -> Vec<u8> {
219
        // Some text-like bytes, a long run of zeros (as an ext4 image has),
220
        // then more bytes.
221
        let mut v: Vec<u8> = (0..300_000u32).map(|i| ((i * 7 + seed as u32) % 251) as u8).collect();
222
        v.extend(std::iter::repeat_n(0u8, 3 << 20));
223
        v.extend((0..200_000u32).map(|i| ((i * 13 + seed as u32) % 241) as u8));
224
        v
225
    }
226
227
    #[test]
228
    fn a_scanned_image_comes_back_whole_and_zeros_are_never_fetched() {
229
        let data = image(1);
230
        let mut stored = std::collections::HashMap::new();
231
        let m = scan(&data[..], Some("x86_64".into()), |c, b| {
232
            stored.insert(c.hash().to_string(), b.to_vec());
233
            Ok(())
234
        })
235
        .unwrap();
236
        m.check().unwrap();
237
        assert_eq!(m.hash, blake3::hash(&data).to_hex().to_string(), "named as pipelines pin it");
238
        assert_eq!(m.size, data.len() as u64);
239
        assert_eq!(
240
            m.chunks.iter().map(|c| c.len()).collect::<Vec<_>>(),
241
            oxy_core::chunk::chunks(&data).iter().map(|c| c.len()).collect::<Vec<_>>(),
242
            "the same cuts as Oxygen's chunker over the whole"
243
        );
244
        assert!(m.chunks.iter().any(|c| c.is_zeros()));
245
        let dir = tempfile::tempdir().unwrap();
246
        let out = dir.path().join("img.ext4");
247
        let mut fetched = 0;
248
        assemble(&m, &out, |c| {
249
            fetched += 1;
250
            assert!(!c.is_zeros());
251
            Ok(stored[c.hash()].clone())
252
        })
253
        .unwrap();
254
        assert_eq!(std::fs::read(&out).unwrap(), data);
255
        assert!(fetched <= m.needed().len() + 1);
256
257
        // A rebuild with a small change shares nearly every chunk.
258
        let mut changed = data.clone();
259
        changed[100] ^= 0xff;
260
        let m2 = scan(&changed[..], None, |_, _| Ok(())).unwrap();
261
        let have: HashSet<&str> = m.chunks.iter().map(|c| c.hash()).collect();
262
        let new = m2.needed().into_iter().filter(|h| !have.contains(h)).count();
263
        assert!(new <= 2, "{new} new chunks for a one-byte change");
264
    }
265
266
    #[test]
267
    fn bad_bytes_and_bad_manifests_are_refused() {
268
        let data = image(2);
269
        let m = scan(&data[..], None, |_, _| Ok(())).unwrap();
270
        let dir = tempfile::tempdir().unwrap();
271
        let out = dir.path().join("img.ext4");
272
        let e = assemble(&m, &out, |c| Ok(vec![1u8; c.len()])).unwrap_err();
273
        assert!(e.to_string().contains("other bytes"), "{e}");
274
        assert!(!out.exists() && !out.with_extension("part").exists());
275
        let mut short = m.clone();
276
        short.size += 1;
277
        assert!(short.check().unwrap_err().to_string().contains("chunks hold"));
278
        let mut odd = m.clone();
279
        odd.hash = "nope".into();
280
        assert!(odd.check().is_err());
281
    }
282
}