plugins/native/znippy-plugin-git/tests/archive_write.rs
//! Guards for the push path: [`ArchiveWrite`]'s three arms and the shared
//! per-account indexer.
//!
//! **Every assertion here is on applied output** — bytes read back off the disk
//! through a *fresh* file handle, journal rows decoded out of the Arrow IPC
//! stream, index rows decoded out of the built `RecordBatch`. Not one of them
//! asserts that a call returned `Ok`, because `Ok` is exactly what a writer that
//! silently dropped the payload also returns.
//!
//! **LAW 2 — every guard below was seen RED.** Each test's doc comment records
//! the exact edit that broke it and what the runner then printed. Runs on
//! 2026-08-07, oden, `--no-default-features`, `/proc/loadavg` 0.26–2.1 across
//! the session.
use std::fs::File;
use std::io::Read;
use std::os::unix::fs::FileExt;
use std::path::Path;
use std::sync::Arc;
use znippy_common::arrow::array::{StringArray, UInt32Array, UInt64Array};
use znippy_plugin_git::archive_write::{
ArchiveWrite, FastWriter, Faults, SafeWriter, read_journal,
};
use znippy_plugin_git::indexer::{AccountIndexer, IndexJob, Lookup, PushPath};
use znippy_plugin_git::uring_write::UringWriter;
// ── fixtures ────────────────────────────────────────────────────────────────
/// Bytes shaped like a v2 packfile: the 12-byte header a real pack opens with
/// (`PACK`, version, object count) followed by an incompressible body.
///
/// Incompressible on purpose. These bytes are stored **verbatim**, and a body of
/// zeroes would let a writer that quietly deflated the payload still pass a
/// length check.
fn pack_shaped(objects: u32, body: usize) -> Vec<u8> {
let mut v = Vec::with_capacity(12 + body);
v.extend_from_slice(b"PACK");
v.extend_from_slice(&2u32.to_be_bytes());
v.extend_from_slice(&objects.to_be_bytes());
// xorshift64*, so the body is deterministic across runs and has no
// compressible structure.
let mut s: u64 = 0x9E37_79B9_7F4A_7C15 ^ (objects as u64) << 32 ^ body as u64;
while v.len() < 12 + body {
s ^= s >> 12;
s ^= s << 25;
s ^= s >> 27;
v.extend_from_slice(&s.wrapping_mul(0x2545_F491_4F6C_DD1D).to_le_bytes());
}
v.truncate(12 + body);
v
}
/// Read `len` bytes at `offset` through a **fresh** handle. Never the writer's
/// own `File` — that would let a writer that never issued the syscall still pass
/// by handing back its own buffer.
fn read_back(path: &Path, offset: u64, len: u64) -> Vec<u8> {
let f = File::open(path).expect("reopen archive");
let mut buf = vec![0u8; len as usize];
f.read_exact_at(&mut buf, offset).expect("pread extent");
buf
}
fn sha1_hex(bytes: &[u8]) -> String {
use sha1::{Digest, Sha1};
let mut h = Sha1::new();
h.update(bytes);
hex::encode(h.finalize())
}
fn loadavg() -> String {
let mut s = String::new();
File::open("/proc/loadavg")
.and_then(|mut f| f.read_to_string(&mut s))
.map(|_| ())
.unwrap_or_default();
s.split_whitespace().take(3).collect::<Vec<_>>().join(" ")
}
// ── impl 1 ──────────────────────────────────────────────────────────────────
/// `FastWriter` puts the caller's bytes, unaltered, at the extent it reported.
///
/// Asserts the bytes read back through a second handle, not the return value:
/// the extent has to *address* the payload, and the payload has to be
/// byte-identical (verbatim — no deflate, no framing, no length prefix).
///
/// Seen RED by `self.blobs.write_all_at(bytes, offset)?;` →
/// `self.blobs.write_all_at(&bytes[..bytes.len() / 2], offset)?;` in
/// `FastWriter::append`: observed
/// `assertion `left == right` failed: pack A verbatim on disk` at
/// `tests/archive_write.rs:107` — the extent still said 4108 B, `append` still
/// returned `Ok`, and the trailing 2054 B read back as a hole. Restored.
#[test]
fn fast_writer_stores_the_pack_verbatim_at_the_returned_extent() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
let w = FastWriter::create(&archive).unwrap();
let a = pack_shaped(3, 4096);
let b = pack_shaped(17, 200);
let (ao, al) = w.append(&a).unwrap();
let (bo, bl) = w.append(&b).unwrap();
assert_eq!(al, a.len() as u64, "extent length is the input length");
assert_eq!(bl, b.len() as u64);
assert_eq!(bo, ao + al, "second append starts where the first ended");
assert_eq!(read_back(&archive, ao, al), a, "pack A verbatim on disk");
assert_eq!(read_back(&archive, bo, bl), b, "pack B verbatim on disk");
}
// ── impl 2 ──────────────────────────────────────────────────────────────────
/// `SafeWriter`'s journal row is **on disk** and **addresses the stored bytes**.
///
/// Both halves matter. The row is decoded out of the on-disk Arrow IPC stream
/// through a fresh handle, and the extent it carries is then used to `pread` the
/// payload back and compare it to what was pushed. A journal that recorded the
/// wrong offset would pass a row-count assertion and fail this one.
///
/// Seen RED by `journal_batch(offset, len)` → `journal_batch(offset + 1, len)`
/// in `SafeWriter::append`: observed
/// `assertion `left == right` failed: journal rows are the extents append
/// returned` at `tests/archive_write.rs:139` — every row was present and the
/// row *count* was still right; only the offsets were off by one. Restored.
#[test]
fn safe_writer_journal_row_is_on_disk_and_addresses_the_stored_bytes() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
let w = SafeWriter::create(&archive).unwrap();
let jpath = w.journal_file();
let packs: Vec<Vec<u8>> = vec![pack_shaped(1, 188), pack_shaped(9, 65536)];
let mut extents = Vec::new();
for p in &packs {
extents.push(w.append(p).unwrap());
}
let rows = read_journal(&jpath).expect("decode journal");
assert_eq!(rows.len(), packs.len(), "one journal row per append");
assert_eq!(rows, extents, "journal rows are the extents append returned");
for (p, (off, len)) in packs.iter().zip(&rows) {
assert_eq!(
&read_back(&archive, *off, *len),
p,
"the journal's extent addresses the payload"
);
}
}
/// The **BufWriter trap**: `flush()` before `sync_all()` is load-bearing.
///
/// `SafeWriter` writes its journal through a `BufWriter`, so the row lives in a
/// userspace `Vec` until `flush()`. `sync_all()` on its own
/// succeeds, fsyncs a real fd, and syncs **nothing that was just written**. This
/// guard reads the journal back through a separate handle — which can only see
/// bytes the kernel has — and demands the row be there.
///
/// Seen RED by deleting `j.writer.flush()...?;` (step 4) from
/// `SafeWriter::append`, leaving step 5's `sync_all()` in place: observed
/// `assertion `left == right` failed: the row must be past userspace before the
/// fsync / left: 0 / right: 1` at `tests/archive_write.rs:174`. `append`
/// returned `Ok`, the `fsync` succeeded, and the row was not on disk. Restored.
#[test]
fn journal_flush_before_sync_is_load_bearing() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
let w = SafeWriter::create(&archive).unwrap();
let jpath = w.journal_file();
let p = pack_shaped(2, 512);
let extent = w.append(&p).unwrap();
// A fresh handle sees only what reached the kernel.
let rows = read_journal(&jpath).expect("decode journal");
assert_eq!(
rows.len(),
1,
"the row must be past userspace before the fsync"
);
assert_eq!(rows[0], extent);
}
/// **A reopen appends to the journal; it does not truncate it.**
///
/// The journal is the durable half of §13.12's `indexed` bit — a pack is
/// unabsorbed *iff* its extent is here and its rows are not in the index — so a
/// writer opened over an existing archive has to be able to read what the last
/// process acked. Three writers in succession, each closed before the next
/// opens, and every extent from all three must still decode out of the one file.
///
/// Asserted on applied output: the rows read back through a fresh handle *are*
/// the extents the three `append` calls returned, in order, and each one still
/// addresses its own payload in the blob file.
///
/// Seen RED by restoring `File::create(&path)` in `SafeWriter::create_inner`
/// (which is what it did before this guard existed): observed
/// `assertion `left == right` failed: a reopen dropped the earlier rows / left:
/// [(65560, 524)] / right: [(0, 4108), (4108, 61452), (65560, 524)]` — the bytes
/// of the first two packs were still on disk and nothing on disk pointed at them
/// any more. Restored.
///
/// Seen RED a second time by emitting the schema message on every open rather
/// than only for an empty file (`if already == 0` → unconditional): observed
/// `assertion `left == right` failed: a reopen dropped the earlier rows / left:
/// [(0, 4108)] / right: [(0, 4108), (4108, 61452), (65560, 524)]` — arrow's
/// `StreamReader` stops at the second schema message, so every row written after
/// the first reopen was silently invisible while the file kept growing.
#[test]
fn a_reopened_safe_writer_appends_to_the_journal_it_finds() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
let packs: Vec<Vec<u8>> = vec![
pack_shaped(1, 4096),
pack_shaped(2, 61440),
pack_shaped(3, 512),
];
let mut extents = Vec::new();
let mut jpath = std::path::PathBuf::new();
for p in &packs {
// A new writer per pack: this is a process restart in every way that
// matters to the file, which is the only thing the journal knows about.
let w = SafeWriter::create(&archive).unwrap();
jpath = w.journal_file();
extents.push(w.append(p).unwrap());
drop(w);
}
let rows = read_journal(&jpath).expect("decode journal");
assert_eq!(rows, extents, "a reopen dropped the earlier rows");
for (p, (off, len)) in packs.iter().zip(&rows) {
assert_eq!(
&read_back(&archive, *off, *len),
p,
"the journal's extent no longer addresses the payload it was written for"
);
}
}
/// **A store that has acked nothing is rebuilt, not fsynced.**
///
/// `SafeWriter::create` no longer barriers the schema message it writes into a
/// fresh journal — on gunnar's create path that was 8.9 ms of a 68.5 ms
/// push-create spent making durable a file that references nothing and that a
/// transactional create discards on failure anyway. The property the barrier
/// used to hold is held here instead, by checking on the way in: a journal whose
/// header is not this build's schema message, in a file no longer than one, was
/// interrupted before any row could exist behind it, and is written again.
///
/// Every torn length is tried, because "it works for the length I happened to
/// pick" is not the claim. The assertion is on applied output: a pack acked
/// through the repaired writer is addressable through a journal decoded from a
/// fresh handle.
///
/// Seen RED by dropping the repair (leaving only `if already == 0`): observed
/// `decode journal: Err(... Unable to get root as message ...)` at the first
/// torn length — the exact state the removed fsync used to prevent, left
/// unhandled.
#[test]
fn an_interrupted_schema_message_is_rewritten_rather_than_inherited() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
// The schema message this build writes, at its real length.
let jpath = SafeWriter::journal_path(&archive);
drop(SafeWriter::create(&archive).unwrap());
let schema_len = std::fs::metadata(&jpath).unwrap().len();
assert!(schema_len > 0, "a fresh journal must hold the schema message");
for cut in 0..schema_len {
let f = File::options().write(true).open(&jpath).unwrap();
f.set_len(cut).unwrap();
drop(f);
// Both durable arms, alternating, because LAW 5 claims they are
// indistinguishable on disk and one repair that only the safe arm knew
// about is exactly how that would stop being true.
let pack = pack_shaped(1, 4096);
let (off, len) = if cut % 2 == 0 {
let w = SafeWriter::create(&archive).unwrap();
let e = w.append(&pack).unwrap();
drop(w);
e
} else {
match UringWriter::create(&archive) {
Ok(w) => {
let e = w.append(&pack).unwrap();
drop(w);
e
}
// No ring on this kernel: the safe arm still has to hold.
Err(_) => {
let w = SafeWriter::create(&archive).unwrap();
let e = w.append(&pack).unwrap();
drop(w);
e
}
}
};
let rows = read_journal(&jpath).expect("decode journal");
assert_eq!(
rows.last().copied(),
Some((off, len)),
"cut at {cut}: the ack after the repair is not in the journal"
);
assert_eq!(
&read_back(&archive, off, len),
&pack,
"cut at {cut}: the journal's extent does not address the payload"
);
std::fs::remove_file(&jpath).ok();
std::fs::remove_file(&archive).ok();
drop(SafeWriter::create(&archive).unwrap());
}
}
/// **The crash-ordering contract.** An interruption between the two fsyncs
/// leaves **orphan bytes**, never a dangling reference.
///
/// Injected at exactly the point a machine would die: after the blob `fsync`,
/// before the journal row. The on-disk state a real power loss leaves there is
/// byte-for-byte what this produces — `append` returns `Err` only because a Rust
/// function must return something.
///
/// Two assertions, and the second is the one that matters:
/// * the pack bytes ARE on disk at the extent that was reserved — orphan payload
/// the seal will drop;
/// * the journal names **no** extent — nothing points into a hole.
///
/// Seen RED by moving the journal block (steps 3–5) *above* the
/// `die_between_fsyncs` bail in `SafeWriter::append` — i.e. writing the row
/// first, the ordering this contract forbids: observed
/// `the journal must not reference bytes whose fsync never happened / journal
/// rows: [(0, 4108)]` at `tests/archive_write.rs:228`. A durable, live
/// reference to an extent whose bytes were never fsynced: the dangling
/// reference this ordering exists to make impossible. Restored.
#[test]
fn crash_between_fsyncs_leaves_orphan_bytes_not_a_dangling_reference() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
let w = SafeWriter::create_with_faults(&archive, Faults {
die_between_fsyncs: true,
..Default::default()
})
.unwrap();
let jpath = w.journal_file();
let p = pack_shaped(5, 4096);
let err = w.append(&p).unwrap_err();
assert!(
err.to_string().contains("injected crash"),
"the fault fired: {err}"
);
drop(w);
// Orphan bytes: the payload is on the platter at the reserved extent…
assert_eq!(
read_back(&archive, 0, p.len() as u64),
p,
"the fsynced blob bytes survived the crash"
);
// …and nothing references them.
let rows = read_journal(&jpath).expect("decode journal");
assert!(
rows.is_empty(),
"the journal must not reference bytes whose fsync never happened / journal rows: {rows:?}"
);
}
// ── impl 3 ──────────────────────────────────────────────────────────────────
/// The io_uring chain lands the pack **and** its journal row.
///
/// One `io_uring_enter` carrying `Write → Fsync → WriteFixed → Fsync`, linked
/// with `IOSQE_IO_LINK`. If this kernel cannot run it the test **fails loudly**
/// rather than skipping — a silent skip is a guard that can never be red.
///
/// The journal is asserted through the same `read_journal` the `SafeWriter`
/// guard uses, which is the point: the io_uring arm writes the *same* Arrow IPC
/// journal format, built by hand into a registered buffer instead of through
/// arrow's `BufWriter`.
///
/// Seen RED by `.offset(journal_at)` → `.offset(journal_at + 8)` on the
/// `WriteFixed` SQE in `UringWriter::append` — same op, same byte count, wrong
/// place: observed `assertion `left == right` failed: one journal row per
/// append / left: 0 / right: 2` at `tests/archive_write.rs:278`. Both appends
/// returned `Ok`, the writer's own short-write check passed (224 of 224 B), the
/// pack bytes were on disk, and the journal decoded to nothing. Restored.
///
/// (A cruder break — deleting the `WriteFixed` SQE entirely — was tried first
/// and is *not* what is recorded here: `UringWriter`'s own short-journal-write
/// check caught it before the test could, so it proved the writer, not the
/// guard.)
#[test]
fn uring_writer_chain_lands_bytes_and_journal_row() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
let w = match UringWriter::create(&archive) {
Ok(w) => w,
Err(e) => panic!(
"io_uring arm unavailable on this kernel — not skipping, this is a real failure: {e}"
),
};
let jpath = w.journal_file().to_path_buf();
let packs = vec![pack_shaped(4, 200), pack_shaped(64, 1 << 20)];
let mut extents = Vec::new();
for p in &packs {
extents.push(w.append(p).unwrap());
}
for (p, (off, len)) in packs.iter().zip(&extents) {
assert_eq!(
&read_back(&archive, *off, *len),
p,
"io_uring wrote the pack verbatim"
);
}
let rows = read_journal(&jpath).expect("decode journal");
assert_eq!(rows.len(), packs.len(), "one journal row per append");
assert_eq!(rows, extents);
}
// ── the three arms agree on what they store ─────────────────────────────────
/// All three arms store **byte-identical** payload. They differ in durability,
/// never in content.
///
/// Seen RED by making `FastWriter::append` write `&bytes[1..]`: observed
/// `pread extent: Error { kind: UnexpectedEof, message: "failed to fill whole
/// buffer" }` at `tests/archive_write.rs:60` — the extent claimed 4108 B and
/// the file held 4107. Restored.
#[test]
fn all_three_arms_store_byte_identical_payload() {
let dir = tempfile::tempdir().unwrap();
let p = pack_shaped(11, 4096);
let mut stored: Vec<(&'static str, Vec<u8>)> = Vec::new();
for (name, make) in [
("FastWriter", 0u8),
("SafeWriter", 1),
("UringWriter", 2),
] {
let archive = dir.path().join(format!("{name}.znippy"));
let w: Box<dyn ArchiveWrite> = match make {
0 => Box::new(FastWriter::create(&archive).unwrap()),
1 => Box::new(SafeWriter::create(&archive).unwrap()),
_ => Box::new(UringWriter::create(&archive).unwrap()),
};
assert_eq!(w.name(), name);
let (off, len) = w.append(&p).unwrap();
assert_eq!(len, p.len() as u64, "{name} stored a different length");
stored.push((name, read_back(&archive, off, len)));
}
for (name, bytes) in &stored {
assert_eq!(
bytes, &p,
"{name} stored something other than the pushed pack"
);
}
}
// ── the shared indexer ──────────────────────────────────────────────────────
/// The index tables are built **after** the bytes are down, and they carry rows
/// derived from the bytes — not echoes of the job that was submitted.
///
/// `pack_id`, `blob_offset` and `blob_size` came in on the channel; `object_count`,
/// `pack_version` and `pack_sha1` did **not** — they can only be produced by
/// reading the extent back out of the archive, which is the whole point of the
/// deferred index. So the assertion is on those three.
///
/// Seen RED by `h.update(&buf);` → `h.update(&buf[..1.min(buf.len())]);` in
/// `indexer::build_row`: observed `assertion `left == right` failed: pack_sha1
/// is a hash of the stored bytes / left: "511993d3c99719e38a6779073019dacd7178ddb9"
/// right: "8dc6cbb5a1129b90d8f44d307c492746aa1fcdb8"` at
/// `tests/archive_write.rs:375`. Five rows still appeared, with the right ids,
/// offsets, lengths and object counts. Restored.
#[test]
fn index_tables_are_built_after_the_bytes_and_carry_derived_rows() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
let w = SafeWriter::create(&archive).unwrap();
let jpath = w.journal_file();
let path = PushPath::new(Box::new(w), &archive, Some(jpath)).unwrap();
let packs: Vec<Vec<u8>> = (0..5).map(|i| pack_shaped(100 + i, 1024 * (i as usize + 1))).collect();
let mut ids = Vec::new();
for p in &packs {
ids.push(path.push_pack("acct-alpha", p).unwrap());
}
let idx = path.indexer("acct-alpha");
idx.wait_caught_up();
assert_eq!(idx.rows(), packs.len(), "one index row per pack");
let tables = idx.tables();
assert!(!tables.is_empty(), "index tables exist");
let mut seen = 0usize;
for t in &tables {
let id = t.column(0).as_any().downcast_ref::<UInt64Array>().unwrap();
let off = t.column(1).as_any().downcast_ref::<UInt64Array>().unwrap();
let len = t.column(2).as_any().downcast_ref::<UInt64Array>().unwrap();
let ver = t.column(3).as_any().downcast_ref::<UInt32Array>().unwrap();
let cnt = t.column(4).as_any().downcast_ref::<UInt32Array>().unwrap();
let sha = t.column(5).as_any().downcast_ref::<StringArray>().unwrap();
for r in 0..t.num_rows() {
let i = id.value(r) as usize;
let (pid, (o, l)) = ids[i];
assert_eq!(id.value(r), pid);
assert_eq!(off.value(r), o);
assert_eq!(len.value(r), l);
// Derived — could only come from reading the stored bytes back.
assert_eq!(ver.value(r), 2, "pack_version parsed from the header");
assert_eq!(
cnt.value(r),
100 + i as u32,
"object_count parsed from the header"
);
assert_eq!(
sha.value(r),
sha1_hex(&packs[i]),
"pack_sha1 is a hash of the stored bytes"
);
seen += 1;
}
}
assert_eq!(seen, packs.len());
assert_eq!(loadavg().split(' ').count(), 3, "loadavg recorded: {}", loadavg());
}
/// A read that arrives **before** the index is ready is slower, never wrong.
///
/// Deterministic by construction: the extent is appended but no [`IndexJob`] is
/// submitted, so the indexed bit is provably unset. In that state
/// [`AccountIndexer::lookup`] must hand back the journal — the durable record
/// written on the ack path — and the bytes at the journal's extent must be the
/// pushed pack. Only after the job is submitted and drained may it answer
/// [`Lookup::Indexed`].
///
/// Seen RED by replacing the journal fallback in `AccountIndexer::lookup` with
/// a bare `Lookup::Unknowable`: observed
/// `pre-index read must fall back to scanning, got Unknowable` at
/// `tests/archive_write.rs:431`. Restored.
///
/// Recorded because it is the honest version of the story: the *first* break
/// tried here was deleting the `indexed.contains(&pack_id)` check that used to
/// sit in front of the row map, and the test **stayed green** — the two
/// structures could never disagree, so the check was decoration. It was removed
/// rather than guarded: the indexed bit is now membership in the published-row
/// map itself (LAW 5, fix by construction). What is left to break is the
/// fallback, and that does go red.
#[test]
fn read_before_the_index_falls_back_to_scanning_and_is_not_wrong() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
let w = SafeWriter::create(&archive).unwrap();
let jpath = w.journal_file();
let p = pack_shaped(7, 3000);
let (off, len) = w.append(&p).unwrap();
let idx = AccountIndexer::start(
"acct-beta",
Arc::new(File::open(&archive).unwrap()),
Some(jpath),
);
// Nothing submitted: the indexed bit is unset, by construction.
assert!(!idx.is_indexed(0));
let before = idx.lookup(0);
match &before {
Lookup::ScanJournal(extents) => {
assert!(
extents.contains(&(off, len)),
"the scan fallback must expose the extent: {extents:?}"
);
assert_eq!(
read_back(&archive, off, len),
p,
"scanning the fallback extent yields the real pack"
);
}
other => panic!("pre-index read must fall back to scanning, got {other:?}"),
}
assert!(
matches!(before, Lookup::ScanJournal(_)),
"pre-index read must not claim an index hit"
);
idx.submit(IndexJob {
pack_id: 0,
offset: off,
len,
})
.unwrap();
idx.wait_caught_up();
match idx.lookup(0) {
Lookup::Indexed(row) => {
assert_eq!((row.offset, row.len), (off, len));
assert_eq!(row.object_count, 7);
assert_eq!(row.sha1, sha1_hex(&p));
}
other => panic!("after draining, the index must answer: {other:?}"),
}
}
/// Idle indexers **sleep**. They do not poll and they do not spin.
///
/// One indexer per account only works if an idle account is free. Measured as
/// process CPU out of `/proc/self/stat` across 400 ms of wall time with 64
/// account indexers parked — a spin loop over `try_recv` would burn 64 cores'
/// worth of it.
///
/// Seen RED by replacing `let Ok(first) = rx.recv() else { return };` in
/// `index_worker` with a `loop { match rx.try_recv() { Ok(j) => break j, Empty
/// => continue, Disconnected => return } }` spin: observed
/// `64 idle indexers burned 12920 ms of CPU in 400 ms of wall time (loadavg
/// 6.39 1.91 1.05) — they are spinning, not sleeping` at
/// `tests/archive_write.rs:498`, against a 200 ms budget. 32× over, and the
/// box's 1-minute load went from ~0.3 to 6.39 while doing no work at all.
/// Restored to the blocking `recv()`.
#[test]
fn idle_indexers_sleep_they_do_not_spin() {
fn cpu_ms() -> u64 {
let mut s = String::new();
File::open("/proc/self/stat")
.unwrap()
.read_to_string(&mut s)
.unwrap();
// utime and stime are fields 14 and 15, after the (possibly
// paren-wrapped) comm field.
let tail = &s[s.rfind(')').unwrap() + 1..];
let f: Vec<&str> = tail.split_whitespace().collect();
let ticks: u64 = f[11].parse::<u64>().unwrap() + f[12].parse::<u64>().unwrap();
let hz = unsafe { libc::sysconf(libc::_SC_CLK_TCK) } as u64;
ticks * 1000 / hz.max(1)
}
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
std::fs::write(&archive, b"").unwrap();
let shared = Arc::new(File::open(&archive).unwrap());
let idlers: Vec<AccountIndexer> = (0..64)
.map(|i| AccountIndexer::start(&format!("acct-{i}"), shared.clone(), None))
.collect();
let before = cpu_ms();
std::thread::sleep(std::time::Duration::from_millis(400));
let burned = cpu_ms() - before;
assert!(
burned < 200,
"64 idle indexers burned {burned} ms of CPU in 400 ms of wall time \
(loadavg {}) — they are spinning, not sleeping",
loadavg()
);
drop(idlers);
}
/// Accounts get **separate** indexers: one account's queue is not another's.
///
/// Asserts the isolation as applied state — each account's index holds exactly
/// its own packs and none of the other's, and the two `Arc`s are different
/// objects.
///
/// Seen RED by making `IndexerPool::indexer` key every account to the same
/// entry (`m.entry(account.to_string())` → `m.entry("shared".to_string())`):
/// observed `two accounts must not share one indexer` at
/// `tests/archive_write.rs:536` — the `Arc::ptr_eq` fired before the row counts
/// were even reached. Restored.
#[test]
fn accounts_do_not_share_an_indexer() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
let w = SafeWriter::create(&archive).unwrap();
let jpath = w.journal_file();
let path = PushPath::new(Box::new(w), &archive, Some(jpath)).unwrap();
let mut one = Vec::new();
let mut two = Vec::new();
for i in 0..2 {
one.push(path.push_pack("acct-one", &pack_shaped(1 + i, 300)).unwrap());
}
for i in 0..3 {
two.push(path.push_pack("acct-two", &pack_shaped(50 + i, 700)).unwrap());
}
let a = path.indexer("acct-one");
let b = path.indexer("acct-two");
assert!(
!Arc::ptr_eq(&a, &b),
"two accounts must not share one indexer"
);
a.wait_caught_up();
b.wait_caught_up();
assert_eq!(a.rows(), 2, "acct-one indexed only its own packs");
assert_eq!(b.rows(), 3, "acct-two indexed only its own packs");
for (id, _) in &one {
assert!(a.is_indexed(*id), "acct-one's bit set for {id}");
assert!(!b.is_indexed(*id), "acct-two must not claim {id}");
}
for (id, _) in &two {
assert!(b.is_indexed(*id));
assert!(!a.is_indexed(*id));
}
assert_eq!(path.pool().accounts(), 2);
}
// ── the append-side gc ──────────────────────────────────────────────────────
/// **Orphan bytes are given back at the next open, and the live pack is not.**
///
/// The crash-ordering contract above leaves bytes on the platter that no extent
/// addresses (`crash_between_fsyncs_leaves_orphan_bytes_not_a_dangling_reference`),
/// and the append cursor is taken from the blob file's LENGTH — so before this
/// they were stepped over by the next append and unreachable for ever.
/// `SafeWriter::create_inner`'s own comment names the shape: "a reopen that
/// dropped them would leave their bytes unreferenced for ever".
///
/// Nothing referenced them, so nothing is lost by taking them back: a push whose
/// journal row never landed was never acked and the client sends it again.
///
/// Asserted on **applied output** — the file's length on disk and the live
/// pack's bytes read back through a fresh handle — never on a return value.
///
/// RED-FIRST: with `reclaim_orphan_tail` returning `len` unconditionally the
/// file stays at both packs' length and the assertion reads
/// "orphan bytes were not reclaimed".
#[test]
fn orphan_bytes_past_the_last_journal_row_are_reclaimed_at_the_next_open() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("repo.znippy");
// A good pack, acked: bytes AND journal row.
let good = pack_shaped(5, 4096);
let (off, len) = {
let w = SafeWriter::create(&archive).unwrap();
let e = w.append(&good).unwrap();
e
};
let after_good = std::fs::metadata(&archive).unwrap().len();
assert_eq!(after_good, off + len, "the good pack is the whole file");
// A second pack that dies between the two fsyncs: bytes on disk, no row.
let orphan = pack_shaped(9, 8192);
{
let w = SafeWriter::create_with_faults(&archive, Faults {
die_between_fsyncs: true,
..Default::default()
})
.unwrap();
let err = w.append(&orphan).unwrap_err();
assert!(err.to_string().contains("injected crash"), "the fault fired: {err}");
}
let with_orphan = std::fs::metadata(&archive).unwrap().len();
assert!(
with_orphan > after_good,
"the orphan bytes are not on disk, so this proves nothing: {with_orphan} vs {after_good}"
);
// The reopen is the gc.
drop(SafeWriter::create(&archive).unwrap());
assert_eq!(
std::fs::metadata(&archive).unwrap().len(),
after_good,
"orphan bytes were not reclaimed"
);
// And the pack that WAS acked is untouched, byte for byte.
assert_eq!(
read_back(&archive, off, len),
good,
"the reclaim cut into a live extent"
);
let rows = read_journal(&SafeWriter::journal_path(&archive)).expect("decode journal");
assert_eq!(rows, vec![(off, len)], "the journal lost or gained a row: {rows:?}");
}
/// **An archive with no journal is left alone.** [`FastWriter`] keeps none, so
/// an absent journal is a normal arm and not evidence that anything is orphaned.
/// Truncating on it would delete every pack a `FastWriter` ever stored.
///
/// RED-FIRST: with the `!journal.exists()` guard removed, `read_journal` fails
/// on the missing path and `FastWriter::create` returns `Err` — the arm stops
/// opening at all.
#[test]
fn an_archive_with_no_journal_is_never_truncated() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("fast.znippy");
let p = pack_shaped(4, 2048);
let (off, len) = {
let w = FastWriter::create(&archive).unwrap();
w.append(&p).unwrap()
};
assert!(!SafeWriter::journal_path(&archive).exists(), "FastWriter wrote a journal");
drop(FastWriter::create(&archive).unwrap());
assert_eq!(
std::fs::metadata(&archive).unwrap().len(),
off + len,
"a journal-less archive was truncated"
);
assert_eq!(read_back(&archive, off, len), p);
}
/// **A journal that reads as empty is not evidence either.** `read_journal`
/// answers `Ok(vec![])` both for a genuinely empty log and for one whose SCHEMA
/// message is torn, and those are indistinguishable from here. Truncating on
/// that reading would turn one bad message into the loss of every pack.
///
/// RED-FIRST: with the `None` arm truncating to 0 instead of returning `len`,
/// the whole archive is destroyed and the assertion reads "an unreadable
/// journal cost the archive its packs".
#[test]
fn an_unreadable_journal_leaves_the_blobs_alone() {
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("torn.znippy");
let p = pack_shaped(6, 4096);
let (off, len) = {
let w = SafeWriter::create(&archive).unwrap();
w.append(&p).unwrap()
};
// Tear the journal down to nothing — the shape a crash in the schema
// message leaves, and the shape an empty log has.
std::fs::write(SafeWriter::journal_path(&archive), b"").unwrap();
drop(SafeWriter::create(&archive).unwrap());
assert_eq!(
std::fs::metadata(&archive).unwrap().len(),
off + len,
"an unreadable journal cost the archive its packs"
);
assert_eq!(read_back(&archive, off, len), p, "the payload did not survive");
}