images.rs 92.8%

270 of 291 lines run, 48 of 103 functions · run #516 on main · run missed

//! Images by hash (oxygenflow#39, OxygenFlow docs/jetstream.md §8): rootfs
//! images kept site-wide for runners to fetch, so nobody copies an image
//! between machines. An image arrives as chunks (Oxygen's FastCDC, sent
//! only when this hub lacks them) and then its manifest, which is accepted
//! only when every chunk is here and the image they make has the hash it
//! is named by. Chunks are kept zstd-compressed under
//! `<data>/images/chunks`, shared by every image that has them.

use crate::db::{now_ms, Pool};
use jet::images::{is_hash, Manifest};
use std::io::Write;
use std::path::{Path, PathBuf};

/// Where this hub keeps images.
#[derive(Clone, Debug)]
pub struct Store {
    root: PathBuf,
}

impl Store {
    pub fn new(data: &Path) -> Store {
        Store { root: data.join("images") }
    }

    fn chunk_path(&self, hash: &str) -> PathBuf {
        self.root.join("chunks").join(&hash[..2]).join(hash)
    }

    fn manifest_path(&self, hash: &str) -> PathBuf {
        self.root.join("manifests").join(format!("{hash}.json"))
    }

    pub fn has_chunk(&self, hash: &str) -> bool {
        is_hash(hash) && self.chunk_path(hash).exists()
    }

    /// Keep chunk `hash`, whose bytes the caller has checked.
    pub fn put_chunk(&self, hash: &str, bytes: &[u8]) -> std::io::Result<()> {
        let path = self.chunk_path(hash);
        if path.exists() {
            return Ok(());
        }
        std::fs::create_dir_all(path.parent().expect("a chunk's directory"))?;
        let tmp = path.with_extension(format!("tmp{}", std::process::id()));
        let mut f = std::fs::File::create(&tmp)?;
        f.write_all(&zstd::encode_all(bytes, 3)?)?;
        f.sync_all()?;
        std::fs::rename(&tmp, &path)
    }

    pub fn chunk(&self, hash: &str) -> std::io::Result<Vec<u8>> {
        if !is_hash(hash) {
            return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "not a chunk name"));
        }
        zstd::decode_all(std::fs::File::open(self.chunk_path(hash))?)
    }

    pub fn manifest(&self, hash: &str) -> Option<Manifest> {
        if !is_hash(hash) {
            return None;
        }
        serde_json::from_slice(&std::fs::read(self.manifest_path(hash)).ok()?).ok()
    }

    /// Accept manifest `m`: well formed, every chunk here, and the image
    /// they make hashing to its name. Reads the whole image once.
    pub fn verify_and_keep(&self, m: &Manifest) -> Result<(), String> {
        m.check().map_err(|e| e.to_string())?;
        let mut whole = blake3::Hasher::new();
        for c in &m.chunks {
            if c.is_zeros() {
                whole.update(&vec![0u8; c.len()]);
                continue;
            }
            let bytes = self.chunk(c.hash()).map_err(|_| format!("chunk {} was not sent; send it first", &c.hash()[..12]))?;
            if bytes.len() != c.len() {
                return Err(format!("chunk {} is {} bytes, not {}", &c.hash()[..12], bytes.len(), c.len()));
            }
            whole.update(&bytes);
        }
        let got = whole.finalize().to_hex().to_string();
        if got != m.hash {
            return Err(format!("the chunks make an image named {}, not {}", &got[..12], &m.hash[..12]));
        }
        self.keep_manifest(m)
    }

    /// Keep manifest `m` of an image only runners hold (oxygenflow#94): well
    /// formed, its chunks not here, so not checked against its hash; a
    /// fetcher checks the image it assembles against the pinned hash.
    /// Never over a manifest already kept.
    pub fn keep_peer_manifest(&self, m: &Manifest) -> Result<(), String> {
        m.check().map_err(|e| e.to_string())?;
        if self.manifest_path(&m.hash).exists() {
            return Ok(());
        }
        self.keep_manifest(m)
    }

    /// Write manifest `m`, whose image is here.
    fn keep_manifest(&self, m: &Manifest) -> Result<(), String> {
        let path = self.manifest_path(&m.hash);
        std::fs::create_dir_all(path.parent().expect("the manifests directory")).map_err(|e| e.to_string())?;
        let tmp = path.with_extension("json.tmp");
        std::fs::write(&tmp, serde_json::to_vec(m).map_err(|e| e.to_string())?).map_err(|e| e.to_string())?;
        std::fs::rename(&tmp, &path).map_err(|e| e.to_string())
    }

    /// What image `m` is for, read from its own files (OxygenFlow spec
    /// §11.2, oxygenhub#173): its shell's ELF header, found through its ext4
    /// out of the kept chunks, a few of them, not the whole image. `None`
    /// when it is not ext4 or has no shell to read.
    pub fn read_arch(&self, m: &Manifest) -> Option<&'static str> {
        let mut starts = Vec::with_capacity(m.chunks.len());
        let mut end = 0u64;
        for c in &m.chunks {
            starts.push(end);
            end += c.len() as u64;
        }
        let mut cache: std::collections::HashMap<usize, Vec<u8>> = std::collections::HashMap::new();
        let short = || std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "past the image's end");
        jet::rootfs::arch(&mut |at: u64, buf: &mut [u8]| {
            let mut done = 0usize;
            while done < buf.len() {
                let pos = at + done as u64;
                let i = starts.partition_point(|&s| s <= pos).checked_sub(1).ok_or_else(short)?;
                let c = &m.chunks[i];
                let within = (pos - starts[i]) as usize;
                if within >= c.len() {
                    return Err(short());
                }
                let n = (c.len() - within).min(buf.len() - done);
                if c.is_zeros() {
                    buf[done..done + n].fill(0);
                } else {
                    if !cache.contains_key(&i) {
                        if cache.len() >= 32 {
                            cache.clear();
                        }
                        cache.insert(i, self.chunk(c.hash())?);
                    }
                    let bytes = &cache[&i];
                    if bytes.len() != c.len() {
                        return Err(short());
                    }
                    buf[done..done + n].copy_from_slice(&bytes[within..within + n]);
                }
                done += n;
            }
            Ok(())
        })
    }

    /// Keep image `m`, checked as [`Store::verify_and_keep`], with the
    /// architecture its own files name: the manifest's word is taken only
    /// when the image cannot be read. The manifest as kept.
    pub fn keep_reading_arch(&self, m: &Manifest) -> Result<Manifest, String> {
        self.verify_and_keep(m)?;
        match self.read_arch(m) {
            Some(a) if m.arch.as_deref() != Some(a) => {
                let mut kept = m.clone();
                kept.arch = Some(a.to_string());
                self.keep_manifest(&kept)?;
                Ok(kept)
            }
            _ => Ok(m.clone()),
        }
    }

    /// Record `arch` in kept image `hash`'s manifest.
    fn set_arch(&self, hash: &str, arch: &str) -> Result<(), String> {
        let mut m = self.manifest(hash).ok_or("no such image")?;
        m.arch = Some(arch.to_string());
        self.keep_manifest(&m)
    }
}

