clawhdf5/crates/clawhdf5-agent/tests/wal_properties.rs

//! 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<u32>, String, String, String, u64),
    Update(usize, String, Vec<u32>, 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<u32> = 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<Logged> {
    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<Vec<Logged>> {
    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<u8>, 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}"
    );
}