entropyfs 0.7.0

//! Aggressive simulated power-failure tests (§38, `docs/recovery/
//! crash-consistency.md`): kill at every durability boundary — for both
//! regular commits and GC — then verify the recovered store is an
//! admissible pre/post state, passes fsck, and remains writable.

#![forbid(unsafe_code)]

use tempfile::TempDir;

use crate::core::candidate::{pick_cheapest, raw_candidate, zero_candidate};
use crate::core::extent::ChunkId;
use crate::entropy::palette::PaletteEncoder;
use crate::entropy::periodic::PeriodicEncoder;
use crate::entropy::sparse::SparseEncoder;
use crate::fsck::{FsckOptions, fsck};
use crate::rans::residual::RansEncoder;
use crate::store::io::IoBackendKind;
use crate::store::transaction::{CrashHooks, CrashPoint};
use crate::store::{ExtentUpdate, Store, StoreConfig};

fn create_store(dir: &TempDir, kind: IoBackendKind) -> Store {
    let cfg = StoreConfig {
        segment_size: 1024 * 1024,
        io_backend: kind,
        ..Default::default()
    };
    Store::create(dir.path(), &cfg, [0x33; 16]).unwrap()
}

fn encode_chunks(content: &[u8], store: &Store) -> Vec<ExtentUpdate> {
    let limits = store.limits();
    let policy = store.policy();
    let chunk_class = limits.chunk_class as usize;
    let mut updates = Vec::new();
    let mut off = 0usize;
    while off < content.len() {
        let end = (off + chunk_class).min(content.len());
        let chunk = &content[off..end];
        let cid = ChunkId::of(chunk);
        let ctx = crate::core::candidate::CandidateContext {
            limits,
            policy,
            content_id: cid,
            bases: &[],
            dedup: None,
        };
        let mut cands = Vec::new();
        if let Some(z) = zero_candidate(chunk, cid, limits) {
            cands.push(z);
        }
        for enc in [
            Box::new(SparseEncoder) as Box<dyn crate::core::candidate::Encoder>,
            Box::new(PaletteEncoder),
            Box::new(PeriodicEncoder),
            Box::new(RansEncoder),
        ] {
            cands.extend(enc.encode(chunk, &ctx));
        }
        if let Some(r) = raw_candidate(chunk, cid, limits) {
            cands.push(r);
        }
        let best = pick_cheapest(&cands, policy).expect("at least raw");
        updates.push(ExtentUpdate {
            offset: off as u64,
            descriptor: best.representation.clone(),
            content_id: cid,
            objects: best.objects.clone(),
        });
        off = end;
    }
    updates
}

fn write_file(store: &Store, ino: u64, content: &[u8]) {
    let updates = encode_chunks(content, store);
    store
        .commit_file_extents(
            ino,
            updates,
            Some(content.len() as u64),
            &CrashHooks::none(),
        )
        .unwrap();
}

fn create_file(store: &Store, ino: u64) {
    let inode = crate::store::inode::Inode::new_file(1000, 1000, 0o644);
    let mut tx = store.begin_tx().unwrap();
    Store::put_inode_in_tx(&mut tx, ino, &inode).unwrap();
    tx.commit(&CrashHooks::none()).unwrap();
}

/// Every commit boundary (regular path).
fn commit_crash_points() -> Vec<CrashPoint> {
    vec![
        CrashPoint::AfterRootWrite,
        CrashPoint::AfterRecordAppend,
        CrashPoint::AfterSegmentFdatasync,
        CrashPoint::AfterSegmentDirFsync,
        CrashPoint::AfterSuperblockWrite,
        CrashPoint::AfterSuperblockFsync,
    ]
}

/// Every GC boundary.
fn gc_crash_points() -> Vec<CrashPoint> {
    vec![
        CrashPoint::AfterRootWrite,
        CrashPoint::AfterSuperblockWrite,
        CrashPoint::AfterSuperblockFsync,
        CrashPoint::BeforeOldSegmentDelete,
    ]
}

/// The recovery contract for one backend.
fn assert_recovered(dir: &TempDir, expected: &[u8], kind: IoBackendKind) -> Store {
    let report = fsck(dir.path(), &FsckOptions::default())
        .unwrap_or_else(|e| panic!("fsck after crash: {e}"));
    assert!(
        report.is_clean(),
        "fsck after crash must be clean:\n{}",
        report.render()
    );
    let store = Store::open(dir.path(), &config_for(kind))
        .unwrap_or_else(|e| panic!("reopen after crash: {e}"));
    let read = store
        .read_file(3, 0, expected.len() as u64)
        .unwrap_or_else(|e| panic!("read after crash: {e}"));
    assert_eq!(read, expected, "recovered content must be exact");
    // The recovered store must accept new writes.
    write_file(
        &store,
        3,
        b"post-crash-recovery-write".repeat(64).as_slice(),
    );
    store
}

fn config_for(kind: IoBackendKind) -> StoreConfig {
    StoreConfig {
        io_backend: kind,
        ..Default::default()
    }
}