/// Record an image this hub now keeps, its chunks and all: one only
/// runners held before becomes the hub's.
pub async fn record(pool: &Pool, m: &Manifest, uploader: &str) -> sqlx::Result<()> {
    sqlx::query(
        "INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms) VALUES ($1, $2, $3, $4, $5, $6) \
         ON CONFLICT (hash) DO UPDATE SET source = 'hub', holder = '', \
         arch = CASE WHEN excluded.arch != '' THEN excluded.arch ELSE jet_images.arch END",
    )
    .bind(&m.hash)
    .bind(m.size as i64)
    .bind(m.arch.clone().unwrap_or_default())
    .bind(m.chunks.len() as i64)
    .bind(uploader)
    .bind(now_ms())
    .execute(pool)
    .await?;
    if let Some(a) = m.arch.as_ref().filter(|a| !a.is_empty()) {
        crate::db::flow::record_image_archs(pool, &[(m.hash.clone(), a.clone())].into_iter().collect()).await?;
    }
    Ok(())
}

/// Learn what every image is for (oxygenhub#173), once at start: read
/// each kept image that has no architecture from its own files, then give
/// any image still unknown the architecture its jobs passed on. A runner's
/// own report, read from the image it holds, wins over what was learned.
/// How many images were read.
pub async fn learn_archs(app: &crate::Shared) -> sqlx::Result<usize> {
    let mut read = 0;
    for k in kept(&app.db).await?.into_iter().filter(|k| k.arch.is_empty()) {
        let store = app.images();
        let hash = k.hash.clone();
        let arch = tokio::task::spawn_blocking(move || {
            let a = store.read_arch(&store.manifest(&hash)?)?;
            store.set_arch(&hash, a).ok()?;
            Some(a)
        })
        .await
        .ok()
        .flatten();
        if let Some(a) = arch {
            sqlx::query("UPDATE jet_images SET arch = $1 WHERE hash = $2").bind(a).bind(&k.hash).execute(&app.db).await?;
            crate::db::flow::record_image_archs(&app.db, &[(k.hash.clone(), a.to_string())].into_iter().collect()).await?;
            read += 1;
        }
    }
    crate::db::flow::learn_archs_from_runs(&app.db).await?;
    Ok(read)
}

