traits/src/znippy_base.rs
//! **`ZnippyBase`** — what every read-only `znippy-*` repository backend holds
//! and does the same way: the name, the archive (typed and as a plain reader),
//! the change-token index, the repo-prefix strip, the 404/text replies and the
//! one SHA-256 helper. A backend keeps a `base: ZnippyBase` and supplies only
//! its format logic; [`znippy_read_only!`](crate::znippy_read_only) writes the
//! five trait methods that are the same everywhere.
use std::collections::HashMap;
use std::fs::File;
use std::path::Path;
use std::sync::{Arc, OnceLock};
use znippy_common::index::{locate_file, ChunkLoc};
use znippy_common::meta_sink_append::read_delta_map;
use znippy_common::{ZnippyArchive, ZnippyReader};
use crate::backend::{unsupported, Archive, Cap, ChunkIter, HttpReply, StreamedGet};
use crate::{ArchiveTokens, ArtifactEntry, ArtifactId, Bytes};
/// A file at least this large is answered in pieces ([`ZnippyBase::stream_get`]); a
/// smaller one is read whole, which costs less than a stream and holds little.
pub const STREAM_MIN_BYTES: u64 = 1024 * 1024;
pub struct ZnippyBase {
name: String,
/// The typed archive, when the backend was opened from a file (the typed
/// views — `as_maven()`, `as_python()`, … — live on it).
archive: Option<Arc<ZnippyArchive>>,
/// The archive as a plain reader: the typed archive above, or the reader
/// handed to [`with_reader`](Self::with_reader).
reader: Option<Arc<dyn ZnippyReader>>,
tokens: ArchiveTokens,
/// What a read in pieces needs (lane HOL-RSS); only with an archive file.
pieces: Option<Pieces>,
}
/// **Reads in pieces** (lane HOL-RSS, 2026-11-01). `extract_file` hands back a file's
/// bytes whole, so a download held the artifact in memory until the client had it:
/// holperf on oden (the real pip set, 2026-11-01) peaked at 570 MiB of live heap for 32
/// downloads, each holding its wheel. A file's chunks are where the archive's lookup
/// sub-index says ([`locate_file`]); they are read one piece at a time, a compressed
/// chunk decoded alone, so a download holds one piece or one chunk.
///
/// A delta chunk (an archive's delta map names it: a superseded generation stored as
/// instructions against another file) is rebuilt alone too, against its base, and its
/// checksum checked, as `extract_file` does: what it holds is the base and that chunk
/// (lane HOL-RSS part 2).
struct Pieces {
/// The archive file, opened beside the typed archive.
file: Arc<File>,
/// Its length when opened: every chunk's extent is checked against it.
len: u64,
/// `(device, inode)` when opened: the lookup is read by path, so a path that names
/// another file by now (a re-sealed archive) is not read in pieces.
id: (u64, u64),
/// The archive's delta map, read once: each file's delta chunks and their bases.
/// `None` when it cannot be read, and then nothing is read in pieces.
deltas: OnceLock<Option<DeltaMap>>,
/// Each streamed file's chunks, looked up once.
plans: DocMemo<Arc<Plan>>,
}
/// Each file's delta chunks: `(chunk_seq, base path)`.
type DeltaMap = HashMap<String, Vec<(u32, String)>>;
/// One file's chunks in file order, and the base of each one that is a delta.
struct Plan {
locs: Vec<ChunkLoc>,
bases: Vec<Option<String>>,
}
fn file_id(m: &std::fs::Metadata) -> (u64, u64) {
use std::os::unix::fs::MetadataExt;
(m.dev(), m.ino())
}
impl Pieces {
fn open(path: &Path) -> Option<Self> {
let file = File::open(path).ok()?;
let m = file.metadata().ok()?;
Some(Self {
id: file_id(&m),
len: m.len(),
file: Arc::new(file),
deltas: OnceLock::new(),
plans: DocMemo::new(4096),
})
}
/// The chunks of `rel` (`size` bytes) in file order, when they can be read one
/// after the other into exactly the file: contiguous, each stored chunk as long as
/// what it holds, all inside the archive.
fn plan(&self, path: &Path, rel: &str, size: u64) -> Option<Arc<Plan>> {
let deltas = self
.deltas
.get_or_init(|| {
let rows = read_delta_map(path).ok()?;
let mut by_file: DeltaMap = HashMap::new();
for (p, seq, base) in rows {
by_file.entry(p).or_default().push((seq, base));
}
Some(by_file)
})
.as_ref()?;
self.plans.get_or(rel, || {
let same = std::fs::metadata(path).ok().map(|m| file_id(&m)) == Some(self.id);
if !same {
return None;
}
let locs = locate_file(path, rel).ok()?;
let of_file = deltas.get(rel);
let bases: Vec<Option<String>> = locs
.iter()
.map(|c| {
of_file
.and_then(|d| d.iter().find(|(seq, _)| *seq == c.chunk_seq))
.map(|(_, base)| base.clone())
})
.collect();
plan_holds(&locs, &bases, size, self.len).then(|| Arc::new(Plan { locs, bases }))
})
}
}
/// `locs` are `size` bytes laid end to end from offset 0, in `chunk_seq` order, every
/// blob inside an archive of `archive_len` bytes. `bases[i]` names chunk `i`'s base
/// when it is a delta chunk (its blob is instructions, any length).
fn plan_holds(locs: &[ChunkLoc], bases: &[Option<String>], size: u64, archive_len: u64) -> bool {
let mut at = 0u64;
let mut seq = None;
for (c, base) in locs.iter().zip(bases) {
let in_bounds = c
.blob_offset
.checked_add(c.blob_size)
.is_some_and(|end| end <= archive_len);
let stored_whole = base.is_some() || c.compressed || c.blob_size == c.uncompressed_size;
if !in_bounds
|| !stored_whole
|| c.fdata_offset != at
|| seq.is_some_and(|s| c.chunk_seq <= s)
{
return false;
}
seq = Some(c.chunk_seq);
at += c.uncompressed_size;
}
at == size && (size == 0 || !locs.is_empty())
}
/// Slack a delta chunk's instruction stream may have over the bytes it produces
/// (znippy's own bound: the two size varints, one opcode byte per 127 inserted bytes).
const DELTA_FRAME_SLACK: u64 = 64 * 1024;
/// One file's pieces, read as they are pulled: a stored chunk `piece` bytes at a time,
/// a compressed one decoded whole and handed out in pieces, a delta chunk rebuilt
/// against its base and handed out in pieces.
struct ArchivePieces {
file: Arc<File>,
/// The archive, which reads a delta chunk's base (itself perhaps a delta).
archive: Arc<ZnippyArchive>,
plan: Arc<Plan>,
/// The chunk being read, and how far into it.
chunk: usize,
at: u64,
piece: usize,
/// A decoded chunk and how much of it is out.
decoded: Option<(Bytes, usize)>,
done: bool,
}
impl ArchivePieces {
fn step(&mut self) -> anyhow::Result<Option<Bytes>> {
use std::os::unix::fs::FileExt;
loop {
if let Some((whole, out)) = &mut self.decoded {
if *out < whole.len() {
let end = (*out + self.piece).min(whole.len());
let p = whole.slice(*out..end);
*out = end;
return Ok(Some(p));
}
self.decoded = None;
self.chunk += 1;
continue;
}
let Some(c) = self.plan.locs.get(self.chunk) else {
return Ok(None);
};
let base = self.plan.bases[self.chunk].as_deref();
if base.is_none() && !c.compressed {
if self.at >= c.blob_size {
self.chunk += 1;
self.at = 0;
continue;
}
let n = (c.blob_size - self.at).min(self.piece as u64) as usize;
let mut buf = vec![0u8; n];
self.file.read_exact_at(&mut buf, c.blob_offset + self.at)?;
self.at += n as u64;
return Ok(Some(Bytes::from(buf)));
}
let chunk = decode_chunk(&self.file, &self.archive, c, base)?;
self.decoded = Some((Bytes::from(chunk), 0));
}
}
}
/// Chunk `c` decoded whole: a compressed chunk decompressed and its length checked; a
/// delta chunk (`base` names its base) its instructions (decoded when compressed) applied
/// to `base`, checked against the checksum its index row carries, as `extract_file` does.
/// A stored chunk is its blob.
fn decode_chunk(
file: &File,
archive: &ZnippyArchive,
c: &ChunkLoc,
base: Option<&str>,
) -> anyhow::Result<Vec<u8>> {
use std::os::unix::fs::FileExt;
let mut blob = vec![0u8; c.blob_size as usize];
file.read_exact_at(&mut blob, c.blob_offset)?;
let Some(base) = base else {
if !c.compressed {
return Ok(blob);
}
let mut out = Vec::new();
let limit = usize::try_from(c.uncompressed_size).unwrap_or(usize::MAX);
znippy_common::codec::decompress_into_max(&blob, &mut out, limit)?;
anyhow::ensure!(
out.len() as u64 == c.uncompressed_size,
"a chunk decoded to {} bytes, its index row says {}",
out.len(),
c.uncompressed_size
);
return Ok(out);
};
let instructions = if c.compressed {
let limit = c
.uncompressed_size
.saturating_add(c.uncompressed_size / 64)
.saturating_add(DELTA_FRAME_SLACK);
let mut out = Vec::new();
znippy_common::codec::decompress_into_max(
&blob,
&mut out,
usize::try_from(limit).unwrap_or(usize::MAX),
)?;
out
} else {
blob
};
let base_bytes = archive.extract_file(base)?;
let rebuilt = znippy_common::archive::apply_delta(&base_bytes, &instructions)?;
drop(base_bytes);
anyhow::ensure!(
rebuilt.len() as u64 == c.uncompressed_size,
"a delta chunk rebuilt {} bytes against {base}, its index row says {}",
rebuilt.len(),
c.uncompressed_size
);
anyhow::ensure!(
c.checksum == [0u8; 32]
|| znippy_common::blake3::hash(&rebuilt).as_bytes()[..] == c.checksum[..],
"a delta chunk rebuilt against {base} does not match its checksum"
);
Ok(rebuilt)
}
/// **A file of the archive read at any offset** (lane HOL-RSS-3): what reading one
/// member of a zip needs — its end, its central directory, the member — without the
/// rest of it. A read touches the chunks that hold the bytes asked for: a stored chunk
/// is read where it lies, a compressed or delta chunk is decoded alone and kept for the
/// next read inside it, so a [`RangedFile`] holds at most one chunk.
pub struct RangedFile {
file: Arc<File>,
archive: Arc<ZnippyArchive>,
plan: Arc<Plan>,
len: u64,
/// The last compressed or delta chunk decoded: its index in the plan and its bytes.
held: Option<(usize, Vec<u8>)>,
}
impl RangedFile {
/// The file's length in bytes.
pub fn len(&self) -> u64 {
self.len
}
/// Whether the file is empty.
pub fn is_empty(&self) -> bool {
self.len == 0
}
/// Exactly `buf.len()` bytes from `offset`; an error past the end, never a short read.
pub fn read_at(&mut self, offset: u64, buf: &mut [u8]) -> anyhow::Result<()> {
use std::os::unix::fs::FileExt;
let end = offset
.checked_add(buf.len() as u64)
.filter(|&e| e <= self.len)
.ok_or_else(|| {
anyhow::anyhow!(
"a read of {} bytes at {offset} is past the {} byte file",
buf.len(),
self.len
)
})?;
let (mut at, mut out) = (offset, 0usize);
while at < end {
// The chunk holding `at`: the last one starting at or before it.
let i = self
.plan
.locs
.partition_point(|c| c.fdata_offset <= at)
.checked_sub(1)
.ok_or_else(|| anyhow::anyhow!("no chunk holds offset {at}"))?;
let c = &self.plan.locs[i];
let into = at - c.fdata_offset;
let n = c.uncompressed_size.saturating_sub(into).min(end - at) as usize;
anyhow::ensure!(n > 0, "no chunk holds offset {at}");
let base = self.plan.bases[i].as_deref();
if base.is_none() && !c.compressed {
self.file
.read_exact_at(&mut buf[out..out + n], c.blob_offset + into)?;
} else {
if self.held.as_ref().map(|(h, _)| *h) != Some(i) {
// The chunk held before is let go before the next is decoded.
self.held = None;
let bytes = decode_chunk(&self.file, &self.archive, c, base)?;
self.held = Some((i, bytes));
}
let Some((_, bytes)) = &self.held else {
unreachable!("a chunk was just decoded")
};
let from = into as usize;
buf[out..out + n].copy_from_slice(&bytes[from..from + n]);
}
at += n as u64;
out += n;
}
Ok(())
}
}
impl Iterator for ArchivePieces {
type Item = anyhow::Result<Bytes>;
fn next(&mut self) -> Option<Self::Item> {
if self.done {
return None;
}
match self.step() {
Ok(Some(p)) => Some(Ok(p)),
Ok(None) => {
self.done = true;
None
}
Err(e) => {
self.done = true;
Some(Err(e))
}
}
}
}
impl ZnippyBase {
/// No archive: every read misses and [`Caps::archive`](crate::Caps::archive) is off.
pub fn new(name: impl Into<String>) -> Self {
Self {
name: name.into(),
archive: None,
reader: None,
tokens: ArchiveTokens::none(),
pieces: None,
}
}
/// Open the archive at `path`; its lookup sub-index answers change tokens.
pub fn with_archive(name: impl Into<String>, path: impl AsRef<Path>) -> anyhow::Result<Self> {
let path = path.as_ref();
let archive = Arc::new(ZnippyArchive::open(path)?);
Ok(Self {
name: name.into(),
reader: Some(Arc::clone(&archive) as Arc<dyn ZnippyReader>),
archive: Some(archive),
tokens: ArchiveTokens::at(path),
pieces: Pieces::open(path),
})
}
/// Any reader. No typed views, no path, so no change tokens.
pub fn with_reader(name: impl Into<String>, reader: Arc<dyn ZnippyReader>) -> Self {
Self {
name: name.into(),
archive: None,
reader: Some(reader),
tokens: ArchiveTokens::none(),
pieces: None,
}
}
/// A writable archive at `path` that may not exist yet: no reader is held
/// (the backend re-opens per read), but change tokens are read from `path`.
pub fn writable(name: impl Into<String>, path: impl AsRef<Path>) -> Self {
Self {
name: name.into(),
archive: None,
reader: None,
tokens: ArchiveTokens::at(path.as_ref()),
pieces: None,
}
}
pub fn name(&self) -> &str {
&self.name
}
/// The typed archive, when opened from a file.
pub fn typed(&self) -> Option<&Arc<ZnippyArchive>> {
self.archive.as_ref()
}
/// The archive as a plain reader.
pub fn reader(&self) -> Option<&Arc<dyn ZnippyReader>> {
self.reader.as_ref()
}
pub fn has_archive(&self) -> bool {
self.reader.is_some()
}
pub fn tokens(&self) -> &ArchiveTokens {
&self.tokens
}
/// The archive's file table, or empty with no archive.
pub fn list_files(&self) -> Vec<String> {
self.reader
.as_ref()
.and_then(|r| r.list_files().ok())
.unwrap_or_default()
}
/// One file's bytes out of the archive.
pub fn get_file(&self, relative_path: &str) -> anyhow::Result<Vec<u8>> {
let reader = self
.reader
.as_ref()
.ok_or_else(|| anyhow::anyhow!("No reader configured"))?;
reader.extract_file(relative_path)
}
/// **`rel` in pieces of at most `piece` bytes** (lane HOL-RSS): `None` when it is not
/// in the archive, when the backend has no archive file, or when its chunks cannot be
/// read one after the other (see [`Pieces`]); the caller then reads it whole.
pub fn file_pieces(&self, rel: &str, piece: usize) -> Option<ChunkIter> {
let pieces = self.pieces.as_ref()?;
let path = self.tokens.path()?;
let archive = self.archive.as_ref()?;
let size = archive.file_size(rel)?;
let plan = pieces.plan(path, rel, size)?;
Some(ChunkIter {
len: Some(size),
pieces: Box::new(ArchivePieces {
file: Arc::clone(&pieces.file),
archive: Arc::clone(archive),
plan,
chunk: 0,
at: 0,
piece: piece.max(1),
decoded: None,
done: false,
}),
})
}
/// **`rel` read at any offset** (lane HOL-RSS-3): `None` exactly where
/// [`file_pieces`](Self::file_pieces) is; the caller then reads it whole.
pub fn file_at(&self, rel: &str) -> Option<RangedFile> {
let pieces = self.pieces.as_ref()?;
let path = self.tokens.path()?;
let archive = self.archive.as_ref()?;
let size = archive.file_size(rel)?;
let plan = pieces.plan(path, rel, size)?;
Some(RangedFile {
file: Arc::clone(&pieces.file),
archive: Arc::clone(archive),
plan,
len: size,
held: None,
})
}
/// **A download answered in pieces** (lane HOL-RSS): the trait's
/// [`stream_get`](crate::RepositoryBackendTrait::stream_get) for archive file `rel`,
/// served as `content_type`. Only a file of at least [`STREAM_MIN_BYTES`] is
/// streamed; `Ok(None)` (the whole-body door) for a smaller one and for anything
/// [`file_pieces`](Self::file_pieces) cannot read in pieces.
pub fn stream_get(
&self,
rel: &str,
content_type: &str,
piece: usize,
) -> anyhow::Result<Option<StreamedGet>> {
let big = self
.reader
.as_ref()
.and_then(|r| r.file_size(rel))
.is_some_and(|n| n >= STREAM_MIN_BYTES);
if !big {
return Ok(None);
}
Ok(self.file_pieces(rel, piece).map(|body| StreamedGet {
headers: vec![("Content-Type".into(), content_type.into())],
body,
}))
}
/// **Every byte of `rel`, handed to `f` a piece at a time** (lane HOL-RSS): what a
/// digest over a file needs, without the file in memory. Read whole where it cannot
/// be read in pieces.
pub fn for_each_piece(&self, rel: &str, mut f: impl FnMut(&[u8])) -> anyhow::Result<()> {
if let Some(pieces) = self.file_pieces(rel, 1024 * 1024) {
for p in pieces {
f(&p?);
}
return Ok(());
}
f(&self.get_file(rel)?);
Ok(())
}
/// SHA-256 (lowercase hex) of archive file `rel`, hashed a piece at a time.
pub fn sha256_hex_of(&self, rel: &str) -> anyhow::Result<String> {
let mut h = nornir_hash::Sha256Hasher::new();
self.for_each_piece(rel, |p| h.update(p))?;
Ok(h.finalize_hex())
}
/// The trait's `archive()`: the paths and sizes, `Unsupported` with no archive.
pub fn archive(&self) -> anyhow::Result<Archive> {
match &self.reader {
Some(r) => Archive::read(&self.name, r.as_ref()),
None => unsupported(&self.name, Cap::Archive),
}
}
/// The trait's `put()` on a read-only archive.
pub fn refuse_put(&self) -> anyhow::Result<()> {
anyhow::bail!("znippy repository '{}' is read-only", self.name)
}
/// `suburl` without its leading `/` and `/<repo>/` prefix.
pub fn strip_repo<'a>(&self, suburl: &'a str) -> &'a str {
strip_repo(&self.name, suburl)
}
/// The shared archive listing: every path `coordinate` names, with a change
/// token where the archive's lookup sub-index can answer one.
pub fn list_entries<F>(
&self,
name_filter: Option<&str>,
limit: usize,
content_type: &str,
coordinate: F,
) -> anyhow::Result<Vec<ArtifactEntry>>
where
F: Fn(&str) -> Option<ArtifactId>,
{
let Some(reader) = &self.reader else {
return Ok(Vec::new());
};
let tokens = self.tokens.open();
crate::list_archive_entries(
reader.as_ref(),
tokens.as_ref(),
name_filter,
limit,
content_type,
coordinate,
)
}
}
/// `suburl` without its query (`?…`): the path a download names.
pub fn path_only(suburl: &str) -> &str {
suburl.split('?').next().unwrap_or(suburl)
}
/// `suburl` without its leading `/` and `/<repo>/` prefix.
pub fn strip_repo<'a>(repo: &str, suburl: &'a str) -> &'a str {
let path = suburl.trim_start_matches('/');
path.strip_prefix(repo)
.map(|r| r.trim_start_matches('/'))
.unwrap_or(path)
}
/// A `404` with a plain-text reason.
pub fn not_found(reason: &str) -> HttpReply {
(404, Vec::new(), reason.as_bytes().to_vec())
}
/// A `200` with one `Content-Type`.
pub fn ok(content_type: &str, body: Vec<u8>) -> HttpReply {
(
200,
vec![("Content-Type".into(), content_type.into())],
body,
)
}
/// **A bounded memo of rendered documents** (lane HOL-PERF-2): an index a backend over an
/// immutable archive renders (a `maven-metadata.xml`, a checksum, a `@v/list`, a simple
/// page, a listing), kept by its key and handed out as the same `Bytes`. Only a document
/// that exists is kept; past `cap` keys the memo starts again, so request paths cannot
/// grow it without bound.
pub struct DocMemo<V: Clone = crate::Bytes> {
cap: usize,
map: std::sync::Mutex<std::collections::HashMap<String, V>>,
}
impl<V: Clone> DocMemo<V> {
pub fn new(cap: usize) -> Self {
Self {
cap,
map: std::sync::Mutex::new(std::collections::HashMap::new()),
}
}
/// The document at `key`, built by `build` the first time; `None` (nothing there) is
/// never kept.
pub fn get_or(&self, key: &str, build: impl FnOnce() -> Option<V>) -> Option<V> {
if let Some(hit) = self.map.lock().unwrap_or_else(|p| p.into_inner()).get(key) {
return Some(hit.clone());
}
let built = build()?;
let mut map = self.map.lock().unwrap_or_else(|p| p.into_inner());
if map.len() >= self.cap {
map.clear();
}
map.insert(key.to_string(), built.clone());
Some(built)
}
}
impl<V: Clone> Default for DocMemo<V> {
fn default() -> Self {
Self::new(16_384)
}
}
/// [`ok`] with a refcounted body (lane HOL-PERF-2).
pub fn ok_bytes(content_type: &str, body: crate::Bytes) -> crate::HttpBytesReply {
(
200,
vec![("Content-Type".into(), content_type.into())],
body,
)
}
/// [`not_found`] as an [`crate::HttpBytesReply`].
pub fn not_found_bytes(reason: &str) -> crate::HttpBytesReply {
(
404,
Vec::new(),
crate::Bytes::copy_from_slice(reason.as_bytes()),
)
}
/// SHA-256 of `data`, lowercase hex — the digest pip, apt, dnf, cargo and OCI
/// clients verify served bytes against. The ONE helper the backends share.
pub fn sha256_hex(data: &[u8]) -> String {
nornir_hash::sha256_hex(data)
}
/// The five trait methods every read-only znippy backend spells the same way:
/// `name`, `is_writable` (false), `put` (refused), `caps` (a listing, plus the
/// archive while one is loaded) and `archive`. Invoke inside the
/// `impl RepositoryBackendTrait` block with the `ZnippyBase` field's name.
#[macro_export]
macro_rules! znippy_read_only {
($base:ident) => {
fn name(&self) -> &str {
self.$base.name()
}
fn is_writable(&self) -> bool {
false
}
fn put(&self, _id: &$crate::ArtifactId, _data: &[u8]) -> $crate::anyhow::Result<()> {
self.$base.refuse_put()
}
fn caps(&self) -> $crate::Caps {
$crate::Caps::znippy(self.$base.has_archive())
}
fn archive(&self) -> $crate::anyhow::Result<$crate::Archive> {
self.$base.archive()
}
};
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_repo_prefix_is_stripped_once() {
assert_eq!(strip_repo("pkgs", "/pkgs/a/b"), "a/b");
assert_eq!(strip_repo("pkgs", "pkgs/a"), "a");
assert_eq!(strip_repo("pkgs", "/other/a"), "other/a");
}
#[test]
fn no_archive_is_unsupported_not_empty() {
let b = ZnippyBase::new("r");
assert!(!b.has_archive());
assert!(crate::is_unsupported(&b.archive().unwrap_err()));
assert!(b.refuse_put().is_err());
assert!(b.list_entries(None, 0, "x", |_| None).unwrap().is_empty());
}
#[test]
fn sha256_hex_is_the_known_vector() {
assert_eq!(
sha256_hex(b"abc"),
"ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
);
}
}