holger-traits 0.6.10

//! **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};

    use znippy_common::arrow::ipc::reader::StreamReader;

    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()?;
        let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None).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());
    }

    #[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");
    }
}