/// Record an image only runners hold, whose manifest `holder` sent
/// (oxygenflow#94). Never over one the hub keeps.
pub async fn record_peer(pool: &Pool, m: &Manifest, holder: &str) -> sqlx::Result<()> {
    sqlx::query(
        "INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms, source, holder) VALUES ($1, $2, $3, $4, $5, $6, 'peer', $7) \
         ON CONFLICT (hash) DO NOTHING",
    )
    .bind(&m.hash)
    .bind(m.size as i64)
    .bind(m.arch.clone().unwrap_or_default())
    .bind(m.chunks.len() as i64)
    .bind(holder)
    .bind(now_ms())
    .bind(holder)
    .execute(pool)
    .await?;
    if let Some(a) = m.arch.as_ref().filter(|a| !a.is_empty()) {
        crate::db::flow::record_image_archs(pool, &[(m.hash.clone(), a.clone())].into_iter().collect()).await?;
    }
    Ok(())
}

/// Every image this hub keeps a manifest of, by hash: its own and those
/// only runners hold.
pub async fn held(pool: &Pool) -> sqlx::Result<std::collections::HashSet<String>> {
    Ok(sqlx::query_scalar::<_, String>("SELECT hash FROM jet_images").fetch_all(pool).await?.into_iter().collect())
}

/// Where the images jobs need can come from (oxygenflow#94): the hub's
/// own, pushed and verified, and those only runners hold, each with the
/// swarm sites of the online runners holding it now.
#[derive(Debug, Default)]
pub struct Reach {
    pub hub: std::collections::HashSet<String>,
    pub peers: std::collections::HashMap<String, Vec<String>>,
    /// Each runner in a swarm now, and its site.
    pub sites: std::collections::HashMap<i64, String>,
}

impl Reach {
    /// Whether a runner that fetches images can get `image`: from the hub,
    /// or from a runner of its own site.
    pub fn fetchable(&self, runner: i64, image: &str) -> bool {
        self.hub.contains(image) || self.sites.get(&runner).is_some_and(|s| self.peers.get(image).is_some_and(|v| v.contains(s)))
    }

    /// Every image `runner` can fetch.
    pub fn for_runner(&self, runner: i64) -> std::collections::HashSet<String> {
        let mut out = self.hub.clone();
        out.extend(self.peers.keys().filter(|i| self.fetchable(runner, i)).cloned());
        out
    }

    /// Whether only runners hold `image`.
    pub fn peer_only(&self, image: &str) -> bool {
        !self.hub.contains(image) && self.peers.contains_key(image)
    }
}

/// [`Reach`] now.
pub async fn reach(app: &crate::Shared) -> sqlx::Result<Reach> {
    let rows: Vec<(String, String)> = sqlx::query_as("SELECT hash, source FROM jet_images").fetch_all(&app.db).await?;
    let mut reach = Reach::default();
    for (hash, source) in rows {
        if source == "peer" {
            reach.peers.insert(hash, Vec::new());
        } else {
            reach.hub.insert(hash);
        }
    }
    for r in crate::db::runners::list(&app.db).await? {
        if r.state == "disabled" || !r.online() {
            continue;
        }
        let Some(addr) = app.swarm.addr(r.id) else { continue };
        for image in r.image_list().unwrap_or_default() {
            if let Some(sites) = reach.peers.get_mut(&image.to_ascii_lowercase()) {
                if !sites.contains(&addr.site) {
                    sites.push(addr.site.clone());
                }
            }
        }
        reach.sites.insert(r.id, addr.site);
    }
    Ok(reach)
}

/// An image this hub keeps (`jet_images`).
#[derive(sqlx::FromRow)]
pub struct Kept {
    pub hash: String,
    pub size: i64,
    pub arch: String,
    pub chunks: i64,
    pub uploader: String,
    pub created_ms: i64,
    /// 'hub', or 'peer' for one only runners hold (oxygenflow#94).
    pub source: String,
    pub holder: String,
}

pub async fn kept(pool: &Pool) -> sqlx::Result<Vec<Kept>> {
    sqlx::query_as("SELECT hash, size, arch, chunks, uploader, created_ms, source, holder FROM jet_images").fetch_all(pool).await
}

/// One repository's use of one image: its jobs since `since_ms`, and when
/// it last ran one.
#[derive(sqlx::FromRow)]
pub struct Use {
    pub image: String,
    pub owner: String,
    pub name: String,
    pub jobs: i64,
    pub last_ms: i64,
}

