From 24afcdc70f1b7933be2b9911da659aa9955afe24 Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 19 Sep 2026 06:43:34 -0700 Subject: [PATCH] test(agent): WAL property tests, crash-recovery matrix, WAL fuzz target MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tests/wal_properties.rs — deterministic generator, reproducible by seed: everything appended is read back intact (300 cases), and after ANY damage to the file (bit flips, truncation, inserted/deleted bytes, duplicated or rotated regions, overwritten ranges; 1500 cases) reading never panics and yields an exact prefix of what was written — the guarantee the chained CRC exists to give. Opening for append then repairs the tail and a new entry lands right behind the surviving prefix. - tests/crash_recovery.rs — builds the on-disk images a process crash can leave and reopens each against a model of what was acknowledged: an image after every operation (random saves, in-place updates, checkpoints, small wal_max_entries), the checkpoint window (new .h5 + not-yet-truncated WAL) over several rounds, and the WAL torn at every byte length, which must recover the checkpoint plus a prefix of the operations logged since. - fuzz/fuzz_wal_replay — arbitrary bytes as a WAL: read and open-for-append must not panic, and open() must not change what is replayable. Based on the target from the clawmates mission branch (4aee2fa), with the repair property added. The CI fuzz step now covers both fuzz crates. Co-Authored-By: Claude Fable 5.1 --- crates/clawhdf5-agent/fuzz/.gitignore | 3 + crates/clawhdf5-agent/fuzz/Cargo.toml | 23 ++ .../fuzz/fuzz_targets/fuzz_wal_replay.rs | 36 +++ crates/clawhdf5-agent/tests/crash_recovery.rs | 187 +++++++++++++++ crates/clawhdf5-agent/tests/wal_properties.rs | 213 ++++++++++++++++++ scripts/ci-test.sh | 14 +- 6 files changed, 470 insertions(+), 6 deletions(-) create mode 100644 crates/clawhdf5-agent/fuzz/.gitignore create mode 100644 crates/clawhdf5-agent/fuzz/Cargo.toml create mode 100644 crates/clawhdf5-agent/fuzz/fuzz_targets/fuzz_wal_replay.rs create mode 100644 crates/clawhdf5-agent/tests/crash_recovery.rs create mode 100644 crates/clawhdf5-agent/tests/wal_properties.rs diff --git a/crates/clawhdf5-agent/fuzz/.gitignore b/crates/clawhdf5-agent/fuzz/.gitignore new file mode 100644 index 0000000..c937f05 --- /dev/null +++ b/crates/clawhdf5-agent/fuzz/.gitignore @@ -0,0 +1,3 @@ +target/ +artifacts/ +coverage/ diff --git a/crates/clawhdf5-agent/fuzz/Cargo.toml b/crates/clawhdf5-agent/fuzz/Cargo.toml new file mode 100644 index 0000000..6f8bc01 --- /dev/null +++ b/crates/clawhdf5-agent/fuzz/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "clawhdf5-agent-fuzz" +version = "0.0.0" +publish = false +edition = "2024" + +[package.metadata] +cargo-fuzz = true + +[dependencies] +libfuzzer-sys = "0.4" +tempfile = "3" + +[dependencies.clawhdf5-agent] +path = ".." + +[workspace] +members = ["."] + +[[bin]] +name = "fuzz_wal_replay" +path = "fuzz_targets/fuzz_wal_replay.rs" +doc = false diff --git a/crates/clawhdf5-agent/fuzz/fuzz_targets/fuzz_wal_replay.rs b/crates/clawhdf5-agent/fuzz/fuzz_targets/fuzz_wal_replay.rs new file mode 100644 index 0000000..a4a81a9 --- /dev/null +++ b/crates/clawhdf5-agent/fuzz/fuzz_targets/fuzz_wal_replay.rs @@ -0,0 +1,36 @@ +#![no_main] +//! Arbitrary bytes as a WAL file. Reading, and opening for append (which scans +//! the chain and truncates an unverifiable tail), must never panic, hang, or +//! allocate without bound — and after `open` repairs the file, everything +//! `read_entries` returned before must still be returned. +//! +//! The deterministic counterpart that runs in ordinary CI is +//! `tests/wal_properties.rs`; this target explores inputs it cannot reach. + +use std::io::Write as _; + +use clawhdf5_agent::wal::WalFile; +use libfuzzer_sys::fuzz_target; + +fuzz_target!(|data: &[u8]| { + let Ok(mut tmp) = tempfile::NamedTempFile::new() else { + return; + }; + if tmp.write_all(data).and_then(|()| tmp.flush()).is_err() { + return; + } + let before = WalFile::read_entries(tmp.path()).map(|e| e.len()); + // Only the chained formats (header versions 3 and 4) are repaired in + // place. `open` deliberately recreates a legacy-format file from scratch: + // `HDF5Memory::open` has already replayed its entries by then. + let chained = matches!(data.get(4), Some(3 | 4)); + let opened = WalFile::open(tmp.path()); + if !chained { + return; + } + if let (Ok(before), Ok(wal)) = (before, opened) { + drop(wal); + let after = WalFile::read_entries(tmp.path()).map(|e| e.len()); + assert_eq!(after.ok(), Some(before), "open() changed what is replayable"); + } +}); diff --git a/crates/clawhdf5-agent/tests/crash_recovery.rs b/crates/clawhdf5-agent/tests/crash_recovery.rs new file mode 100644 index 0000000..bd91179 --- /dev/null +++ b/crates/clawhdf5-agent/tests/crash_recovery.rs @@ -0,0 +1,187 @@ +//! Crash-recovery matrix for `HDF5Memory`. +//! +//! A process crash leaves whatever reached the OS on disk. These tests build +//! the on-disk images such a crash can leave behind — after every operation, +//! inside the checkpoint window (new `.h5` in place, WAL not yet truncated), +//! and with the WAL torn at every possible length — then reopen each image +//! and check the recovered store against a model of what was acknowledged. +//! +//! Invariants: +//! * never a duplicated or invented record; +//! * an image taken between operations recovers *exactly* the acknowledged +//! state; +//! * a torn WAL recovers the last checkpoint plus a prefix of the operations +//! logged since. + +use std::path::{Path, PathBuf}; + +use clawhdf5_agent::{AgentMemory, HDF5Memory, MemoryConfig, MemoryEntry}; +use tempfile::TempDir; + +struct Rng(u64); + +impl Rng { + fn next(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + fn below(&mut self, n: usize) -> usize { + (self.next() % n.max(1) as u64) as usize + } +} + +fn entry(chunk: &str, tags: &str) -> MemoryEntry { + MemoryEntry { + chunk: chunk.to_string(), + embedding: vec![1.0, 0.0, 0.0, 0.0], + source_channel: "test".into(), + timestamp: 1.0, + session_id: "s".into(), + tags: tags.to_string(), + } +} + +fn wal_path(h5: &Path) -> PathBuf { + h5.with_extension("h5.wal") +} + +/// Copy the store (`.h5` + WAL) into a fresh directory, as a crash image. +fn image(h5: &Path, into: &TempDir, name: &str) -> PathBuf { + let dest = into.path().join(format!("{name}.h5")); + std::fs::copy(h5, &dest).unwrap(); + if wal_path(h5).exists() { + std::fs::copy(wal_path(h5), wal_path(&dest)).unwrap(); + } + dest +} + +fn recovered(h5: &Path) -> Vec { + // Read-only: the image must not be modified, and no lock is needed. + HDF5Memory::open_read_only(h5).unwrap().cache.chunks.clone() +} + +/// Apply one random operation to the store and to the model. +fn step(mem: &mut HDF5Memory, model: &mut Vec, rng: &mut Rng, n: usize) { + match rng.below(6) { + 0 => mem.flush_wal().unwrap(), + 1 if !model.is_empty() => { + // Update an existing record in place, addressed by its tag. + let idx = rng.below(model.len()); + let chunk = format!("u{n}"); + assert_eq!( + mem.save_or_update(entry(&chunk, &format!("tag{idx}"))) + .unwrap(), + idx + ); + model[idx] = chunk; + } + _ => { + let chunk = format!("c{n}"); + mem.save(entry(&chunk, &format!("tag{}", model.len()))) + .unwrap(); + model.push(chunk); + } + } +} + +#[test] +fn image_after_every_operation_recovers_the_acknowledged_state() { + for seed in 0..40u64 { + let mut rng = Rng(seed); + let dir = TempDir::new().unwrap(); + let images = TempDir::new().unwrap(); + let mut config = MemoryConfig::new(dir.path().join("store.h5"), "agent", 4); + config.wal_enabled = true; + config.wal_max_entries = 1 + rng.below(6); // force frequent checkpoints + let h5 = config.path.clone(); + let mut mem = HDF5Memory::create(config).unwrap(); + let mut model = Vec::new(); + + for n in 0..30 { + step(&mut mem, &mut model, &mut rng, n); + let img = image(&h5, &images, &format!("s{seed}-{n}")); + assert_eq!(recovered(&img), model, "seed {seed}, after op {n}"); + } + } +} + +#[test] +fn crash_inside_the_checkpoint_window_never_duplicates() { + for seed in 0..40u64 { + let mut rng = Rng(seed ^ 0xABCD); + let dir = TempDir::new().unwrap(); + let images = TempDir::new().unwrap(); + let mut config = MemoryConfig::new(dir.path().join("store.h5"), "agent", 4); + config.wal_enabled = true; + config.wal_max_entries = 1000; // checkpoints only when we ask + let h5 = config.path.clone(); + let mut mem = HDF5Memory::create(config).unwrap(); + let mut model = Vec::new(); + + for round in 0..4 { + for n in 0..(1 + rng.below(6)) { + step(&mut mem, &mut model, &mut rng, round * 100 + n); + } + // The WAL as it is just before the checkpoint... + let stale_wal = images.path().join(format!("stale-{seed}-{round}.wal")); + if wal_path(&h5).exists() { + std::fs::copy(wal_path(&h5), &stale_wal).unwrap(); + } + mem.flush_wal().unwrap(); + // ...put back next to the NEW .h5: the crash-in-the-window image. + let img = image(&h5, &images, &format!("w{seed}-{round}")); + if stale_wal.exists() { + std::fs::copy(&stale_wal, wal_path(&img)).unwrap(); + } + assert_eq!(recovered(&img), model, "seed {seed}, round {round}"); + } + } +} + +#[test] +fn torn_wal_recovers_checkpoint_plus_a_prefix() { + let dir = TempDir::new().unwrap(); + let images = TempDir::new().unwrap(); + let mut config = MemoryConfig::new(dir.path().join("store.h5"), "agent", 4); + config.wal_enabled = true; + config.wal_max_entries = 1000; + let h5 = config.path.clone(); + let mut mem = HDF5Memory::create(config).unwrap(); + + for name in ["a", "b"] { + mem.save(entry(name, name)).unwrap(); + } + mem.flush_wal().unwrap(); + let checkpointed = vec!["a".to_string(), "b".to_string()]; + + // States the store passes through as each later op is logged. + let mut states = vec![checkpointed.clone()]; + let mut model = checkpointed.clone(); + mem.save(entry("c", "c")).unwrap(); + model.push("c".into()); + states.push(model.clone()); + mem.save_or_update(entry("a2", "a")).unwrap(); + model[0] = "a2".into(); + states.push(model.clone()); + mem.save(entry("d", "d")).unwrap(); + model.push("d".into()); + states.push(model.clone()); + + let full_wal = std::fs::read(wal_path(&h5)).unwrap(); + let mut seen = std::collections::BTreeSet::new(); + for len in 0..=full_wal.len() { + let img = image(&h5, &images, &format!("t{len}")); + std::fs::write(wal_path(&img), &full_wal[..len]).unwrap(); + let got = recovered(&img); + let which = states + .iter() + .position(|s| *s == got) + .unwrap_or_else(|| panic!("WAL torn at {len} bytes recovered {got:?}")); + seen.insert(which); + } + // Every intermediate state is reachable, and the full WAL gives the last. + assert_eq!(seen.into_iter().collect::>(), [0, 1, 2, 3]); +} diff --git a/crates/clawhdf5-agent/tests/wal_properties.rs b/crates/clawhdf5-agent/tests/wal_properties.rs new file mode 100644 index 0000000..ad714d5 --- /dev/null +++ b/crates/clawhdf5-agent/tests/wal_properties.rs @@ -0,0 +1,213 @@ +//! Property tests for the write-ahead log. +//! +//! A deterministic generator (no external crates, reproducible from the seed +//! printed on failure) drives thousands of cases through two properties: +//! +//! 1. **Round trip** — whatever was appended is read back, in order, intact. +//! 2. **Prefix under corruption** — after *any* damage to the file (bit flips, +//! truncation, inserted or deleted bytes, duplicated or reordered regions), +//! reading never panics and yields an exact *prefix* of what was written. +//! This is the guarantee the chained CRC exists to provide: replay may stop +//! early, but it never returns a corrupted, reordered, or invented entry. + +use clawhdf5_agent::wal::{WalEntry, WalEntryType, WalFile}; + +/// SplitMix64: tiny, well-distributed, and fully determined by its seed. +struct Rng(u64); + +impl Rng { + fn next(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + + fn below(&mut self, n: usize) -> usize { + (self.next() % n.max(1) as u64) as usize + } + + fn string(&mut self, max_len: usize) -> String { + const ALPHABET: &[char] = &['a', 'Z', '0', ' ', '\n', '\0', 'é', '漢', '🦀', '"']; + (0..self.below(max_len + 1)) + .map(|_| ALPHABET[self.below(ALPHABET.len())]) + .collect() + } +} + +/// What a test appended, in a form comparable with what is read back. +#[derive(Debug, Clone, PartialEq)] +enum Logged { + Save(String, Vec, String, String, String, u64), + Update(usize, String, Vec, u64), + Tombstone(usize, u64), +} + +fn logged(entry: &WalEntry) -> Logged { + // Compare floats by bit pattern so NaN payloads and -0.0 count as intact. + let bits: Vec = entry.embedding.iter().map(|f| f.to_bits()).collect(); + let ts = entry.timestamp.to_bits(); + match entry.entry_type { + WalEntryType::Save => Logged::Save( + entry.chunk.clone(), + bits, + entry.source_channel.clone(), + entry.session_id.clone(), + entry.tags.clone(), + ts, + ), + WalEntryType::Update => { + Logged::Update(entry.update_index.unwrap(), entry.chunk.clone(), bits, ts) + } + WalEntryType::Tombstone => Logged::Tombstone(entry.tombstone_index.unwrap(), ts), + WalEntryType::ActivationUpdate => unreachable!("never written by these tests"), + } +} + +/// Append a random mix of records; return what was written. +fn write_random_wal(path: &std::path::Path, rng: &mut Rng) -> Vec { + let mut wal = WalFile::open(path).unwrap(); + let mut written = Vec::new(); + for _ in 0..rng.below(12) { + let timestamp = f64::from_bits(rng.next()); + if rng.below(5) == 0 { + let index = rng.below(1000); + wal.append_tombstone(index, timestamp).unwrap(); + written.push(Logged::Tombstone(index, timestamp.to_bits())); + continue; + } + let update_index = (rng.below(4) == 0).then(|| rng.below(1000)); + let entry = WalEntry { + entry_type: if update_index.is_some() { + WalEntryType::Update + } else { + WalEntryType::Save + }, + timestamp, + chunk: rng.string(40), + embedding: (0..rng.below(9)) + .map(|_| f32::from_bits(rng.next() as u32)) + .collect(), + source_channel: rng.string(8), + session_id: rng.string(8), + tags: rng.string(8), + tombstone_index: None, + update_index, + }; + wal.append_save(&entry).unwrap(); + written.push(logged(&entry)); + } + written +} + +fn read_back(path: &std::path::Path) -> Option> { + WalFile::read_entries(path) + .ok() + .map(|entries| entries.iter().map(logged).collect()) +} + +#[test] +fn everything_appended_is_read_back_intact() { + let dir = tempfile::TempDir::new().unwrap(); + for seed in 0..300u64 { + let path = dir.path().join(format!("rt-{seed}.wal")); + let written = write_random_wal(&path, &mut Rng(seed)); + assert_eq!(read_back(&path).unwrap(), written, "seed {seed}"); + // Reopening (which scans and repositions) must not disturb anything. + drop(WalFile::open(&path).unwrap()); + assert_eq!( + read_back(&path).unwrap(), + written, + "seed {seed} after reopen" + ); + } +} + +/// Damage `bytes` in one of several ways. +fn corrupt(bytes: &mut Vec, rng: &mut Rng) { + if bytes.is_empty() { + return; + } + match rng.below(7) { + 0 => { + let i = rng.below(bytes.len()); + bytes[i] ^= 1 << rng.below(8); + } + 1 => bytes.truncate(rng.below(bytes.len())), + 2 => { + let i = rng.below(bytes.len() + 1); + bytes.insert(i, rng.next() as u8); + } + 3 => { + let i = rng.below(bytes.len()); + bytes.remove(i); + } + 4 => { + // Duplicate a region in place (a replayed/duplicated entry). + let a = rng.below(bytes.len()); + let b = a + rng.below(bytes.len() - a); + let region = bytes[a..b].to_vec(); + let at = rng.below(bytes.len() + 1); + bytes.splice(at..at, region); + } + 5 => { + // Swap two regions (reordered entries). + let mid = rng.below(bytes.len()); + bytes.rotate_left(mid); + } + _ => { + let i = rng.below(bytes.len()); + let n = rng.below(bytes.len() - i + 1); + for b in &mut bytes[i..i + n] { + *b = rng.next() as u8; + } + } + } +} + +#[test] +fn any_corruption_yields_a_prefix_never_a_wrong_entry() { + let dir = tempfile::TempDir::new().unwrap(); + let mut shortened = 0u32; + for seed in 0..1500u64 { + let mut rng = Rng(seed ^ 0xC0FF_EE00); + let path = dir.path().join("c.wal"); + let _ = std::fs::remove_file(&path); + let written = write_random_wal(&path, &mut rng); + + let mut bytes = std::fs::read(&path).unwrap(); + for _ in 0..=rng.below(3) { + corrupt(&mut bytes, &mut rng); + } + std::fs::write(&path, &bytes).unwrap(); + + // An unreadable header is a clean error; anything else is a prefix. + if let Some(read) = read_back(&path) { + assert!( + read.len() <= written.len() && read[..] == written[..read.len()], + "seed {seed}: read {read:?}\nis not a prefix of {written:?}" + ); + if read.len() < written.len() { + shortened += 1; + } + // Opening for append repairs the tail; what was readable stays so, + // and a new entry lands right after it. + if let Ok(mut wal) = WalFile::open(&path) { + wal.append_tombstone(7, 1.0).unwrap(); + drop(wal); + let mut expected = read.clone(); + expected.push(Logged::Tombstone(7, 1.0f64.to_bits())); + assert_eq!( + read_back(&path).unwrap(), + expected, + "seed {seed} after repair" + ); + } + } + } + assert!( + shortened > 100, + "corruption rarely took effect: {shortened}" + ); +} diff --git a/scripts/ci-test.sh b/scripts/ci-test.sh index 65dc9ea..aa7146d 100755 --- a/scripts/ci-test.sh +++ b/scripts/ci-test.sh @@ -95,12 +95,14 @@ run_step "check-nostd.sh" "$SCRIPT_DIR/check-nostd.sh" # 8. Optional fuzz smoke run if [ -n "${CLAWHDF5_FUZZ_SECONDS:-}" ]; then fuzz_smoke() { - local target - cd "$SCRIPT_DIR/../crates/clawhdf5-format" || return 1 - for target in $(cargo +nightly fuzz list); do - echo "--- fuzz: $target" - cargo +nightly fuzz run "$target" -- \ - -max_total_time="$CLAWHDF5_FUZZ_SECONDS" || return 1 + local crate target + for crate in clawhdf5-format clawhdf5-agent; do + cd "$SCRIPT_DIR/../crates/$crate" || return 1 + for target in $(cargo +nightly fuzz list); do + echo "--- fuzz: $crate/$target" + cargo +nightly fuzz run "$target" -- \ + -max_total_time="$CLAWHDF5_FUZZ_SECONDS" || return 1 + done done } run_step "fuzz smoke (${CLAWHDF5_FUZZ_SECONDS}s/target)" fuzz_smoke