images.rs 27.2%

68 of 250 lines run, 14 of 88 functions · run #206 on main · run missed

//! Images by hash with a hub (oxygenflow#39, docs/jetstream.md §8):
//! `flow rootfs push` sends an image's missing chunks and its manifest;
//! `flow rootfs pull`, and a runner that lacks a job's image, fetch the
//! manifest and the chunks they lack and write the image back, checked
//! against its hash. Fetched chunks stay in the home's chunk cache
//! (`chunks/`), so a rebuilt image costs only the chunks that changed.

use crate::home::Home;
use anyhow::{bail, Context, Result};
use jet::images::{self, Chunk, Manifest};
use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::time::{Duration, Instant};

/// Requests in flight at once.
const PARALLEL: usize = 8;
/// Chunk names per `missing` question.
const ASK: usize = 20_000;

/// A hub's image endpoints, with a token.
pub struct Client {
    root: String,
    base: String,
    token: String,
    agent: ureq::Agent,
}

/// What a push or a fetch moved.
pub struct Moved {
    pub chunks: usize,
    pub bytes: u64,
    /// Of `bytes`, those that came from peers rather than the hub.
    pub peer_bytes: u64,
    pub elapsed: Duration,
}

impl Client {
    pub fn new(hub: &str, token: &str, ca: Option<&Path>) -> Result<Client> {
        let tls = oxy_core::protocol::tls_client_config(ca).map_err(|e| anyhow::anyhow!("{e}"))?;
        let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(120)).build();
        let root = hub.trim_end_matches('/').to_string();
        Ok(Client { base: format!("{root}/api/v1"), root, token: token.into(), agent })
    }

    fn auth(&self) -> String {
        format!("Bearer {}", self.token)
    }

    fn fail(url: &str, e: ureq::Error) -> anyhow::Error {
        match e {
            ureq::Error::Status(code, r) => {
                let text = r.into_string().unwrap_or_default();
                let detail = serde_json::from_str::<serde_json::Value>(&text)
                    .ok()
                    .and_then(|v| v["detail"].as_str().map(String::from))
                    .unwrap_or(text);
                anyhow::anyhow!("{url}: HTTP {code}: {detail}")
            }
            e => anyhow::anyhow!("{url}: {e}"),
        }
    }

    /// The hub's manifest of image `hash`, `None` when it does not hold it.
    pub fn manifest(&self, hash: &str) -> Result<Option<Manifest>> {
        let url = format!("{}/flow/images/{hash}", self.base);
        match self.agent.get(&url).set("Authorization", &self.auth()).call() {
            Ok(r) => Ok(Some(r.into_json()?)),
            Err(ureq::Error::Status(404, _)) => Ok(None),
            Err(e) => Err(Self::fail(&url, e)),
        }
    }

    /// The hub's identity (`/.well-known/oxyhub-key`): the key its swarm
    /// tickets are signed with.
    pub fn hub_identity(&self) -> Result<oxy_core::identity::Identity> {
        let url = format!("{}/.well-known/oxyhub-key", self.root);
        let r = self.agent.get(&url).call().map_err(|e| Self::fail(&url, e))?;
        let mut out = Vec::new();
        std::io::Read::read_to_end(&mut r.into_reader(), &mut out)?;
        oxy_core::identity::Identity::decode(&out).map_err(|e| anyhow::anyhow!("{url}: not a hub identity: {e}"))
    }

    /// Tell the hub's tracker this runner is fetching image `hash`, or has
    /// stopped: the peers to ask and the ticket they admit.
    pub fn announce(&self, hash: &str, state: &str) -> Result<flow_core::api::Announced> {
        let url = format!("{}/flow/swarm/announce", self.base);
        let body = flow_core::api::Announce { image: hash.into(), state: state.into() };
        Ok(self.agent.post(&url).set("Authorization", &self.auth()).send_json(&body).map_err(|e| Self::fail(&url, e))?.into_json()?)
    }

    pub(crate) fn chunk(&self, hash: &str) -> Result<Vec<u8>> {
        let url = format!("{}/flow/images/chunks/{hash}", self.base);
        let r = self.agent.get(&url).set("Authorization", &self.auth()).call().map_err(|e| Self::fail(&url, e))?;
        let mut out = Vec::new();
        std::io::Read::read_to_end(&mut r.into_reader(), &mut out)?;
        Ok(out)
    }

    /// Send image `hash` from `home`: only the chunks the hub lacks, then
    /// the manifest, which the hub checks against the hash.
    pub fn push(&self, home: &Home, hash: &str) -> Result<Moved> {
        let start = Instant::now();
        let path = home.rootfs(hash);
        if !path.is_file() {
            bail!("{} is not in this Flow home ({}); import it first", &hash[..12.min(hash.len())], path.display());
        }
        let m = images::scan_file(&path, home.image_arch(hash), |_, _| Ok(()))?;
        if m.hash != hash {
            bail!("{} holds an image named {}, not {}: the file was changed", path.display(), &m.hash[..12], &hash[..12]);
        }
        let mut missing: HashSet<String> = HashSet::new();
        // 20,000 names is about 1.3 MB of JSON: under a hub's 2 MB default
        // body limit, which refused 50,000 at a time (oxygenhub#147).
        for ask in m.needed().chunks(ASK) {
            let url = format!("{}/flow/images/missing", self.base);
            let r: serde_json::Value = self
                .agent
                .post(&url)
                .set("Authorization", &self.auth())
                .send_json(serde_json::json!({ "chunks": ask }))
                .map_err(|e| Self::fail(&url, e))?
                .into_json()?;
            missing.extend(r["missing"].as_array().into_iter().flatten().filter_map(|h| h.as_str().map(String::from)));
        }
        // Read the image again, sending the missing chunks as they come.
        let (count, bytes) = (missing.len(), Mutex::new(0u64));
        let queue: Mutex<Vec<(String, Vec<u8>)>> = Mutex::new(Vec::new());
        let failed: Mutex<Option<anyhow::Error>> = Mutex::new(None);
        let sent = Mutex::new(HashSet::new());
        let send = |c: &str, b: &[u8]| -> Result<()> {
            let url = format!("{}/flow/images/chunks/{c}", self.base);
            self.agent.put(&url).set("Authorization", &self.auth()).send_bytes(b).map_err(|e| Self::fail(&url, e))?;
            *bytes.lock().unwrap() += b.len() as u64;
            Ok(())
        };
        images::scan_file(&path, None, |c: &Chunk, b: &[u8]| {
            if missing.contains(c.hash()) && sent.lock().unwrap().insert(c.hash().to_string()) {
                queue.lock().unwrap().push((c.hash().to_string(), b.to_vec()));
            }
            if queue.lock().unwrap().len() >= PARALLEL * 4 {
                let batch: Vec<_> = std::mem::take(&mut *queue.lock().unwrap());
                send_batch(&batch, &send, &failed);
            }
            match failed.lock().unwrap().take() {
                Some(e) => Err(images::Error::Bad(format!("{e:#}"))),
                None => Ok(()),
            }
        })?;
        let rest: Vec<_> = std::mem::take(&mut *queue.lock().unwrap());
        send_batch(&rest, &send, &failed);
        if let Some(e) = failed.lock().unwrap().take() {
            return Err(e);
        }
        let url = format!("{}/flow/images/{hash}", self.base);
        // The hub reads the whole image before it answers, which on a busy
        // disk takes far longer than the agent's 120 s (oxygenhub#341).
        self.agent
            .put(&url)
            .timeout(Duration::from_secs(3600))
            .set("Authorization", &self.auth())
            .send_json(&m)
            .map_err(|e| Self::fail(&url, e))?;
        let bytes = *bytes.lock().unwrap();
        Ok(Moved { chunks: count, bytes, peer_bytes: 0, elapsed: start.elapsed() })
    }

    /// Fetch image `hash` into `home`: the chunks its cache lacks, the
    /// image written and checked, its architecture recorded. `Ok(None)`
    /// when the hub does not hold it.
    pub fn fetch(&self, home: &Home, hash: &str) -> Result<Option<Moved>> {
        let start = Instant::now();
        let Some(m) = self.manifest(hash)? else { return Ok(None) };
        if m.hash != hash {
            bail!("the hub's manifest for {} names {}", &hash[..12], &m.hash[..12]);
        }
        let cache = chunk_cache(home);
        save_manifest(&home.root, &m)?;
        let wanted: Vec<String> = m.needed().into_iter().filter(|h| !cache_path(&cache, h).is_file()).map(String::from).collect();
        let (bytes, failed) = (Mutex::new(0u64), Mutex::new(None::<anyhow::Error>));
        let next = Mutex::new(wanted.iter());
        std::thread::scope(|s| {
            for _ in 0..PARALLEL {
                s.spawn(|| loop {
                    let Some(h) = next.lock().unwrap().next() else { break };
                    if failed.lock().unwrap().is_some() {
                        break;
                    }
                    let got = self.chunk(h).and_then(|b| {
                        if blake3::hash(&b).to_hex().as_str() != h {
                            bail!("chunk {} came back with other bytes", &h[..12]);
                        }
                        let p = cache_path(&cache, h);
                        std::fs::create_dir_all(p.parent().expect("a chunk directory"))?;
                        let tmp = p.with_extension("part");
                        std::fs::write(&tmp, &b)?;
                        std::fs::rename(&tmp, &p)?;
                        *bytes.lock().unwrap() += b.len() as u64;
                        Ok(())
                    });
                    if let Err(e) = got {
                        *failed.lock().unwrap() = Some(e);
                        break;
                    }
                });
            }
        });
        if let Some(e) = failed.into_inner().unwrap() {
            return Err(e);
        }
        write_image(home, &m)?;
        Ok(Some(Moved { chunks: wanted.len(), bytes: bytes.into_inner().unwrap(), peer_bytes: 0, elapsed: start.elapsed() }))
    }
}

