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;
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());
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
}
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(" ")
}
#[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");
}
#[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"
);
}
}
#[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();
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);
}
#[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 {
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"
);
}
}
#[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);
assert_eq!(
read_back(&archive, 0, p.len() as u64),
p,
"the fsynced blob bytes survived the crash"
);
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:?}"
);
}
#[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);
}
#[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"
);
}
}
#[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);
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());
}
#[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),
);
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:?}"),
}
}
#[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();
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);
}
#[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);
}