traits/src/znippy_format.rs
//! **Reader-side znippy format-version gate** — holger's own pin on the on-disk
//! archive format it is willing to parse.
//!
//! ## Why this exists (the defect it closes)
//!
//! A znippy archive records the on-disk format version it was written with, in the
//! Arrow schema metadata of every sub-index (`znippy_format_version`). Holger is an
//! artifact registry: an archive can arrive from *anywhere* — an agent across the
//! airgap, a USB stick, a pull-through cache, a client `PUT`. If a **newer** znippy
//! wrote it, an older reader parsing it is reading a layout it does not know: at
//! best a silent mis-parse, at worst an arrow downcast panic or a wild allocation
//! inside the request handler. `.nornir/artifactory-parity-design.md` records the
//! observed consequence — *"a newer-written archive crashes an older reader and
//! takes the server down"* — and rates it P0. A registry any client can kill by
//! uploading is not a registry.
//!
//! ## The contract
//!
//! * The archive's recorded version is read **before** any of its rows are parsed.
//! * A version **greater** than [`MAX_SUPPORTED_ZNIPPY_FORMAT`], or a recorded
//! version that is not a plain number, is **refused** with an
//! [`ArchiveFormatError`] — a typed, request-facing error that names the archive,
//! what it recorded, and what this build reads.
//! * A version equal to or lower than the pin, and an archive with **no** recorded
//! version (legacy / pre-version archives), read **exactly as before** — this gate
//! is additive and refuses nothing that works today.
//! * Anything that stops us *determining* the version (unreadable file, corrupt
//! manifest, a panic inside the index reader) is **not** this gate's verdict: it
//! returns "undetermined" and the normal open path reports its own error as
//! before. The gate never converts an existing failure mode into a new one.
//!
//! ## Deliberately independent of znippy's own check
//!
//! `znippy-common` grew an internal `check_format_version` on its read path, and
//! that is welcome — but holger must not *depend* on it. holger pins znippy by
//! semver range and is built against published crates in release (the local
//! `[patch.crates-io]` is a dev convenience), so the reader linked into a given
//! holger binary may predate that check entirely. This gate is holger's, sits in
//! front of the reader, and holds regardless of which znippy is linked. Defence in
//! depth on the one path a hostile input reaches first.
//!
//! ## Cost
//!
//! [`recorded_format_version`] reads the manifest and **one** sub-index's Arrow
//! stream schema — not the archive, not the whole index. For the per-request
//! re-open path (writable dev-mode repos re-open their `.znippy` on every read)
//! use [`FormatGate`], which memoises the verdict against the archive's
//! `(len, mtime)` so the steady-state cost of the gate is a single `stat`.
use std::fmt;
use std::path::Path;
use std::sync::Mutex;
/// The highest on-disk znippy archive format version this holger build parses.
///
/// Bump this **only** together with a reader that actually understands the new
/// layout. It is deliberately a holger-side constant rather than a re-export of
/// `znippy_common::index::ZNIPPY_FORMAT_VERSION`: the pin must be a statement
/// about *what holger can read*, and it must keep compiling (and refusing) when
/// holger is built against a znippy that has no such constant.
pub const MAX_SUPPORTED_ZNIPPY_FORMAT: u32 = 3;
/// Arrow schema-metadata key under which a znippy archive records its on-disk
/// format version. Matches `znippy_common::index::FORMAT_VERSION_KEY`.
pub const ZNIPPY_FORMAT_VERSION_KEY: &str = "znippy_format_version";
/// A znippy archive this build refuses to read: it records a format version that
/// is newer than [`MAX_SUPPORTED_ZNIPPY_FORMAT`], or one that is not a number.
///
/// Carried as a typed error (not a bare string) so a request handler can
/// `downcast_ref` it and answer the *request* — 503 with this message — instead of
/// letting a mis-parse loose in the process.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ArchiveFormatError {
/// The archive that was refused.
pub path: String,
/// Exactly what the archive recorded, verbatim (it may not be a number).
pub recorded: String,
/// The highest format version this build reads.
pub max_supported: u32,
}
impl fmt::Display for ArchiveFormatError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"holger refuses znippy archive {}: it records format version {}, \
which this build does not read (supports up to v{}) — re-pack the \
archive with a matching znippy, or upgrade holger",
self.path, self.recorded, self.max_supported
)
}
}
impl std::error::Error for ArchiveFormatError {}
/// The archive's recorded format version, verbatim, or `None` when it cannot be
/// determined (no recorded version, unreadable file, corrupt manifest, or a panic
/// inside the index reader — all of which the normal open path reports itself).
///
/// Reads the manifest plus the Arrow stream **schema** of the first non-reserved
/// sub-index; no archive rows and no blob bytes are touched. The whole read runs
/// under [`std::panic::catch_unwind`] because `path` is attacker-controlled input
/// and the point of this function is that hostile bytes must not escape it.
pub fn recorded_format_version(path: &Path) -> Option<String> {
let path = path.to_path_buf();
std::panic::catch_unwind(move || read_recorded_version(&path))
.ok()
.flatten()
}
fn read_recorded_version(path: &Path) -> Option<String> {
use std::io::{Read, Seek, SeekFrom};
let entries = znippy_common::read_znippy_manifest(path).ok()?;
let mut file = std::fs::File::open(path).ok()?;
let file_len = file.metadata().ok()?.len();
for entry in &entries {
// Reserved modules are derived structures (lookup sub-index, trie blob,
// signature sections) — not Arrow file-row sub-indexes; the trie is not
// even Arrow IPC. Never parse them.
if znippy_common::is_reserved_module(&entry.module_name) {
continue;
}
// The declared extent comes out of the archive, so bound it against the
// real file size BEFORE allocating — a corrupt manifest must not be able
// to ask for a multi-gigabyte zero-fill.
let end = entry.index_offset.checked_add(entry.index_len)?;
if end > file_len {
return None;
}
file.seek(SeekFrom::Start(entry.index_offset)).ok()?;
let mut bytes = vec![0u8; entry.index_len as usize];
file.read_exact(&mut bytes).ok()?;
// znippy's IPC guard checks the framing, body lengths and buffers
// before arrow allocates anything: a hostile body length is an abort
// that the catch_unwind above cannot stop.
let reader = znippy_common::ipc_guard::stream_reader(bytes).ok()?;
return reader
.schema()
.metadata()
.get(ZNIPPY_FORMAT_VERSION_KEY)
.cloned();
}
None
}
/// Refuse the archive at `path` if it records a format version this build cannot
/// read. `Ok(())` for a supported version, an unrecorded version, and for anything
/// that cannot be determined (see [`recorded_format_version`]) — the gate only ever
/// *adds* a refusal, it never masks an existing error.
pub fn ensure_supported_format(path: &Path) -> Result<(), ArchiveFormatError> {
let Some(recorded) = recorded_format_version(path) else {
return Ok(());
};
let refuse = || ArchiveFormatError {
path: path.display().to_string(),
recorded: recorded.clone(),
max_supported: MAX_SUPPORTED_ZNIPPY_FORMAT,
};
match recorded.parse::<u32>() {
Ok(v) if v <= MAX_SUPPORTED_ZNIPPY_FORMAT => Ok(()),
// Newer than we read, or not a number at all: both are archives whose
// layout this build cannot vouch for.
_ => Err(refuse()),
}
}
/// [`ensure_supported_format`] with the verdict memoised against the archive's
/// identity on disk — its `(length, mtime)`.
///
/// The writable dev-mode backends re-open their `.znippy` on **every** read (a
/// `put` invalidates any cached view), so the gate sits on a hot path. A file
/// whose length and mtime are unchanged is the same file, and its recorded format
/// version cannot have changed; the memo turns the steady-state cost into one
/// `stat`. A file that *has* changed — including one swapped underneath a running
/// server — is re-gated.
#[derive(Debug, Default)]
pub struct FormatGate {
/// `(len, mtime_nanos)` of the last archive that passed. `None` until the
/// first successful check.
passed: Mutex<Option<(u64, i128)>>,
}
impl FormatGate {
pub fn new() -> Self {
Self::default()
}
/// Gate `path`, skipping the read when it is byte-identical (by `(len, mtime)`)
/// to the last archive that passed.
pub fn check(&self, path: &Path) -> Result<(), ArchiveFormatError> {
let stamp = file_stamp(path);
if let Some(s) = stamp {
if self.passed.lock().is_ok_and(|g| *g == Some(s)) {
return Ok(());
}
}
ensure_supported_format(path)?;
if let (Some(s), Ok(mut g)) = (stamp, self.passed.lock()) {
*g = Some(s);
}
Ok(())
}
}
/// `(len, mtime_nanos)` identity of a file, or `None` when it cannot be stat'ed —
/// in which case the caller simply does not memoise.
fn file_stamp(path: &Path) -> Option<(u64, i128)> {
let md = std::fs::metadata(path).ok()?;
let mtime = md
.modified()
.ok()?
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as i128)
.unwrap_or(-1);
Some((md.len(), mtime))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_missing_or_garbage_archive_is_undetermined_not_refused() {
// Nothing there at all.
assert_eq!(
recorded_format_version(Path::new("/nonexistent/x.znippy")),
None
);
assert!(ensure_supported_format(Path::new("/nonexistent/x.znippy")).is_ok());
// Garbage that is not a znippy archive: the gate stays silent and lets the
// normal open path report its own (already-graceful) error.
let dir = tempfile::tempdir().unwrap();
let junk = dir.path().join("junk.znippy");
std::fs::write(&junk, b"this is not a znippy archive at all").unwrap();
assert_eq!(recorded_format_version(&junk), None);
assert!(ensure_supported_format(&junk).is_ok());
}
/// A sub-index whose first IPC message claims a 1 TiB body. arrow-ipc's
/// `StreamReader::try_new` allocates a message's body before it reads it, so
/// this was an allocation ABORT of the process — which the `catch_unwind`
/// above cannot stop (lane arrowcrash).
#[test]
fn a_sub_index_claiming_a_terabyte_body_is_undetermined_not_an_abort() {
use std::sync::Arc;
use znippy_common::arrow::array::{ArrayRef, UInt64Array};
use znippy_common::arrow::ipc::writer::StreamWriter;
use znippy_common::arrow::record_batch::RecordBatch;
// An honest stream, then drop its schema message so the record batch
// (which has a body) is the first message the reader allocates for.
let col: ArrayRef = Arc::new(UInt64Array::from(vec![1u64, 2, 3]));
let batch = RecordBatch::try_from_iter(vec![("x", col)]).unwrap();
let mut ipc = Vec::new();
{
let mut w = StreamWriter::try_new(&mut ipc, &batch.schema()).unwrap();
w.write(&batch).unwrap();
w.finish().unwrap();
}
let schema_len = u32::from_le_bytes(ipc[4..8].try_into().unwrap()) as usize;
let mut sub = ipc[8 + schema_len..].to_vec();
let meta_len = u32::from_le_bytes(sub[4..8].try_into().unwrap()) as usize;
let msg = znippy_common::arrow::ipc::root_as_message(&sub[8..8 + meta_len]).unwrap();
let needle = msg.bodyLength().to_le_bytes();
let at = sub[8..8 + meta_len]
.windows(8)
.position(|w| w == needle)
.unwrap()
+ 8;
sub[at..at + 8].copy_from_slice(&(1i64 << 40).to_le_bytes());
// [sub-index][manifest][ZNPYMIDX][manifest offset]
let manifest = znippy_common::write_manifest_bytes(&[znippy_common::ManifestEntry {
pkg_type: 1,
repo: "r".into(),
module_name: "rust".into(),
index_offset: 0,
index_len: sub.len() as u64,
row_count: 3,
}])
.unwrap();
let mut file = sub.clone();
let manifest_offset = file.len() as u64;
file.extend_from_slice(&manifest);
file.extend_from_slice(&znippy_common::MULTI_INDEX_MAGIC);
file.extend_from_slice(&manifest_offset.to_le_bytes());
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("terabyte.znippy");
std::fs::write(&path, &file).unwrap();
assert_eq!(recorded_format_version(&path), None);
assert!(ensure_supported_format(&path).is_ok());
}
#[test]
fn the_pin_refuses_newer_and_admits_current_and_older() {
// The decision function, exercised directly on recorded values.
let verdict = |raw: &str| -> bool {
raw.parse::<u32>()
.map(|v| v <= MAX_SUPPORTED_ZNIPPY_FORMAT)
.unwrap_or(false)
};
assert!(verdict(&MAX_SUPPORTED_ZNIPPY_FORMAT.to_string()));
assert!(verdict("1"));
assert!(!verdict(&(MAX_SUPPORTED_ZNIPPY_FORMAT + 1).to_string()));
assert!(!verdict("99"));
assert!(!verdict("not-a-version"));
}
#[test]
fn the_refusal_names_the_archive_the_version_and_the_ceiling() {
let e = ArchiveFormatError {
path: "/srv/holger/drift.znippy".into(),
recorded: "39".into(),
max_supported: MAX_SUPPORTED_ZNIPPY_FORMAT,
};
let msg = e.to_string();
assert!(msg.contains("holger refuses"), "{msg}");
assert!(msg.contains("/srv/holger/drift.znippy"), "{msg}");
assert!(msg.contains("39"), "{msg}");
assert!(
msg.contains(&format!("v{MAX_SUPPORTED_ZNIPPY_FORMAT}")),
"{msg}"
);
// It must be usable as a real error (the request handler downcasts it).
let boxed: Box<dyn std::error::Error> = Box::new(e);
assert!(boxed.downcast_ref::<ArchiveFormatError>().is_some());
}
#[test]
fn the_gate_memo_is_keyed_on_the_file_not_the_path() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("a.znippy");
std::fs::write(&p, b"garbage-v1").unwrap();
let gate = FormatGate::new();
assert!(gate.check(&p).is_ok());
let first = *gate.passed.lock().unwrap();
assert!(first.is_some(), "a stat-able file must be memoised");
// Rewriting the file with different content changes the stamp, so the gate
// re-reads instead of trusting the memo.
std::thread::sleep(std::time::Duration::from_millis(10));
std::fs::write(&p, b"garbage-v2-longer").unwrap();
assert!(gate.check(&p).is_ok());
assert_ne!(
first,
*gate.passed.lock().unwrap(),
"a changed file must be re-gated"
);
}
}