/// Write image `m` into `home` from the chunk cache, checked against its
/// hash, and record its architecture.
pub(crate) fn write_image(home: &Home, m: &Manifest) -> Result<()> {
    let cache = chunk_cache(home);
    let out = home.rootfs(&m.hash);
    std::fs::create_dir_all(out.parent().expect("the cas directory"))?;
    images::assemble(m, &out, |c| Ok(std::fs::read(cache_path(&cache, c.hash()))?))
        .with_context(|| format!("write image {}", &m.hash[..12]))?;
    if let Some(arch) = &m.arch {
        let manifest = serde_json::json!({ "arch": arch, "imported_ms": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0), "from": "hub" });
        std::fs::write(home.image_manifest(&m.hash), serde_json::to_vec(&manifest)?)?;
    }
    Ok(())
}

impl Client {
    /// Tell the hub which images this home holds, and send the manifest of
    /// each it keeps none for (oxygenflow#94, jetstream.md §8.1.1): the
    /// swarm can then carry an image nobody pushed. A manifest not kept here
    /// yet is made by reading the image once, and kept. How many were sent;
    /// none from a hub that predates it.
    pub fn offer_held(&self, home: &Home) -> Result<usize> {
        let images = home.images();
        if images.is_empty() {
            return Ok(0);
        }
        let url = format!("{}/flow/images/held", self.base);
        let wanted: flow_core::api::WantedManifests =
            match self.agent.post(&url).set("Authorization", &self.auth()).send_json(flow_core::api::HeldImages { images }) {
                Ok(r) => r.into_json()?,
                Err(ureq::Error::Status(404, _)) => return Ok(0),
                Err(e) => return Err(Self::fail(&url, e)),
            };
        let mut sent = 0;
        for hash in wanted.wanted.iter().filter(|h| images::is_hash(h)) {
            let path = home.rootfs(hash);
            if !path.is_file() {
                continue;
            }
            let kept = std::fs::read(manifest_path(&home.root, hash)).ok().and_then(|b| serde_json::from_slice::<Manifest>(&b).ok());
            let m = match kept.filter(|m| m.hash == *hash) {
                Some(m) => m,
                None => {
                    let m = images::scan_file(&path, home.image_arch(hash), |_, _| Ok(()))?;
                    if m.hash != *hash {
                        continue;
                    }
                    save_manifest(&home.root, &m)?;
                    m
                }
            };
            let url = format!("{}/flow/images/{hash}/held", self.base);
            self.agent.put(&url).set("Authorization", &self.auth()).send_json(&m).map_err(|e| Self::fail(&url, e))?;
            sent += 1;
        }
        Ok(sent)
    }
}

