znippy-common/src/hot.rs
//! `HotArchive` — the **dynamic half** of a znippy archive: append blobs and
//! index rows at O(1) per append, and pay the static seal exactly once.
//!
//! # The defect this exists to remove, measured
//!
//! [`append_files`](crate::append_files) re-seals the *whole* metadata tail on
//! every call: [`ArrowIpcSinkAppend::open_existing`](crate::ArrowIpcSinkAppend::open_existing)
//! decodes every pre-existing row out of the lookup section, `finish()` re-emits
//! them as a data sub-index, re-sorts all of them, rebuilds the fst trie over all
//! of them, and rewrites manifest and footer. So one append costs O(rows already
//! in the archive) and N appends cost **O(N²)**.
//!
//! That is not a theory. **MEASURED on oden 2026-08-04**, loadavg 14–17, 32
//! cores / 512 GB, release build, archive on **md1** (`/dev/md1`, not tmpfs — so
//! the two fsyncs per append are real), one archive, two entries appended per
//! call (a `pack-*.pack` and its `.idx`, ~1.5 KiB together),
//! `SkipPolicy::already_compressed()`. Same fixture, same rungs, A/B in one
//! binary:
//!
//! | appends | CPU/append, re-seal | CPU/append, hot | wall total, re-seal | wall total, hot |
//! |---|---|---|---|---|
//! | 800 | 1.88 ms | — | | |
//! | 2 000 | 2.66 ms | 0.075 ms | | |
//! | 4 000 | 5.38 ms | 0.075 ms | | |
//! | 8 000 | **10.18 ms** | **0.060 ms** | **61.16 s** | **2.29 s** |
//! | 32 000 | (not run) | 0.061 ms | | 5.62 s |
//!
//! Re-seal's per-append cost doubles every time N doubles — a straight line
//! through the origin, the signature of a quadratic total. The hot path's is
//! **flat from 2 000 to 32 000 appends**, which is the claim. At 8 000 appends
//! that is **26.7× on wall clock and 170× on CPU per append**, and both ratios
//! grow with N because only one side of them grows.
//!
//! INFERRED, not measured: extrapolating re-seal's line, the 100 000th append
//! costs ~127 ms and getting there costs ~1.8 h of CPU. The hot path's 100 000th
//! append costs what its 2 000th did.
//!
//! The seal is the once-only price: **0.08 s of CPU for 64 000 rows**. At 8 000
//! appends the two paths land on the same archive — 15 929 871 B sealed from hot
//! against 15 932 909 B re-sealed, the 3 038 B being the seed entry
//! `append_files` needs and the hot path does not (it can start from nothing).
//!
//! # The two-phase shape
//!
//! ```text
//! HOT COLD
//! foo.znippy blobs only, pure append foo.znippy [blobs][index][lookup][trie][manifest][footer]
//! foo.znippy.hot.NNNNN Arrow IPC journal (journal removed)
//! ```
//!
//! * [`HotArchive::append`] writes the blob bytes past the blob cursor and
//! *one* Arrow IPC RecordBatch of the new index rows to the journal. It never
//! reads, sorts or rewrites a row it wrote earlier, so its cost depends only on
//! what is being appended. **O(1) in the archive's size.**
//! * [`HotArchive::seal`] pays the static cost once: one sorted lookup, one fst
//! trie, one manifest, one footer.
//!
//! This is znippy's own `znippy-iceberg` architecture — dynamic metadata store,
//! then `seal()` freezing it into the static `.znippy` with the blobs copied
//! verbatim — with the metadata store made local and cheap instead of an Iceberg
//! warehouse. Iceberg's `fast_append` rewrites its manifest list on every commit,
//! which is the same O(N²) in a different format.
//!
//! # Why the sealed output is byte-identical to `create_archive`
//!
//! The blobs are written in the same order at the same offsets, and the seal
//! pushes exactly one data sub-index carrying the rows in insertion order —
//! literally the same call [`create_archive`](crate::create_archive) makes. So
//! *appending N files one at a time and sealing* produces the **same bytes** as
//! *creating the archive from all N at once*. `seal_matches_create_archive_byte_for_byte`
//! asserts that on real archives rather than asserting a row count, because a
//! seal that silently dropped the trie would still be able to report the right
//! number of rows.
//!
//! # Journal segments, and why there is more than one file
//!
//! An Arrow IPC *stream* begins with a schema message, so appending batches to
//! one across process restarts is not possible without re-emitting the schema —
//! 424 B of flatbuffer metadata on a four-row batch, paid per push. Instead each
//! **process** that opens the archive for append starts a new segment
//! `foo.znippy.hot.NNNNN`; the schema is written once per segment, and a push
//! costs one RecordBatch message. Segment count grows with restarts, not with
//! pushes.
//!
//! A torn segment (killed mid-write) is read up to its last complete message and
//! the rest is discarded, which is what a log reader must do; the blob bytes
//! whose rows were lost stay in the file as dead payload the seal drops, exactly
//! like [`AppendReport::rows_replaced`](crate::AppendReport::rows_replaced).
//!
//! # Ordering, so a row never points at a blob that is not there
//!
//! `append` fsyncs the blob bytes **before** the journal batch that references
//! them. A crash between the two loses the append and nothing else; a crash the
//! other way round would leave an index row pointing into a hole.
use std::collections::HashMap;
use std::fs::File;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock, RwLockReadGuard};
use anyhow::{Result, anyhow, bail};
use arrow::ipc::reader::StreamReader;
use arrow::ipc::writer::StreamWriter;
use crate::archive::ZnippyReader;
use crate::codec::CompressCtx;
use crate::index::{ChunkLoc, data_subindex_schema};
use crate::meta_sink::{ArchiveMetaSink, ArrowIpcSink, GroupKey, ReservedSectionBuilder};
use crate::meta_sink_append::{base_batch_from_rows, decode_base_rows, write_blobs};
/// Suffix of a journal segment: `<archive>.hot.<NNNNN>`.
const JOURNAL_INFIX: &str = ".hot.";
/// What one [`HotArchive::append`] did. Deliberately the same three counts
/// [`AppendReport`](crate::AppendReport) reports, so a caller can swap paths
/// without re-reading its own accounting.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct HotAppendReport {
/// Index rows written by this append (one per chunk).
pub rows_added: u64,
/// Rows already in the journal that this append superseded, by
/// `relative_path`. Their blob bytes stay in the file as dead payload.
pub rows_replaced: u64,
/// Byte offset the appended blob region started at.
pub blob_append_offset: u64,
/// Blob payload bytes written (compressed or stored raw).
pub blob_bytes_added: u64,
/// Bytes the journal batch cost — the index-side price of this append.
pub journal_bytes_added: u64,
}
/// What one [`HotArchive::seal`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HotSealReport {
pub archive: PathBuf,
/// Rows carried out of the journal into the static index.
pub rows_sealed: u64,
/// Journal segments removed.
pub segments_removed: usize,
/// Journal bytes reclaimed.
pub journal_bytes_reclaimed: u64,
/// Final size of the sealed archive.
pub sealed_total_bytes: u64,
}
/// A znippy archive in its **hot** phase: blobs appended in place, index rows in
/// a journal beside it.
///
/// [`open`](Self::open) reads the journal once (O(rows)); every [`append`](Self::append)
/// after that is O(what is appended). Holding the rows in memory is what buys
/// that, and it is the same set of rows the old path decoded *per append*.
///
/// # Why an append no longer stops the readers
///
/// `append` used to take `&mut self`, so a caller sharing one archive across
/// threads had to put it behind an `RwLock` and take the EXCLUSIVE guard for the
/// whole call — and the whole call is two `fsync`s. MEASURED in holger
/// (`.nornir/stinky/12-dataplane-finished.md` §5.4, oden 2026-09-20, 32 readers
/// against 4 publishers): readers kept **2.2–3.0 %** of their throughput, and it
/// got WORSE when group commit doubled the write rate, because the write side
/// then held that guard for a larger share of the window. Nothing the caller
/// could write shortened it; the `&mut` was the mechanism.
///
/// The exclusion was never needed for the durable part. An append writes blob
/// bytes **past** the blob cursor and a journal message **past** the journal's
/// end — two regions no reader can name, because no index row points at them
/// yet. The only instant a reader must not observe is the index swap, and that
/// is a `Vec::push` and a `HashMap::insert` per row appended.
///
/// So the three states are guarded separately:
///
/// | state | guard | held for |
/// |---|---|---|
/// | blob cursor · journal writer · codec context | [`HotWriter`] behind a `Mutex` | the whole append — WRITERS ONLY |
/// | the live index rows | [`HotIndex`] behind an `RwLock` | the swap, after both fsyncs |
/// | blob end, for the accessor | `AtomicU64` | nothing |
///
/// `append` therefore takes `&self`, and a reader that is not naming a row this
/// append creates never waits on either fsync. [`fsyncs`](Self::fsyncs) counts them so
/// the claim is countable rather than argued.
pub struct HotArchive {
archive: PathBuf,
blobs: Arc<File>,
/// Reader-visible state, and the ONLY thing an append takes an exclusive
/// guard on — after both fsyncs are already durable.
index: RwLock<HotIndex>,
/// Published mirror of [`HotWriter::blob_end`], stored under the index write
/// guard so a reader holding that guard can never see the two disagree.
blob_end: AtomicU64,
/// Append-side state: one appender at a time, and no reader waits on it.
writer: Mutex<HotWriter>,
/// `fsync`s this archive has paid inside [`append`](Self::append), blobs and
/// journal together. An append costs exactly TWO, whatever it appends, and
/// that has not changed — what changed is that no reader is inside them.
/// Exposed by [`fsyncs`](Self::fsyncs) so a bench can divide it by its own
/// append count and STATE the ratio instead of asserting it.
fsyncs: AtomicU64,
journal_path: PathBuf,
policy: crate::SkipPolicy,
}
/// The live index — what a reader reads, and the whole of what an append has to
/// exclude a reader from.
struct HotIndex {
/// Every live row, in insertion order. The seal emits exactly this.
paths: Vec<String>,
locs: Vec<ChunkLoc>,
/// `relative_path` -> index into `paths`/`locs`, for replace semantics.
by_path: HashMap<String, usize>,
}
/// The append side, behind a `Mutex` that serialises WRITERS against each other
/// and nothing else.
struct HotWriter {
/// Byte offset one past the last blob written — the authoritative cursor.
/// [`HotArchive::blob_end`] is its published mirror.
blob_end: u64,
/// The segment this process appends to, schema already written.
journal: StreamWriter<File>,
journal_len: u64,
/// Long-lived, per Design Law 2 — a codec context built per push would put
/// the setup cost back into the hot path this module exists to empty. It
/// lives here rather than in a `Mutex` of its own because it is writer
/// state, and the append lock already owns it. (The raw OpenZL context is
/// `Send` but not `Sync`, which is what the old separate `Mutex` was for.)
ctx: CompressCtx,
}
impl HotArchive {
/// Open `archive` for hot appends, creating it if absent.
///
/// Three entry states, all handled here so no caller has to know which it is:
///
/// * **already hot** — journal segments exist; their rows are loaded and the
/// blob cursor is taken from the last one.
/// * **sealed** — a static `.znippy`; its rows are recovered into a fresh
/// segment and the (rebuildable) metadata tail is dropped. O(rows), once.
/// * **absent** — a new archive at blob offset 0.
pub fn open(archive: &Path, level: i32, policy: crate::SkipPolicy) -> Result<Self> {
let archive = archive.to_path_buf();
let segments = journal_segments(&archive)?;
let (mut paths, mut locs) = (Vec::new(), Vec::new());
let mut reheated = false;
if !segments.is_empty() {
for (_, seg) in &segments {
let (p, l) = read_segment(seg)?;
paths.extend(p);
locs.extend(l);
}
} else if archive.is_file() && crate::index::read_znippy_full_manifest(&archive).is_ok() {
// Sealed → hot. Recover the rows out of the static index, then drop
// the tail. The journal is made durable BEFORE the truncation, so a
// crash in between leaves the rows recoverable and the stale tail is
// simply overwritten by the next append (and `set_len` below).
let (entries, _) = crate::index::read_znippy_full_manifest(&archive)?;
let (p, l) = crate::meta_sink_append::recover_rows(&archive, &entries)?;
paths = p;
locs = l;
reheated = true;
}
// Dedup to live rows: last writer for a path wins, insertion order kept.
let (paths, locs) = live_rows(paths, locs);
// The blob cursor is the end of the last live blob — NOT the file length,
// which on the sealed path still includes the metadata tail and after a
// torn append may include payload no row references.
let blob_end = locs
.iter()
.map(|l| l.blob_offset + l.blob_size)
.max()
.unwrap_or(0);
let next = segments.last().map(|(n, _)| n + 1).unwrap_or(0);
let journal_path = segment_path(&archive, next);
let mut journal = new_segment(&journal_path)?;
if reheated {
// Segment 0 of a re-heated archive carries the recovered rows, so the
// journal alone is a complete statement of the archive's index.
let batch = base_batch_from_rows(&paths, &locs)?;
journal
.write(&batch)
.map_err(|e| anyhow!("hot: seeding journal from sealed index: {e}"))?;
journal.flush().map_err(|e| anyhow!("hot: journal flush: {e}"))?;
journal
.get_ref()
.sync_all()
.map_err(|e| anyhow!("hot: journal fsync: {e}"))?;
sync_parent_dir(&journal_path);
}
let journal_len = std::fs::metadata(&journal_path).map(|m| m.len()).unwrap_or(0);
let blobs = Arc::new(
std::fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&archive)
.map_err(|e| anyhow!("hot: open {} : {e}", archive.display()))?,
);
// Drop anything past the live blob region — a stale metadata tail from
// the sealed path, or a torn append's orphan payload. Idempotent.
if blobs.metadata()?.len() > blob_end {
blobs.set_len(blob_end)?;
blobs.sync_all()?;
}
let mut by_path = HashMap::with_capacity(paths.len());
for (i, p) in paths.iter().enumerate() {
by_path.insert(p.clone(), i);
}
Ok(Self {
archive,
blobs,
blob_end: AtomicU64::new(blob_end),
index: RwLock::new(HotIndex { paths, locs, by_path }),
writer: Mutex::new(HotWriter {
blob_end,
journal,
journal_len,
ctx: CompressCtx::new(level)?,
}),
journal_path,
policy,
fsyncs: AtomicU64::new(0),
})
}
/// Whether `archive` currently has a journal — i.e. is in its hot phase.
pub fn is_hot(archive: &Path) -> bool {
journal_segments(archive).map(|s| !s.is_empty()).unwrap_or(false)
}
/// The archive path.
pub fn archive(&self) -> &Path {
&self.archive
}
/// Live index rows.
pub fn rows(&self) -> u64 {
self.index_read().paths.len() as u64
}
/// Byte offset one past the last blob.
pub fn blob_end(&self) -> u64 {
self.blob_end.load(Ordering::Acquire)
}
/// `fsync`s paid by [`append`](Self::append) on THIS archive so far — two
/// per non-empty append, none for an empty one.
pub fn fsyncs(&self) -> u64 {
self.fsyncs.load(Ordering::Relaxed)
}
/// The live index, shared.
///
/// A poisoned lock means a thread panicked holding it. The index is a
/// `Vec`/`HashMap` pair mutated only under the write guard, in a loop with
/// no fallible step, so there is nothing half-applied to recover from and
/// refusing every later read would be the worse answer.
fn index_read(&self) -> RwLockReadGuard<'_, HotIndex> {
self.index.read().unwrap_or_else(|e| e.into_inner())
}
/// Append `files`, paying only for `files`.
///
/// Replace semantics match [`append_files`](crate::append_files): a
/// `relative_path` already present is superseded and its old blob bytes
/// become dead payload.
///
/// Takes `&self`. The blob write with its fsync, and the journal commit with
/// its fsync, are guarded against other WRITERS only; the live index is
/// taken exclusively once, at the end, for the rows this call adds. A reader
/// on a region this append is not writing never waits on either fsync — and
/// it cannot be on one that it IS writing, because the row naming those
/// bytes does not exist until after both.
pub fn append(&self, files: &[(String, Vec<u8>)]) -> Result<HotAppendReport> {
if files.is_empty() {
return Ok(HotAppendReport {
blob_append_offset: self.blob_end(),
..Default::default()
});
}
// ── writers only, from here down to the swap ─────────────────────────
let mut guard = self.writer.lock().unwrap_or_else(|e| e.into_inner());
let w = &mut *guard;
let blob_append_offset = w.blob_end;
// Blobs first, and durable before any row points at them. Written PAST
// the cursor, into a region no live row names.
let (new_paths, new_locs, cursor) =
write_blobs(&self.blobs, w.blob_end, files, &mut w.ctx, self.policy)?;
let blob_bytes_added = cursor - blob_append_offset;
self.blobs.sync_all()?;
self.fsyncs.fetch_add(1, Ordering::Relaxed);
w.blob_end = cursor;
// One RecordBatch, one message on the journal. No re-sort, no trie, no
// manifest — that is the entire difference from `append_files`.
let batch = base_batch_from_rows(&new_paths, &new_locs)?;
w.journal
.write(&batch)
.map_err(|e| anyhow!("hot: journal write: {e}"))?;
w.journal.flush().map_err(|e| anyhow!("hot: journal flush: {e}"))?;
w.journal
.get_ref()
.sync_all()
.map_err(|e| anyhow!("hot: journal fsync: {e}"))?;
self.fsyncs.fetch_add(1, Ordering::Relaxed);
let journal_len = std::fs::metadata(&self.journal_path)?.len();
let journal_bytes_added = journal_len - w.journal_len;
w.journal_len = journal_len;
// ── the index swap: the ONLY part a reader is excluded from ──────────
// Both halves are on disk already, so a reader taking the guard on
// either side of this sees a consistent archive — before, without these
// rows; after, with them AND with their bytes durable.
let rows_replaced = {
let mut idx = self.index.write().unwrap_or_else(|e| e.into_inner());
let mut replaced = 0u64;
for (p, l) in new_paths.into_iter().zip(new_locs) {
match idx.by_path.get(&p).copied() {
Some(i) => {
replaced += 1;
idx.locs[i] = l;
}
None => {
let at = idx.paths.len();
idx.by_path.insert(p.clone(), at);
idx.paths.push(p);
idx.locs.push(l);
}
}
}
self.blob_end.store(cursor, Ordering::Release);
replaced
};
Ok(HotAppendReport {
rows_added: batch.num_rows() as u64,
rows_replaced,
blob_append_offset,
blob_bytes_added,
journal_bytes_added,
})
}
/// Freeze the hot archive into a static `.znippy` and drop the journal.
///
/// The blobs are already in place and are not touched: the seal writes the
/// metadata tail only. `reserved` is the same hook
/// [`ArrowIpcSink::with_reserved_builder`](crate::ArrowIpcSink::with_reserved_builder)
/// offers, so a caller with its own sections (gunnar's `__gunnar_refs__`)
/// seals them here rather than needing a second seal.
pub fn seal(self, reserved: Option<ReservedSectionBuilder>) -> Result<HotSealReport> {
let Self {
archive,
blobs,
index,
writer,
journal_path,
..
} = self;
// `self` is owned, so neither guard can be contended: no other reference
// to this archive exists. Both are taken by value.
let HotIndex { paths, locs, .. } = index.into_inner().unwrap_or_else(|e| e.into_inner());
let HotWriter { blob_end, journal, .. } =
writer.into_inner().unwrap_or_else(|e| e.into_inner());
drop(journal);
// The ORIGINAL sink, not the append clone: the seal has every row in
// hand, so it needs none of the resume machinery — and this is the sink
// that already carries the reserved-section hook gunnar's
// `__gunnar_refs__` needs. The two are proven byte-identical on the fresh
// path by `clone_fresh_path_is_byte_identical_to_original`.
let rows_sealed = paths.len() as u64;
let mut sink = ArrowIpcSink::new(blobs, blob_end);
if let Some(b) = reserved {
sink = sink.with_reserved_builder(b);
}
if !paths.is_empty() {
let batch = base_batch_from_rows(&paths, &locs)?;
let schema = data_subindex_schema();
sink.push_subindex(schema.as_ref(), &[batch], GroupKey {
pkg_type: 0,
repo: String::new(),
module_name: String::new(),
})?;
}
let sealed_total_bytes = Box::new(sink).finish()?;
// The static archive is durable; only now is the journal redundant.
let segments = journal_segments(&archive)?;
let mut journal_bytes_reclaimed = 0u64;
for (_, seg) in &segments {
journal_bytes_reclaimed += std::fs::metadata(seg).map(|m| m.len()).unwrap_or(0);
std::fs::remove_file(seg)
.map_err(|e| anyhow!("hot: removing journal {}: {e}", seg.display()))?;
}
sync_parent_dir(&journal_path);
Ok(HotSealReport {
archive,
rows_sealed,
segments_removed: segments.len(),
journal_bytes_reclaimed,
sealed_total_bytes,
})
}
}
/// The hot archive reads without being sealed — that is the other half of "worth
/// having". Same trait the static [`ZnippyArchive`](crate::ZnippyArchive)
/// implements, so a caller does not branch on phase.
///
/// Every one of these takes the index's SHARED guard, and none of them holds it
/// across I/O: `extract_file` copies the [`ChunkLoc`] out and does its `pread`
/// with the guard already released. So the index lock is held for a hash lookup
/// and nothing else, on the read side as on the write side.
impl ZnippyReader for HotArchive {
fn list_files(&self) -> Result<Vec<String>> {
Ok(self.index_read().paths.clone())
}
fn contains(&self, relative_path: &str) -> bool {
self.index_read().by_path.contains_key(relative_path)
}
fn file_size(&self, relative_path: &str) -> Option<u64> {
let idx = self.index_read();
idx.by_path
.get(relative_path)
.map(|&i| idx.locs[i].uncompressed_size)
}
fn extract_file(&self, relative_path: &str) -> Result<Vec<u8>> {
let loc = {
let idx = self.index_read();
let &i = idx
.by_path
.get(relative_path)
.ok_or_else(|| anyhow!("file not found in archive: {relative_path}"))?;
idx.locs[i].clone()
};
let mut blob = vec![0u8; loc.blob_size as usize];
self.blobs.read_exact_at(&mut blob, loc.blob_offset)?;
if loc.compressed {
let mut out = Vec::new();
crate::codec::decompress_into(&blob, &mut out)?;
Ok(out)
} else {
Ok(blob)
}
}
}
/// Keep the last row written for each `relative_path`, in first-insertion order —
/// the same last-writer-wins the re-sealing path gets from `drop_carried_paths`.
fn live_rows(paths: Vec<String>, locs: Vec<ChunkLoc>) -> (Vec<String>, Vec<ChunkLoc>) {
let mut at: HashMap<&str, usize> = HashMap::with_capacity(paths.len());
let mut order: Vec<usize> = Vec::with_capacity(paths.len());
for (i, p) in paths.iter().enumerate() {
match at.get(p.as_str()).copied() {
Some(slot) => order[slot] = i,
None => {
at.insert(p.as_str(), order.len());
order.push(i);
}
}
}
let order: Vec<usize> = order;
let mut out_p = Vec::with_capacity(order.len());
let mut out_l = Vec::with_capacity(order.len());
for &i in &order {
out_p.push(paths[i].clone());
out_l.push(locs[i].clone());
}
(out_p, out_l)
}
fn segment_path(archive: &Path, n: u32) -> PathBuf {
let mut s = archive.as_os_str().to_os_string();
s.push(format!("{JOURNAL_INFIX}{n:05}"));
PathBuf::from(s)
}
/// Every journal segment for `archive`, sorted by sequence number.
fn journal_segments(archive: &Path) -> Result<Vec<(u32, PathBuf)>> {
let dir = archive.parent().filter(|p| !p.as_os_str().is_empty());
let dir = dir.map(|d| d.to_path_buf()).unwrap_or_else(|| PathBuf::from("."));
let stem = archive
.file_name()
.and_then(|n| n.to_str())
.ok_or_else(|| anyhow!("hot: {} has no usable file name", archive.display()))?;
let prefix = format!("{stem}{JOURNAL_INFIX}");
let mut out = Vec::new();
let rd = match std::fs::read_dir(&dir) {
Ok(rd) => rd,
Err(_) => return Ok(out),
};
for e in rd.flatten() {
let name = e.file_name();
let Some(name) = name.to_str() else { continue };
let Some(tail) = name.strip_prefix(&prefix) else { continue };
let Ok(n) = tail.parse::<u32>() else { continue };
out.push((n, e.path()));
}
out.sort_by_key(|(n, _)| *n);
Ok(out)
}
/// Buffer alignment for the journal, in bytes.
///
/// **8, not arrow-rs's default 64.** The 64 is arrow-rs's choice, not Arrow's —
/// arrow-cpp and pyarrow have always written 8 — so this is what the C++
/// reference implementation does and not a bent format. It matters here and
/// almost nowhere else, because a journal batch is *small*: the base schema has
/// eight columns and a two-row batch therefore carries a dozen-plus buffers that
/// are each rounded up. MEASURED on oden, 2026-08-04, two rows per append
/// (a `pack-*.pack` and its `.idx`, 40-hex names): **1 664 B/append at 64-byte
/// alignment, 824 B at 8** — 2.02×, for byte-identical rows. Flat at 824 B all
/// the way to 32 000 appends, so it is a per-append constant and not a curve.
///
/// The **sealed** archive is untouched by this: it is written by `ArrowIpcSink`
/// with arrow-rs's default, so every `.znippy` on disk keeps the bytes its
/// readers already expect. This is the transient half only.
const JOURNAL_ALIGNMENT: usize = 8;
fn new_segment(path: &Path) -> Result<StreamWriter<File>> {
let f = File::create(path)
.map_err(|e| anyhow!("hot: create journal segment {}: {e}", path.display()))?;
let schema = data_subindex_schema();
let opts = arrow::ipc::writer::IpcWriteOptions::try_new(
JOURNAL_ALIGNMENT,
false,
arrow::ipc::MetadataVersion::V5,
)
.map_err(|e| anyhow!("hot: journal write options: {e}"))?;
let w = StreamWriter::try_new_with_options(f, schema.as_ref(), opts)
.map_err(|e| anyhow!("hot: journal schema: {e}"))?;
sync_parent_dir(path);
Ok(w)
}
/// Read one segment, tolerating a torn tail.
///
/// A killed process leaves a partial final message. Every complete message
/// before it is real and is kept; the partial one is dropped, which is what makes
/// the journal a log rather than a document. `decode_base_rows` cannot be reused
/// here for exactly that reason — it fails the whole stream on the first bad
/// batch, which is right for a sealed section and wrong for a log.
fn read_segment(path: &Path) -> Result<(Vec<String>, Vec<ChunkLoc>)> {
let bytes = std::fs::read(path)
.map_err(|e| anyhow!("hot: reading journal {}: {e}", path.display()))?;
if bytes.is_empty() {
return Ok((Vec::new(), Vec::new()));
}
let reader = match StreamReader::try_new(std::io::Cursor::new(crate::ipc_guard::valid_prefix(&bytes)), None) {
Ok(r) => r,
// A segment whose schema message never landed carries nothing.
Err(_) => return Ok((Vec::new(), Vec::new())),
};
let mut paths = Vec::new();
let mut locs = Vec::new();
let mut complete: Vec<arrow::record_batch::RecordBatch> = Vec::new();
for batch in reader {
match batch {
Ok(b) => complete.push(b),
Err(_) => break, // torn tail
}
}
for b in &complete {
let mut ipc: Vec<u8> = Vec::new();
{
let mut w = StreamWriter::try_new(&mut ipc, b.schema().as_ref())
.map_err(|e| anyhow!("hot: re-encode: {e}"))?;
w.write(b).map_err(|e| anyhow!("hot: re-encode write: {e}"))?;
w.finish().map_err(|e| anyhow!("hot: re-encode finish: {e}"))?;
}
let (p, l) = decode_base_rows(&ipc)?;
paths.extend(p);
locs.extend(l);
}
Ok((paths, locs))
}
/// A newly-created or removed file is only durable once its directory entry is.
fn sync_parent_dir(path: &Path) {
let parent = match path.parent() {
Some(p) if !p.as_os_str().is_empty() => p.to_path_buf(),
_ => PathBuf::from("."),
};
if let Ok(f) = File::open(parent) {
let _ = f.sync_all();
}
}
/// `HotArchive::seal` as a free function, for a caller that has a path and not a
/// handle. Refuses an archive with no journal rather than silently producing an
/// empty one.
pub fn seal_hot(archive: &Path, level: i32, policy: crate::SkipPolicy) -> Result<HotSealReport> {
if !HotArchive::is_hot(archive) {
bail!("{} has no journal; it is not hot", archive.display());
}
HotArchive::open(archive, level, policy)?.seal(None)
}
// ── tests ────────────────────────────────────────────────────────────────────
// Every test here seals a real archive through a `CompressCtx`, so all of them
// need the codec — the same gate `meta_sink_append::tests` carries.
#[cfg(all(test, feature = "openzl"))]
mod tests {
use super::*;
use crate::{ZnippyArchive, create_archive};
use std::time::{SystemTime, UNIX_EPOCH};
fn unique_dir(tag: &str) -> PathBuf {
let ns = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
let d = std::env::temp_dir()
.join(format!("znippy_hot_{tag}_{ns}_{:?}", std::thread::current().id()));
std::fs::create_dir_all(&d).unwrap();
d
}
/// Deterministic, lexicographically spread, and *compressible* — so the codec
/// really runs and a byte-identity claim is about compressed output, not about
/// two stored-raw copies of the same input.
fn synth(n: usize, salt: u64) -> Vec<(String, Vec<u8>)> {
(0..n)
.map(|i| {
let g = (i.wrapping_mul(2_654_435_761) ^ salt as usize) % 997;
let p = format!("repo/grp{g:03}/file{i:08}_{salt}.bin");
let body = format!("payload {i} salt {salt} {}\n", "z".repeat(8 + (i % 40)));
(p, body.into_bytes())
})
.collect()
}
/// **The correctness claim the whole module rests on.** Appending N files one
/// at a time through the hot journal and then sealing must produce the SAME
/// BYTES as creating the archive from all N at once — same blob offsets, same
/// data sub-index, same sorted lookup, same fst trie, same manifest, same
/// footer.
///
/// This is an applied-output assertion, not a round trip: a seal that dropped
/// the trie, or emitted rows in journal-segment order instead of insertion
/// order, would still list N files and still read every one of them back
/// correctly. Only the bytes catch it.
///
/// Seen RED by emitting the rows sorted instead of in insertion order in
/// `HotArchive::seal` — the data sub-index differs and the assertion fires
/// with a first-difference offset inside it.
#[test]
fn seal_matches_create_archive_byte_for_byte() {
let dir = unique_dir("identity");
let files = synth(400, 11);
let reference = dir.join("ref.znippy");
create_archive(&reference, &files, 3).unwrap();
let hot = dir.join("hot.znippy");
{
let h = HotArchive::open(&hot, 3, crate::SkipPolicy::resolve()).unwrap();
for f in &files {
h.append(std::slice::from_ref(f)).unwrap();
}
let report = h.seal(None).unwrap();
assert_eq!(report.rows_sealed, files.len() as u64);
assert_eq!(report.segments_removed, 1, "one process, one segment");
}
let a = std::fs::read(&reference).unwrap();
let b = std::fs::read(&hot).unwrap();
assert_eq!(
a.len(),
b.len(),
"sealed length differs: create_archive {} vs hot-then-seal {}",
a.len(),
b.len()
);
let first_diff = a.iter().zip(&b).position(|(x, y)| x != y);
assert!(
first_diff.is_none(),
"hot-then-seal is NOT byte-identical to create_archive; first difference at byte {:?}",
first_diff
);
// …and the journal is gone, so the archive is genuinely cold.
assert!(!HotArchive::is_hot(&hot), "seal must remove the journal");
let _ = std::fs::remove_dir_all(&dir);
}
/// The hot archive is READABLE before it is sealed — otherwise a two-phase
/// tier would be write-only until compaction, which is not "worth having".
#[test]
fn a_hot_archive_serves_reads_before_the_seal() {
let dir = unique_dir("read");
let path = dir.join("a.znippy");
let files = synth(120, 3);
let h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
for f in &files {
h.append(std::slice::from_ref(f)).unwrap();
}
for (p, bytes) in &files {
assert!(h.contains(p), "hot archive must contain {p}");
assert_eq!(h.file_size(p), Some(bytes.len() as u64));
assert_eq!(&h.extract_file(p).unwrap(), bytes, "hot read mismatch for {p}");
}
assert_eq!(h.list_files().unwrap().len(), files.len());
// And after the seal the ORDINARY static reader answers the same bytes.
h.seal(None).unwrap();
let ar = ZnippyArchive::open(&path).unwrap();
for (p, bytes) in &files {
assert_eq!(&ar.extract_file(p).unwrap(), bytes, "sealed read mismatch for {p}");
}
// Random access through the rebuilt trie+lookup.
assert!(!crate::locate_file(&path, &files[77].0).unwrap().is_empty());
let _ = std::fs::remove_dir_all(&dir);
}
/// A restart must not lose the journal, and must not corrupt the blob cursor.
/// Three processes' worth of appends (three segments), then one seal.
#[test]
fn reopening_recovers_every_segment() {
let dir = unique_dir("reopen");
let path = dir.join("a.znippy");
let mut all: Vec<(String, Vec<u8>)> = Vec::new();
for round in 0..3u64 {
let files = synth(40, round + 1);
let h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
assert_eq!(h.rows(), all.len() as u64, "reopen must recover prior rows");
for f in &files {
h.append(std::slice::from_ref(f)).unwrap();
}
all.extend(files);
}
let h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
assert_eq!(h.rows(), all.len() as u64);
let report = h.seal(None).unwrap();
assert_eq!(report.segments_removed, 4, "three appending opens + the final one");
let ar = ZnippyArchive::open(&path).unwrap();
for (p, bytes) in &all {
assert_eq!(&ar.extract_file(p).unwrap(), bytes, "byte mismatch after restarts for {p}");
}
let _ = std::fs::remove_dir_all(&dir);
}
/// A SEALED archive goes hot again without losing anything: the static rows
/// are recovered into segment 0 and the rebuildable tail is dropped. This is
/// the path gunnar's cold tier takes after a `repack()` fold.
#[test]
fn a_sealed_archive_reheats_and_keeps_its_rows() {
let dir = unique_dir("reheat");
let path = dir.join("a.znippy");
let sealed = synth(60, 5);
create_archive(&path, &sealed, 3).unwrap();
let cold_len = std::fs::metadata(&path).unwrap().len();
let h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
assert_eq!(h.rows(), sealed.len() as u64, "re-heat must recover the sealed rows");
assert!(
h.blob_end() < cold_len,
"the metadata tail must be dropped: blob_end {} vs sealed length {cold_len}",
h.blob_end()
);
let extra = synth(20, 6);
for f in &extra {
h.append(std::slice::from_ref(f)).unwrap();
}
h.seal(None).unwrap();
let ar = ZnippyArchive::open(&path).unwrap();
for (p, bytes) in sealed.iter().chain(extra.iter()) {
assert_eq!(&ar.extract_file(p).unwrap(), bytes, "byte mismatch after re-heat for {p}");
}
let mut listed = ar.list_files().unwrap();
listed.sort();
let mut want: Vec<String> =
sealed.iter().chain(extra.iter()).map(|(p, _)| p.clone()).collect();
want.sort();
assert_eq!(listed, want);
let _ = std::fs::remove_dir_all(&dir);
}
/// Last writer wins, exactly as `append_files` does — one live row per path,
/// and the bytes read back are the NEW ones. The old blob stays in the file as
/// dead payload, which is the documented cost, so the assertion is on the row
/// count and the bytes, not on the file size.
#[test]
fn re_appending_a_path_replaces_it() {
let dir = unique_dir("replace");
let path = dir.join("a.znippy");
let h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
h.append(&[("x.bin".into(), b"first version".to_vec())]).unwrap();
h.append(&[("y.bin".into(), b"other".to_vec())]).unwrap();
let r = h.append(&[("x.bin".into(), b"SECOND version, longer".to_vec())]).unwrap();
assert_eq!(r.rows_replaced, 1, "re-appending x.bin must replace, not duplicate");
assert_eq!(h.rows(), 2);
assert_eq!(h.extract_file("x.bin").unwrap(), b"SECOND version, longer");
h.seal(None).unwrap();
let ar = ZnippyArchive::open(&path).unwrap();
assert_eq!(ar.list_files().unwrap().len(), 2, "the seal must carry ONE row per path");
assert_eq!(ar.extract_file("x.bin").unwrap(), b"SECOND version, longer");
assert_eq!(ar.extract_file("y.bin").unwrap(), b"other");
let _ = std::fs::remove_dir_all(&dir);
}
/// A killed process leaves a partial final message. Everything before it is
/// real and must survive; the partial one must be dropped rather than failing
/// the open. Simulated by truncating the segment inside its last message.
#[test]
fn a_torn_journal_tail_is_dropped_not_fatal() {
let dir = unique_dir("torn");
let path = dir.join("a.znippy");
let files = synth(30, 2);
{
let h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
for f in &files {
h.append(std::slice::from_ref(f)).unwrap();
}
}
let seg = segment_path(&path, 0);
let len = std::fs::metadata(&seg).unwrap().len();
// Cut 40 bytes: inside the last batch message, past many complete ones.
std::fs::OpenOptions::new().write(true).open(&seg).unwrap().set_len(len - 40).unwrap();
let h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
let rows = h.rows();
assert!(
rows > 0 && rows < files.len() as u64,
"a torn tail must cost SOME rows and not all of them; got {rows} of {}",
files.len()
);
// Every surviving row still reads back byte-exact.
for p in h.list_files().unwrap() {
let want = &files.iter().find(|(q, _)| *q == p).unwrap().1;
assert_eq!(&h.extract_file(&p).unwrap(), want, "torn-journal read mismatch for {p}");
}
let _ = std::fs::remove_dir_all(&dir);
}
/// **The O(1) claim, as an applied output.** The index-side cost of an append
/// is the journal bytes it writes. If that is flat across a 100× growth in
/// archive size, the append is not paying for the archive; if the old
/// re-sealing path were behind this API the number would grow with N.
///
/// Bytes, not CPU: instructions and wall clock are unusable as absolutes on a
/// contended box, and a byte count is exact.
#[test]
fn the_index_cost_of_an_append_does_not_grow_with_the_archive() {
let dir = unique_dir("flat");
let path = dir.join("a.znippy");
let h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
let mut early = 0u64;
let mut late = 0u64;
for i in 0..2_000usize {
// Constant-shape rows so the comparison is about N, not about payload.
let f = vec![(format!("pack-{i:040x}.pack"), vec![b'q'; 64])];
let r = h.append(&f).unwrap();
if i == 10 {
early = r.journal_bytes_added;
}
if i == 1_999 {
late = r.journal_bytes_added;
}
}
assert!(early > 0, "an append must cost SOME journal bytes; the probe read 0");
assert_eq!(
early, late,
"the index cost of an append grew from {early} B at row 10 to {late} B at row 2000 — \
the append is paying for the archive again"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// `HotArchive` is shared across threads by every real caller (holger holds
/// exactly one per writable repo and serves every request off it), so the
/// auto traits are part of the contract, not an accident of the fields.
#[test]
fn hot_archive_is_send_and_sync() {
fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<HotArchive>();
}
/// **The property this module's locking exists for: a reader on a region the
/// append is not touching does not wait for the append.**
///
/// Before this shape `append` took `&mut self`, so a shared archive lived
/// behind an `RwLock` and an append took the EXCLUSIVE guard for its whole
/// body — compression, the blob write, `sync_all`, the journal batch, the
/// journal `sync_all`. Every reader of that archive stopped for all of it.
/// MEASURED downstream (holger `.nornir/stinky/12-dataplane-finished.md`
/// §5.4): 32 readers against 4 publishers kept **2.2–3.0 %** of their
/// throughput.
///
/// The test drives the two halves against each other and counts. A reader
/// thread hammers a row that was in the archive before the append started —
/// so the append cannot be writing its bytes, by construction, since an
/// append only ever writes PAST the blob cursor. The main thread waits until
/// the reader is warm, samples its counter, performs ONE large append, and
/// samples again.
///
/// The assertion is on the reads completed strictly *inside* that one
/// append. Under the old shape that number is zero by construction. The
/// floor here is deliberately far below what a working build produces
/// (tens of thousands on oden) so the test fails on the MECHANISM coming
/// back, not on the box being busy.
#[test]
fn a_reader_on_an_untouched_region_proceeds_during_an_append() {
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as O};
let dir = unique_dir("reader_during_append");
let path = dir.join("a.znippy");
// A row that exists BEFORE the append, at a blob offset the append
// cannot reach.
let cold: Vec<(String, Vec<u8>)> = synth(64, 7);
let probe = cold[0].0.clone();
let probe_body = cold[0].1.clone();
let h = Arc::new(HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap());
h.append(&cold).unwrap();
let stop = Arc::new(AtomicBool::new(false));
let reads = Arc::new(AtomicU64::new(0));
let bad = Arc::new(AtomicU64::new(0));
let reader = {
let (h, stop, reads, bad, probe, probe_body) = (
Arc::clone(&h),
Arc::clone(&stop),
Arc::clone(&reads),
Arc::clone(&bad),
probe.clone(),
probe_body.clone(),
);
std::thread::spawn(move || {
while !stop.load(O::Relaxed) {
match h.extract_file(&probe) {
Ok(b) if b == probe_body => {
reads.fetch_add(1, O::Relaxed);
}
_ => {
bad.fetch_add(1, O::Relaxed);
}
}
}
})
};
// Warm: do not start measuring until the reader is actually running, so
// a slow thread spawn cannot be mistaken for a blocked reader.
let warm_deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
while reads.load(O::Relaxed) < 1_000 {
assert!(
std::time::Instant::now() < warm_deadline,
"reader never warmed up: {} reads, {} failures",
reads.load(O::Relaxed),
bad.load(O::Relaxed)
);
std::thread::yield_now();
}
// ONE append, big enough that the window is unambiguous. The body is
// LCG-generated so the codec has real work to do and the blob write is
// a real 16 MiB — the point being that none of it is a reason for the
// reader to stop.
let mut body = vec![0u8; 16 * 1024 * 1024];
let mut s: u64 = 0x2545_F491_4F6C_DD1D;
for b in body.iter_mut() {
s = s.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1_442_695_040_888_963_407);
*b = (s >> 33) as u8;
}
let before = reads.load(O::Relaxed);
let t0 = std::time::Instant::now();
let report = h.append(&[("hot/big.bin".to_string(), body)]).unwrap();
let append_took = t0.elapsed();
let during = reads.load(O::Relaxed) - before;
stop.store(true, O::Relaxed);
reader.join().unwrap();
assert_eq!(bad.load(O::Relaxed), 0, "a concurrent reader saw a wrong or missing body");
assert_eq!(report.rows_added, 1);
assert!(
report.blob_append_offset >= 1,
"the append must start past the cold rows, not on top of them"
);
eprintln!(
"reader completed {during} reads on an untouched row during one {append_took:?} append"
);
assert!(
during >= 100,
"a reader on an untouched row completed only {during} reads during a \
{append_took:?} append — the append is excluding readers again"
);
// And the archive is correct afterwards: the cold row and the new one.
assert_eq!(h.extract_file(&probe).unwrap(), probe_body);
assert_eq!(h.extract_file("hot/big.bin").unwrap().len(), 16 * 1024 * 1024);
assert_eq!(h.rows(), 65);
let _ = std::fs::remove_dir_all(&dir);
}
/// The COUNTED half of the same claim: an append costs exactly two `fsync`s
/// — the blob region, then the journal message that names it — whatever it
/// is appending, and an empty append costs none.
///
/// Per-archive, not process-wide, so the count is EXACT and this test cannot
/// be made to pass or fail by whatever else the test binary is doing on
/// another thread. It is what makes "fsyncs per append" a number rather than
/// an argument downstream.
#[test]
fn an_append_pays_exactly_two_fsyncs_and_an_empty_one_pays_none() {
let dir = unique_dir("fsync_count");
let path = dir.join("a.znippy");
let h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
assert_eq!(h.fsyncs(), 0, "a freshly opened archive has appended nothing");
for f in synth(8, 11) {
h.append(std::slice::from_ref(&f)).unwrap();
}
assert_eq!(h.fsyncs(), 16, "8 one-at-a-time appends are 8 x (blob, journal)");
h.append(&[]).unwrap();
assert_eq!(h.fsyncs(), 16, "an empty append must not touch the disk");
// Batched: 8 files in ONE call is still two fsyncs. That is the whole
// reason holger group-commits N publishes into one append — N entries
// cost two fsyncs, not 2N.
h.append(&synth(8, 12)).unwrap();
assert_eq!(h.fsyncs(), 18, "8 files in one append are two fsyncs, not sixteen");
assert_eq!(h.rows(), 16);
let _ = std::fs::remove_dir_all(&dir);
}
}