znippy-common/src/index.rs
// index.rs — v0.6 format: blobs stored inline, Arrow IPC is a pure metadata index.
//
// File layout:
// [blob_0][blob_1]...[blob_N] — compressed/raw chunk bytes, written as produced
// [Arrow IPC stream] — metadata index, written after all blobs
// [8 bytes LE u64] — byte offset where Arrow IPC starts (footer)
//
// Arrow schema columns:
// relative_path, chunk_seq, fdata_offset, checksum_group,
// compressed, uncompressed_size, blob_offset, blob_size, checksum
use std::collections::HashMap;
use std::fs::File;
use std::io::{Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use crate::common_config::StrategicConfig;
use crate::meta::BlobMeta;
use crate::plugin::ExtensionRow;
use crate::{decompress_archive};
use anyhow::Result;
use arrow::array::{
Array, ArrayRef, BooleanBuilder, FixedSizeBinaryBuilder, Int8Builder, ListBuilder, StringBuilder,
UInt32Builder, UInt64Builder,
};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use once_cell::sync::Lazy;
/// Per-file extension metadata carried into the Arrow index.
/// (plugin_type_id, extracted fields) — None for files with no matching plugin.
pub type FileExtMeta = Option<(i8, ExtensionRow)>;
/// v0.6 schema: Arrow IPC is a pure metadata index; blobs are stored inline before it.
/// Base index columns — present in every archive, type-agnostic.
/// Package-type modules contribute their own columns on top via `schema_fields()`;
/// the writer composes the on-disk schema with [`compose_index_schema`].
pub static ZNIPPY_INDEX_SCHEMA: Lazy<Arc<Schema>> = Lazy::new(|| {
Arc::new(Schema::new(base_index_fields()))
});
fn base_index_fields() -> Vec<Field> {
vec![
Field::new("relative_path", DataType::Utf8, false),
Field::new("chunk_seq", DataType::UInt32, false),
Field::new("fdata_offset", DataType::UInt64, false),
Field::new("compressed", DataType::Boolean, false),
Field::new("uncompressed_size", DataType::UInt64, false),
Field::new("blob_offset", DataType::UInt64, false),
Field::new("blob_size", DataType::UInt64, false),
Field::new("checksum", DataType::FixedSizeBinary(32), false),
]
}
pub fn znippy_index_schema() -> &'static Arc<Schema> {
&ZNIPPY_INDEX_SCHEMA
}
/// Compose the on-disk index schema: base columns, plus — when a module contributes columns —
/// a `pkg_type` discriminator followed by the module's own `ext_fields`.
/// With no module fields, this is exactly the base schema (v0.6 layout, directly DuckDB-queryable).
pub fn compose_index_schema(ext_fields: &[Field]) -> Arc<Schema> {
let mut fields = base_index_fields();
if !ext_fields.is_empty() {
fields.push(Field::new("pkg_type", DataType::Int8, true));
fields.extend(ext_fields.iter().cloned());
}
Arc::new(Schema::new(fields))
}
/// On-disk archive format version, recorded in the Arrow index schema metadata
/// under `znippy_format_version`. The reader refuses any archive whose recorded
/// version is greater than this — a newer format it cannot safely parse — with a
/// clear error instead of panicking or mis-parsing. Bump only on a real on-disk
/// format change.
pub const ZNIPPY_FORMAT_VERSION: u32 = 3;
/// Schema-metadata key holding the on-disk [`ZNIPPY_FORMAT_VERSION`].
pub const FORMAT_VERSION_KEY: &str = "znippy_format_version";
/// Enforce that an archive's recorded format version is one this reader supports.
/// `metadata` is an Arrow schema's key/value metadata. A *newer* version yields a
/// clear error instead of a panic or silent mis-parse; equal/older versions — and
/// archives with no recorded version — read exactly as before.
pub fn check_format_version(metadata: &HashMap<String, String>) -> Result<()> {
if let Some(raw) = metadata.get(FORMAT_VERSION_KEY) {
let version: u32 = raw
.parse()
.map_err(|_| anyhow::anyhow!("invalid znippy archive format version {raw:?}"))?;
anyhow::ensure!(
version <= ZNIPPY_FORMAT_VERSION,
"znippy archive format v{version} is newer than this reader supports \
(max v{ZNIPPY_FORMAT_VERSION}) — upgrade znippy",
);
}
Ok(())
}
/// Build Arrow schema metadata containing config (no checksum entries — those live in column).
pub fn build_arrow_metadata_for_config(config: &StrategicConfig) -> HashMap<String, String> {
let mut m = HashMap::new();
m.insert(FORMAT_VERSION_KEY.into(), ZNIPPY_FORMAT_VERSION.to_string());
m.insert("max_core_in_flight".into(), config.max_core_in_flight.to_string());
m.insert("max_core_in_compress".into(), config.max_core_in_compress.to_string());
m.insert("max_mem_allowed".into(), config.max_mem_allowed.to_string());
m.insert("min_free_memory_ratio".into(), config.min_free_memory_ratio.to_string());
m.insert("file_split_block_size".into(), config.file_split_block_size.to_string());
m.insert("max_chunks".into(), config.max_chunks.to_string());
m.insert("compression_level".into(), config.compression_level.to_string());
m.insert("zstd_output_buffer_size".into(), config.zstd_output_buffer_size.to_string());
m
}
pub fn extract_config_from_arrow_metadata(
metadata: &HashMap<String, String>,
) -> anyhow::Result<StrategicConfig> {
Ok(StrategicConfig {
max_core_allowed: 0,
max_core_in_flight: metadata
.get("max_core_in_flight")
.ok_or_else(|| anyhow::anyhow!("Missing 'max_core_in_flight'"))?
.parse()?,
max_core_in_compress: metadata
.get("max_core_in_compress")
.ok_or_else(|| anyhow::anyhow!("Missing 'max_core_in_compress'"))?
.parse()?,
max_mem_allowed: metadata
.get("max_mem_allowed")
.ok_or_else(|| anyhow::anyhow!("Missing 'max_mem_allowed'"))?
.parse()?,
min_free_memory_ratio: metadata
.get("min_free_memory_ratio")
.ok_or_else(|| anyhow::anyhow!("Missing 'min_free_memory_ratio'"))?
.parse()?,
file_split_block_size: metadata
.get("file_split_block_size")
.ok_or_else(|| anyhow::anyhow!("Missing 'file_split_block_size'"))?
.parse()?,
max_chunks: metadata
.get("max_chunks")
.ok_or_else(|| anyhow::anyhow!("Missing 'max_chunks'"))?
.parse()?,
compression_level: metadata
.get("compression_level")
.ok_or_else(|| anyhow::anyhow!("Missing 'compression_level'"))?
.parse()?,
zstd_output_buffer_size: metadata
.get("zstd_output_buffer_size")
.ok_or_else(|| anyhow::anyhow!("Missing 'zstd_output_buffer_size'"))?
.parse()?,
})
}
/// Build the Arrow metadata index batch from blob positions.
///
/// Every row carries its own per-slice BLAKE3 in the `checksum` column
/// (over the chunk's uncompressed bytes).
pub fn build_metadata_batch<F>(
blobs: &[BlobMeta],
path_resolver: F,
ext_meta: &[FileExtMeta],
ext_fields: &[Field],
) -> arrow::error::Result<RecordBatch>
where
F: Fn(u64) -> String,
{
let len = blobs.len();
let mut path_builder = StringBuilder::with_capacity(len, len * 64);
let mut seq_builder = UInt32Builder::with_capacity(len);
let mut fdata_builder = UInt64Builder::with_capacity(len);
let mut compressed_builder = BooleanBuilder::with_capacity(len);
let mut size_builder = UInt64Builder::with_capacity(len);
let mut blob_offset_builder = UInt64Builder::with_capacity(len);
let mut blob_size_builder = UInt64Builder::with_capacity(len);
let mut checksum_builder = FixedSizeBinaryBuilder::with_capacity(len, 32);
for blob in blobs {
let m = &blob.chunk_meta;
path_builder.append_value(path_resolver(m.file_index));
seq_builder.append_value(m.chunk_seq);
fdata_builder.append_value(m.fdata_offset);
compressed_builder.append_value(m.compressed);
size_builder.append_value(m.uncompressed_size);
blob_offset_builder.append_value(blob.blob_offset);
blob_size_builder.append_value(blob.blob_size);
checksum_builder.append_value(m.checksum)?;
}
let mut columns: Vec<ArrayRef> = vec![
Arc::new(path_builder.finish()),
Arc::new(seq_builder.finish()),
Arc::new(fdata_builder.finish()),
Arc::new(compressed_builder.finish()),
Arc::new(size_builder.finish()),
Arc::new(blob_offset_builder.finish()),
Arc::new(blob_size_builder.finish()),
Arc::new(checksum_builder.finish()),
];
// Module-contributed columns: a pkg_type discriminator + one column per ext field.
if !ext_fields.is_empty() {
let mut pkg_type_builder = Int8Builder::with_capacity(len);
for blob in blobs {
match ext_meta.get(blob.chunk_meta.file_index as usize).and_then(|x| x.as_ref()) {
Some((type_id, _)) => pkg_type_builder.append_value(*type_id),
None => pkg_type_builder.append_null(),
}
}
columns.push(Arc::new(pkg_type_builder.finish()));
for field in ext_fields {
columns.push(build_ext_column(field, blobs, ext_meta));
}
}
RecordBatch::try_new(compose_index_schema(ext_fields), columns)
}
/// Build one extension column from the per-file `ExtensionRow`, keyed by the field name.
/// Supports the Arrow types modules declare (Utf8, UInt32, List<Utf8>); other types yield nulls.
fn build_ext_column(field: &Field, blobs: &[BlobMeta], ext_meta: &[FileExtMeta]) -> ArrayRef {
use crate::plugin::ExtensionValue;
let len = blobs.len();
let value_for = |blob: &BlobMeta| -> Option<&ExtensionValue> {
ext_meta
.get(blob.chunk_meta.file_index as usize)
.and_then(|x| x.as_ref())
.and_then(|(_, row)| row.fields.get(field.name()))
};
match field.data_type() {
DataType::UInt32 => {
let mut b = UInt32Builder::with_capacity(len);
for blob in blobs {
match value_for(blob) {
Some(ExtensionValue::U32(n)) => b.append_value(*n),
_ => b.append_null(),
}
}
Arc::new(b.finish())
}
// A list of strings (StrList): cargo `deps`, python `requires_dist`.
DataType::List(item) if item.data_type() == &DataType::Utf8 => {
let mut b = ListBuilder::new(StringBuilder::new()).with_field(Arc::clone(item));
for blob in blobs {
match value_for(blob) {
Some(ExtensionValue::StrList(v)) => {
for s in v {
b.values().append_value(s);
}
b.append(true);
}
_ => b.append(false),
}
}
Arc::new(b.finish())
}
// Default to Utf8 for string-like fields (Str / OptStr).
_ => {
let mut b = StringBuilder::with_capacity(len, len * 16);
for blob in blobs {
match value_for(blob) {
Some(ExtensionValue::Str(s)) => b.append_value(s),
Some(ExtensionValue::OptStr(Some(s))) => b.append_value(s),
_ => b.append_null(),
}
}
Arc::new(b.finish())
}
}
}
// ─── Multi-index container codec (planned v0.7, see design.md §6) ──────────────
//
// A multi-type archive holds several Arrow IPC index streams (one per (pkg_type, repo)
// sub-znippy, each with its own narrow schema), followed by a manifest stream that points
// at them, and a footer. The footer layout still *distinguishes* the legacy v0.6 trailer
// from the v0.7 one, but v0.6 archives are no longer readable — they are detected only to
// emit a clear "unsupported, re-compress with v0.7" error (see `read_znippy_index_filtered`):
//
// v0.6 single index: [...index...] [8-byte LE u64 index_offset]
// v0.7 multi index: [...sub-indexes...][manifest] [8-byte MAGIC] [8-byte LE u64 manifest_offset]
//
// A reader peeks the 8 bytes preceding the trailing offset: if they equal MAGIC it's a
// multi-index (v0.7) archive; otherwise it's a legacy v0.6 single index, which the reader
// rejects rather than parses.
/// Magic preceding the trailing offset that marks a multi-index (v0.7) archive.
pub const MULTI_INDEX_MAGIC: [u8; 8] = *b"ZNPYMIDX";
/// One entry in the multi-index manifest: a sub-znippy's identity + byte range.
#[derive(Debug, Clone, PartialEq)]
pub struct ManifestEntry {
pub pkg_type: i8,
pub repo: String,
pub module_name: String,
pub index_offset: u64,
pub index_len: u64,
pub row_count: u64,
}
/// Reserved `module_name` for the sorted random-access lookup sub-index.
/// Its rows are the base index columns re-sorted by `(relative_path, chunk_seq)`,
/// so external tools can `SELECT … WHERE relative_path = …` in O(log n) and the
/// native reader can binary-search it. Manifest readers filter this entry out of
/// the data sub-index set; [`read_znippy_lookup`] reads it explicitly.
pub const LOOKUP_MODULE: &str = "__znippy_lookup__";
/// Reserved `module_name` for the fst trie blob: an `fst::Map` of
/// `relative_path → first row index in the lookup sub-index`. Not Arrow IPC —
/// raw fst bytes — so it must never be parsed as a sub-index.
pub const TRIE_MODULE: &str = "__znippy_trie__";
/// Reserved `module_name` for the per-artifact detached CMS signatures
/// (feature `sign`). An Arrow IPC sub-index of `(relative_path, cms)` rows — one
/// detached CMS `SignedData` per file, over that file's merkle-folded chunk
/// hashes. Additive: readers that don't know it simply skip it (it is reserved),
/// so unsigned archives are byte-identical and old readers ignore signed ones.
pub const SIGN_ARTIFACTS_MODULE: &str = "__znippy_sign_artifacts__";
/// Reserved `module_name` for the per-archive detached CMS signature (feature
/// `sign`): raw CMS `SignedData` (DER) over the archive root digest.
pub const SIGN_ARCHIVE_MODULE: &str = "__znippy_sign_archive__";
/// Reserved `module_name` for the **searchable metadata sub-index**: typed
/// key/value rows, per entry and per archive, sorted by `(key, relative_path)`.
/// An Arrow IPC sub-index like the lookup — so "which entries carry key `X`" is
/// answered by one seek to the footer plus one read of this section, at a cost
/// set by the number of metadata rows and **not** by the payload size.
///
/// Reserved, therefore additive: a reader that predates it skips the entry, and
/// an archive that predates it simply has no such entry — which
/// [`crate::meta_index::read_archive_meta`] reports as
/// [`ArchiveMeta::NoMetadata`](crate::meta_index::ArchiveMeta::NoMetadata),
/// a state distinct from a present-but-empty index.
pub const META_MODULE: &str = "__znippy_meta__";
/// Reserved `module_name` for the **git oid index** written by the `git` package
/// format (`znippy-plugin-git`): a raw (non-Arrow) `stree` payload over the first
/// eight bytes of every object id, mapping to that object's first lookup row.
/// Raw bytes, never an Arrow IPC stream — it must never be parsed as a sub-index.
pub const GUNNAR_OID_MODULE: &str = "__gunnar_oid__";
/// Reserved `module_name` for the **git commit graph** written by the `git`
/// package format: an Arrow IPC sub-index of
/// `(oid, parents[], tree, committer_time, generation)`. Reserved so ordinary
/// `list`/`decompress`/iceberg readers skip it; queryable from DuckDB by slicing
/// its manifest byte range.
pub const GUNNAR_GRAPH_MODULE: &str = "__gunnar_graph__";
/// Reserved `module_name` for the **git reachability bitmaps** written by the
/// `git` package format: an Arrow IPC sub-index of `(commit_oid, bitmap)` where
/// `bitmap` is a serialized roaring bitmap over object ordinals.
pub const GUNNAR_REACH_MODULE: &str = "__gunnar_reach__";
/// Reserved `module_name` for the **git ref log** written by the `git` package
/// format: an Arrow IPC sub-index carrying **one RecordBatch per push**.
///
/// The batch boundary *is* the transaction boundary. There is no database here
/// (D18: redb was a database wedged between two archive formats, and it is
/// gone). A push is durable exactly when its IPC frame is complete on disk; a
/// frame torn by a crash is not half a push, it is a trailing byte run that no
/// reader will accept. Recovery is therefore "read forward while frames parse",
/// never a journal replay.
pub const GUNNAR_REFS_MODULE: &str = "__gunnar_refs__";
/// Reserved `module_name` for the **git secrets log** written by the `git`
/// package format: the same one-RecordBatch-per-push shape as
/// [`GUNNAR_REFS_MODULE`], carrying already-encrypted material only. znippy
/// never sees plaintext, and ciphertext is never compressed.
pub const GUNNAR_SECRETS_MODULE: &str = "__gunnar_secrets__";
/// Reserved `module_name` for the **delta map**: which stored chunks are delta
/// chunks, and what each one's base entry is.
///
/// # Why a reserved section and not a column
///
/// The obvious place is a `delta_base` column on the data sub-index. It is the
/// wrong place, and the reason is worth recording because the column is the
/// first thing anyone reaches for. `base_index_fields()` is the schema of EVERY
/// archive znippy has ever written; adding a column to it means a
/// [`ZNIPPY_FORMAT_VERSION`] bump, a new column every existing writer must
/// supply, and a changed on-disk shape for the iceberg sink and for anyone
/// pointing DuckDB at an archive — all to describe a property that almost no row
/// has.
///
/// A reserved section is additive by construction: the manifest readers already
/// skip reserved modules, so an archive without this one is exactly what it is
/// today and needs no version bump. `ZnippyArchive::open` joins it onto the
/// chunks it names; absent, every chunk is [`Stored`](crate::archive) and the
/// read path is byte-for-byte the one the golden digest pins.
///
/// The payload is an Arrow IPC stream of `(relative_path, chunk_seq, base_path)`.
/// It is independent of the objects, so an append must carry it — but it travels
/// as **decoded rows** rather than raw bytes (like `__meta__`, and unlike
/// `__gunnar_refs__`), because an append may add rows to it and two sections with
/// one module name is not a thing the manifest can express. That is why it is
/// NOT in [`CARRIED_RESERVED_MODULES`].
pub const ZNIPPY_DELTA_MODULE: &str = "__znippy_delta__";
/// Reserved `module_name` for the **filesystem-shape** section: the entries the
/// data index structurally cannot hold, because the data index is a table of
/// *chunks* and a chunk belongs to a file.
///
/// # What is missing without it
///
/// `base_index_fields()` is `(relative_path, chunk_seq, fdata_offset, compressed,
/// uncompressed_size, blob_offset, blob_size, checksum)`. Every row is a slice of
/// a regular file's bytes. Two facts of a real tree have no row shape there at
/// all, and both were silently dropped on the way in:
///
/// * a **directory** — most sharply an EMPTY one, which owns no file and so
/// contributes no row. Extract only ever ran `create_dir_all` on a file's
/// parent (`crate::decompress`), so a directory with no files under it did not
/// come back. A container-registry mirror (`talosctl image cache-create`) has
/// them, and a git checkout, and a build tree.
/// * a **unix mode** — the executable bit above all. An extracted file got
/// `0666 & ~umask` regardless of what went in, so a script round-tripped
/// non-executable.
///
/// # Why a reserved section rather than columns or dir rows
///
/// Same argument as [`ZNIPPY_DELTA_MODULE`], and the alternatives are worse here.
/// A `mode` column on the base schema is a [`ZNIPPY_FORMAT_VERSION`] bump plus a
/// column every existing writer (the iceberg sink, the append sink, gunnar's
/// generation-0 seal) must start supplying. Encoding a directory as a zero-byte
/// data row is worse still: every reader — `list`, `verify`, `get`, the typed
/// views, DuckDB over the Arrow file — would then see a *file* of length 0 where
/// a directory is, and extract would create one.
///
/// Reserved is additive by construction: `is_reserved_module` already makes every
/// data reader skip it, so an archive written before this section existed reads
/// exactly as it did, and one written with it opens unchanged in an older znippy
/// — minus the empty dirs and modes it cannot know about.
///
/// The payload is an Arrow IPC stream of `(relative_path, kind, mode)`; see
/// [`crate::fsmeta`].
pub const ZNIPPY_FSMETA_MODULE: &str = "__znippy_fsmeta__";
/// Schema of the [`ZNIPPY_DELTA_MODULE`] section.
pub fn delta_map_schema() -> Arc<Schema> {
Arc::new(Schema::new(vec![
Field::new("relative_path", DataType::Utf8, false),
Field::new("chunk_seq", DataType::UInt32, false),
Field::new("base_path", DataType::Utf8, false),
]))
}
/// `pkg_type` discriminant carried by reserved (non-data) manifest entries.
pub const RESERVED_PKG_TYPE: i8 = i8::MIN;
/// Every reserved `module_name` this reader knows, in one place.
///
/// Kept as a slice rather than a hand-written `||` chain so that adding a
/// reserved module is a single edit that both [`is_reserved_module`] and the
/// guard test see — the failure mode being avoided is a new module name that is
/// written as reserved by the sink but classified as *data* by the manifest
/// readers, which silently corrupts `list`, `decompress`, the iceberg sink and
/// the manifest-count assertions.
pub const RESERVED_MODULES: &[&str] = &[
LOOKUP_MODULE,
TRIE_MODULE,
SIGN_ARTIFACTS_MODULE,
SIGN_ARCHIVE_MODULE,
META_MODULE,
GUNNAR_OID_MODULE,
GUNNAR_GRAPH_MODULE,
GUNNAR_REACH_MODULE,
GUNNAR_REFS_MODULE,
GUNNAR_SECRETS_MODULE,
ZNIPPY_DELTA_MODULE,
ZNIPPY_FSMETA_MODULE,
];
/// A reserved manifest entry holds a derived structure (lookup / trie / detached
/// signatures / the git oid-index, commit-graph and reachability sections), not
/// file rows. The merge + manifest readers skip these so data consumers are
/// unaffected — which is exactly what makes those sections additive and
/// backward-compatible.
pub fn is_reserved_module(module_name: &str) -> bool {
RESERVED_MODULES.contains(&module_name)
}
/// The reserved sections an **append must carry forward**, because they are
/// independent logs rather than derivations of the blobs beside them.
///
/// MEASURED 2026-08-04, gunnar: an object-carrying push removed
/// `__gunnar_refs__` from the archive (144 928 -> 146 396 bytes, section gone),
/// because [`crate::ArrowIpcSinkAppend::open_existing`] truncates the whole
/// metadata tail and re-supplies no reserved section it was not handed. All
/// reserved sections were treated alike, and they are not alike:
///
/// * **derived** — `__gunnar_oid__`, `__gunnar_graph__`, `__gunnar_reach__`,
/// plus `__lookup__` / `__trie__` — are functions of the objects in the
/// archive. When objects change they are *wrong*, and dropping them is
/// correct; the writer rebuilds them.
/// * **independent** — `__gunnar_refs__` and `__gunnar_secrets__` — are logs,
/// one RecordBatch per push. Nothing in a new push's objects reproduces the
/// ref history that came before it, so dropping them destroys data.
///
/// `__meta__` is carried by its own typed path (`open_existing` decodes it into
/// a `MetaTable`) and so is deliberately not listed here: it would then be
/// written twice.
///
/// This distinction is what the append path could not previously express, and it
/// blocks serving refs from the archive for most of a repository's life.
/// [`ZNIPPY_FSMETA_MODULE`] joins the independent half for the same reason: the
/// tree's shape — its empty directories, its modes — is a fact about the walk
/// that produced the archive, and NOTHING in a later append's file list
/// reproduces it. `append_files` is handed `(relative_path, bytes)` pairs, never
/// a directory; if the section were dropped on append, the first `znippy append`
/// after a `compress` would silently delete every empty directory from the
/// archive. Carried verbatim, an append leaves the shape exactly as the compress
/// that established it recorded it.
pub const CARRIED_RESERVED_MODULES: &[&str] =
&[GUNNAR_REFS_MODULE, GUNNAR_SECRETS_MODULE, ZNIPPY_FSMETA_MODULE];
/// Whether an append must carry this reserved section forward verbatim.
/// See [`CARRIED_RESERVED_MODULES`].
pub fn is_carried_reserved_module(module_name: &str) -> bool {
CARRIED_RESERVED_MODULES.contains(&module_name)
}
/// Read the raw bytes of one reserved sub-section by module name, if present.
/// Public entry point for the signature layer (feature `sign`) and any other
/// reader that needs a reserved section's bytes without re-implementing the
/// manifest walk.
pub fn read_reserved_section_bytes(path: &Path, module_name: &str) -> Result<Option<Vec<u8>>> {
read_reserved_section(path, module_name)
}
/// Schema of the lookup sub-index — identical to the base index columns. The
/// lookup is the same per-chunk rows, re-sorted by `(relative_path, chunk_seq)`
/// and stripped of any plugin columns, so one path's chunks are contiguous.
///
/// **Carries no schema metadata**, and that is correct for the *lookup*: the
/// lookup is a RESERVED sub-index, which every reader skips. It is NOT correct
/// for a **data** sub-index — use [`data_subindex_schema`] there, or the archive
/// records no format version and [`check_format_version`] has nothing to check.
pub fn lookup_schema() -> Arc<Schema> {
Arc::new(Schema::new(base_index_fields()))
}
/// Schema for a base-column **data** sub-index: the same columns as
/// [`lookup_schema`], stamped with the archive metadata that
/// [`build_arrow_metadata_for_config`] produces — including
/// [`FORMAT_VERSION_KEY`].
///
/// The reader (and holger's independent reader-side pin) determines an archive's
/// on-disk format version from the Arrow schema metadata of the **first
/// non-reserved sub-index**. A writer that seals its data sub-index with the bare
/// [`lookup_schema`] therefore produces an archive that records no version at
/// all, and the version pin degrades to "undetermined → read as before" for
/// exactly those archives. Every data sub-index must be sealed with this.
pub fn data_subindex_schema() -> Arc<Schema> {
Arc::new(Schema::new_with_metadata(
base_index_fields(),
build_arrow_metadata_for_config(&crate::common_config::CONFIG),
))
}
/// One chunk's location for single-file random access.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChunkLoc {
pub chunk_seq: u32,
pub fdata_offset: u64,
pub blob_offset: u64,
pub blob_size: u64,
pub uncompressed_size: u64,
pub compressed: bool,
pub checksum: [u8; 32],
}
/// What the trailing footer of an archive points at.
#[derive(Debug, Clone, PartialEq)]
pub enum IndexFooter {
/// Legacy v0.6: a single Arrow IPC index began at this offset. Still detected
/// so the reader can reject v0.6 archives with a clear "re-compress with v0.7"
/// error — it is no longer a live read path.
Single { index_offset: u64 },
/// v0.7: the manifest stream begins at this offset.
Multi { manifest_offset: u64 },
}
/// Interpret an archive's trailing bytes. `tail` must be the last 16 bytes of the file
/// (or last 8 for tiny v0.6 files — then it's always Single).
pub fn interpret_footer(tail: &[u8]) -> IndexFooter {
let n = tail.len();
let offset = u64::from_le_bytes(tail[n - 8..].try_into().unwrap());
if n >= 16 && tail[n - 16..n - 8] == MULTI_INDEX_MAGIC {
IndexFooter::Multi { manifest_offset: offset }
} else {
IndexFooter::Single { index_offset: offset }
}
}
/// The error for an archive whose last 16 bytes do not carry the v0.7
/// `MULTI_INDEX_MAGIC`.
///
/// Before 2026-11-01 every such file was reported as "v0.6 archives are not
/// supported", so a TRUNCATED archive (the common case: an interrupted copy or
/// download) sent the reader looking for a format version it never had. A v0.6
/// archive is recognisable: its last 8 bytes point inside the file at an Arrow
/// IPC index (`ARROW1` file magic or the `0xFFFFFFFF` stream continuation
/// marker). Only that keeps the v0.6 message; anything else is called what it
/// almost always is — truncated, or not a znippy archive at all.
pub(crate) fn missing_footer_error(file: &mut File, file_len: u64, tail: &[u8; 16]) -> anyhow::Error {
let offset = u64::from_le_bytes(tail[8..].try_into().unwrap());
let looks_v06 = offset.checked_add(6).is_some_and(|end| end <= file_len.saturating_sub(8)) && {
let mut head = [0u8; 6];
file.seek(SeekFrom::Start(offset)).is_ok()
&& file.read_exact(&mut head).is_ok()
&& (&head == b"ARROW1" || head[..4] == [0xFF; 4])
};
if looks_v06 {
anyhow::anyhow!("v0.6 archives are not supported; re-compress with v0.7")
} else {
anyhow::anyhow!(
"not a complete znippy archive: the v0.7 footer is missing from the last 16 bytes \
of this {file_len}-byte file — it is truncated, or not a znippy archive"
)
}
}
/// The error for a file shorter than the 16-byte v0.7 footer.
fn too_small_error(file_len: u64) -> anyhow::Error {
anyhow::anyhow!(
"not a complete znippy archive: {file_len} bytes is shorter than the 16-byte v0.7 \
footer — it is truncated, or not a znippy archive"
)
}
fn manifest_schema() -> Arc<Schema> {
Arc::new(Schema::new(vec![
Field::new("pkg_type", DataType::Int8, false),
Field::new("repo", DataType::Utf8, false),
Field::new("module_name", DataType::Utf8, false),
Field::new("index_offset", DataType::UInt64, false),
Field::new("index_len", DataType::UInt64, false),
Field::new("row_count", DataType::UInt64, false),
]))
}
/// Serialize manifest entries to an Arrow IPC stream (itself DuckDB-readable).
pub fn write_manifest_bytes(entries: &[ManifestEntry]) -> Result<Vec<u8>> {
use arrow::ipc::writer::StreamWriter;
let len = entries.len();
let mut pkg_type = Int8Builder::with_capacity(len);
let mut repo = StringBuilder::with_capacity(len, len * 16);
let mut module_name = StringBuilder::with_capacity(len, len * 16);
let mut index_offset = UInt64Builder::with_capacity(len);
let mut index_len = UInt64Builder::with_capacity(len);
let mut row_count = UInt64Builder::with_capacity(len);
for e in entries {
pkg_type.append_value(e.pkg_type);
repo.append_value(&e.repo);
module_name.append_value(&e.module_name);
index_offset.append_value(e.index_offset);
index_len.append_value(e.index_len);
row_count.append_value(e.row_count);
}
let schema = manifest_schema();
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(pkg_type.finish()),
Arc::new(repo.finish()),
Arc::new(module_name.finish()),
Arc::new(index_offset.finish()),
Arc::new(index_len.finish()),
Arc::new(row_count.finish()),
],
)?;
let mut buf = Vec::new();
{
let mut w = StreamWriter::try_new(&mut buf, &schema)?;
w.write(&batch)?;
w.finish()?;
}
Ok(buf)
}
/// Parse a manifest Arrow IPC stream back into entries.
pub fn read_manifest_bytes(bytes: &[u8]) -> Result<Vec<ManifestEntry>> {
use arrow::array::{Int8Array, StringArray, UInt64Array};
let reader = crate::ipc_guard::stream_reader(bytes)?;
let mut out = Vec::new();
for batch in reader {
let batch = batch?;
let col = |name: &str| batch.column_by_name(name)
.ok_or_else(|| anyhow::anyhow!("manifest missing column {name}"));
let pkg_type = col("pkg_type")?.as_any().downcast_ref::<Int8Array>()
.ok_or_else(|| anyhow::anyhow!("pkg_type type"))?;
let repo = col("repo")?.as_any().downcast_ref::<StringArray>()
.ok_or_else(|| anyhow::anyhow!("repo type"))?;
let module_name = col("module_name")?.as_any().downcast_ref::<StringArray>()
.ok_or_else(|| anyhow::anyhow!("module_name type"))?;
let index_offset = col("index_offset")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("index_offset type"))?;
let index_len = col("index_len")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("index_len type"))?;
let row_count = col("row_count")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("row_count type"))?;
for i in 0..batch.num_rows() {
out.push(ManifestEntry {
pkg_type: pkg_type.value(i),
repo: repo.value(i).to_string(),
module_name: module_name.value(i).to_string(),
index_offset: index_offset.value(i),
index_len: index_len.value(i),
row_count: row_count.value(i),
});
}
}
Ok(out)
}
/// Read the Arrow IPC index from a v0.7 .znippy file.
///
/// Reads the 16-byte footer (8-byte `ZNPYMIDX` magic + 8-byte LE u64 manifest_offset),
/// parses the manifest, reads every sub-index, and merges all batches into one so callers
/// need no format-version awareness.
pub fn read_znippy_index(path: &Path) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
read_znippy_index_filtered(path, &IndexFilter::default())
}
/// Selective sub-index filter: keep only sub-indexes whose manifest entry
/// matches. A `None` field matches anything, so `IndexFilter::default()` reads
/// the whole archive (what [`read_znippy_index`] does). Filtering happens at the
/// `(pkg_type, repo)` sub-index granularity — whole Arrow IPC streams are
/// skipped, never read — so a selective extract touches only the matching rows.
#[derive(Debug, Clone, Default)]
pub struct IndexFilter {
/// Keep only this package-type discriminant (`ManifestEntry::pkg_type`).
pub pkg_type: Option<i8>,
/// Keep only this repo (`ManifestEntry::repo`).
pub repo: Option<String>,
}
impl IndexFilter {
pub fn is_empty(&self) -> bool {
self.pkg_type.is_none() && self.repo.is_none()
}
fn matches(&self, e: &ManifestEntry) -> bool {
self.pkg_type.is_none_or(|t| e.pkg_type == t)
&& self.repo.as_deref().is_none_or(|r| e.repo == r)
}
}
/// Like [`read_znippy_index`] but keeps only sub-indexes matching `filter`.
pub fn read_znippy_index_filtered(
path: &Path,
filter: &IndexFilter,
) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
let mut file = File::open(path)?;
let file_len = file.metadata()?.len();
if file_len < 16 {
return Err(too_small_error(file_len));
}
file.seek(SeekFrom::End(-16))?;
let mut tail = [0u8; 16];
file.read_exact(&mut tail)?;
match interpret_footer(&tail) {
IndexFooter::Multi { manifest_offset } => {
read_multi_index(&mut file, file_len, manifest_offset, filter)
}
IndexFooter::Single { .. } => Err(missing_footer_error(&mut file, file_len, &tail)),
}
}
/// Read all sub-indexes from a v0.7 multi-index archive and concatenate them.
fn read_multi_index(
file: &mut File,
file_len: u64,
manifest_offset: u64,
filter: &IndexFilter,
) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
// manifest lives between manifest_offset and (file_len − 16): 8-byte magic + 8-byte offset
let manifest_end = file_len.checked_sub(16)
.ok_or_else(|| anyhow::anyhow!("v0.7 archive too small"))?;
anyhow::ensure!(manifest_offset <= manifest_end, "corrupt v0.7 manifest_offset");
let manifest_len = (manifest_end - manifest_offset) as usize;
file.seek(SeekFrom::Start(manifest_offset))?;
let mut manifest_bytes = vec![0u8; manifest_len];
file.read_exact(&mut manifest_bytes)?;
let entries = read_manifest_bytes(&manifest_bytes)?;
let mut all_batches: Vec<RecordBatch> = Vec::new();
let mut schema: Option<Arc<Schema>> = None;
for entry in &entries {
// Skip derived structures (lookup sub-index, trie blob) — they are not
// file-row data and must not be merged into the index.
if is_reserved_module(&entry.module_name) {
continue;
}
// Selective read: skip whole sub-indexes that don't match the filter.
if !filter.matches(entry) {
continue;
}
// Bound the attacker-controlled declared length against the real file
// size before allocating, so a corrupt manifest can't force a huge alloc.
anyhow::ensure!(
entry.index_offset.checked_add(entry.index_len)
.is_some_and(|end| end <= file_len),
"sub-index for module {} out of bounds (offset={}, len={}, file_len={})",
entry.module_name, entry.index_offset, entry.index_len, file_len
);
file.seek(SeekFrom::Start(entry.index_offset))?;
let mut sub_bytes = vec![0u8; entry.index_len as usize];
file.read_exact(&mut sub_bytes)?;
let reader = crate::ipc_guard::stream_reader(sub_bytes)?;
if schema.is_none() {
let sub_schema = reader.schema();
// Refuse archives written by a newer znippy than this reader supports,
// before parsing any sub-index batches we might mis-interpret.
check_format_version(sub_schema.metadata())?;
schema = Some(sub_schema);
}
for batch in reader {
all_batches.push(batch.map_err(|e| anyhow::anyhow!("sub-index read error: {}", e))?);
}
}
let schema = schema.unwrap_or_else(|| Arc::new(Schema::new(base_index_fields())));
// Merge all sub-index batches into one so callers stay format-agnostic.
let merged = if all_batches.len() <= 1 {
all_batches
} else {
// `concat_batches` takes column i of every batch for each field of
// `schema` (the FIRST sub-index's) and panics on a batch with fewer
// columns; a hostile second sub-index must be an error instead.
if let Some(b) = all_batches.iter().find(|b| b.num_columns() < schema.fields().len()) {
anyhow::bail!(
"concat sub-indexes: a sub-index has {} columns, the first has {}",
b.num_columns(),
schema.fields().len()
);
}
let batch = arrow_select::concat::concat_batches(&schema, all_batches.iter())
.map_err(|e| anyhow::anyhow!("concat sub-indexes: {}", e))?;
vec![batch]
};
Ok((schema, merged))
}
/// Read the manifest from a v0.7 multi-index archive.
/// Returns an error if the file is a plain v0.6 single-index archive.
pub fn read_znippy_manifest(path: &Path) -> Result<Vec<ManifestEntry>> {
let mut file = File::open(path)?;
let file_len = file.metadata()?.len();
if file_len < 16 {
return Err(too_small_error(file_len));
}
file.seek(SeekFrom::End(-16))?;
let mut tail = [0u8; 16];
file.read_exact(&mut tail)?;
match interpret_footer(&tail) {
IndexFooter::Single { .. } => Err(missing_footer_error(&mut file, file_len, &tail)),
IndexFooter::Multi { manifest_offset } => {
let manifest_end = file_len - 16;
anyhow::ensure!(manifest_offset <= manifest_end, "corrupt manifest_offset");
let manifest_len = (manifest_end - manifest_offset) as usize;
file.seek(SeekFrom::Start(manifest_offset))?;
let mut manifest_bytes = vec![0u8; manifest_len];
file.read_exact(&mut manifest_bytes)?;
let mut entries = read_manifest_bytes(&manifest_bytes)?;
// Hide reserved (lookup/trie) entries from data-manifest consumers.
entries.retain(|e| !is_reserved_module(&e.module_name));
Ok(entries)
}
}
}
/// Read **all** manifest entries of a sealed v0.7 archive, *including* the
/// reserved (lookup/trie) ones, plus the manifest's byte offset.
///
/// Public entrypoint used by the append/resume sink
/// ([`ArrowIpcSinkAppend::open_existing`](crate::ArrowIpcSinkAppend::open_existing)):
/// it needs the reserved sections (to find the blob-region end and to recover
/// the sorted lookup) which `read_znippy_manifest` hides. Read-only — does not
/// touch the original write path.
pub fn read_znippy_full_manifest(path: &Path) -> Result<(Vec<ManifestEntry>, u64)> {
let mut file = File::open(path)?;
let file_len = file.metadata()?.len();
read_full_manifest(&mut file, file_len)
}
/// Read all manifest entries, *including* reserved (lookup/trie) ones.
fn read_full_manifest(file: &mut File, file_len: u64) -> Result<(Vec<ManifestEntry>, u64)> {
if file_len < 16 {
return Err(too_small_error(file_len));
}
file.seek(SeekFrom::End(-16))?;
let mut tail = [0u8; 16];
file.read_exact(&mut tail)?;
let manifest_offset = match interpret_footer(&tail) {
IndexFooter::Multi { manifest_offset } => manifest_offset,
IndexFooter::Single { .. } => return Err(missing_footer_error(file, file_len, &tail)),
};
let manifest_end = file_len.checked_sub(16)
.ok_or_else(|| anyhow::anyhow!("v0.7 archive too small"))?;
anyhow::ensure!(manifest_offset <= manifest_end, "corrupt manifest_offset");
let manifest_len = (manifest_end - manifest_offset) as usize;
file.seek(SeekFrom::Start(manifest_offset))?;
let mut manifest_bytes = vec![0u8; manifest_len];
file.read_exact(&mut manifest_bytes)?;
Ok((read_manifest_bytes(&manifest_bytes)?, manifest_offset))
}
/// Read the raw bytes of one reserved sub-section (lookup or trie), if present.
fn read_reserved_section(path: &Path, module_name: &str) -> Result<Option<Vec<u8>>> {
let mut file = File::open(path)?;
let file_len = file.metadata()?.len();
let (entries, _) = read_full_manifest(&mut file, file_len)?;
let Some(entry) = entries.iter().find(|e| e.module_name == module_name) else {
return Ok(None);
};
// Bound the attacker-controlled declared length against the real file size
// before allocating, so a corrupt manifest can't force a huge allocation.
anyhow::ensure!(
entry.index_offset.checked_add(entry.index_len)
.is_some_and(|end| end <= file_len),
"reserved section {} out of bounds (offset={}, len={}, file_len={})",
entry.module_name, entry.index_offset, entry.index_len, file_len
);
file.seek(SeekFrom::Start(entry.index_offset))?;
let mut bytes = vec![0u8; entry.index_len as usize];
file.read_exact(&mut bytes)?;
Ok(Some(bytes))
}
/// Decode the lookup sub-index bytes into parallel column vectors (already sorted
/// by `(relative_path, chunk_seq)` on disk).
/// Read row `i` of an index/lookup `checksum` column as a 32-byte blake3 digest.
///
/// `FixedSizeBinaryArray` is one concrete Arrow type for *every* width, so a
/// `downcast_ref::<FixedSizeBinaryArray>()` succeeds just as happily for
/// `FixedSizeBinary(16)` as for `(32)`. Copying such a row straight into a
/// `[u8; 32]` panics ("source slice length (16) does not match destination slice
/// length (32)") — a process abort driven by an attacker-supplied schema, which
/// is exactly what the downcast-or-error style everywhere else in this file
/// exists to prevent. Check the declared width and return a clean `Err` instead.
/// (`ZnippyArchive::build_file_index` already guards this with the same test.)
fn checksum32(col: &arrow::array::FixedSizeBinaryArray, i: usize) -> Result<[u8; 32]> {
if col.value_length() != 32 {
return Err(anyhow::anyhow!(
"checksum column has width {}, expected 32",
col.value_length()
));
}
let mut ck = [0u8; 32];
ck.copy_from_slice(col.value(i));
Ok(ck)
}
fn decode_lookup(bytes: &[u8]) -> Result<LookupColumns> {
use arrow::array::{BooleanArray, FixedSizeBinaryArray, StringArray, UInt32Array, UInt64Array};
let reader = crate::ipc_guard::stream_reader(bytes)?;
let mut cols = LookupColumns::default();
for batch in reader {
let batch = batch?;
let get = |n: &str| batch.column_by_name(n)
.ok_or_else(|| anyhow::anyhow!("lookup missing column {n}"));
let paths = get("relative_path")?.as_any().downcast_ref::<StringArray>()
.ok_or_else(|| anyhow::anyhow!("relative_path type"))?;
let chunk_seq = get("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()
.ok_or_else(|| anyhow::anyhow!("chunk_seq type"))?;
let fdata = get("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("fdata_offset type"))?;
let compressed = get("compressed")?.as_any().downcast_ref::<BooleanArray>()
.ok_or_else(|| anyhow::anyhow!("compressed type"))?;
let usize_col = get("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("uncompressed_size type"))?;
let blob_offset = get("blob_offset")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("blob_offset type"))?;
let blob_size = get("blob_size")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("blob_size type"))?;
let checksum = get("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()
.ok_or_else(|| anyhow::anyhow!("checksum type"))?;
for i in 0..batch.num_rows() {
cols.paths.push(paths.value(i).to_string());
let ck = checksum32(checksum, i)?;
cols.locs.push(ChunkLoc {
chunk_seq: chunk_seq.value(i),
fdata_offset: fdata.value(i),
blob_offset: blob_offset.value(i),
blob_size: blob_size.value(i),
uncompressed_size: usize_col.value(i),
compressed: compressed.value(i),
checksum: ck,
});
}
}
Ok(cols)
}
#[derive(Default)]
struct LookupColumns {
paths: Vec<String>,
locs: Vec<ChunkLoc>,
}
/// Locate every chunk of `target` for single-file random access.
///
/// Fast paths, in order: the fst trie (O(key length)) → binary search of the
/// sorted lookup sub-index (O(log n)) → linear scan of the merged main index
/// (O(n), for archives written before the lookup layer existed). Returns the
/// chunks sorted by `chunk_seq`, or an empty vec if `target` is not in the archive.
pub fn locate_file(path: &Path, target: &str) -> Result<Vec<ChunkLoc>> {
if let Some(lookup_bytes) = read_reserved_section(path, LOOKUP_MODULE)? {
let cols = decode_lookup(&lookup_bytes)?;
let n = cols.paths.len();
// Find any row whose path == target.
let hit = if let Some(trie_bytes) = read_reserved_section(path, TRIE_MODULE)? {
let map = fst::Map::new(trie_bytes)
.map_err(|e| anyhow::anyhow!("trie open: {e}"))?;
map.get(target.as_bytes()).map(|v| v as usize)
} else {
// Lookup is sorted by path; binary-search the path column.
match cols.paths.binary_search_by(|p| p.as_str().cmp(target)) {
Ok(i) => Some(i),
Err(_) => None,
}
};
let Some(hit) = hit else { return Ok(Vec::new()); };
// The trie value is attacker-controlled: a corrupt/malicious archive can
// map `target` to a row index past the (smaller) lookup table. Treat an
// out-of-range hit as "not found" instead of indexing OOB below — this is
// a remote-DoS guard on the random-access read path.
if hit >= n { return Ok(Vec::new()); }
// Expand to the contiguous run of rows sharing this path.
let mut start = hit;
while start > 0 && cols.paths[start - 1] == target { start -= 1; }
let mut end = hit + 1;
while end < n && cols.paths[end] == target { end += 1; }
let mut out: Vec<ChunkLoc> = cols.locs[start..end].to_vec();
out.sort_by_key(|c| c.chunk_seq);
return Ok(out);
}
// Fallback: no lookup layer — scan the merged main index.
locate_file_via_index(path, target)
}
/// Slim per-file (per-artifact) metadata for browsing an archive without
/// reading file bytes. One row per file, aggregated across its chunks.
#[derive(Debug, Clone, PartialEq)]
pub struct ArtifactMeta {
pub relative_path: String,
/// Total uncompressed size across all of the file's chunks.
pub uncompressed_size: u64,
pub chunk_count: u32,
/// Whether the file's data was compressed (false on the stored-raw skip path).
pub compressed: bool,
}
/// Metadata for **every** file in the archive, sorted by `relative_path`.
///
/// Reads only the lookup sub-index (not the file bytes); falls back to the main
/// index for archives written before the lookup layer.
pub fn get_all_files_meta(path: &Path) -> Result<Vec<ArtifactMeta>> {
files_meta_impl(path, None)
}
/// Metadata for the files whose `relative_path` starts with `prefix` — for when
/// you don't want all files (e.g. one `group/artifact/` subtree).
///
/// When the trie is present this jumps straight to the first matching key via the
/// fst's ordered range (O(prefix) to seek, then O(matches)); otherwise it binary
/// -searches the sorted lookup; otherwise it scans the legacy main index.
pub fn get_files_meta_with_prefix(path: &Path, prefix: &str) -> Result<Vec<ArtifactMeta>> {
files_meta_impl(path, Some(prefix))
}
fn files_meta_impl(path: &Path, prefix: Option<&str>) -> Result<Vec<ArtifactMeta>> {
use fst::{IntoStreamer, Streamer};
if let Some(lookup_bytes) = read_reserved_section(path, LOOKUP_MODULE)? {
let cols = decode_lookup(&lookup_bytes)?; // sorted by (relative_path, chunk_seq)
let n = cols.paths.len();
// Resolve the contiguous row window [lo, hi) to aggregate.
let (lo, hi) = match prefix {
None | Some("") => (0, n),
Some(pre) => {
// Seek the first key >= prefix. Use the trie's ordered range when
// present (the "trie search"); else binary-search the sorted paths.
let lo = if let Some(trie_bytes) = read_reserved_section(path, TRIE_MODULE)? {
let map = fst::Map::new(trie_bytes)
.map_err(|e| anyhow::anyhow!("trie open: {e}"))?;
let mut stream = map.range().ge(pre.as_bytes()).into_stream();
match stream.next() {
Some((k, v)) if k.starts_with(pre.as_bytes()) => v as usize,
_ => return Ok(Vec::new()),
}
} else {
cols.paths.partition_point(|p| p.as_str() < pre)
};
if lo >= n || !cols.paths[lo].starts_with(pre) {
return Ok(Vec::new());
}
let mut hi = lo;
while hi < n && cols.paths[hi].starts_with(pre) {
hi += 1;
}
(lo, hi)
}
};
// Aggregate contiguous chunk rows per path.
let mut out = Vec::new();
let mut i = lo;
while i < hi {
let p = &cols.paths[i];
let mut total = 0u64;
let mut count = 0u32;
let mut compressed = false;
while i < hi && &cols.paths[i] == p {
total += cols.locs[i].uncompressed_size;
count += 1;
compressed |= cols.locs[i].compressed;
i += 1;
}
out.push(ArtifactMeta {
relative_path: p.clone(),
uncompressed_size: total,
chunk_count: count,
compressed,
});
}
return Ok(out);
}
files_meta_via_index(path, prefix)
}
/// A cached, open-once random-access reader over a sealed v0.7 archive.
///
/// The free functions [`locate_file`] and [`get_files_meta_with_prefix`] re-open
/// the file and re-read the footer + manifest + both reserved sections (the
/// lookup sub-index and the trie) on **every** call, and re-`decode_lookup` every
/// row each time — three manifest reads per lookup (analysis §1.5). For a
/// long-lived reader (a browsing UI, holger's dynamic side, selective restore)
/// that is pure repeated I/O.
///
/// `ArchiveReader::open` reads the footer, manifest, lookup sub-index and trie
/// **once**, decodes the lookup columns once, and builds the fst map once. Every
/// subsequent [`locate`](ArchiveReader::locate),
/// [`files_meta`](ArchiveReader::files_meta) and
/// [`files_meta_with_prefix`](ArchiveReader::files_meta_with_prefix) query is then
/// served from memory — an fst `get` (O(key)) or a binary search (O(log n)) over
/// the already-decoded columns — with **zero** further I/O and **zero**
/// per-call manifest reads. The on-disk format is untouched; results are
/// identical to the equivalent free function.
pub struct ArchiveReader {
/// Relative paths, sorted by `(relative_path, chunk_seq)` (lookup order).
paths: Vec<String>,
/// Chunk locators, index-aligned with `paths`.
locs: Vec<ChunkLoc>,
/// `path -> first row index` fst over the sorted paths, when the archive
/// carries a trie section (it always does for v0.7 seals). Absent ⇒ fall back
/// to binary search over `paths`.
trie: Option<fst::Map<Vec<u8>>>,
/// The archive handle kept open across calls, so [`read_file`](Self::read_file)
/// serves selective restore with **one** open + cached locate (positioned
/// `pread`s only — never seeks a shared cursor).
file: File,
/// Real archive length, for the same pre-alloc bounds check the free
/// `get_file` does against attacker-controlled blob offsets/sizes.
file_len: u64,
}
impl ArchiveReader {
/// Open a sealed v0.7 archive and cache its manifest + lookup + trie. Reads
/// the manifest exactly once (vs three re-reads per free-function call).
pub fn open(path: &Path) -> Result<Self> {
let mut file = File::open(path)?;
let file_len = file.metadata()?.len();
let (entries, _) = read_full_manifest(&mut file, file_len)?;
// Read one reserved section by module name from the already-open handle,
// bounding the attacker-controlled declared length against the real file
// size before allocating (same guard as `read_reserved_section`).
let mut read_section = |module: &str| -> Result<Option<Vec<u8>>> {
let Some(entry) = entries.iter().find(|e| e.module_name == module) else {
return Ok(None);
};
anyhow::ensure!(
entry.index_offset.checked_add(entry.index_len)
.is_some_and(|end| end <= file_len),
"reserved section {} out of bounds (offset={}, len={}, file_len={})",
entry.module_name, entry.index_offset, entry.index_len, file_len
);
file.seek(SeekFrom::Start(entry.index_offset))?;
let mut bytes = vec![0u8; entry.index_len as usize];
file.read_exact(&mut bytes)?;
Ok(Some(bytes))
};
let lookup_bytes = read_section(LOOKUP_MODULE)?.ok_or_else(|| {
anyhow::anyhow!("archive has no lookup sub-index (not a v0.7 sealed archive)")
})?;
let cols = decode_lookup(&lookup_bytes)?;
let trie = match read_section(TRIE_MODULE)? {
Some(tb) => Some(fst::Map::new(tb).map_err(|e| anyhow::anyhow!("trie open: {e}"))?),
None => None,
};
Ok(Self { paths: cols.paths, locs: cols.locs, trie, file, file_len })
}
/// Total number of chunk rows cached (across all files).
pub fn row_count(&self) -> usize {
self.paths.len()
}
/// Read a single file's bytes by relative path — the cached, held-open
/// equivalent of [`get_file`](crate::get_file). Uses the in-memory locate
/// (no manifest re-read, no per-call `decode_lookup`) and the archive handle
/// opened once in [`open`](Self::open), then `pread`s + decompresses +
/// blake3-verifies each chunk. This is the selective-restore entry point:
/// hold one reader and call `read_file` per artifact instead of paying a full
/// index re-parse on every `get_file`.
///
/// Returns an error if `target` is not present in the archive.
pub fn read_file(&self, target: &str) -> Result<Vec<u8>> {
let chunks = self.locate(target);
anyhow::ensure!(!chunks.is_empty(), "file not found in archive: {target}");
crate::decompress::reassemble_file(&self.file, self.file_len, target, &chunks)
}
/// Locate every chunk of `target`, sorted by `chunk_seq` — the cached
/// equivalent of [`locate_file`], but with no I/O after `open`.
pub fn locate(&self, target: &str) -> Vec<ChunkLoc> {
let n = self.paths.len();
let hit = match &self.trie {
Some(map) => map.get(target.as_bytes()).map(|v| v as usize),
None => self.paths.binary_search_by(|p| p.as_str().cmp(target)).ok(),
};
let Some(hit) = hit else { return Vec::new(); };
// The trie value is attacker-controlled (see `locate_file`): a malicious
// archive can map `target` to a row index past the cached lookup table.
// Treat an out-of-range hit as "not found" instead of panicking below.
if hit >= n { return Vec::new(); }
// Expand to the contiguous run of rows sharing this path.
let mut start = hit;
while start > 0 && self.paths[start - 1] == target { start -= 1; }
let mut end = hit + 1;
while end < n && self.paths[end] == target { end += 1; }
let mut out: Vec<ChunkLoc> = self.locs[start..end].to_vec();
out.sort_by_key(|c| c.chunk_seq);
out
}
/// Per-file metadata for **every** file, sorted by `relative_path` — the
/// cached equivalent of [`get_all_files_meta`].
pub fn files_meta(&self) -> Vec<ArtifactMeta> {
self.aggregate(0, self.paths.len())
}
/// Per-file metadata for files whose path starts with `prefix` — the cached
/// equivalent of [`get_files_meta_with_prefix`].
pub fn files_meta_with_prefix(&self, prefix: &str) -> Vec<ArtifactMeta> {
let (lo, hi) = self.window(prefix);
self.aggregate(lo, hi)
}
/// Resolve the contiguous `[lo, hi)` row window matching `prefix` — trie
/// range-seek when present, else binary search over the sorted paths. Empty
/// prefix selects all rows; a non-matching prefix returns `(0, 0)`.
fn window(&self, prefix: &str) -> (usize, usize) {
use fst::{IntoStreamer, Streamer};
let n = self.paths.len();
if prefix.is_empty() {
return (0, n);
}
let lo = match &self.trie {
Some(map) => {
let mut stream = map.range().ge(prefix.as_bytes()).into_stream();
match stream.next() {
Some((k, v)) if k.starts_with(prefix.as_bytes()) => v as usize,
_ => return (0, 0),
}
}
None => self.paths.partition_point(|p| p.as_str() < prefix),
};
if lo >= n || !self.paths[lo].starts_with(prefix) {
return (0, 0);
}
let mut hi = lo;
while hi < n && self.paths[hi].starts_with(prefix) {
hi += 1;
}
(lo, hi)
}
/// Aggregate the contiguous chunk rows in `[lo, hi)` into one `ArtifactMeta`
/// per file (rows are already grouped by path in lookup order).
fn aggregate(&self, lo: usize, hi: usize) -> Vec<ArtifactMeta> {
let mut out = Vec::new();
let mut i = lo;
while i < hi {
let p = &self.paths[i];
let mut total = 0u64;
let mut count = 0u32;
let mut compressed = false;
while i < hi && &self.paths[i] == p {
total += self.locs[i].uncompressed_size;
count += 1;
compressed |= self.locs[i].compressed;
i += 1;
}
out.push(ArtifactMeta {
relative_path: p.clone(),
uncompressed_size: total,
chunk_count: count,
compressed,
});
}
out
}
}
/// Legacy fallback: aggregate per-file metadata from the merged main index.
fn files_meta_via_index(path: &Path, prefix: Option<&str>) -> Result<Vec<ArtifactMeta>> {
use arrow::array::{BooleanArray, StringArray, UInt64Array};
let (schema, batches) = read_znippy_index(path)?;
let batch = match batches.len() {
0 => return Ok(Vec::new()),
1 => batches.into_iter().next().unwrap(),
_ => arrow_select::concat::concat_batches(&schema, batches.iter())?,
};
// Downcast-or-error: the schema is attacker-controlled, so a wrong column
// type must surface as an `Err`, never an `unwrap` panic (DoS on `get`/meta).
// Downcast-or-error: the schema is attacker-controlled, so a wrong column
// type must surface as an `Err`, never an `unwrap` panic (DoS on `get`/meta).
let col = |n: &str| batch.column_by_name(n)
.ok_or_else(|| anyhow::anyhow!("index missing column {n}"));
let paths = col("relative_path")?.as_any().downcast_ref::<StringArray>()
.ok_or_else(|| anyhow::anyhow!("index column relative_path has unexpected type"))?;
let usize_col = col("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("index column uncompressed_size has unexpected type"))?;
let compressed = col("compressed")?.as_any().downcast_ref::<BooleanArray>()
.ok_or_else(|| anyhow::anyhow!("index column compressed has unexpected type"))?;
// Main index rows are not path-sorted, so aggregate via a map.
let mut agg: HashMap<&str, (u64, u32, bool)> = HashMap::new();
for i in 0..batch.num_rows() {
let p = paths.value(i);
if let Some(pre) = prefix {
if !p.starts_with(pre) { continue; }
}
let e = agg.entry(p).or_insert((0, 0, false));
e.0 += usize_col.value(i);
e.1 += 1;
e.2 |= compressed.value(i);
}
let mut out: Vec<ArtifactMeta> = agg.into_iter().map(|(p, (sz, c, comp))| ArtifactMeta {
relative_path: p.to_string(),
uncompressed_size: sz,
chunk_count: c,
compressed: comp,
}).collect();
out.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));
Ok(out)
}
/// O(n) fallback for archives written before the lookup sub-index existed.
fn locate_file_via_index(path: &Path, target: &str) -> Result<Vec<ChunkLoc>> {
use arrow::array::{BooleanArray, FixedSizeBinaryArray, StringArray, UInt32Array, UInt64Array};
let (schema, batches) = read_znippy_index(path)?;
let batch = match batches.len() {
0 => return Ok(Vec::new()),
1 => batches.into_iter().next().unwrap(),
_ => arrow_select::concat::concat_batches(&schema, batches.iter())?,
};
// Downcast-or-error: the schema is attacker-controlled, so a wrong column
// type must surface as an `Err`, never an `unwrap` panic (DoS on `get`).
let col = |n: &str| batch.column_by_name(n)
.ok_or_else(|| anyhow::anyhow!("index missing column {n}"));
let paths = col("relative_path")?.as_any().downcast_ref::<StringArray>()
.ok_or_else(|| anyhow::anyhow!("index column relative_path has unexpected type"))?;
let chunk_seq = col("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()
.ok_or_else(|| anyhow::anyhow!("index column chunk_seq has unexpected type"))?;
let fdata = col("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("index column fdata_offset has unexpected type"))?;
let compressed = col("compressed")?.as_any().downcast_ref::<BooleanArray>()
.ok_or_else(|| anyhow::anyhow!("index column compressed has unexpected type"))?;
let usize_col = col("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("index column uncompressed_size has unexpected type"))?;
let blob_offset = col("blob_offset")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("index column blob_offset has unexpected type"))?;
let blob_size = col("blob_size")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow::anyhow!("index column blob_size has unexpected type"))?;
let checksum = col("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()
.ok_or_else(|| anyhow::anyhow!("index column checksum has unexpected type"))?;
let mut out = Vec::new();
for i in 0..batch.num_rows() {
if paths.value(i) == target {
let ck = checksum32(checksum, i)?;
out.push(ChunkLoc {
chunk_seq: chunk_seq.value(i),
fdata_offset: fdata.value(i),
blob_offset: blob_offset.value(i),
blob_size: blob_size.value(i),
uncompressed_size: usize_col.value(i),
compressed: compressed.value(i),
checksum: ck,
});
}
}
out.sort_by_key(|c| c.chunk_seq);
Ok(out)
}
pub fn is_probably_compressed(path: &Path) -> bool {
if let Some(ext) = path.extension().and_then(|e| e.to_str()) {
let ext = ext.to_ascii_lowercase();
matches!(
ext.as_str(),
"zip" | "gz" | "bz2" | "xz" | "lz" | "lzma" | "7z" | "rar" | "cab"
| "jar" | "war" | "ear" | "zst" | "sz" | "lz4" | "tgz" | "txz"
| "tbz" | "apk" | "dmg" | "deb" | "rpm" | "arrow" | "mpeg" | "mpg"
| "jpeg" | "jpg" | "gif" | "bmp" | "png" | "crate" | "znippy"
| "zdata" | "parquet" | "webp" | "webm"
// already-compressed / high-entropy payloads (Skidbladnir bundles
// carry these): age = encrypted (never shrinks), iso = squashfs
// (already xz/zstd), pdf = Flate streams. NB: pg_dump plain-SQL
// (`.dump`/`.sql`) is deliberately NOT here — it compresses well.
| "age" | "iso" | "pdf"
// git's pack directory — the bulk of any real `.git`, and of
// gunnar's cold tier. `pack` holds deflated objects; `idx`,
// `midx` and `bitmap` hold object ids (hash output) and EWAH
// bitmaps, all incompressible. NB: `.rev` is deliberately NOT
// here — it is a permutation of 0..N as big-endian u32s, whose
// high bytes are mostly zero, so it genuinely does compress.
| "pack" | "idx" | "midx" | "bitmap"
)
} else {
false
}
}
/// The name-only decision, kept as the fast path it has always been.
///
/// This is step 2 of the policy in [`crate::precompressed`] — it sees a path and
/// nothing else, so it cannot reach an entry that carries no extension (a git
/// loose object is named for its oid) and it believes a name that lies. Callers
/// holding real bytes should go through
/// [`SkipPolicy`](crate::precompressed::SkipPolicy), which runs this first and
/// then refines it with a magic-byte probe.
pub fn should_skip_compression(path: &Path) -> bool {
is_probably_compressed(path)
}
#[cfg(test)]
mod skip_compression_tests {
use super::*;
#[test]
fn skips_already_compressed_and_encrypted() {
for p in ["secrets/secrets.age", "infra/talos/talos-1.13.iso", "s3/doc.pdf", "x.PDF"] {
assert!(should_skip_compression(Path::new(p)), "should skip {p}");
}
}
#[test]
fn compresses_plain_dump_and_config() {
// pg_dump plain-SQL + declarative config must still be compressed.
for p in ["dbdump/njord.dump", "njord.sql", "lakespec.json", "njordconf/serverspec.json"] {
assert!(!should_skip_compression(Path::new(p)), "should compress {p}");
}
}
/// A packfile and its index are the two largest things in any real `.git`,
/// and neither was on the extension table.
#[test]
fn skips_git_pack_directory_artefacts() {
for p in [
".git/objects/pack/pack-9f2c.pack",
".git/objects/pack/pack-9f2c.idx",
".git/objects/pack/multi-pack-index.midx",
".git/objects/pack/pack-9f2c.bitmap",
] {
assert!(should_skip_compression(Path::new(p)), "should skip {p}");
}
}
/// The pack directory is not uniformly incompressible, and the table must
/// not be widened to the whole directory. `.rev` is a permutation of
/// `0..N` as big-endian u32s — its high bytes are mostly zero.
#[test]
fn still_compresses_the_compressible_git_artefacts() {
for p in [
".git/objects/pack/pack-9f2c.rev",
".git/objects/pack/pack-9f2c.promisor",
".git/COMMIT_EDITMSG",
".git/config",
] {
assert!(!should_skip_compression(Path::new(p)), "should compress {p}");
}
}
}
#[derive(Debug, Default)]
pub struct VerifyReport {
pub total_files: usize,
pub verified_files: usize,
pub corrupt_files: usize,
pub total_bytes: u64,
pub verified_bytes: u64,
pub corrupt_bytes: u64,
pub chunks: u64,
}
pub fn list_archive_contents(path: &Path) -> Result<()> {
let (_schema, batches) = read_znippy_index(path)?;
for batch in &batches {
let paths = batch
.column_by_name("relative_path")
.and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>())
.ok_or_else(|| anyhow::anyhow!("missing relative_path column"))?;
let sizes = batch
.column_by_name("uncompressed_size")
.and_then(|c| c.as_any().downcast_ref::<arrow::array::UInt64Array>())
.ok_or_else(|| anyhow::anyhow!("missing uncompressed_size column"))?;
let chunk_seqs = batch
.column_by_name("chunk_seq")
.and_then(|c| c.as_any().downcast_ref::<arrow::array::UInt32Array>());
let group_ids = batch
.column_by_name("group_id")
.and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>());
let artifact_ids = batch
.column_by_name("artifact_id")
.and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>());
let versions = batch
.column_by_name("version")
.and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>());
for i in 0..batch.num_rows() {
// Only print once per file (chunk_seq == 0)
if let Some(seqs) = chunk_seqs {
if seqs.value(i) != 0 {
continue;
}
}
if let (Some(g), Some(a), Some(v)) = (group_ids, artifact_ids, versions) {
if !g.is_null(i) {
println!(
"{}\t{}\t{}:{}:{}",
paths.value(i),
sizes.value(i),
g.value(i),
a.value(i),
v.value(i)
);
continue;
}
}
println!("{}\t{}", paths.value(i), sizes.value(i));
}
}
Ok(())
}
pub fn verify_archive_integrity(path: &Path) -> Result<VerifyReport> {
let out_dir = PathBuf::from("/dev/null");
decompress_archive(path, false, &out_dir)
}
#[cfg(test)]
mod version_tests {
use super::*;
#[test]
fn current_and_older_versions_are_accepted() {
let mut m = HashMap::new();
// No version recorded (legacy / pre-version archives): read as before.
check_format_version(&m).unwrap();
// Exactly the supported version.
m.insert(FORMAT_VERSION_KEY.into(), ZNIPPY_FORMAT_VERSION.to_string());
check_format_version(&m).unwrap();
// An older version.
m.insert(FORMAT_VERSION_KEY.into(), (ZNIPPY_FORMAT_VERSION - 1).to_string());
check_format_version(&m).unwrap();
}
#[test]
fn newer_version_is_rejected_clearly() {
let mut m = HashMap::new();
m.insert(FORMAT_VERSION_KEY.into(), (ZNIPPY_FORMAT_VERSION + 1).to_string());
let err = check_format_version(&m).unwrap_err().to_string();
assert!(err.contains("newer than this reader supports"), "got: {err}");
assert!(err.contains("upgrade znippy"), "got: {err}");
}
#[test]
fn writer_records_the_current_version() {
let meta = build_arrow_metadata_for_config(&crate::common_config::CONFIG);
assert_eq!(
meta.get(FORMAT_VERSION_KEY).map(String::as_str),
Some(ZNIPPY_FORMAT_VERSION.to_string().as_str())
);
}
}
#[cfg(test)]
mod checksum_width_tests {
use super::*;
use arrow::array::FixedSizeBinaryArray;
/// `FixedSizeBinaryArray` is ONE concrete Arrow type for every width, so the
/// `downcast_ref::<FixedSizeBinaryArray>()` in `decode_lookup` /
/// `locate_file_via_index` succeeds just as happily for a hostile archive
/// declaring `FixedSizeBinary(16)`. The old code then did
/// `[0u8; 32].copy_from_slice(col.value(i))`, which panics — a process abort
/// driven purely by an attacker-supplied schema, on the `znippy get` path.
#[test]
fn narrow_checksum_column_errors_instead_of_panicking() {
let narrow = FixedSizeBinaryArray::try_from_iter([[0u8; 16]].into_iter()).unwrap();
assert_eq!(narrow.value_length(), 16);
let err = checksum32(&narrow, 0).expect_err("a 16-byte checksum column must be an Err");
assert!(
err.to_string().contains("width 16"),
"error must name the bad width, got: {err}"
);
}
#[test]
fn wide_checksum_column_errors_instead_of_truncating() {
let wide = FixedSizeBinaryArray::try_from_iter([[7u8; 64]].into_iter()).unwrap();
assert!(checksum32(&wide, 0).is_err(), "a 64-byte checksum column must be an Err");
}
#[test]
fn correct_width_checksum_is_read_verbatim() {
let mut digest = [0u8; 32];
for (i, b) in digest.iter_mut().enumerate() {
*b = i as u8;
}
let col = FixedSizeBinaryArray::try_from_iter([digest].into_iter()).unwrap();
assert_eq!(checksum32(&col, 0).unwrap(), digest);
}
}
#[cfg(test)]
mod reserved_module_tests {
use super::*;
/// The regression this guards is specific and it has teeth: a module name
/// that the sink writes with `RESERVED_PKG_TYPE` but that
/// [`is_reserved_module`] does not recognise is classified as a **data**
/// sub-index. `read_multi_index` then merges its rows into the file index and
/// `read_znippy_manifest` reports it as a data section — corrupting `list`,
/// `decompress`, the iceberg sink and every manifest-count assertion.
///
/// So the assertion is not "the constant exists"; it is "every reserved
/// module name in the catalog is classified reserved, and nothing else is".
#[test]
fn every_reserved_module_is_classified_reserved() {
for m in RESERVED_MODULES {
assert!(
is_reserved_module(m),
"'{m}' is in RESERVED_MODULES but is_reserved_module says it is DATA — \
its rows would be merged into the file index"
);
}
}
/// The three `git` package-format modules, named literally. If someone
/// removes one from [`RESERVED_MODULES`], the loop above still passes
/// (it iterates whatever is left) — this is the test that goes red.
#[test]
fn the_git_format_modules_are_reserved() {
for m in [
GUNNAR_OID_MODULE,
GUNNAR_GRAPH_MODULE,
GUNNAR_REACH_MODULE,
GUNNAR_REFS_MODULE,
GUNNAR_SECRETS_MODULE,
] {
assert!(is_reserved_module(m), "git package-format module '{m}' must be reserved");
assert!(
RESERVED_MODULES.contains(&m),
"git package-format module '{m}' must be in the catalog"
);
}
assert_eq!(GUNNAR_OID_MODULE, "__gunnar_oid__");
assert_eq!(GUNNAR_GRAPH_MODULE, "__gunnar_graph__");
assert_eq!(GUNNAR_REACH_MODULE, "__gunnar_reach__");
assert_eq!(GUNNAR_REFS_MODULE, "__gunnar_refs__");
assert_eq!(GUNNAR_SECRETS_MODULE, "__gunnar_secrets__");
}
/// The other direction: a data module must NOT be swept up as reserved, or
/// its file rows would silently vanish from `list` and `decompress`.
#[test]
fn ordinary_module_names_stay_data() {
for m in ["maven", "git", "rust", "", "__gunnar__", "gunnar_oid", "__gunnar_oid", "objects"] {
assert!(!is_reserved_module(m), "'{m}' must be treated as a DATA sub-index");
}
}
/// No duplicates and no accidental empty entry in the catalog.
#[test]
fn the_catalog_is_well_formed() {
let mut seen = std::collections::HashSet::new();
for m in RESERVED_MODULES {
assert!(!m.is_empty(), "empty reserved module name");
assert!(seen.insert(*m), "duplicate reserved module name '{m}'");
}
assert_eq!(seen.len(), RESERVED_MODULES.len());
}
}
#[cfg(test)]
mod missing_footer_tests {
use super::*;
fn err_for(bytes: &[u8]) -> String {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("a.znippy");
std::fs::write(&p, bytes).unwrap();
read_znippy_index_filtered(&p, &IndexFilter::default())
.expect_err("no v0.7 footer must be an Err")
.to_string()
}
/// A file whose tail is blob bytes (a truncated archive) is called truncated,
/// not v0.6.
#[test]
fn truncated_tail_is_reported_as_truncated() {
let junk: Vec<u8> = (0..4096u32).map(|i| (i * 2654435761u32 >> 13) as u8).collect();
let msg = err_for(&junk);
assert!(msg.contains("truncated") && !msg.contains("v0.6"), "{msg}");
let msg = err_for(b"short");
assert!(msg.contains("truncated"), "{msg}");
}
/// A real v0.6 shape — trailing u64 pointing at an Arrow IPC index — keeps the
/// "re-compress with v0.7" message.
#[test]
fn v06_shape_keeps_the_v06_message() {
let mut bytes = vec![0u8; 100];
bytes.extend_from_slice(b"ARROW1\0\0");
bytes.extend_from_slice(&[0u8; 32]);
bytes.extend_from_slice(&100u64.to_le_bytes());
let msg = err_for(&bytes);
assert!(msg.contains("v0.6"), "{msg}");
}
}