/// What each repository's jobs ran on since `since_ms`, newest first.
pub async fn uses(pool: &Pool, since_ms: i64) -> sqlx::Result<Vec<Use>> {
    let rootfs = crate::db::json_text("j.sandbox", "rootfs");
    sqlx::query_as(&format!(
        "SELECT lower({rootfs}) AS image, p.owner, p.name, COUNT(*) AS jobs,
                MAX(COALESCE(j.started_ms, r.created_ms)) AS last_ms
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id
         WHERE r.created_ms >= $1 AND COALESCE({rootfs}, '') != ''
         GROUP BY lower({rootfs}), p.id, p.owner, p.name ORDER BY last_ms DESC"
    ))
    .bind(since_ms)
    .fetch_all(pool)
    .await
}

/// What one repository's jobs ran on since `since_ms`, newest first
/// (its Images page, oxygenhub#254).
pub async fn uses_of(pool: &Pool, repo_id: i64, since_ms: i64) -> sqlx::Result<Vec<Use>> {
    let rootfs = crate::db::json_text("j.sandbox", "rootfs");
    sqlx::query_as(&format!(
        "SELECT lower({rootfs}) AS image, p.owner, p.name, COUNT(*) AS jobs,
                MAX(COALESCE(j.started_ms, r.created_ms)) AS last_ms
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id
         WHERE r.created_ms >= $1 AND p.id = $2 AND COALESCE({rootfs}, '') != ''
         GROUP BY lower({rootfs}), p.id, p.owner, p.name ORDER BY last_ms DESC"
    ))
    .bind(since_ms)
    .bind(repo_id)
    .fetch_all(pool)
    .await
}

/// The architecture runners reported for each image (`flow_images`).
pub async fn reported_archs(pool: &Pool) -> sqlx::Result<std::collections::HashMap<String, String>> {
    Ok(sqlx::query_as::<_, (String, String)>("SELECT hash, arch FROM flow_images").fetch_all(pool).await?.into_iter().collect())
}

#[cfg(test)]
mod reach_tests {
    use super::Reach;

    #[test]
    fn an_image_only_runners_hold_is_fetchable_in_their_site_alone() {
        let mut r = Reach::default();
        r.hub.insert("hub-image".into());
        r.peers.insert("peer-image".into(), vec!["home".into()]);
        r.sites.insert(1, "home".into());
        r.sites.insert(2, "cloud".into());
        assert!(r.fetchable(1, "hub-image") && r.fetchable(2, "hub-image") && r.fetchable(3, "hub-image"));
        assert!(r.fetchable(1, "peer-image"), "a runner of the holder's site");
        assert!(!r.fetchable(2, "peer-image"), "another site");
        assert!(!r.fetchable(3, "peer-image"), "a runner in no swarm");
        assert!(r.peer_only("peer-image") && !r.peer_only("hub-image"));
        assert_eq!(r.for_runner(1).len(), 2);
        assert_eq!(r.for_runner(2).len(), 1);
    }
}