#[test]
fn commit_crash_matrix_then_fsck_and_rewrite() {
    for kind in IoBackendKind::ALL {
        for point in commit_crash_points() {
            let dir = TempDir::new().unwrap();
            let store = create_store(&dir, kind);
            create_file(&store, 3);
            let pre: Vec<u8> = b"pre-crash-payload".repeat(512);
            write_file(&store, 3, &pre);
            let pre_len = pre.len() as u64;

            let post: Vec<u8> = b"post-crash-payload-DIFFERENT".repeat(512);
            let updates = encode_chunks(&post, &store);
            let res = store.commit_file_extents(
                3,
                updates,
                Some(post.len() as u64),
                &CrashHooks::crash_at(point),
            );
            assert!(res.is_err(), "crash point {point:?} must report");
            drop(store);

            // The recovered state is the pre-state or the post-state — never a
            // hybrid — and must pass fsck and remain writable.
            let store2 = Store::open(dir.path(), &config_for(kind))
                .unwrap_or_else(|e| panic!("reopen at {point:?}: {e}"));
            let after = store2
                .read_file(3, 0, pre_len.max(post.len() as u64))
                .unwrap();
            let pre_ok = after == pre;
            let post_ok = after == post;
            assert!(
                pre_ok || post_ok,
                "point {point:?} ({kind:?}): hybrid or corrupt state (len {})",
                after.len()
            );
            let report = fsck(dir.path(), &FsckOptions::default()).unwrap();
            assert!(
                report.is_clean(),
                "point {point:?} ({kind:?}): fsck must be clean:\n{}",
                report.render()
            );
            write_file(&store2, 3, b"post-recovery".repeat(128).as_slice());
        }
    }
}

#[test]
fn gc_crash_matrix_preserves_live_data() {
    for kind in IoBackendKind::ALL {
        for point in gc_crash_points() {
            let dir = TempDir::new().unwrap();
            let store = create_store(&dir, kind);
            create_file(&store, 3);
            // Generate garbage + a well-defined final state.
            for i in 0..8 {
                let content = format!("gc-version-{i}:{}", "g".repeat(1500)).into_bytes();
                write_file(&store, 3, &content);
            }
            let final_content = format!("gc-version-{}:{}", 8, "g".repeat(1500)).into_bytes();
            write_file(&store, 3, &final_content);
            assert_eq!(
                store.read_file(3, 0, final_content.len() as u64).unwrap(),
                final_content
            );
            // Arm the crash at a GC boundary.
            let res = crate::store::gc::collect(&store, &CrashHooks::crash_at(point));
            assert!(res.is_err(), "GC crash point {point:?} must report");
            drop(store);
            // The live data must survive exactly; fsck clean; writable.
            assert_recovered(&dir, &final_content, kind);
        }
    }
}

#[test]
fn gc_crash_at_delete_still_reclaims_later() {
    for kind in IoBackendKind::ALL {
        let dir = TempDir::new().unwrap();
        let store = create_store(&dir, kind);
        create_file(&store, 3);
        for i in 0..10 {
            let content = format!("v{i}:{}", "h".repeat(1200)).into_bytes();
            write_file(&store, 3, &content);
        }
        let final_content = format!("v{}:{}", 10, "h".repeat(1200)).into_bytes();
        write_file(&store, 3, &final_content);
        // Crash AFTER the new root is durable but BEFORE old segment deletion.
        let res = crate::store::gc::collect(
            &store,
            &CrashHooks::crash_at(CrashPoint::BeforeOldSegmentDelete),
        );
        assert!(res.is_err());
        drop(store);
        // Reopen, verify content, and run GC to completion: it must reclaim
        // the garbage left by the interrupted run (both old and new).
        let store = assert_recovered(&dir, &final_content, kind);
        let reclaimed = crate::store::gc::collect(&store, &CrashHooks::none()).unwrap();
        assert!(reclaimed > 0, "interrupted GC must leave reclaimable data");
        // `assert_recovered` wrote this exact content after recovery; it must
        // still be intact after the completed GC.
        let recovery_content: Vec<u8> = b"post-crash-recovery-write".repeat(64);
        assert_eq!(
            store
                .read_file(3, 0, recovery_content.len() as u64)
                .unwrap(),
            recovery_content
        );
    }
}

#[test]
fn crash_between_commits_is_linearizable() {
    for kind in IoBackendKind::ALL {
        // A sequence of commits with a crash mid-sequence: the store must
        // expose exactly one of the committed prefixes (write-atomicity).
        let dir = TempDir::new().unwrap();
        let store = create_store(&dir, kind);
        create_file(&store, 3);
        let mut committed: Vec<Vec<u8>> = Vec::new();
        for i in 0..6 {
            let content = format!("commit-{i}:{}", "q".repeat(800)).into_bytes();
            write_file(&store, 3, &content);
            committed.push(content.clone());
            assert_eq!(
                store.read_file(3, 0, content.len() as u64).unwrap(),
                content
            );
        }
        // Crash on the 7th commit at the superblock-write boundary.
        let content7 = format!("commit-{}:{}", 6, "q".repeat(800)).into_bytes();
        let updates = encode_chunks(&content7, &store);
        let res = store.commit_file_extents(
            3,
            updates,
            Some(content7.len() as u64),
            &CrashHooks::crash_at(CrashPoint::AfterSuperblockWrite),
        );
        assert!(res.is_err());
        drop(store);
        // Recovered content must be one of the committed prefixes.
        let store2 = Store::open(dir.path(), &config_for(kind)).unwrap();
        let read = store2
            .read_file(
                3,
                0,
                committed.last().unwrap().len().max(content7.len()) as u64,
            )
            .unwrap();
        let admissible = committed.contains(&read)
            || read == content7
            || (read.len() == committed.last().unwrap().len()
                && read == *committed.last().unwrap());
        assert!(admissible, "recovered state is not a committed prefix");
    }
}