znippy-iceberg/tests/lifecycle_seal.rs
//! The iceberg write → restart → resume → seal → static lifecycle test.
//!
//! This is the proof the seal proposal hinges on. It exercises the full
//! lifecycle of a *dynamic* iceberg-backed znippy archive and its
//! materialisation into a *static* native `.znippy`:
//!
//! 1. **WRITE** — compress real-ish fixtures with the iceberg metadata sink
//! (`IcebergSink`) → a durable skade `catalog.redb` warehouse + a pure-blob
//! `.znippy` sidecar. This is the dynamic store.
//! 2. **RESTART** — drop *all* handles (sink, catalog, reader): a process
//! boundary. Re-`skade::open` the warehouse fresh through a new
//! `IcebergZnippyReader`.
//! 3. **RESTART-SURVIVAL** — assert the already-written entries read back
//! byte-correct after the reopen (the durable catalog survived).
//! 4. **RESUME** — append more entries to the same iceberg-backed store as a new
//! iceberg commit (blobs appended to the same sidecar, a new sub-index
//! `table.append`ed via a fresh `IcebergSink` over the same warehouse +
//! namespace), then restart + read again.
//! 5. **SEAL** — `seal()` the post-resume iceberg archive → a static native
//! `.znippy`.
//! 6. **STATIC PROOF** — the sealed `.znippy` round-trips byte-correct through
//! the ordinary arrow-ipc reader (`ZnippyArchive`), its index lists exactly
//! the original+resumed files, and a random blob `get_file`s correctly.
//!
//! Every step inject-asserts real input → real output (the "tests inject
//! values, not just no-crash" LAW): real bytes in, byte-exact bytes out, with
//! the measured values reported into the nornir test matrix.
use std::fs;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use znippy_common::arrow::record_batch::RecordBatch;
use znippy_common::codec::CompressCtx;
use znippy_common::index::{build_metadata_batch, lookup_schema};
use znippy_common::meta::{BlobMeta, ChunkMeta};
use znippy_common::{ArchiveMetaSink, GroupKey, ZnippyArchive, ZnippyReader, get_file};
use znippy_compress::compress_dir;
use znippy_iceberg::{IcebergSink, IcebergZnippyReader, seal};
/// Emit one functional-status row into the nornir test matrix. No-op unless the
/// `testmatrix` feature is on (the release / `cargo publish` path compiles it
/// out). Mirrors `round_trip.rs::fstatus`.
#[cfg(feature = "testmatrix")]
#[inline]
fn fstatus(component: &str, check: &str, ok: bool, detail: &str) {
nornir_testmatrix::functional_status(component, check, ok, detail);
}
#[cfg(not(feature = "testmatrix"))]
#[inline(always)]
fn fstatus(_component: &str, _check: &str, _ok: bool, _detail: &str) {}
/// Initial fixtures (the WRITE phase) — a mix of compressible, incompressible,
/// empty, and a multi-chunk file (> the split block) plus a nested path.
fn write_fixtures() -> Vec<(&'static str, Vec<u8>)> {
let mut rng: u64 = 0x1234_5678_9ABC_DEF0;
let mut random = |n: usize| -> Vec<u8> {
(0..n)
.map(|_| {
rng ^= rng << 13;
rng ^= rng >> 7;
rng ^= rng << 17;
(rng >> 24) as u8
})
.collect()
};
vec![
("hello.txt", b"hello iceberg lifecycle\n".to_vec()),
("empty.txt", Vec::new()),
("repeat.txt", b"znippy".repeat(40_000)),
("noise.bin", random(48 * 1024)),
("big.bin", random(12 * 1024 * 1024)),
("sub/dir/nested.dat", b"nested payload ".repeat(2_000)),
]
}
/// Resume fixtures (appended after the restart, as a new iceberg commit).
fn resume_fixtures() -> Vec<(&'static str, Vec<u8>)> {
let mut rng: u64 = 0x0FED_CBA9_8765_4321;
let mut random = |n: usize| -> Vec<u8> {
(0..n)
.map(|_| {
rng ^= rng << 13;
rng ^= rng >> 7;
rng ^= rng << 17;
(rng >> 24) as u8
})
.collect()
};
vec![
("resume/added_a.txt", b"appended after restart\n".repeat(500)),
("resume/added_b.bin", random(96 * 1024)),
("resume/added_c.txt", b"second iceberg commit\n".to_vec()),
]
}
#[test]
fn iceberg_write_restart_resume_seal_static() {
let tmp = tempfile::tempdir().unwrap();
let src = tmp.path().join("src");
let sidecar: PathBuf = tmp.path().join("store.znippy");
let warehouse = tmp.path().join("wh");
let namespace = "store"; // file stem, as the CLI derives the namespace.
// ───────────────────────────────────────────────────────────────────────
// 1. WRITE — compress fixtures with the iceberg metadata sink.
// ───────────────────────────────────────────────────────────────────────
let write = write_fixtures();
for (rel, bytes) in &write {
let p = src.join(rel);
fs::create_dir_all(p.parent().unwrap()).unwrap();
fs::write(&p, bytes).unwrap();
}
{
let wh = warehouse.clone();
let ns = namespace.to_string();
let factory: znippy_common::MetaSinkFactory =
Box::new(move |_file, _off| Box::new(IcebergSink::new(wh, ns)));
let report = compress_dir(&src, &sidecar, false, None, None, Some(factory)).unwrap();
assert_eq!(report.total_files, write.len() as u64);
// The sidecar produced by the iceberg sink is pure blobs (no footer).
assert!(sidecar.exists(), "blob sidecar must exist");
} // <- sink + its catalog handle dropped here.
let catalog_durable = warehouse.join("catalog.redb").exists();
assert!(catalog_durable, "expected durable catalog.redb");
fstatus(
"iceberg_lifecycle",
"write_durable_catalog",
catalog_durable,
&format!("{} files written via IcebergSink → catalog.redb", write.len()),
);
// Track the current end of the blob region for the resume append. The
// iceberg-format sidecar is pure blobs, so its length == blob region end.
let mut blob_end = fs::metadata(&sidecar).unwrap().len();
// ───────────────────────────────────────────────────────────────────────
// 2 + 3. RESTART (fresh skade::open via a new reader) + RESTART-SURVIVAL.
// ───────────────────────────────────────────────────────────────────────
{
let reader = IcebergZnippyReader::open(&sidecar, &warehouse, namespace).unwrap();
let mut all_ok = true;
for (rel, bytes) in &write {
let got = reader.extract_file(rel).unwrap();
all_ok &= &got == bytes;
assert_eq!(&got, bytes, "post-restart byte mismatch for {rel}");
}
assert_eq!(reader.file_count(), write.len());
fstatus(
"iceberg_lifecycle",
"restart_survival_byte_exact",
all_ok && reader.file_count() == write.len(),
&format!(
"{} files read byte-exact after dropping all handles + fresh skade::open",
write.len()
),
);
} // <- reader dropped (another handle boundary).
// ───────────────────────────────────────────────────────────────────────
// 4. RESUME — append more entries as a NEW iceberg commit on the durable
// catalog. Blobs are appended to the SAME sidecar past `blob_end`; the
// new sub-index is `table.append`ed via a fresh IcebergSink over the same
// warehouse + namespace (skade `table_or_create` reuses the table and
// `append`s a new commit — the resume mechanism).
// ───────────────────────────────────────────────────────────────────────
let resume = resume_fixtures();
let resume_batch = append_blobs_and_build_batch(&sidecar, &mut blob_end, &resume);
{
let mut sink: Box<dyn ArchiveMetaSink> =
Box::new(IcebergSink::new(warehouse.clone(), namespace.to_string()));
let schema = lookup_schema();
sink.push_subindex(
schema.as_ref(),
&[resume_batch],
GroupKey {
pkg_type: 0,
repo: String::new(),
module_name: String::new(),
},
)
.unwrap();
sink.finish().unwrap();
} // <- resume sink + catalog handle dropped (restart again).
// Restart once more and assert BOTH the original and resumed entries read
// back byte-correct (the append landed as a durable, reopenable commit).
let all_after_resume: Vec<(String, Vec<u8>)> = write
.iter()
.chain(resume.iter())
.map(|(r, b)| (r.to_string(), b.clone()))
.collect();
{
let reader = IcebergZnippyReader::open(&sidecar, &warehouse, namespace).unwrap();
let mut all_ok = true;
for (rel, bytes) in &all_after_resume {
let got = reader.extract_file(rel).unwrap();
all_ok &= &got == bytes;
assert_eq!(&got, bytes, "post-resume byte mismatch for {rel}");
}
let count_ok = reader.file_count() == all_after_resume.len();
assert!(count_ok, "resume should add {} files", resume.len());
fstatus(
"iceberg_lifecycle",
"resume_append_new_commit_byte_exact",
all_ok && count_ok,
&format!(
"resume appended {} files as a new iceberg commit; {} total read byte-exact after restart",
resume.len(),
all_after_resume.len()
),
);
}
// ───────────────────────────────────────────────────────────────────────
// 5. SEAL — materialise the post-resume iceberg archive → static .znippy.
// ───────────────────────────────────────────────────────────────────────
let sealed: PathBuf = tmp.path().join("sealed.znippy");
let blob_region_before = fs::metadata(&sidecar).unwrap().len();
let report = seal(&sidecar, &warehouse, namespace, &sealed).unwrap();
// Seal is metadata-only: the blob region is copied verbatim (no recompress),
// and the sealed file is strictly larger only by the metadata tail + footer.
let blobs_reused = report.blob_bytes_copied == blob_region_before;
let tail_only_growth = report.sealed_total_bytes > report.blob_bytes_copied;
assert!(blobs_reused, "seal must reuse the blob region verbatim");
assert!(tail_only_growth, "sealed file = blobs + metadata tail");
assert_eq!(report.files, all_after_resume.len() as u64);
fstatus(
"iceberg_seal",
"metadata_only_blobs_reused",
blobs_reused && tail_only_growth,
&format!(
"{} blob bytes copied verbatim (no recompress); tail+footer added {} bytes for {} files / {} rows",
report.blob_bytes_copied,
report.sealed_total_bytes - report.blob_bytes_copied,
report.files,
report.rows,
),
);
// ───────────────────────────────────────────────────────────────────────
// 6. STATIC PROOF — the sealed .znippy is a valid immutable v0.7 archive.
// ───────────────────────────────────────────────────────────────────────
// 6a. Re-open with the ORDINARY arrow-ipc reader (no warehouse needed).
let archive = ZnippyArchive::open(&sealed).unwrap();
let mut listed = archive.list_files().unwrap();
listed.sort();
let mut expected: Vec<String> = all_after_resume.iter().map(|(r, _)| r.clone()).collect();
expected.sort();
let index_matches = listed == expected;
assert!(index_matches, "sealed index must list all original + resumed files");
// 6b. Byte-exact round-trip of EVERY file through the native reader.
let mut all_ok = true;
let mut total_bytes = 0u64;
for (rel, bytes) in &all_after_resume {
let got = archive.extract_file(rel).unwrap();
all_ok &= &got == bytes;
total_bytes += got.len() as u64;
assert_eq!(&got, bytes, "sealed byte mismatch for {rel}");
}
fstatus(
"iceberg_seal",
"sealed_native_reader_byte_exact",
all_ok && index_matches,
&format!(
"sealed .znippy re-opened with ZnippyArchive: {} files / {} bytes byte-exact",
all_after_resume.len(),
total_bytes
),
);
// 6c. Random-access a blob via the native lookup path (get_file = locate +
// pread + decompress + verify) — proves the sealed index + trie + a
// random blob are all valid.
let probe = "big.bin";
let probe_bytes = &write.iter().find(|(r, _)| *r == probe).unwrap().1;
let got = get_file(&sealed, probe).unwrap();
let random_blob_ok = &got == probe_bytes;
assert!(random_blob_ok, "random-access get_file('{probe}') must be byte-exact");
fstatus(
"iceberg_seal",
"sealed_random_access_get_file",
random_blob_ok,
&format!("get_file('{probe}') byte-exact via sealed lookup+trie ({} bytes)", got.len()),
);
}
/// Compress each `(rel, bytes)` fixture with the znippy codec, append the
/// compressed blob to `sidecar` at the running `*blob_end`, and build the
/// base-schema [`RecordBatch`] the iceberg sink ingests. Mirrors the real
/// compress path: blake3 over the ORIGINAL bytes, store-raw when the codec frame
/// is not smaller, offsets into the appended region. Single chunk per file (the
/// fixtures are small) — `chunk_seq = 0`.
fn append_blobs_and_build_batch(
sidecar: &Path,
blob_end: &mut u64,
fixtures: &[(&'static str, Vec<u8>)],
) -> RecordBatch {
let file = fs::OpenOptions::new().write(true).open(sidecar).unwrap();
let mut ctx = CompressCtx::new(3).unwrap();
let mut blobs: Vec<BlobMeta> = Vec::with_capacity(fixtures.len());
let mut paths: Vec<String> = Vec::with_capacity(fixtures.len());
for (file_index, (rel, bytes)) in fixtures.iter().enumerate() {
let checksum = *blake3::hash(bytes).as_bytes(); // ORIGINAL bytes.
let compressed_frame = ctx.compress(bytes).unwrap();
// Store raw unless compression actually shrinks the payload.
let (on_disk, compressed): (&[u8], bool) = if compressed_frame.len() < bytes.len() {
(&compressed_frame, true)
} else {
(bytes, false)
};
let blob_offset = *blob_end;
file.write_all_at(on_disk, blob_offset).unwrap();
*blob_end += on_disk.len() as u64;
paths.push(rel.to_string());
blobs.push(BlobMeta {
blob_offset,
blob_size: on_disk.len() as u64,
chunk_meta: ChunkMeta {
fdata_offset: 0,
file_index: file_index as u64,
chunk_seq: 0,
checksum,
compressed,
uncompressed_size: bytes.len() as u64,
compressed_size: on_disk.len() as u64,
},
});
}
file.sync_all().unwrap();
let resolver = {
let paths = paths.clone();
move |fi: u64| paths[fi as usize].clone()
};
let batch = build_metadata_batch(&blobs, resolver, &[], &[]).unwrap();
// The sink ingests Arrow-58 base-schema batches.
assert_eq!(batch.schema().fields(), lookup_schema().fields());
batch
}