src/pool.rs
//! **The pool — ONE content-addressed store, keyed by sha256.**
//!
//! Extracted from the moln collector (2026-08), where it was born to fix what the Go
//! `airgap-kubernetes` measurably lacks: `grep -niE 'sha256|checksum|digest|sign'` over
//! its `*.go` → **0 hits**, and its container leg writes one docker-archive tar per
//! image, so a base layer shared by five images is stored five times. Here blobs land
//! in one content-addressed pool keyed by digest — **not** one archive per artefact —
//! so shared bytes are stored once and dedup is free rather than engineered.
//!
//! ## Layout
//!
//! ```text
//! <root>/blobs/sha256/<aa>/<bb>/<64-hex> the bytes, named by what they are
//! <root>/names/<binding>.json a stable name → the digest it points at
//! ```
//!
//! Two levels of fan-out, `aa`/`bb`, because a real pool is thousands of OCI layers,
//! not tens, and one flat directory of those is a `readdir` nobody wants.
//!
//! ## What this is NOT
//!
//! It is **not a bundle format**. Skidbladnir's `Archive` + `BundleManifest`
//! (`archive.rs:45-75`) is the packer and owns manifests, roots and signatures; this
//! crate deliberately writes none of those. `names/*.json` is the pool's own catalogue
//! — the thing that lets a cut ask for "the Talos ISO" without already knowing its
//! digest — and it is two fields wide on purpose so nobody mistakes it for a manifest.
//!
//! ## No delta here
//!
//! The pool once carried a `--since` snapshot family (`snapshot`, `delta_since`,
//! `PoolDelta`); nothing outside its own tests called it, so it was removed
//! (2026-10-22, holger stinky 9). A bundle's delta is `archive.rs`'s; sub-file delta at
//! ISO scale lives in znippy's `RegionDeltaSplitter`.
use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result, bail};
use serde::{Deserialize, Serialize};
/// A lowercase-hex SHA-256, validated on construction.
///
/// A newtype and not a `String` because every path in this module is derived from it:
/// an unvalidated digest with a `/` or a `..` in it is a directory traversal with the
/// pool's own hand on the pen.
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(try_from = "String", into = "String")]
pub struct Digest(String);
impl Digest {
/// Validate and normalise. Accepts an OCI-style `sha256:<hex>` prefix, because that
/// is the literal string a registry manifest hands over and stripping it at each
/// call site is how one call site forgets.
pub fn parse(s: &str) -> Result<Digest> {
let hex = s.strip_prefix("sha256:").unwrap_or(s).trim();
if hex.len() != 64 || !hex.bytes().all(|b| b.is_ascii_hexdigit()) {
bail!(
"`{s}` is not a sha256 digest (want 64 hex chars, optionally `sha256:`-prefixed; \
got {} chars)",
hex.len()
);
}
Ok(Digest(hex.to_ascii_lowercase()))
}
/// The bytes' digest.
pub fn of_bytes(bytes: &[u8]) -> Digest {
Digest(nornir_hash::sha256_hex(bytes))
}
/// The file's digest, streamed — a 4 GB ISO must be hashable without being loaded.
pub fn of_file(path: &Path) -> Result<Digest> {
let hex =
nornir_hash::sha256_file(path).with_context(|| format!("hash {}", path.display()))?;
Ok(Digest(hex))
}
pub fn hex(&self) -> &str {
&self.0
}
/// The OCI spelling, for the registry arm and for anything that logs it.
pub fn prefixed(&self) -> String {
format!("sha256:{}", self.0)
}
fn shard(&self) -> (&str, &str) {
(&self.0[0..2], &self.0[2..4])
}
}
impl fmt::Display for Digest {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.0)
}
}
impl TryFrom<String> for Digest {
type Error = anyhow::Error;
fn try_from(s: String) -> Result<Digest> {
Digest::parse(&s)
}
}
impl From<Digest> for String {
fn from(d: Digest) -> String {
d.0
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The binding — a stable name for a digest
// ─────────────────────────────────────────────────────────────────────────────
/// What a name points at, and where it came from.
///
/// `source` is kept because a pool full of anonymous digests cannot answer the one
/// question an operator always asks first — *where did this come from* — and a
/// content-addressed store that has to be re-derived to answer it is a store that gets
/// re-derived.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Binding {
pub digest: Digest,
pub len: u64,
/// The URL or image coordinate this content was fetched from.
pub source: String,
}
// ─────────────────────────────────────────────────────────────────────────────
// The pool
// ─────────────────────────────────────────────────────────────────────────────
/// A content-addressed blob pool on a directory.
#[derive(Clone, Debug)]
pub struct Pool {
root: PathBuf,
}
/// What an ingest did. `stored` is separate from `digest` because "we already had these
/// bytes" is the dedup win, and a caller that cannot see it cannot report it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Ingested {
pub digest: Digest,
pub len: u64,
/// `false` when the pool already held this content — the deduplicated case.
pub stored: bool,
}
impl Pool {
/// Open (creating) a pool rooted at `root`.
pub fn open(root: impl Into<PathBuf>) -> Result<Pool> {
let root = root.into();
for sub in ["blobs/sha256", "names"] {
let d = root.join(sub);
std::fs::create_dir_all(&d)
.with_context(|| format!("create pool directory {}", d.display()))?;
}
Ok(Pool { root })
}
pub fn root(&self) -> &Path {
&self.root
}
/// Where `digest`'s bytes live, whether or not they are there yet.
pub fn blob_path(&self, digest: &Digest) -> PathBuf {
let (a, b) = digest.shard();
self.root
.join("blobs/sha256")
.join(a)
.join(b)
.join(digest.hex())
}
pub fn has(&self, digest: &Digest) -> bool {
self.blob_path(digest).is_file()
}
pub fn len_of(&self, digest: &Digest) -> Result<u64> {
let p = self.blob_path(digest);
Ok(std::fs::metadata(&p)
.with_context(|| format!("stat blob {}", p.display()))?
.len())
}
pub fn read(&self, digest: &Digest) -> Result<Vec<u8>> {
let p = self.blob_path(digest);
std::fs::read(&p).with_context(|| format!("read blob {}", p.display()))
}
/// **Take `src` into the pool by content.** Idempotent, and cheap: the file is
/// **hardlinked** into the blob path, not copied, so a 4 GB ISO that is also kept
/// under a human name in a staging directory costs 4 GB once and two directory
/// entries. A cross-device `src` falls back to a copy.
///
/// `src` is left where it is. That is what makes the staging directory a working
/// download cache: the next run sees the file, `HttpRange` short-circuits, and no
/// byte crosses the network.
pub fn ingest_file(&self, src: &Path) -> Result<Ingested> {
let digest = Digest::of_file(src)?;
let len = std::fs::metadata(src)
.with_context(|| format!("stat {}", src.display()))?
.len();
let dest = self.blob_path(&digest);
if dest.is_file() {
return Ok(Ingested {
digest,
len,
stored: false,
});
}
let parent = dest.parent().expect("blob path always has a parent");
std::fs::create_dir_all(parent)
.with_context(|| format!("create blob shard {}", parent.display()))?;
// Stage under a temp name in the SAME shard directory, then rename. A reader
// that lists the pool mid-ingest must never see a short file wearing a digest
// name — the whole point of naming a file after its content is that the name is
// a promise.
let staging = parent.join(format!(".{}.incoming", digest.hex()));
let _ = std::fs::remove_file(&staging);
if std::fs::hard_link(src, &staging).is_err() {
std::fs::copy(src, &staging)
.with_context(|| format!("copy {} → {}", src.display(), staging.display()))?;
}
std::fs::rename(&staging, &dest)
.with_context(|| format!("finalise blob {}", dest.display()))?;
Ok(Ingested {
digest,
len,
stored: true,
})
}
/// Take `bytes` into the pool. The OCI arm's shape, when it exists — a manifest or a
/// small config blob never needs a file on the way in.
pub fn ingest_bytes(&self, bytes: &[u8]) -> Result<Ingested> {
let digest = Digest::of_bytes(bytes);
let dest = self.blob_path(&digest);
if dest.is_file() {
return Ok(Ingested {
digest,
len: bytes.len() as u64,
stored: false,
});
}
let parent = dest.parent().expect("blob path always has a parent");
std::fs::create_dir_all(parent)?;
let staging = parent.join(format!(".{}.incoming", digest.hex()));
std::fs::write(&staging, bytes).with_context(|| format!("write {}", staging.display()))?;
std::fs::rename(&staging, &dest)?;
Ok(Ingested {
digest,
len: bytes.len() as u64,
stored: true,
})
}
/// Point a stable `name` at a digest the pool holds.
///
/// Refuses a digest that is not in the pool. A dangling binding is worse than a
/// missing one: a cut that resolves it gets a path, opens nothing, and reports a
/// bundle that is short one ISO.
pub fn bind(&self, name: &str, digest: &Digest, source: &str) -> Result<()> {
if !self.has(digest) {
bail!(
"refusing to bind `{name}` → {digest}: the pool does not hold those bytes. \
A binding that resolves to nothing is how a cut ships a bundle with a hole in it"
);
}
let len = self.len_of(digest)?;
let b = Binding {
digest: digest.clone(),
len,
source: source.to_string(),
};
let p = self.name_path(name)?;
let tmp = p.with_extension("json.incoming");
std::fs::write(&tmp, serde_json::to_vec_pretty(&b)?)
.with_context(|| format!("write {}", tmp.display()))?;
std::fs::rename(&tmp, &p).with_context(|| format!("finalise {}", p.display()))?;
Ok(())
}
/// Resolve a name, or `None` if nothing is bound to it.
pub fn resolve(&self, name: &str) -> Result<Option<Binding>> {
let p = self.name_path(name)?;
if !p.is_file() {
return Ok(None);
}
let raw = std::fs::read(&p).with_context(|| format!("read {}", p.display()))?;
Ok(Some(serde_json::from_slice(&raw).with_context(|| {
format!("parse binding {}", p.display())
})?))
}
/// Every binding, by name.
pub fn bindings(&self) -> Result<BTreeMap<String, Binding>> {
let dir = self.root.join("names");
let mut out = BTreeMap::new();
for e in std::fs::read_dir(&dir).with_context(|| format!("list {}", dir.display()))? {
let e = e?;
let file = e.file_name();
let file = file.to_string_lossy();
let Some(name) = file.strip_suffix(".json") else {
continue;
};
if let Some(b) = self.resolve(name)? {
out.insert(name.to_string(), b);
}
}
Ok(out)
}
/// **Every digest the pool holds.**
///
/// Reads the directory names, not the file contents: the pool's whole premise is
/// that a blob's name IS its digest, so re-hashing to enumerate would be paying for
/// the property twice. [`Pool::verify`] is the separate, explicit act of checking
/// that the promise held.
pub fn digests(&self) -> Result<BTreeSet<Digest>> {
let base = self.root.join("blobs/sha256");
let mut out = BTreeSet::new();
for a in read_dir_sorted(&base)? {
for b in read_dir_sorted(&a)? {
for f in read_dir_sorted(&b)? {
let Some(name) = f.file_name().and_then(|n| n.to_str()) else {
continue;
};
// `.<hex>.incoming` staging files are deliberately skipped: they are
// not yet content, and counting them would make a torn ingest look
// like a pool member.
if let Ok(d) = Digest::parse(name) {
out.insert(d);
}
}
}
}
Ok(out)
}
/// Total bytes held.
pub fn total_bytes(&self) -> Result<u64> {
let mut n = 0u64;
for d in self.digests()? {
n += self.len_of(&d)?;
}
Ok(n)
}
/// **Re-hash every blob and report the ones whose bytes no longer match their name.**
///
/// Separate from [`Pool::digests`] and never called by it, because a store that
/// verified on every listing would be too slow to list and would therefore stop
/// being listed. Empty result = clean.
pub fn verify(&self) -> Result<Vec<Digest>> {
let mut bad = Vec::new();
for d in self.digests()? {
match Digest::of_file(&self.blob_path(&d)) {
Ok(actual) if actual == d => {}
_ => bad.push(d),
}
}
Ok(bad)
}
fn name_path(&self, name: &str) -> Result<PathBuf> {
safe_component(name, "binding name")?;
Ok(self.root.join("names").join(format!("{name}.json")))
}
}
/// A name that becomes a filename must not become a path.
pub(crate) fn safe_component(s: &str, what: &str) -> Result<()> {
if s.is_empty()
|| s == "."
|| s == ".."
|| s.contains('/')
|| s.contains('\\')
|| s.contains('\0')
{
bail!("`{s}` is not a usable {what} — it must be a single path component");
}
Ok(())
}
fn read_dir_sorted(p: &Path) -> Result<Vec<PathBuf>> {
if !p.is_dir() {
return Ok(Vec::new());
}
let mut v: Vec<PathBuf> = std::fs::read_dir(p)
.with_context(|| format!("list {}", p.display()))?
.filter_map(|e| e.ok().map(|e| e.path()))
.collect();
v.sort();
Ok(v)
}
// ─────────────────────────────────────────────────────────────────────────────
// RED PROBES
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
fn pool() -> (tempfile::TempDir, Pool) {
let d = tempfile::tempdir().unwrap();
let p = Pool::open(d.path()).unwrap();
(d, p)
}
/// ★ The whole reason the pool exists: **the same bytes under two names are stored
/// ONCE.** Asserted on the filesystem — one blob file, one total byte count — not on
/// a bookkeeping counter that could agree with itself while the disk disagreed.
///
/// **SEEN RED 2026-08-07 — probe E**: `ingest_file` re-pointed at
/// `Digest::of_bytes(src.file_name())`, i.e. keyed by NAME instead of by content —
/// which is precisely what `airgap-kubernetes` does with its one-tar-per-image layout
/// (`bundle.go:433-456`). `pool.rs — assertion left == right failed` on the two
/// digests, and four sibling tests fell with it.
#[test]
fn identical_content_under_two_names_is_stored_once() {
let (d, p) = pool();
let layer = vec![7u8; 64 * 1024];
let a = d.path().join("image-a-layer.tar");
let b = d.path().join("image-b-layer.tar");
std::fs::write(&a, &layer).unwrap();
std::fs::write(&b, &layer).unwrap();
let ia = p.ingest_file(&a).unwrap();
let ib = p.ingest_file(&b).unwrap();
assert_eq!(ia.digest, ib.digest);
assert!(ia.stored, "the first ingest stores");
assert!(!ib.stored, "the second ingest deduplicates");
// APPLIED OUTPUT: exactly one blob on disk, and the pool's total is one layer,
// not two. This is the number `bundle.go:433-456` cannot produce.
assert_eq!(p.digests().unwrap().len(), 1);
assert_eq!(p.total_bytes().unwrap(), layer.len() as u64);
p.bind("image-a", &ia.digest, "ghcr.io/a:1").unwrap();
p.bind("image-b", &ib.digest, "quay.io/b:2").unwrap();
let names = p.bindings().unwrap();
assert_eq!(names.len(), 2, "two names");
assert_eq!(names["image-a"].digest, names["image-b"].digest, "one blob");
assert_eq!(names["image-a"].source, "ghcr.io/a:1");
assert_eq!(
p.digests().unwrap().len(),
1,
"still one blob after binding"
);
}
/// The bytes come back out byte-identical, and the blob really is at the sharded
/// path the digest names.
#[test]
fn ingested_bytes_round_trip_at_the_digest_path() {
let (_d, p) = pool();
let content = b"metal-amd64.iso would go here".to_vec();
let i = p.ingest_bytes(&content).unwrap();
assert_eq!(p.read(&i.digest).unwrap(), content);
let expected = p
.root()
.join("blobs/sha256")
.join(&i.digest.hex()[0..2])
.join(&i.digest.hex()[2..4])
.join(i.digest.hex());
assert!(expected.is_file(), "blob at {}", expected.display());
assert_eq!(p.blob_path(&i.digest), expected);
assert_eq!(
i.digest.hex(),
nornir_hash::sha256_hex(&content),
"the key is the content's sha256 and nothing else"
);
}
/// ★ **RED — a binding to a digest the pool does not hold is REFUSED.** A dangling
/// binding is how a cut ships a bundle with a hole in it and reports success.
#[test]
fn a_binding_to_an_absent_digest_is_refused() {
let (_d, p) = pool();
let ghost = Digest::of_bytes(b"never ingested");
let e = p
.bind("talos-iso", &ghost, "https://example/iso")
.expect_err("a dangling binding must refuse");
assert!(e.to_string().contains("does not hold those bytes"), "{e}");
assert!(
p.resolve("talos-iso").unwrap().is_none(),
"and nothing was written"
);
}
/// ★ **RED — a torn ingest never appears as a pool member.** The staging file lives
/// in the shard directory next to real blobs, so a listing that globbed the
/// directory would count it. `digests()` parses the name, so it cannot.
#[test]
fn a_torn_ingest_staging_file_is_not_a_pool_member() {
let (_d, p) = pool();
let real = p.ingest_bytes(b"complete").unwrap();
let shard = p.blob_path(&real.digest).parent().unwrap().to_path_buf();
let torn = shard.join(format!(".{}.incoming", "ab".repeat(32)));
std::fs::write(&torn, b"half a").unwrap();
assert!(torn.is_file(), "the torn file is genuinely there");
assert_eq!(
p.digests().unwrap().len(),
1,
"a `.incoming` file is not content"
);
}
/// ★ **RED — `verify` finds a blob whose bytes stopped matching its name.** A
/// content-addressed store whose verify cannot fail is a filename convention.
#[test]
fn verify_catches_a_blob_whose_bytes_no_longer_match_its_name() {
let (_d, p) = pool();
let good = p.ingest_bytes(b"the real thing").unwrap();
let also = p.ingest_bytes(b"another blob entirely").unwrap();
assert!(
p.verify().unwrap().is_empty(),
"a fresh pool verifies clean"
);
std::fs::write(p.blob_path(&good.digest), b"tampered").unwrap();
let bad = p.verify().unwrap();
assert_eq!(bad, vec![good.digest.clone()], "exactly the tampered blob");
assert!(!bad.contains(&also.digest), "the untouched blob is fine");
}
/// ★ **RED — a digest is validated, so a name can never be a path.** The pool derives
/// every path from a digest; an unvalidated one is a traversal with the pool's own
/// hand on the pen.
#[test]
fn a_digest_is_validated_and_a_binding_name_cannot_escape_the_pool() {
assert!(Digest::parse("../../etc/passwd").is_err());
assert!(Digest::parse("").is_err());
assert!(Digest::parse(&"g".repeat(64)).is_err(), "not hex");
assert!(Digest::parse(&"a".repeat(63)).is_err(), "too short");
// The OCI spelling is accepted and normalised.
let d = Digest::parse(&format!("sha256:{}", "AB".repeat(32))).unwrap();
assert_eq!(d.hex(), "ab".repeat(32));
assert_eq!(d.prefixed(), format!("sha256:{}", "ab".repeat(32)));
let (_t, p) = pool();
let i = p.ingest_bytes(b"x").unwrap();
for bad in ["../escape", "a/b", "..", ""] {
assert!(
p.bind(bad, &i.digest, "src").is_err(),
"`{bad}` must be refused as a binding name"
);
}
}
/// A pool re-opened on the same directory sees what the last process left. This is
/// what "self-heals across restarts" rests on — if the pool were not durable, the
/// daemon's restart-safety would be a claim about an empty directory.
#[test]
fn a_reopened_pool_sees_the_previous_processs_blobs_and_bindings() {
let d = tempfile::tempdir().unwrap();
let digest = {
let p = Pool::open(d.path()).unwrap();
let i = p.ingest_bytes(b"survives a restart").unwrap();
p.bind("iso", &i.digest, "https://example/iso").unwrap();
i.digest
};
let p2 = Pool::open(d.path()).unwrap();
assert!(p2.has(&digest));
assert_eq!(p2.resolve("iso").unwrap().unwrap().digest, digest);
}
/// stinky (holger) 9: the pool keeps no `--since` snapshot family. Nothing cut a
/// bundle `--since` a snapshot and nothing differenced two pool states outside
/// these tests; holger's twin follows its primary by manifest and watermark, and a
/// bundle's own delta is `archive.rs`'s. A primitive that exists only in its own
/// tests is a promise nobody keeps, so it is gone.
#[test]
fn the_pool_keeps_no_uncalled_since_snapshot_family() {
let src = include_str!("pool.rs");
for dead in [
["pub fn delta", "_since("].concat(),
["pub struct Pool", "Delta"].concat(),
["pub fn read", "_snapshot("].concat(),
["pub fn bytes", "_of("].concat(),
] {
assert!(!src.contains(&dead), "pool.rs still carries `{dead}`");
}
}
}