/// Where an image's chunk manifest is kept (`manifests/<hash>.json`), so
/// the runner can serve its chunks to peers out of the image itself.
pub(crate) fn manifest_path(root: &Path, hash: &str) -> PathBuf {
    root.join("manifests").join(format!("{hash}.json"))
}

pub(crate) fn save_manifest(root: &Path, m: &Manifest) -> Result<()> {
    let p = manifest_path(root, &m.hash);
    std::fs::create_dir_all(p.parent().expect("the manifests directory"))?;
    let tmp = p.with_extension("part");
    std::fs::write(&tmp, serde_json::to_vec(m)?)?;
    std::fs::rename(&tmp, &p)?;
    Ok(())
}

fn send_batch(batch: &[(String, Vec<u8>)], send: &(dyn Fn(&str, &[u8]) -> Result<()> + Sync), failed: &Mutex<Option<anyhow::Error>>) {
    std::thread::scope(|s| {
        for part in batch.chunks(batch.len().div_ceil(PARALLEL).max(1)) {
            s.spawn(move || {
                for (h, b) in part {
                    if let Err(e) = send(h, b) {
                        *failed.lock().unwrap() = Some(e);
                        return;
                    }
                }
            });
        }
    });
}

/// The home's chunk cache.
pub fn chunk_cache(home: &Home) -> PathBuf {
    home.root.join("chunks")
}