`/`	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 (oxygenflow#39, OxygenFlow docs/jetstream.md §8): rootfs
2		//! images kept site-wide for runners to fetch, so nobody copies an image
3		//! between machines. An image arrives as chunks (Oxygen's FastCDC, sent
4		//! only when this hub lacks them) and then its manifest, which is accepted
5		//! only when every chunk is here and the image they make has the hash it
6		//! is named by. Chunks are kept zstd-compressed under
7		//! `<data>/images/chunks`, shared by every image that has them.
8
9		use crate::db::{now_ms, Pool};
10		use jet::images::{is_hash, Manifest};
11		use std::io::Write;
12		use std::path::{Path, PathBuf};
13
14		/// Where this hub keeps images.
15		#[derive(Clone, Debug)]
16		pub struct Store {
17		root: PathBuf,
18		}
19
20		impl Store {
21	31×	pub fn new(data: &Path) -> Store {
22	31×	Store { root: data.join("images") }
23	31×	}
24
25	60039×	fn chunk_path(&self, hash: &str) -> PathBuf {
26	60039×	self.root.join("chunks").join(&hash[..2]).join(hash)
27	60039×	}
28
29	21×	fn manifest_path(&self, hash: &str) -> PathBuf {
30	21×	self.root.join("manifests").join(format!("{hash}.json"))
31	21×	}
32
33	60006×	pub fn has_chunk(&self, hash: &str) -> bool {
34	60006×	is_hash(hash) && self.chunk_path(hash).exists()
35	60006×	}
36
37		/// Keep chunk `hash`, whose bytes the caller has checked.
38	10×	pub fn put_chunk(&self, hash: &str, bytes: &[u8]) -> std::io::Result<()> {
39	10×	let path = self.chunk_path(hash);
40	10×	if path.exists() {
41	0×	return Ok(());
42	10×	}
43	10×	std::fs::create_dir_all(path.parent().expect("a chunk's directory"))?;
44	10×	let tmp = path.with_extension(format!("tmp{}", std::process::id()));
45	10×	let mut f = std::fs::File::create(&tmp)?;
46	10×	f.write_all(&zstd::encode_all(bytes, 3)?)?;
47	10×	f.sync_all()?;
48	10×	std::fs::rename(&tmp, &path)
49	10×	}
50
51	23×	pub fn chunk(&self, hash: &str) -> std::io::Result<Vec<u8>> {
52	23×	if !is_hash(hash) {
53	0×	return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "not a chunk name"));
54	23×	}
55	23×	zstd::decode_all(std::fs::File::open(self.chunk_path(hash))?)
56	23×	}
57
58	14×	pub fn manifest(&self, hash: &str) -> Option<Manifest> {
59	14×	if !is_hash(hash) {
60	0×	return None;
61	14×	}
62	14×	serde_json::from_slice(&std::fs::read(self.manifest_path(hash)).ok()?).ok()
63	14×	}
64
65		/// Accept manifest `m`: well formed, every chunk here, and the image
66		/// they make hashing to its name. Reads the whole image once.
67	5×	pub fn verify_and_keep(&self, m: &Manifest) -> Result<(), String> {
68	5×	m.check().map_err(\|e\| e.to_string())?;
69	5×	let mut whole = blake3::Hasher::new();
70	55×	for c in &m.chunks {
71	55×	if c.is_zeros() {
72	43×	whole.update(&vec![0u8; c.len()]);
73	43×	continue;
74	12×	}
75	12×	let bytes = self.chunk(c.hash()).map_err(\|_\| format!("chunk {} was not sent; send it first", &c.hash()[..12]))?;
76	11×	if bytes.len() != c.len() {
77	0×	return Err(format!("chunk {} is {} bytes, not {}", &c.hash()[..12], bytes.len(), c.len()));
78	11×	}
79	11×	whole.update(&bytes);
80		}
81	4×	let got = whole.finalize().to_hex().to_string();
82	4×	if got != m.hash {
83	1×	return Err(format!("the chunks make an image named {}, not {}", &got[..12], &m.hash[..12]));
84	3×	}
85	3×	self.keep_manifest(m)
86	5×	}
87
88		/// Keep manifest `m` of an image only runners hold (oxygenflow#94): well
89		/// formed, its chunks not here, so not checked against its hash; a
90		/// fetcher checks the image it assembles against the pinned hash.
91		/// Never over a manifest already kept.
92	1×	pub fn keep_peer_manifest(&self, m: &Manifest) -> Result<(), String> {
93	1×	m.check().map_err(\|e\| e.to_string())?;
94	1×	if self.manifest_path(&m.hash).exists() {
95	0×	return Ok(());
96	1×	}
97	1×	self.keep_manifest(m)
98	1×	}
99
100		/// Write manifest `m`, whose image is here.
101	6×	fn keep_manifest(&self, m: &Manifest) -> Result<(), String> {
102	6×	let path = self.manifest_path(&m.hash);
103	6×	std::fs::create_dir_all(path.parent().expect("the manifests directory")).map_err(\|e\| e.to_string())?;
104	6×	let tmp = path.with_extension("json.tmp");
105	6×	std::fs::write(&tmp, serde_json::to_vec(m).map_err(\|e\| e.to_string())?).map_err(\|e\| e.to_string())?;
106	6×	std::fs::rename(&tmp, &path).map_err(\|e\| e.to_string())
107	6×	}
108
109		/// What image `m` is for, read from its own files (OxygenFlow spec
110		/// §11.2, oxygenhub#173): its shell's ELF header, found through its ext4
111		/// out of the kept chunks, a few of them, not the whole image. `None`
112		/// when it is not ext4 or has no shell to read.
113	5×	pub fn read_arch(&self, m: &Manifest) -> Option<&'static str> {
114	5×	let mut starts = Vec::with_capacity(m.chunks.len());
115	5×	let mut end = 0u64;
116	78×	for c in &m.chunks {
117	78×	starts.push(end);
118	78×	end += c.len() as u64;
119	78×	}
120	5×	let mut cache: std::collections::HashMap<usize, Vec<u8>> = std::collections::HashMap::new();
121	5×	let short = \|\| std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "past the image's end");
122	31×	jet::rootfs::arch(&mut \|at: u64, buf: &mut [u8]\| {
123	31×	let mut done = 0usize;
124	62×	while done < buf.len() {
125	31×	let pos = at + done as u64;
126	177×	let i = starts.partition_point(\|&s\| s <= pos).checked_sub(1).ok_or_else(short)?;
127	31×	let c = &m.chunks[i];
128	31×	let within = (pos - starts[i]) as usize;
129	31×	if within >= c.len() {
130	0×	return Err(short());
131	31×	}
132	31×	let n = (c.len() - within).min(buf.len() - done);
133	31×	if c.is_zeros() {
134	0×	buf[done..done + n].fill(0);
135	0×	} else {
136	31×	if !cache.contains_key(&i) {
137	7×	if cache.len() >= 32 {
138	0×	cache.clear();
139	7×	}
140	7×	cache.insert(i, self.chunk(c.hash())?);
141	24×	}
142	31×	let bytes = &cache[&i];
143	31×	if bytes.len() != c.len() {
144	0×	return Err(short());
145	31×	}
146	31×	buf[done..done + n].copy_from_slice(&bytes[within..within + n]);
147		}
148	31×	done += n;
149		}
150	31×	Ok(())
151	31×	})
152	5×	}
153
154		/// Keep image `m`, checked as [`Store::verify_and_keep`], with the
155		/// architecture its own files name: the manifest's word is taken only
156		/// when the image cannot be read. The manifest as kept.
157	5×	pub fn keep_reading_arch(&self, m: &Manifest) -> Result<Manifest, String> {
158	5×	self.verify_and_keep(m)?;
159	3×	match self.read_arch(m) {
160	1×	Some(a) if m.arch.as_deref() != Some(a) => {
161	1×	let mut kept = m.clone();
162	1×	kept.arch = Some(a.to_string());
163	1×	self.keep_manifest(&kept)?;
164	1×	Ok(kept)
165		}
166	2×	_ => Ok(m.clone()),
167		}
168	5×	}
169
170		/// Record `arch` in kept image `hash`'s manifest.
171	1×	fn set_arch(&self, hash: &str, arch: &str) -> Result<(), String> {
172	1×	let mut m = self.manifest(hash).ok_or("no such image")?;
173	1×	m.arch = Some(arch.to_string());
174	1×	self.keep_manifest(&m)
175	1×	}
176		}
177
178		/// Record an image this hub now keeps, its chunks and all: one only
179		/// runners held before becomes the hub's.
180	5×	pub async fn record(pool: &Pool, m: &Manifest, uploader: &str) -> sqlx::Result<()> {
181	5×	sqlx::query(
182	5×	"INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms) VALUES ($1, $2, $3, $4, $5, $6) \
183	5×	ON CONFLICT (hash) DO UPDATE SET source = 'hub', holder = '', \
184	5×	arch = CASE WHEN excluded.arch != '' THEN excluded.arch ELSE jet_images.arch END",
185	5×	)
186	5×	.bind(&m.hash)
187	5×	.bind(m.size as i64)
188	5×	.bind(m.arch.clone().unwrap_or_default())
189	5×	.bind(m.chunks.len() as i64)
190	5×	.bind(uploader)
191	5×	.bind(now_ms())
192	5×	.execute(pool)
193	5×	.await?;
194	5×	if let Some(a) = m.arch.as_ref().filter(\|a\| !a.is_empty()) {
195	4×	crate::db::flow::record_image_archs(pool, &[(m.hash.clone(), a.clone())].into_iter().collect()).await?;
196	1×	}
197	5×	Ok(())
198	5×	}
199
200		/// Learn what every image is for (oxygenhub#173), once at start: read
201		/// each kept image that has no architecture from its own files, then give
202		/// any image still unknown the architecture its jobs passed on. A runner's
203		/// own report, read from the image it holds, wins over what was learned.
204		/// How many images were read.
205	198×	pub async fn learn_archs(app: &crate::Shared) -> sqlx::Result<usize> {
206	198×	let mut read = 0;
207	198×	for k in kept(&app.db).await?.into_iter().filter(\|k\| k.arch.is_empty()) {
208	2×	let store = app.images();
209	2×	let hash = k.hash.clone();
210	2×	let arch = tokio::task::spawn_blocking(move \|\| {
211	2×	let a = store.read_arch(&store.manifest(&hash)?)?;
212	1×	store.set_arch(&hash, a).ok()?;
213	1×	Some(a)
214	2×	})
215	2×	.await
216	2×	.ok()
217	2×	.flatten();
218	2×	if let Some(a) = arch {
219	1×	sqlx::query("UPDATE jet_images SET arch = $1 WHERE hash = $2").bind(a).bind(&k.hash).execute(&app.db).await?;
220	1×	crate::db::flow::record_image_archs(&app.db, &[(k.hash.clone(), a.to_string())].into_iter().collect()).await?;
221	1×	read += 1;
222	1×	}
223		}
224	198×	crate::db::flow::learn_archs_from_runs(&app.db).await?;
225	198×	Ok(read)
226	198×	}
227
228		/// Record an image only runners hold, whose manifest `holder` sent
229		/// (oxygenflow#94). Never over one the hub keeps.
230	2×	pub async fn record_peer(pool: &Pool, m: &Manifest, holder: &str) -> sqlx::Result<()> {
231	2×	sqlx::query(
232	2×	"INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms, source, holder) VALUES ($1, $2, $3, $4, $5, $6, 'peer', $7) \
233	2×	ON CONFLICT (hash) DO NOTHING",
234	2×	)
235	2×	.bind(&m.hash)
236	2×	.bind(m.size as i64)
237	2×	.bind(m.arch.clone().unwrap_or_default())
238	2×	.bind(m.chunks.len() as i64)
239	2×	.bind(holder)
240	2×	.bind(now_ms())
241	2×	.bind(holder)
242	2×	.execute(pool)
243	2×	.await?;
244	2×	if let Some(a) = m.arch.as_ref().filter(\|a\| !a.is_empty()) {
245	2×	crate::db::flow::record_image_archs(pool, &[(m.hash.clone(), a.clone())].into_iter().collect()).await?;
246	0×	}
247	2×	Ok(())
248	2×	}
249
250		/// Every image this hub keeps a manifest of, by hash: its own and those
251		/// only runners hold.
252	140×	pub async fn held(pool: &Pool) -> sqlx::Result<std::collections::HashSet<String>> {
253	140×	Ok(sqlx::query_scalar::<_, String>("SELECT hash FROM jet_images").fetch_all(pool).await?.into_iter().collect())
254	140×	}
255
256		/// Where the images jobs need can come from (oxygenflow#94): the hub's
257		/// own, pushed and verified, and those only runners hold, each with the
258		/// swarm sites of the online runners holding it now.
259		#[derive(Debug, Default)]
260		pub struct Reach {
261		pub hub: std::collections::HashSet<String>,
262		pub peers: std::collections::HashMap<String, Vec<String>>,
263		/// Each runner in a swarm now, and its site.
264		pub sites: std::collections::HashMap<i64, String>,
265		}
266
267		impl Reach {
268		/// Whether a runner that fetches images can get `image`: from the hub,
269		/// or from a runner of its own site.
270	11×	pub fn fetchable(&self, runner: i64, image: &str) -> bool {
271	11×	self.hub.contains(image) \|\| self.sites.get(&runner).is_some_and(\|s\| self.peers.get(image).is_some_and(\|v\| v.contains(s)))
272	11×	}
273
274		/// Every image `runner` can fetch.
275	16×	pub fn for_runner(&self, runner: i64) -> std::collections::HashSet<String> {
276	16×	let mut out = self.hub.clone();
277	16×	out.extend(self.peers.keys().filter(\|i\| self.fetchable(runner, i)).cloned());
278	16×	out
279	16×	}
280
281		/// Whether only runners hold `image`.
282	4×	pub fn peer_only(&self, image: &str) -> bool {
283	4×	!self.hub.contains(image) && self.peers.contains_key(image)
284	4×	}
285		}
286
287		/// [`Reach`] now.
288	249×	pub async fn reach(app: &crate::Shared) -> sqlx::Result<Reach> {
289	249×	let rows: Vec<(String, String)> = sqlx::query_as("SELECT hash, source FROM jet_images").fetch_all(&app.db).await?;
290	249×	let mut reach = Reach::default();
291	249×	for (hash, source) in rows {
292	26×	if source == "peer" {
293	1×	reach.peers.insert(hash, Vec::new());
294	25×	} else {
295	25×	reach.hub.insert(hash);
296	25×	}
297		}
298	309×	for r in crate::db::runners::list(&app.db).await? {
299	309×	if r.state == "disabled" \|\| !r.online() {
300	3×	continue;
301	306×	}
302	306×	let Some(addr) = app.swarm.addr(r.id) else { continue };
303	13×	for image in r.image_list().unwrap_or_default() {
304	5×	if let Some(sites) = reach.peers.get_mut(&image.to_ascii_lowercase()) {
305	0×	if !sites.contains(&addr.site) {
306	0×	sites.push(addr.site.clone());
307	0×	}
308	5×	}
309		}
310	13×	reach.sites.insert(r.id, addr.site);
311		}
312	249×	Ok(reach)
313	249×	}
314
315		/// An image this hub keeps (`jet_images`).
316		#[derive(sqlx::FromRow)]
317		pub struct Kept {
318		pub hash: String,
319		pub size: i64,
320		pub arch: String,
321		pub chunks: i64,
322		pub uploader: String,
323		pub created_ms: i64,
324		/// 'hub', or 'peer' for one only runners hold (oxygenflow#94).
325		pub source: String,
326		pub holder: String,
327		}
328
329	203×	pub async fn kept(pool: &Pool) -> sqlx::Result<Vec<Kept>> {
330	203×	sqlx::query_as("SELECT hash, size, arch, chunks, uploader, created_ms, source, holder FROM jet_images").fetch_all(pool).await
331	203×	}
332
333		/// One repository's use of one image: its jobs since `since_ms`, and when
334		/// it last ran one.
335		#[derive(sqlx::FromRow)]
336		pub struct Use {
337		pub image: String,
338		pub owner: String,
339		pub name: String,
340		pub jobs: i64,
341		pub last_ms: i64,
342		}
343
344		/// What each repository's jobs ran on since `since_ms`, newest first.
345	2×	pub async fn uses(pool: &Pool, since_ms: i64) -> sqlx::Result<Vec<Use>> {
346	2×	let rootfs = crate::db::json_text("j.sandbox", "rootfs");
347	2×	sqlx::query_as(&format!(
348	2×	"SELECT lower({rootfs}) AS image, p.owner, p.name, COUNT(*) AS jobs,
349	2×	MAX(COALESCE(j.started_ms, r.created_ms)) AS last_ms
350	2×	FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id
351	2×	WHERE r.created_ms >= $1 AND COALESCE({rootfs}, '') != ''
352	2×	GROUP BY lower({rootfs}), p.id, p.owner, p.name ORDER BY last_ms DESC"
353	2×	))
354	2×	.bind(since_ms)
355	2×	.fetch_all(pool)
356	2×	.await
357	2×	}
358
359		/// What one repository's jobs ran on since `since_ms`, newest first
360		/// (its Images page, oxygenhub#254).
361	1×	pub async fn uses_of(pool: &Pool, repo_id: i64, since_ms: i64) -> sqlx::Result<Vec<Use>> {
362	1×	let rootfs = crate::db::json_text("j.sandbox", "rootfs");
363	1×	sqlx::query_as(&format!(
364	1×	"SELECT lower({rootfs}) AS image, p.owner, p.name, COUNT(*) AS jobs,
365	1×	MAX(COALESCE(j.started_ms, r.created_ms)) AS last_ms
366	1×	FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id
367	1×	WHERE r.created_ms >= $1 AND p.id = $2 AND COALESCE({rootfs}, '') != ''
368	1×	GROUP BY lower({rootfs}), p.id, p.owner, p.name ORDER BY last_ms DESC"
369	1×	))
370	1×	.bind(since_ms)
371	1×	.bind(repo_id)
372	1×	.fetch_all(pool)
373	1×	.await
374	1×	}
375
376		/// The architecture runners reported for each image (`flow_images`).
377	2×	pub async fn reported_archs(pool: &Pool) -> sqlx::Result<std::collections::HashMap<String, String>> {
378	2×	Ok(sqlx::query_as::<_, (String, String)>("SELECT hash, arch FROM flow_images").fetch_all(pool).await?.into_iter().collect())
379	2×	}
380
381		#[cfg(test)]
382		mod reach_tests {
383		use super::Reach;
384
385		#[test]
386	1×	fn an_image_only_runners_hold_is_fetchable_in_their_site_alone() {
387	1×	let mut r = Reach::default();
388	1×	r.hub.insert("hub-image".into());
389	1×	r.peers.insert("peer-image".into(), vec!["home".into()]);
390	1×	r.sites.insert(1, "home".into());
391	1×	r.sites.insert(2, "cloud".into());
392	1×	assert!(r.fetchable(1, "hub-image") && r.fetchable(2, "hub-image") && r.fetchable(3, "hub-image"));
393	1×	assert!(r.fetchable(1, "peer-image"), "a runner of the holder's site");
394	1×	assert!(!r.fetchable(2, "peer-image"), "another site");
395	1×	assert!(!r.fetchable(3, "peer-image"), "a runner in no swarm");
396	1×	assert!(r.peer_only("peer-image") && !r.peer_only("hub-image"));
397	1×	assert_eq!(r.for_runner(1).len(), 2);
398	1×	assert_eq!(r.for_runner(2).len(), 1);
399	1×	}
400		}

Coverage / crates/hub/src/images.rs 92.8%

Keyboard