pub(crate) fn cache_path(cache: &Path, hash: &str) -> PathBuf {
    cache.join(&hash[..2]).join(hash)
}

/// A size for people: 991 MiB, 1.2 GiB.
pub fn human(bytes: u64) -> String {
    if bytes >= 1 << 30 {
        format!("{:.1} GiB", bytes as f64 / (1u64 << 30) as f64)
    } else {
        format!("{} MiB", bytes >> 20)
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1		//! Images by hash with a hub (oxygenflow#39, docs/jetstream.md §8):
2		//! `flow rootfs push` sends an image's missing chunks and its manifest;
3		//! `flow rootfs pull`, and a runner that lacks a job's image, fetch the
4		//! manifest and the chunks they lack and write the image back, checked
5		//! against its hash. Fetched chunks stay in the home's chunk cache
6		//! (`chunks/`), so a rebuilt image costs only the chunks that changed.
7
8		use crate::home::Home;
9		use anyhow::{bail, Context, Result};
10		use jet::images::{self, Chunk, Manifest};
11		use std::collections::HashSet;
12		use std::path::{Path, PathBuf};
13		use std::sync::Mutex;
14		use std::time::{Duration, Instant};
15
16		/// Requests in flight at once.
17		const PARALLEL: usize = 8;
18		/// Chunk names per `missing` question.
19		const ASK: usize = 20_000;
20
21		/// A hub's image endpoints, with a token.
22		pub struct Client {
23		root: String,
24		base: String,
25		token: String,
26		agent: ureq::Agent,
27		}
28
29		/// What a push or a fetch moved.
30		pub struct Moved {
31		pub chunks: usize,
32		pub bytes: u64,
33		/// Of `bytes`, those that came from peers rather than the hub.
34		pub peer_bytes: u64,
35		pub elapsed: Duration,
36		}
37
38		impl Client {
39	13×	pub fn new(hub: &str, token: &str, ca: Option<&Path>) -> Result<Client> {
40	13×	let tls = oxy_core::protocol::tls_client_config(ca).map_err(\|e\| anyhow::anyhow!("{e}"))?;
41	13×	let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(120)).build();
42	13×	let root = hub.trim_end_matches('/').to_string();
43	13×	Ok(Client { base: format!("{root}/api/v1"), root, token: token.into(), agent })
44	13×	}
45
46	148×	fn auth(&self) -> String {
47	148×	format!("Bearer {}", self.token)
48	148×	}
49
50	0×	fn fail(url: &str, e: ureq::Error) -> anyhow::Error {
51	0×	match e {
52	0×	ureq::Error::Status(code, r) => {
53	0×	let text = r.into_string().unwrap_or_default();
54	0×	let detail = serde_json::from_str::<serde_json::Value>(&text)
55	0×	.ok()
56	0×	.and_then(\|v\| v["detail"].as_str().map(String::from))
57	0×	.unwrap_or(text);
58	0×	anyhow::anyhow!("{url}: HTTP {code}: {detail}")
59		}
60	0×	e => anyhow::anyhow!("{url}: {e}"),
61		}
62	0×	}
63
64		/// The hub's manifest of image `hash`, `None` when it does not hold it.
65	7×	pub fn manifest(&self, hash: &str) -> Result<Option<Manifest>> {
66	7×	let url = format!("{}/flow/images/{hash}", self.base);
67	7×	match self.agent.get(&url).set("Authorization", &self.auth()).call() {
68	7×	Ok(r) => Ok(Some(r.into_json()?)),
69	0×	Err(ureq::Error::Status(404, _)) => Ok(None),
70	0×	Err(e) => Err(Self::fail(&url, e)),
71		}
72	7×	}
73
74		/// The hub's identity (`/.well-known/oxyhub-key`): the key its swarm
75		/// tickets are signed with.
76	12×	pub fn hub_identity(&self) -> Result<oxy_core::identity::Identity> {
77	12×	let url = format!("{}/.well-known/oxyhub-key", self.root);
78	12×	let r = self.agent.get(&url).call().map_err(\|e\| Self::fail(&url, e))?;
79	12×	let mut out = Vec::new();
80	12×	std::io::Read::read_to_end(&mut r.into_reader(), &mut out)?;
81	12×	oxy_core::identity::Identity::decode(&out).map_err(\|e\| anyhow::anyhow!("{url}: not a hub identity: {e}"))
82	12×	}
83
84		/// Tell the hub's tracker this runner is fetching image `hash`, or has
85		/// stopped: the peers to ask and the ticket they admit.
86	17×	pub fn announce(&self, hash: &str, state: &str) -> Result<flow_core::api::Announced> {
87	17×	let url = format!("{}/flow/swarm/announce", self.base);
88	17×	let body = flow_core::api::Announce { image: hash.into(), state: state.into() };
89	17×	Ok(self.agent.post(&url).set("Authorization", &self.auth()).send_json(&body).map_err(\|e\| Self::fail(&url, e))?.into_json()?)
90	17×	}
91
92	124×	pub(crate) fn chunk(&self, hash: &str) -> Result<Vec<u8>> {
93	124×	let url = format!("{}/flow/images/chunks/{hash}", self.base);
94	124×	let r = self.agent.get(&url).set("Authorization", &self.auth()).call().map_err(\|e\| Self::fail(&url, e))?;
95	124×	let mut out = Vec::new();
96	124×	std::io::Read::read_to_end(&mut r.into_reader(), &mut out)?;
97	124×	Ok(out)
98	124×	}
99
100		/// Send image `hash` from `home`: only the chunks the hub lacks, then
101		/// the manifest, which the hub checks against the hash.
102	0×	pub fn push(&self, home: &Home, hash: &str) -> Result<Moved> {
103	0×	let start = Instant::now();
104	0×	let path = home.rootfs(hash);
105	0×	if !path.is_file() {
106	0×	bail!("{} is not in this Flow home ({}); import it first", &hash[..12.min(hash.len())], path.display());
107	0×	}
108	0×	let m = images::scan_file(&path, home.image_arch(hash), \|_, _\| Ok(()))?;
109	0×	if m.hash != hash {
110	0×	bail!("{} holds an image named {}, not {}: the file was changed", path.display(), &m.hash[..12], &hash[..12]);
111	0×	}
112	0×	let mut missing: HashSet<String> = HashSet::new();
113		// 20,000 names is about 1.3 MB of JSON: under a hub's 2 MB default
114		// body limit, which refused 50,000 at a time (oxygenhub#147).
115	0×	for ask in m.needed().chunks(ASK) {
116	0×	let url = format!("{}/flow/images/missing", self.base);
117	0×	let r: serde_json::Value = self
118	0×	.agent
119	0×	.post(&url)
120	0×	.set("Authorization", &self.auth())
121	0×	.send_json(serde_json::json!({ "chunks": ask }))
122	0×	.map_err(\|e\| Self::fail(&url, e))?
123	0×	.into_json()?;
124	0×	missing.extend(r["missing"].as_array().into_iter().flatten().filter_map(\|h\| h.as_str().map(String::from)));
125		}
126		// Read the image again, sending the missing chunks as they come.
127	0×	let (count, bytes) = (missing.len(), Mutex::new(0u64));
128	0×	let queue: Mutex<Vec<(String, Vec<u8>)>> = Mutex::new(Vec::new());
129	0×	let failed: Mutex<Option<anyhow::Error>> = Mutex::new(None);
130	0×	let sent = Mutex::new(HashSet::new());
131	0×	let send = \|c: &str, b: &[u8]\| -> Result<()> {
132	0×	let url = format!("{}/flow/images/chunks/{c}", self.base);
133	0×	self.agent.put(&url).set("Authorization", &self.auth()).send_bytes(b).map_err(\|e\| Self::fail(&url, e))?;
134	0×	*bytes.lock().unwrap() += b.len() as u64;
135	0×	Ok(())
136	0×	};
137	0×	images::scan_file(&path, None, \|c: &Chunk, b: &[u8]\| {
138	0×	if missing.contains(c.hash()) && sent.lock().unwrap().insert(c.hash().to_string()) {
139	0×	queue.lock().unwrap().push((c.hash().to_string(), b.to_vec()));
140	0×	}
141	0×	if queue.lock().unwrap().len() >= PARALLEL * 4 {
142	0×	let batch: Vec<_> = std::mem::take(&mut *queue.lock().unwrap());
143	0×	send_batch(&batch, &send, &failed);
144	0×	}
145	0×	match failed.lock().unwrap().take() {
146	0×	Some(e) => Err(images::Error::Bad(format!("{e:#}"))),
147	0×	None => Ok(()),
148		}
149	0×	})?;
150	0×	let rest: Vec<_> = std::mem::take(&mut *queue.lock().unwrap());
151	0×	send_batch(&rest, &send, &failed);
152	0×	if let Some(e) = failed.lock().unwrap().take() {
153	0×	return Err(e);
154	0×	}
155	0×	let url = format!("{}/flow/images/{hash}", self.base);
156		// The hub reads the whole image before it answers, which on a busy
157		// disk takes far longer than the agent's 120 s (oxygenhub#341).
158	0×	self.agent
159	0×	.put(&url)
160	0×	.timeout(Duration::from_secs(3600))
161	0×	.set("Authorization", &self.auth())
162	0×	.send_json(&m)
163	0×	.map_err(\|e\| Self::fail(&url, e))?;
164	0×	let bytes = *bytes.lock().unwrap();
165	0×	Ok(Moved { chunks: count, bytes, peer_bytes: 0, elapsed: start.elapsed() })
166	0×	}
167
168		/// Fetch image `hash` into `home`: the chunks its cache lacks, the
169		/// image written and checked, its architecture recorded. `Ok(None)`
170		/// when the hub does not hold it.
171	0×	pub fn fetch(&self, home: &Home, hash: &str) -> Result<Option<Moved>> {
172	0×	let start = Instant::now();
173	0×	let Some(m) = self.manifest(hash)? else { return Ok(None) };
174	0×	if m.hash != hash {
175	0×	bail!("the hub's manifest for {} names {}", &hash[..12], &m.hash[..12]);
176	0×	}
177	0×	let cache = chunk_cache(home);
178	0×	save_manifest(&home.root, &m)?;
179	0×	let wanted: Vec<String> = m.needed().into_iter().filter(\|h\| !cache_path(&cache, h).is_file()).map(String::from).collect();
180	0×	let (bytes, failed) = (Mutex::new(0u64), Mutex::new(None::<anyhow::Error>));
181	0×	let next = Mutex::new(wanted.iter());
182	0×	std::thread::scope(\|s\| {
183	0×	for _ in 0..PARALLEL {
184	0×	s.spawn(\|\| loop {
185	0×	let Some(h) = next.lock().unwrap().next() else { break };
186	0×	if failed.lock().unwrap().is_some() {
187	0×	break;
188	0×	}
189	0×	let got = self.chunk(h).and_then(\|b\| {
190	0×	if blake3::hash(&b).to_hex().as_str() != h {
191	0×	bail!("chunk {} came back with other bytes", &h[..12]);
192	0×	}
193	0×	let p = cache_path(&cache, h);
194	0×	std::fs::create_dir_all(p.parent().expect("a chunk directory"))?;
195	0×	let tmp = p.with_extension("part");
196	0×	std::fs::write(&tmp, &b)?;
197	0×	std::fs::rename(&tmp, &p)?;
198	0×	*bytes.lock().unwrap() += b.len() as u64;
199	0×	Ok(())
200	0×	});
201	0×	if let Err(e) = got {
202	0×	*failed.lock().unwrap() = Some(e);
203	0×	break;
204	0×	}
205	0×	});
206		}
207	0×	});
208	0×	if let Some(e) = failed.into_inner().unwrap() {
209	0×	return Err(e);
210	0×	}
211	0×	write_image(home, &m)?;
212	0×	Ok(Some(Moved { chunks: wanted.len(), bytes: bytes.into_inner().unwrap(), peer_bytes: 0, elapsed: start.elapsed() }))
213	0×	}
214		}
215
216		/// Write image `m` into `home` from the chunk cache, checked against its
217		/// hash, and record its architecture.
218	6×	pub(crate) fn write_image(home: &Home, m: &Manifest) -> Result<()> {
219	6×	let cache = chunk_cache(home);
220	6×	let out = home.rootfs(&m.hash);
221	6×	std::fs::create_dir_all(out.parent().expect("the cas directory"))?;
222	228×	images::assemble(m, &out, \|c\| Ok(std::fs::read(cache_path(&cache, c.hash()))?))
223	6×	.with_context(\|\| format!("write image {}", &m.hash[..12]))?;
224	6×	if let Some(arch) = &m.arch {
225	2×	let manifest = serde_json::json!({ "arch": arch, "imported_ms": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(\|d\| d.as_millis() as u64).unwrap_or(0), "from": "hub" });
226	2×	std::fs::write(home.image_manifest(&m.hash), serde_json::to_vec(&manifest)?)?;
227	4×	}
228	6×	Ok(())
229	6×	}
230
231		impl Client {
232		/// Tell the hub which images this home holds, and send the manifest of
233		/// each it keeps none for (oxygenflow#94, jetstream.md §8.1.1): the
234		/// swarm can then carry an image nobody pushed. A manifest not kept here
235		/// yet is made by reading the image once, and kept. How many were sent;
236		/// none from a hub that predates it.
237	0×	pub fn offer_held(&self, home: &Home) -> Result<usize> {
238	0×	let images = home.images();
239	0×	if images.is_empty() {
240	0×	return Ok(0);
241	0×	}
242	0×	let url = format!("{}/flow/images/held", self.base);
243	0×	let wanted: flow_core::api::WantedManifests =
244	0×	match self.agent.post(&url).set("Authorization", &self.auth()).send_json(flow_core::api::HeldImages { images }) {
245	0×	Ok(r) => r.into_json()?,
246	0×	Err(ureq::Error::Status(404, _)) => return Ok(0),
247	0×	Err(e) => return Err(Self::fail(&url, e)),
248		};
249	0×	let mut sent = 0;
250	0×	for hash in wanted.wanted.iter().filter(\|h\| images::is_hash(h)) {
251	0×	let path = home.rootfs(hash);
252	0×	if !path.is_file() {
253	0×	continue;
254	0×	}
255	0×	let kept = std::fs::read(manifest_path(&home.root, hash)).ok().and_then(\|b\| serde_json::from_slice::<Manifest>(&b).ok());
256	0×	let m = match kept.filter(\|m\| m.hash == *hash) {
257	0×	Some(m) => m,
258		None => {
259	0×	let m = images::scan_file(&path, home.image_arch(hash), \|_, _\| Ok(()))?;
260	0×	if m.hash != *hash {
261	0×	continue;
262	0×	}
263	0×	save_manifest(&home.root, &m)?;
264	0×	m
265		}
266		};
267	0×	let url = format!("{}/flow/images/{hash}/held", self.base);
268	0×	self.agent.put(&url).set("Authorization", &self.auth()).send_json(&m).map_err(\|e\| Self::fail(&url, e))?;
269	0×	sent += 1;
270		}
271	0×	Ok(sent)
272	0×	}
273		}
274
275		/// Where an image's chunk manifest is kept (`manifests/<hash>.json`), so
276		/// the runner can serve its chunks to peers out of the image itself.
277	14×	pub(crate) fn manifest_path(root: &Path, hash: &str) -> PathBuf {
278	14×	root.join("manifests").join(format!("{hash}.json"))
279	14×	}
280
281	10×	pub(crate) fn save_manifest(root: &Path, m: &Manifest) -> Result<()> {
282	10×	let p = manifest_path(root, &m.hash);
283	10×	std::fs::create_dir_all(p.parent().expect("the manifests directory"))?;
284	10×	let tmp = p.with_extension("part");
285	10×	std::fs::write(&tmp, serde_json::to_vec(m)?)?;
286	10×	std::fs::rename(&tmp, &p)?;
287	10×	Ok(())
288	10×	}
289
290	0×	fn send_batch(batch: &[(String, Vec<u8>)], send: &(dyn Fn(&str, &[u8]) -> Result<()> + Sync), failed: &Mutex<Option<anyhow::Error>>) {
291	0×	std::thread::scope(\|s\| {
292	0×	for part in batch.chunks(batch.len().div_ceil(PARALLEL).max(1)) {
293	0×	s.spawn(move \|\| {
294	0×	for (h, b) in part {
295	0×	if let Err(e) = send(h, b) {
296	0×	*failed.lock().unwrap() = Some(e);
297	0×	return;
298	0×	}
299		}
300	0×	});
301		}
302	0×	});
303	0×	}
304
305		/// The home's chunk cache.
306	13×	pub fn chunk_cache(home: &Home) -> PathBuf {
307	13×	home.root.join("chunks")
308	13×	}
309
310	826×	pub(crate) fn cache_path(cache: &Path, hash: &str) -> PathBuf {
311	826×	cache.join(&hash[..2]).join(hash)
312	826×	}
313
314		/// A size for people: 991 MiB, 1.2 GiB.
315	38×	pub fn human(bytes: u64) -> String {
316	38×	if bytes >= 1 << 30 {
317	0×	format!("{:.1} GiB", bytes as f64 / (1u64 << 30) as f64)
318		} else {
319	38×	format!("{} MiB", bytes >> 20)
320		}
321	38×	}

Coverage / crates/flow/src/images.rs 27.2%

Keyboard