obj-core 1.1.2

//! Pager tests — unit + property.

use std::collections::HashSet;
use std::fs::OpenOptions;
use std::io::{Read, Seek, SeekFrom, Write};

use proptest::prelude::*;
use tempfile::TempDir;

use crate::error::Error;
use crate::pager::page::{Page, PageId, PAGE_SIZE};
use crate::pager::{Config, Pager};
use crate::wal::Lsn;

fn id(n: u64) -> PageId {
    PageId::new(n).expect("non-zero")
}

/// Test helper — open a file-backed pager and enter a WAL txn so the
/// tests below can call `alloc_page` / `free_page` without tripping
/// the post-#64 `in_txn` debug-assert. The tests pre-#64 relied on
/// `alloc_page` writing the header directly; #64 routes the header
/// through the WAL via `stage_or_write_header`, which requires an
/// open txn. The helper preserves the tests' original flow
/// (open → mutate → maybe-commit → drop) without adding a `begin_txn`
/// at every call site.
fn open_file(path: &std::path::Path, config: Config) -> Pager<crate::platform::FileHandle> {
    let mut p = Pager::open(path, config).expect("open");
    p.begin_txn();
    p
}

#[test]
fn memory_pager_alloc_round_trip() {
    let mut p = Pager::memory(Config::default()).expect("construct");
    let a = p.alloc_page().expect("alloc");
    let mut page = Page::zeroed();
    page.as_bytes_mut()[0] = 0xAB;
    p.write_page(a, &page).expect("write");
    let read = p.read_page(a).expect("read");
    assert_eq!(read.as_bytes()[0], 0xAB);
}

#[test]
fn alloc_and_free_recycles_id() {
    let mut p = Pager::memory(Config::default()).expect("construct");
    let a = p.alloc_page().expect("alloc");
    let b = p.alloc_page().expect("alloc");
    assert_ne!(a, b);
    p.free_page(a).expect("free");
    let c = p.alloc_page().expect("realloc");
    assert_eq!(c, a, "freelist must recycle the most recently freed id");
}

#[test]
fn free_then_alloc_lifo_order() {
    let mut p = Pager::memory(Config::default()).expect("construct");
    let ids: Vec<PageId> = (0..4).map(|_| p.alloc_page().expect("alloc")).collect();
    for &i in &ids {
        p.free_page(i).expect("free");
    }
    // LIFO recycle order.
    let recycled: Vec<PageId> = (0..4).map(|_| p.alloc_page().expect("realloc")).collect();
    let expected: Vec<PageId> = ids.iter().rev().copied().collect();
    assert_eq!(recycled, expected);
}

#[test]
fn file_backend_round_trip() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("test.obj");
    {
        let mut p = open_file(&path, Config::default());
        let a = p.alloc_page().expect("alloc");
        let mut page = Page::zeroed();
        page.as_bytes_mut()[0..4].copy_from_slice(b"WXYZ");
        p.write_page(a, &page).expect("write");
        p.flush().expect("flush");
    }
    // Reopen and read back.
    let mut p = open_file(&path, Config::default());
    let a = id(1);
    let read = p.read_page(a).expect("read");
    assert_eq!(&read.as_bytes()[0..4], b"WXYZ");
}

#[test]
fn close_reopen_preserves_freelist_head() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("test.obj");
    let (a, b) = {
        let mut p = open_file(&path, Config::default());
        let a = p.alloc_page().expect("alloc");
        let b = p.alloc_page().expect("alloc");
        p.free_page(a).expect("free a");
        p.flush().expect("flush");
        let head = p.freelist_head();
        assert_eq!(head, a.get());
        (a, b)
    };
    let mut p = open_file(&path, Config::default());
    assert_eq!(p.freelist_head(), a.get());
    // Allocating must recycle `a`, not give us a fresh page.
    let recycled = p.alloc_page().expect("realloc");
    assert_eq!(recycled, a);
    assert_ne!(recycled, b);
}

#[test]
fn cache_evicts_under_pressure() {
    // Capacity = 2 frames; touch 3 pages; the first one must have
    // been evicted.
    let cfg = Config::default().with_cache_frames(2).expect("cap");
    let mut p = Pager::memory(cfg).expect("construct");
    let a = p.alloc_page().expect("alloc");
    let b = p.alloc_page().expect("alloc");
    let c = p.alloc_page().expect("alloc");
    // Touch all three: this drives evictions through the cache.
    for &x in &[a, b, c] {
        let _ = p.read_page(x).expect("read");
    }
    // All three reads still succeed (the disk-level write-back kicked
    // in during eviction).
    assert!(p.read_page(a).is_ok());
    assert!(p.read_page(b).is_ok());
    assert!(p.read_page(c).is_ok());
}

#[test]
fn dirty_eviction_writes_back() {
    let cfg = Config::default().with_cache_frames(1).expect("cap");
    let mut p = Pager::memory(cfg).expect("construct");
    let a = p.alloc_page().expect("alloc");
    let b = p.alloc_page().expect("alloc"); // evicts the dirty `a` frame
    let mut data = Page::zeroed();
    data.as_bytes_mut()[100] = 0x77;
    p.write_page(a, &data).expect("write");
    // Reading `b` then `a` forces another eviction and exercises the
    // write-back path repeatedly.
    let _ = p.read_page(b).expect("read");
    let back = p.read_page(a).expect("read");
    assert_eq!(back.as_bytes()[100], 0x77);
}

#[test]
fn page_count_grows_only_when_freelist_empty() {
    let mut p = Pager::memory(Config::default()).expect("construct");
    let before = p.page_count();
    let a = p.alloc_page().expect("alloc");
    assert_eq!(p.page_count(), before + 1);
    p.free_page(a).expect("free");
    // Recycle from freelist — page_count must NOT grow.
    let _ = p.alloc_page().expect("realloc");
    assert_eq!(p.page_count(), before + 1);
}

// --- Property tests ---------------------------------------------------

// 10k randomised alloc/free sequences. After each operation we
// assert two invariants:
//
// 1. Every currently-allocated id is unique (no double-issue).
// 2. Every freed id is reusable: after enough alloc calls, the set of
//    issued ids matches the set we have tracked locally.
proptest! {
    #![proptest_config(ProptestConfig::with_cases(10_000))]

    #[test]
    fn alloc_free_sequence_is_consistent(
        ops in proptest::collection::vec(any::<bool>(), 0..32),
    ) {
        let mut p = Pager::memory(Config::default()).expect("construct");
        let mut live: HashSet<u64> = HashSet::new();
        let mut freed: Vec<u64> = Vec::new();

        for op in ops {
            if op || live.is_empty() {
                // alloc
                let id = p.alloc_page().expect("alloc").get();
                prop_assert!(live.insert(id), "double-issued id {id}");
                // If freelist had an id, the new id must come from there.
                if let Some(expected) = freed.last().copied() {
                    prop_assert_eq!(id, expected, "freelist must be LIFO");
                    freed.pop();
                }
            } else {
                // free a random live id
                let &victim = live.iter().next().expect("non-empty");
                prop_assert!(live.remove(&victim));
                freed.push(victim);
                p.free_page(PageId::new(victim).expect("non-zero")).expect("free");
            }
        }
    }
}

// --- Reopen round-trip property test ----------------------------------

proptest! {
    #![proptest_config(ProptestConfig::with_cases(64))]

    #[test]
    fn close_reopen_freelist_head_round_trips(
        n_alloc in 1usize..16,
        seed in any::<u8>(),
    ) {
        let dir = TempDir::new().expect("tempdir");
        let path = dir.path().join("rt.obj");
        let expected_head = {
            let mut p = open_file(&path, Config::default());
            let mut ids = Vec::with_capacity(n_alloc);
            for _ in 0..n_alloc {
                ids.push(p.alloc_page().expect("alloc"));
            }
            // Free a deterministic subset.
            let mut to_free: Vec<PageId> = ids
                .iter()
                .enumerate()
                .filter(|(i, _)| ((seed as usize).wrapping_add(*i)) & 1 == 0)
                .map(|(_, &x)| x)
                .collect();
            // Ensure at least one freed page so the head is non-zero.
            if to_free.is_empty() {
                to_free.push(ids[0]);
            }
            let mut last_freed = 0;
            for fid in &to_free {
                p.free_page(*fid).expect("free");
                last_freed = fid.get();
            }
            p.flush().expect("flush");
            prop_assert_eq!(p.freelist_head(), last_freed);
            last_freed
        };
        let p = open_file(&path, Config::default());
        prop_assert_eq!(p.freelist_head(), expected_head);
    }
}

// --- Corruption-detection tests (issue #6) ----------------------------

/// Build a file with `n_pages` written pages, each filled with a
/// distinctive byte pattern. Returns the path and the list of
/// `PageId`s written.
fn build_corruption_fixture(dir: &TempDir, n_pages: u64) -> (std::path::PathBuf, Vec<PageId>) {
    let path = dir.path().join("corruption.obj");
    let mut p = open_file(&path, Config::default());
    let cap = usize::try_from(n_pages).expect("n_pages fits in usize");
    let mut ids = Vec::with_capacity(cap);
    for i in 0..n_pages {
        let id = p.alloc_page().expect("alloc");
        let mut page = Page::zeroed();
        // Write a per-page pattern that the trailer must cover.
        for (j, b) in page.as_bytes_mut().iter_mut().enumerate().take(64) {
            let j = u64::try_from(j).expect("j < 64");
            let mixed = i.wrapping_mul(13).wrapping_add(j) & 0xFF;
            *b = u8::try_from(mixed).expect("masked to 0..256");
        }
        p.write_page(id, &page).expect("write");
        ids.push(id);
    }
    p.flush().expect("flush");
    drop(p);
    (path, ids)
}

#[test]
fn flipping_a_data_byte_is_detected_as_corruption() {
    let dir = TempDir::new().expect("tempdir");
    let (path, ids) = build_corruption_fixture(&dir, 1);
    // Flip a byte in the middle of page 1.
    let victim = ids[0];
    let offset = victim.byte_offset(0) + 100;
    {
        let mut f = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&path)
            .expect("reopen-rw");
        f.seek(SeekFrom::Start(offset)).expect("seek");
        let mut b = [0u8; 1];
        f.read_exact(&mut b).expect("read");
        b[0] ^= 0x55;
        f.seek(SeekFrom::Start(offset)).expect("seek-back");
        f.write_all(&b).expect("write");
        f.sync_all().expect("sync");
    }
    let mut p = open_file(&path, Config::default());
    match p.read_page(victim) {
        Err(Error::Corruption { page_id }) => assert_eq!(page_id, victim.get()),
        other => panic!("expected Corruption, got {other:?}"),
    }
}

#[test]
fn flipping_the_header_is_detected_at_open() {
    let dir = TempDir::new().expect("tempdir");
    let (path, _) = build_corruption_fixture(&dir, 1);
    {
        // Flip a byte in the page 0 reserved region (which is covered
        // by the header CRC).
        let mut f = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&path)
            .expect("reopen-rw");
        f.seek(SeekFrom::Start(80)).expect("seek");
        let mut b = [0u8; 1];
        f.read_exact(&mut b).expect("read");
        b[0] ^= 0xFF;
        f.seek(SeekFrom::Start(80)).expect("seek-back");
        f.write_all(&b).expect("write");
        f.sync_all().expect("sync");
    }
    match Pager::open(&path, Config::default()) {
        Err(Error::Corruption { page_id }) => assert_eq!(page_id, 0),
        other => panic!("expected Corruption on header CRC mismatch, got {other:?}"),
    }
}

proptest! {
    #![proptest_config(ProptestConfig {
        cases: 1_000,
        // Each iteration creates a temp file; allow a generous timeout
        // because filesystem ops can be slow on heavily loaded CI runners.
        max_shrink_iters: 4,
        .. ProptestConfig::default()
    })]

    /// For any randomly chosen (page, byte-offset), flipping that byte
    /// on disk must cause `read_page` to surface `Error::Corruption`.
    /// 1000 iterations per the issue acceptance criteria.
    #[test]
    fn byte_flip_anywhere_in_a_data_page_is_detected(
        n_pages in 1u64..4,
        page_idx in 0usize..4,
        byte_offset in 0u64..(PAGE_SIZE as u64),
    ) {
        let dir = TempDir::new().expect("tempdir");
        let (path, ids) = build_corruption_fixture(&dir, n_pages);
        let victim = ids[page_idx % ids.len()];
        let file_offset = victim.byte_offset(0) + byte_offset;
        {
            let mut f = OpenOptions::new()
                .read(true)
                .write(true)
                .open(&path)
                .expect("reopen-rw");
            f.seek(SeekFrom::Start(file_offset)).expect("seek");
            let mut b = [0u8; 1];
            f.read_exact(&mut b).expect("read");
            b[0] ^= 0x01;
            f.seek(SeekFrom::Start(file_offset)).expect("seek-back");
            f.write_all(&b).expect("write");
            f.sync_all().expect("sync");
        }
        let mut p = open_file(&path, Config::default());
        match p.read_page(victim) {
            Err(Error::Corruption { page_id }) => prop_assert_eq!(page_id, victim.get()),
            other => prop_assert!(false, "expected Corruption, got {other:?}"),
        }
    }
}

// Silence dead_code for PAGE_SIZE in test helpers.
const _: usize = PAGE_SIZE;

// --- WAL integration tests (issue #14) --------------------------------

#[test]
fn write_then_read_within_same_session_sees_uncommitted_data() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("wal.obj");
    let mut p = open_file(&path, Config::default());
    let a = p.alloc_page().expect("alloc");
    let mut page = Page::zeroed();
    page.as_bytes_mut()[0] = 0x77;
    p.write_page(a, &page).expect("write");
    // No commit yet — the read must STILL see the data (it's in the
    // txn buffer).
    let read = p.read_page(a).expect("read");
    assert_eq!(read.as_bytes()[0], 0x77);
}

#[test]
fn commit_drains_pending_into_view() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("wal.obj");
    let mut p = open_file(&path, Config::default());
    let a = p.alloc_page().expect("alloc");
    let mut page = Page::zeroed();
    page.as_bytes_mut()[100] = 0xAB;
    p.write_page(a, &page).expect("write");
    let lsn = p.commit().expect("commit");
    assert!(lsn >= Lsn::ONE, "commit must assign a positive LSN");
    // After commit, the page is still readable (now from the view).
    let read = p.read_page(a).expect("read");
    assert_eq!(read.as_bytes()[100], 0xAB);
}

#[test]
fn empty_commit_returns_zero_lsn() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("wal.obj");
    let mut p = open_file(&path, Config::default());
    let lsn = p.commit().expect("commit");
    assert_eq!(lsn, Lsn::ZERO);
}

#[test]
fn memory_pager_commit_is_noop() {
    let mut p = Pager::memory(Config::default()).expect("memory");
    let a = p.alloc_page().expect("alloc");
    let mut page = Page::zeroed();
    page.as_bytes_mut()[0] = 0x42;
    p.write_page(a, &page).expect("write");
    let lsn = p.commit().expect("memory commit");
    assert_eq!(lsn, Lsn::ZERO);
    let read = p.read_page(a).expect("read");
    assert_eq!(read.as_bytes()[0], 0x42);
}

#[test]
fn group_commit_assigns_consecutive_lsns() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("wal.obj");
    let mut p = open_file(&path, Config::default());
    let ids: Vec<PageId> = (0..4).map(|_| p.alloc_page().expect("alloc")).collect();
    for (i, &pid) in ids.iter().enumerate() {
        let mut page = Page::zeroed();
        page.as_bytes_mut()[0] = u8::try_from(i & 0xFF).expect("masked");
        p.write_page(pid, &page).expect("write");
    }
    let lsn = p.commit().expect("commit");
    // Four user page frames + one trailing page-0 header frame
    // (#64: `alloc_page` stages the header through the WAL).
    assert_eq!(
        lsn,
        Lsn::new(5),
        "four user frames + one header frame must produce LSNs 1..=5"
    );
    for (i, &pid) in ids.iter().enumerate() {
        let r = p.read_page(pid).expect("read");
        assert_eq!(r.as_bytes()[0], u8::try_from(i & 0xFF).expect("masked"));
    }
}

#[test]
fn commit_then_reopen_recovers_data_without_flush() {
    // This is the "crash after commit but before checkpoint" scenario
    // recovery (#15) handles: write_page, commit, drop without flush,
    // reopen — the data must be readable from the WAL.
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("recover.obj");
    let a_id = {
        let mut p = open_file(&path, Config::default());
        let a = p.alloc_page().expect("alloc");
        let mut page = Page::zeroed();
        page.as_bytes_mut()[0..4].copy_from_slice(b"PQRS");
        p.write_page(a, &page).expect("write");
        let _ = p.commit().expect("commit");
        // NO flush — simulate clean drop with commit-only durability.
        a.get()
    };
    let mut p = open_file(&path, Config::default());
    let a = id(a_id);
    let read = p.read_page(a).expect("read");
    assert_eq!(
        &read.as_bytes()[0..4],
        b"PQRS",
        "recovery must replay the committed write"
    );
}

#[test]
fn open_recovers_multiple_commits() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("recover_multi.obj");
    let (a_id, b_id) = {
        let mut p = open_file(&path, Config::default());
        let a = p.alloc_page().expect("alloc");
        let b = p.alloc_page().expect("alloc");
        // Txn 1: page a = 0xAA.
        let mut pa = Page::zeroed();
        pa.as_bytes_mut()[10] = 0xAA;
        p.write_page(a, &pa).expect("write a");
        let _ = p.commit().expect("commit 1");
        // Txn 2: page a = 0xCC (overwrite), page b = 0xBB.
        let mut pa2 = Page::zeroed();
        pa2.as_bytes_mut()[10] = 0xCC;
        let mut pb = Page::zeroed();
        pb.as_bytes_mut()[20] = 0xBB;
        p.write_page(a, &pa2).expect("write a2");
        p.write_page(b, &pb).expect("write b");
        let _ = p.commit().expect("commit 2");
        (a.get(), b.get())
    };
    let mut p = open_file(&path, Config::default());
    let ra = p.read_page(id(a_id)).expect("read a");
    assert_eq!(ra.as_bytes()[10], 0xCC, "later commit wins");
    let rb = p.read_page(id(b_id)).expect("read b");
    assert_eq!(rb.as_bytes()[20], 0xBB);
}

// --- #86 / #91: fresh-page durability via the WAL ---------------------

/// #91 fault test. Allocate many fresh pages in ONE committed txn,
/// write distinctive content, commit (so the bodies + `page_count` ride
/// the SINGLE WAL group-commit — NOT a checkpoint), drop WITHOUT a
/// checkpoint to model a crash right after the commit, then reopen and
/// read every committed page. None may surface `UnexpectedEof` /
/// `Corruption`: under #91 the fresh pages live only in the WAL until
/// checkpoint, and recovery replays them into the view so `page_count`
/// never outruns what the WAL can produce. The main file is NOT
/// extended at alloc — recovery (or the next checkpoint) heals the
/// length.
#[test]
fn batch_extension_covers_all_committed_pages_after_crash() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("batch_ext.obj");
    // Many allocations in one txn so the recovered view carries a long
    // run of fresh pages the main file never physically covered.
    let n_alloc = 37usize;
    let last = {
        let mut p = open_file(&path, Config::default());
        let mut ids = Vec::with_capacity(n_alloc);
        for k in 0..n_alloc {
            let a = p.alloc_page().expect("alloc");
            let mut page = Page::zeroed();
            // Distinctive per-page byte the trailer must cover.
            page.as_bytes_mut()[0] = u8::try_from(k & 0xFF).expect("masked");
            p.write_page(a, &page).expect("write");
            ids.push(a);
        }
        let _ = p.commit().expect("commit");
        // NO checkpoint / flush — model a crash immediately post-commit.
        ids.last().copied().expect("at least one alloc")
    };
    // Reopen: recovery replays the WAL; every committed page must read
    // back cleanly across the whole [1, page_count) range.
    let mut p = open_file(&path, Config::default());
    let pc = p.page_count();
    for raw in 1..pc {
        let read = p
            .read_page(id(raw))
            .unwrap_or_else(|e| panic!("page {raw} unreadable after crash: {e:?}"));
        let expected = u8::try_from((raw - 1) & 0xFF).expect("masked");
        assert_eq!(
            read.as_bytes()[0],
            expected,
            "page {raw} content lost across batched extension + crash"
        );
    }
    assert_eq!(
        pc,
        u64::try_from(n_alloc + 1).expect("fits"),
        "page_count must equal allocated pages plus page 0"
    );
    // The just-allocated last id is in range and reads clean.
    assert!(p.read_page(last).is_ok());
}

/// #91 fault test — the small-transaction case. Allocate a few pages,
/// write content, commit, drop WITHOUT a checkpoint (crash right after
/// commit), reopen. Every reachable page `[1, page_count)` reads clean
/// (no `UnexpectedEof`) from the recovered WAL view, and `page_count`
/// reflects exactly the committed allocations — proving recovery heals
/// the (un-extended) main-file length without leaking any phantom slot
/// into the recovered authority.
#[test]
fn partial_batch_commit_crash_reopen_reads_clean() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("partial_batch.obj");
    let m_alloc = 5usize;
    {
        let mut p = open_file(&path, Config::default());
        for k in 0..m_alloc {
            let a = p.alloc_page().expect("alloc");
            let mut page = Page::zeroed();
            page.as_bytes_mut()[0] = u8::try_from(0xA0 + k).expect("masked");
            p.write_page(a, &page).expect("write");
        }
        let _ = p.commit().expect("commit");
        // Drop without checkpoint — crash right after the commit.
    }
    let mut p = open_file(&path, Config::default());
    let pc = p.page_count();
    // page_count is exactly the committed allocs + page 0 — the K-M
    // grown-but-uncommitted slack slots did NOT advance the recovered
    // authority.
    assert_eq!(pc, u64::try_from(m_alloc + 1).expect("fits"));
    for raw in 1..pc {
        let read = p
            .read_page(id(raw))
            .unwrap_or_else(|e| panic!("reachable page {raw} unreadable: {e:?}"));
        let expected = u8::try_from(0xA0_u64 + (raw - 1)).expect("masked");
        assert_eq!(read.as_bytes()[0], expected, "page {raw} content lost");
    }
}

#[test]
fn uncommitted_writes_are_not_recovered() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("uncommitted.obj");
    // First open: commit one page so a known id exists in the file
    // header (post-#64 the page-count update rides the WAL and only
    // becomes durable on `commit`). Then on the same handle, stage
    // an uncommitted write to that page and drop without commit.
    let a_id = {
        let mut p = open_file(&path, Config::default());
        let a = p.alloc_page().expect("alloc");
        let mut zero = Page::zeroed();
        zero.as_bytes_mut()[0] = 0x00;
        p.write_page(a, &zero).expect("write zeros");
        let _ = p.commit().expect("commit");
        // Stage an uncommitted overwrite to the same page.
        let mut page = Page::zeroed();
        page.as_bytes_mut()[0] = 0x55;
        p.write_page(a, &page).expect("write");
        // NO second commit — the 0x55 byte exists only in pending_txn.
        a.get()
    };
    let mut p = open_file(&path, Config::default());
    let read = p.read_page(id(a_id)).expect("read");
    assert_ne!(
        read.as_bytes()[0],
        0x55,
        "uncommitted writes MUST NOT survive a drop / reopen"
    );
}

#[test]
fn wal_sidecar_is_created_on_open() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("wal.obj");
    {
        let _p = open_file(&path, Config::default());
    }
    let wal_path = crate::pager::wal_path_for(&path);
    assert!(wal_path.exists(), "WAL sidecar must be created at open");
}

// --- Checkpoint + close tests (issue #16) -----------------------------

#[test]
fn close_removes_wal_sidecar() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("clean.obj");
    let a_id = {
        let mut p = open_file(&path, Config::default());
        let a = p.alloc_page().expect("alloc");
        let mut page = Page::zeroed();
        page.as_bytes_mut()[0] = 0x33;
        p.write_page(a, &page).expect("write");
        let _ = p.commit().expect("commit");
        let aid = a.get();
        p.close().expect("close");
        aid
    };
    let wal_path = crate::pager::wal_path_for(&path);
    assert!(
        !wal_path.exists(),
        "close MUST remove the WAL sidecar (design.md guarantee)"
    );
    // The data must still be readable after a fresh open.
    let mut p = open_file(&path, Config::default());
    let r = p.read_page(id(a_id)).expect("read");
    assert_eq!(r.as_bytes()[0], 0x33);
}

#[test]
fn checkpoint_is_idempotent() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("idem.obj");
    let mut p = open_file(&path, Config::default());
    let a = p.alloc_page().expect("alloc");
    let mut page = Page::zeroed();
    page.as_bytes_mut()[0] = 0x44;
    p.write_page(a, &page).expect("write");
    let _ = p.commit().expect("commit");
    p.checkpoint().expect("checkpoint 1");
    // Second checkpoint must be a no-op.
    p.checkpoint().expect("checkpoint 2");
    let r = p.read_page(a).expect("read");
    assert_eq!(r.as_bytes()[0], 0x44);
}

#[test]
fn auto_checkpoint_fires_at_threshold() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("auto_ckpt.obj");
    let cfg = Config::default()
        .with_checkpoint_threshold(3)
        .with_cache_frames(8)
        .expect("cache");
    let mut p = open_file(&path, cfg);
    let ids: Vec<PageId> = (0..5).map(|_| p.alloc_page().expect("alloc")).collect();
    // Five single-page commits: the third commit should auto-trigger
    // a checkpoint (committed_frames reaches 3), draining the WAL
    // view back into the main file.
    for (i, &pid) in ids.iter().enumerate() {
        let mut page = Page::zeroed();
        page.as_bytes_mut()[0] = u8::try_from(i).expect("masked");
        p.write_page(pid, &page).expect("write");
        let _ = p.commit().expect("commit");
    }
    // Verify all values are readable after the checkpoint storm.
    for (i, &pid) in ids.iter().enumerate() {
        let r = p.read_page(pid).expect("read");
        assert_eq!(r.as_bytes()[0], u8::try_from(i).expect("masked"));
    }
}

#[test]
fn open_after_clean_close_has_no_wal_file() {
    // A second open after close should NOT see a WAL file on disk.
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("noreopen.obj");
    {
        let p = open_file(&path, Config::default());
        p.close().expect("close");
    }
    let wal_path = crate::pager::wal_path_for(&path);
    assert!(!wal_path.exists());
    // After reopen, a new WAL is created (Pager::open always opens
    // one), so the absence guarantee only holds in the *closed* state.
    let p = open_file(&path, Config::default());
    p.close().expect("close 2");
    assert!(!wal_path.exists(), "second close must again leave no WAL");
}

#[test]
fn salt_rotates_on_checkpoint() {
    let dir = TempDir::new().expect("tempdir");
    let path = dir.path().join("salt.obj");
    let mut p = open_file(&path, Config::default());
    let a = p.alloc_page().expect("alloc");
    let mut page = Page::zeroed();
    page.as_bytes_mut()[0] = 0x77;
    p.write_page(a, &page).expect("write");
    let _ = p.commit().expect("commit");
    p.checkpoint().expect("checkpoint");
    // After checkpoint, the WAL view is empty; the header's wal_salt
    // must have rotated (no longer be the original).
    // We re-read the header from disk to verify.
    drop(p);
    let mut p2 = open_file(&path, Config::default());
    // Data still readable.
    let r = p2.read_page(a).expect("read");
    assert_eq!(r.as_bytes()[0], 0x77);
}

// --------------------------------------------------------------------
// MVCC reader-snapshot tests (M6 issue #45).
// --------------------------------------------------------------------

mod snapshot {
    use super::*;
    use crate::pager::PageHandle;
    use crate::platform::FileHandle;
    use std::sync::{Arc, Mutex};
    use std::thread;

    fn stamp(byte: u8) -> Page {
        let mut p = Page::zeroed();
        p.as_bytes_mut()[0] = byte;
        p
    }

    #[test]
    fn snapshot_sees_committed_view_at_pin_time() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap.obj");
        let mut p = open_file(&path, Config::default());
        let a = p.alloc_page().expect("alloc");
        p.write_page(a, &stamp(0x11)).expect("write");
        let _ = p.commit().expect("commit 1");

        // Take a snapshot here.  The frozen view captures byte 0x11.
        let snap = p.reader_snapshot().expect("snap");

        // Writer commits a new version — invisible to the snapshot.
        p.write_page(a, &stamp(0x22)).expect("write 2");
        let _ = p.commit().expect("commit 2");

        // Snapshot still sees 0x11.
        let page = snap.read_page(&p, a).expect("snap read");
        assert_eq!(page.as_bytes()[0], 0x11, "snapshot must see frozen view");
    }

    #[test]
    fn fresh_snapshot_sees_latest_writer_commit() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap.obj");
        let mut p = open_file(&path, Config::default());
        let a = p.alloc_page().expect("alloc");
        p.write_page(a, &stamp(0x11)).expect("write");
        let _ = p.commit().expect("commit 1");
        p.write_page(a, &stamp(0x22)).expect("write 2");
        let _ = p.commit().expect("commit 2");

        let snap = p.reader_snapshot().expect("snap");
        let page = snap.read_page(&p, a).expect("snap read");
        assert_eq!(page.as_bytes()[0], 0x22, "fresh snapshot must see latest");
    }

    #[test]
    fn pending_writes_invisible_to_snapshot() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap.obj");
        let mut p = open_file(&path, Config::default());
        let a = p.alloc_page().expect("alloc");
        p.write_page(a, &stamp(0x11)).expect("write");
        let _ = p.commit().expect("commit");

        // Stage a pending write (no commit).
        p.write_page(a, &stamp(0xFF)).expect("pending");

        let snap = p.reader_snapshot().expect("snap");
        let page = snap.read_page(&p, a).expect("snap read");
        assert_eq!(
            page.as_bytes()[0],
            0x11,
            "pending writes must NOT be visible to a snapshot",
        );
    }

    #[test]
    fn checkpoint_skipped_while_snapshot_pins_old_lsn() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap.obj");
        let cfg = Config::default().with_checkpoint_threshold(u64::MAX); // disable auto-checkpoint
        let mut p = open_file(&path, cfg);
        let a = p.alloc_page().expect("alloc");
        p.write_page(a, &stamp(0x11)).expect("write");
        let _ = p.commit().expect("commit 1");

        // Take a snapshot pinning LSN ~1.
        let _snap = p.reader_snapshot().expect("snap");

        // Writer commits more frames after the pin.
        p.write_page(a, &stamp(0x22)).expect("write 2");
        let _ = p.commit().expect("commit 2");

        // Explicit checkpoint must be a no-op while the snapshot is live.
        let frames_before = p.wal.as_ref().map_or(0, |s| s.wal.committed_frames());
        p.checkpoint().expect("checkpoint");
        let frames_after = p.wal.as_ref().map_or(0, |s| s.wal.committed_frames());
        assert_eq!(
            frames_before, frames_after,
            "checkpoint must defer while snapshot pins old LSN"
        );
    }

    #[test]
    fn dropping_snapshot_allows_checkpoint() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap.obj");
        let cfg = Config::default().with_checkpoint_threshold(u64::MAX);
        let mut p = open_file(&path, cfg);
        let a = p.alloc_page().expect("alloc");
        p.write_page(a, &stamp(0x11)).expect("write");
        let _ = p.commit().expect("commit");

        let snap = p.reader_snapshot().expect("snap");
        assert_eq!(p.live_snapshot_count(), 1);
        drop(snap);
        assert_eq!(p.live_snapshot_count(), 0);
        p.checkpoint()
            .expect("checkpoint must run cleanly after snapshot drop");
    }

    /// Reader thread body: loop `iters` times reading page `a` and
    /// asserting it equals the frozen byte the snapshot was pinned
    /// at (which is `0`).  Borrows `pager` and `snap` by reference so
    /// the thread closure that calls this owns the Arc/snapshot.
    fn reader_loop(
        r: usize,
        pager: &Arc<Mutex<Pager<FileHandle>>>,
        snap: &crate::pager::ReaderSnapshot<FileHandle>,
        a: PageId,
        iters: u32,
    ) {
        for i in 0..iters {
            let p = pager.lock().expect("lock");
            let page = snap.read_page(&p, a).expect("snap read");
            assert_eq!(
                page.as_bytes()[0],
                0,
                "reader {r} iter {i}: snapshot must see frozen byte 0",
            );
            drop(p);
            std::thread::yield_now();
        }
    }

    /// Writer thread body: commit `count` distinct versions, each
    /// stamping a new byte into page `a`.
    fn writer_loop(pager: &Arc<Mutex<Pager<FileHandle>>>, a: PageId, count: u32) {
        for v in 1u32..=count {
            let mut p = pager.lock().expect("lock");
            let byte = u8::try_from((v % 250) + 1).expect("byte fits");
            p.write_page(a, &stamp(byte)).expect("write");
            let _ = p.commit().expect("commit");
            drop(p);
            std::thread::yield_now();
        }
    }

    /// Multi-thread concurrency test (no Miri yet — that's #49).
    /// 8 reader threads each hold a snapshot while a writer commits
    /// 200 page updates concurrently.  Every snapshot read must see
    /// a page consistent with its pinned LSN.
    #[test]
    fn many_readers_one_writer_consistent_view() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap.obj");
        let cfg = Config::default().with_checkpoint_threshold(u64::MAX);
        let mut opened = Pager::<FileHandle>::open(&path, cfg).expect("open");
        opened.begin_txn();
        let pager = Arc::new(Mutex::new(opened));
        let a = {
            let mut p = pager.lock().expect("lock");
            let a = p.alloc_page().expect("alloc");
            p.write_page(a, &stamp(0)).expect("write");
            let _ = p.commit().expect("commit");
            a
        };
        let n_readers = 8usize;
        thread::scope(|scope| {
            let mut handles = Vec::with_capacity(n_readers);
            for r in 0..n_readers {
                let pager = Arc::clone(&pager);
                let snap = {
                    let mut p = pager.lock().expect("lock");
                    p.reader_snapshot().expect("snap")
                };
                handles.push(scope.spawn(move || reader_loop(r, &pager, &snap, a, 1000)));
            }
            let writer_pager = Arc::clone(&pager);
            let writer = scope.spawn(move || writer_loop(&writer_pager, a, 200));
            writer.join().expect("writer join");
            for h in handles {
                h.join().expect("reader join");
            }
        });
        let fresh = {
            let mut p = pager.lock().expect("lock");
            p.reader_snapshot().expect("fresh snap")
        };
        let p = pager.lock().expect("lock");
        let page = fresh.read_page(&p, a).expect("fresh read");
        assert_ne!(
            page.as_bytes()[0],
            0,
            "post-write fresh snapshot must NOT see the original byte 0",
        );
    }

    #[test]
    fn snapshot_id_is_monotonic() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap.obj");
        let mut p = open_file(&path, Config::default());
        let s1 = p.reader_snapshot().expect("s1");
        let s2 = p.reader_snapshot().expect("s2");
        assert!(s2.id().get() > s1.id().get());
    }

    #[test]
    fn min_pinned_lsn_tracks_lowest_live_reader() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap.obj");
        let cfg = Config::default().with_checkpoint_threshold(u64::MAX);
        let mut p = open_file(&path, cfg);
        let a = p.alloc_page().expect("alloc");
        p.write_page(a, &stamp(0x11)).expect("write");
        let _ = p.commit().expect("commit 1");

        let s1 = p.reader_snapshot().expect("s1");
        let s1_lsn = s1.pinned_lsn();

        p.write_page(a, &stamp(0x22)).expect("write 2");
        let _ = p.commit().expect("commit 2");

        let s2 = p.reader_snapshot().expect("s2");
        assert!(s2.pinned_lsn() > s1.pinned_lsn());

        assert_eq!(p.min_pinned_lsn(), Some(s1_lsn));
        drop(s1);
        assert_eq!(p.min_pinned_lsn(), Some(s2.pinned_lsn()));
        drop(s2);
        assert_eq!(p.min_pinned_lsn(), None);
    }

    #[test]
    fn memory_pager_snapshot_works_with_empty_view() {
        let mut p = Pager::memory(Config::default()).expect("mem pager");
        let snap = p.reader_snapshot().expect("snap");
        assert_eq!(snap.pinned_lsn(), Lsn::ZERO, "memory pager has no WAL");
        assert_eq!(p.live_snapshot_count(), 1);
    }

    /// #80 regression: switching the committed view to `Arc<Page>`
    /// must NOT weaken snapshot isolation. A page first written and
    /// committed AFTER a snapshot is pinned must be invisible to that
    /// snapshot — the snapshot's frozen view is its OWN cloned map, so
    /// the writer's later `view.insert` of a fresh `Arc` cannot appear
    /// in it. This guards the isolation the deep clone provided before
    /// the Arc-share change. (Disable auto-checkpoint so the writer's
    /// post-pin commit stays in the WAL view and the only thing
    /// keeping the snapshot consistent is the per-snapshot map clone.)
    #[test]
    fn snapshot_does_not_observe_page_committed_after_pin() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap.obj");
        let cfg = Config::default().with_checkpoint_threshold(u64::MAX);
        let mut p = open_file(&path, cfg);

        // One page committed BEFORE the pin: the snapshot must see it.
        let a = p.alloc_page().expect("alloc a");
        p.write_page(a, &stamp(0x11)).expect("write a");
        let _ = p.commit().expect("commit a");

        // Pin a snapshot. Its frozen view is captured here.
        let snap = p.reader_snapshot().expect("snap");

        // Allocate + commit a BRAND-NEW page id AFTER the pin.
        let b = p.alloc_page().expect("alloc b");
        p.write_page(b, &stamp(0x22)).expect("write b");
        let _ = p.commit().expect("commit b");
        assert_ne!(a, b, "b must be a distinct page id");

        // The pre-pin page is still visible through the snapshot.
        let seen_a = snap.read_page(&p, a).expect("snap read a");
        assert_eq!(
            seen_a.as_bytes()[0],
            0x11,
            "snapshot must still observe the page committed before its pin",
        );

        // The post-pin page b is NOT in the snapshot's frozen view:
        // it was never present when the snapshot cloned `view`. The
        // frozen-view lookup must miss, falling through to the main
        // file, where b was never checkpointed -> all-zero body
        // (definitely NOT the 0x22 the writer committed post-pin).
        let frozen_has_b = snap.frozen_pages().any(|(id, _)| id == b);
        assert!(
            !frozen_has_b,
            "post-pin commit must NOT appear in the snapshot's frozen view",
        );
        let seen_b = snap.read_page(&p, b).expect("snap read b");
        assert_ne!(
            seen_b.as_bytes()[0],
            0x22,
            "snapshot must NOT observe the body committed after its pin",
        );
    }

    /// #81: a frozen-view hit returns `PageHandle::Shared` (an
    /// `Arc::clone`, no 4 KiB body copy), and `into_page` on that arm
    /// materialises the exact frozen body.
    #[test]
    fn frozen_view_hit_is_shared_and_into_page_round_trips() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap.obj");
        let mut p = open_file(&path, Config::default());
        let a = p.alloc_page().expect("alloc");
        p.write_page(a, &stamp(0x11)).expect("write");
        let _ = p.commit().expect("commit 1");

        // Pin: the frozen view captures the 0x11 body for `a`.
        let snap = p.reader_snapshot().expect("snap");
        // Writer commits a new version — invisible to the snapshot.
        p.write_page(a, &stamp(0x22)).expect("write 2");
        let _ = p.commit().expect("commit 2");

        let handle = snap.read_page(&p, a).expect("snap read");
        assert!(
            matches!(handle, PageHandle::Shared(_)),
            "frozen-view hit must return PageHandle::Shared (refcount bump, no body clone)",
        );
        // as_bytes borrows the shared body without copying.
        assert_eq!(
            handle.as_bytes()[0],
            0x11,
            "shared arm must see frozen body"
        );
        // into_page materialises the SAME frozen body (the only place
        // the Shared arm pays a clone).
        let owned = handle.into_page();
        assert_eq!(
            owned.as_bytes()[0],
            0x11,
            "into_page on the Shared arm must round-trip the frozen body",
        );
    }

    /// #81: a frozen-view MISS returns `PageHandle::Owned` via the
    /// checksum-verifying disk read path. First confirm a clean
    /// main-file page reads back as `Owned` with the right body; then
    /// corrupt that page on disk, reopen, and confirm a snapshot read
    /// surfaces `Error::Corruption` — i.e. the `Owned` arm still
    /// verifies integrity (`read_through` -> `page_trailer_valid`).
    #[test]
    fn frozen_view_miss_owned_arm_still_checksum_verifies() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap.obj");
        let a;
        {
            let mut p = open_file(&path, Config::default());
            a = p.alloc_page().expect("alloc");
            p.write_page(a, &stamp(0x11)).expect("write a");
            let _ = p.commit().expect("commit a");
            // Checkpoint so `a`'s bytes live in the MAIN file, then
            // drop the pager so a fresh open has an empty WAL view.
            p.checkpoint().expect("checkpoint");
        }

        // Clean reopen: snapshot's frozen view is empty, so `a` is a
        // frozen-view MISS served via the disk path -> Owned arm.
        {
            let mut p = open_file(&path, Config::default());
            let snap = p.reader_snapshot().expect("snap");
            assert!(
                !snap.frozen_pages().any(|(id, _)| id == a),
                "fresh-open snapshot must not hold `a` in its frozen view",
            );
            let handle = snap.read_page(&p, a).expect("clean owned read");
            assert!(
                matches!(handle, PageHandle::Owned(_)),
                "frozen-view miss must return PageHandle::Owned via the disk path",
            );
            assert_eq!(
                handle.as_bytes()[0],
                0x11,
                "owned arm must see the main-file body",
            );
        }

        // Flip a byte in `a`'s on-disk body (feature_flags == 0 for a
        // default-config file), then reopen and read through a fresh
        // snapshot: the Owned arm's read_through must reject it.
        let file_offset = a.byte_offset(0) + 16;
        {
            let mut f = OpenOptions::new()
                .read(true)
                .write(true)
                .open(&path)
                .expect("reopen-rw");
            f.seek(SeekFrom::Start(file_offset)).expect("seek");
            let mut b = [0u8; 1];
            f.read_exact(&mut b).expect("read");
            b[0] ^= 0x01;
            f.seek(SeekFrom::Start(file_offset)).expect("seek-back");
            f.write_all(&b).expect("write");
            f.sync_all().expect("sync");
        }
        let mut p = open_file(&path, Config::default());
        let snap = p.reader_snapshot().expect("snap2");
        match snap.read_page(&p, a) {
            Err(Error::Corruption { page_id }) => assert_eq!(page_id, a.get()),
            other => panic!("Owned-arm disk miss must checksum-verify; got {other:?}"),
        }
    }

    /// #91 MVCC: a snapshot taken BEFORE a growing commit must not
    /// observe the fresh pages — even though the writer advanced
    /// `page_count` and the fresh body never touched the main file. The
    /// snapshot read must NOT `UnexpectedEof` (the fresh slot is past the
    /// physical high-water) and must return the page's pre-existence
    /// state (a zeroed body), not the writer's post-pin content.
    #[test]
    fn snapshot_before_growing_commit_does_not_observe_fresh_pages() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap_before.obj");
        // checkpoint_threshold = MAX so the growing commit is NOT
        // auto-checkpointed: the fresh page stays WAL-resident, the main
        // file stays short — exactly the #91 window.
        let cfg = Config::default().with_checkpoint_threshold(u64::MAX);
        let mut p = open_file(&path, cfg);
        // One pre-existing committed page so the pin has a non-trivial
        // view.
        let a = p.alloc_page().expect("alloc a");
        p.write_page(a, &stamp(0x11)).expect("write a");
        let _ = p.commit().expect("commit a");
        p.checkpoint().expect("checkpoint a"); // `a` now on the main file

        // Pin BEFORE the growing commit.
        let snap = p.reader_snapshot().expect("snap");

        // Growing commit: allocate + write a brand-new page AFTER the pin.
        let b = p.alloc_page().expect("alloc b");
        p.write_page(b, &stamp(0x22)).expect("write b");
        let _ = p.commit().expect("commit b");
        assert_ne!(a, b);

        // `b` is past the physical high-water (never checkpointed) and is
        // NOT in the snapshot's frozen view. The snapshot must resolve it
        // to a zeroed body — never `UnexpectedEof`, never 0x22.
        assert!(!snap.frozen_pages().any(|(idx, _)| idx == b));
        let seen_b = snap.read_page(&p, b).expect("snap read b must not EOF");
        assert_ne!(
            seen_b.as_bytes()[0],
            0x22,
            "snapshot-before must NOT observe the post-pin fresh page",
        );
        assert_eq!(
            seen_b.as_bytes()[0],
            0x00,
            "post-pin fresh page resolves to its pre-existence (zeroed) state",
        );
        // The pre-pin page is still observable.
        let seen_a = snap.read_page(&p, a).expect("snap read a");
        assert_eq!(seen_a.as_bytes()[0], 0x11);
    }

    /// #91 MVCC: a snapshot taken AFTER a growing commit reads the fresh
    /// page from its `frozen_view` — WITHOUT touching the main file,
    /// which is still too short to hold the page. This is the positive
    /// half of the isolation contract: the fresh body is visible to a
    /// post-commit reader purely via the WAL view.
    #[test]
    fn snapshot_after_growing_commit_reads_fresh_from_frozen_view() {
        let dir = TempDir::new().expect("tmp");
        let path = dir.path().join("snap_after.obj");
        let cfg = Config::default().with_checkpoint_threshold(u64::MAX);
        let mut p = open_file(&path, cfg);
        // Growing commit, NOT checkpointed: `b` lives only in the WAL.
        let b = p.alloc_page().expect("alloc b");
        p.write_page(b, &stamp(0x77)).expect("write b");
        let _ = p.commit().expect("commit b");
        // The main file is physically too short to hold `b`.
        let physical = p.main_physical_page_count().expect("physical");
        assert!(
            b.get() >= physical,
            "fresh page must be beyond the physical high-water before checkpoint \
             (b={}, physical={physical})",
            b.get(),
        );

        // Snapshot AFTER the commit: its frozen view carries `b`.
        let snap = p.reader_snapshot().expect("snap");
        assert!(
            snap.frozen_pages().any(|(idx, _)| idx == b),
            "snapshot-after must capture the fresh page in its frozen view",
        );
        let handle = snap.read_page(&p, b).expect("snap read b");
        assert!(
            matches!(handle, PageHandle::Shared(_)),
            "fresh page must be served from the frozen view (Shared), not the \
             main file (Owned) — the file does not hold it yet",
        );
        assert_eq!(handle.as_bytes()[0], 0x77, "frozen view returns the body");
    }
}

// --------------------------------------------------------------------
// Issue #31: zeroize-on-drop for in-memory key material.
// --------------------------------------------------------------------
#[cfg(feature = "encryption")]
mod zeroize_key_material {
    use super::*;
    use crate::pager::{wrap_master_key, MasterKeyBytes};

    /// Compile-time guarantee: under the `encryption` feature the
    /// master-key storage type wipes its bytes on drop. If a future
    /// change swaps `MasterKeyBytes` back to a bare `[u8; 32]` (which
    /// is not `ZeroizeOnDrop`) this stops compiling.
    #[test]
    fn master_key_bytes_is_zeroize_on_drop() {
        fn assert_zeroize_on_drop<T: zeroize::ZeroizeOnDrop>() {}
        assert_zeroize_on_drop::<MasterKeyBytes>();
    }

    /// The stored key round-trips through the public builder and is
    /// readable internally as a plain `&[u8; 32]` (the `Zeroizing`
    /// wrapper is transparent to the open/derive call sites).
    #[test]
    fn key_round_trips_through_config_storage() {
        let raw = [0x5Au8; 32];
        let cfg = Config::default().with_encryption_key(Some(raw));
        let stored = cfg.master_key().expect("key present");
        assert_eq!(stored, &raw, "stored key must equal the supplied key");

        // Clearing the key drops the `Zeroizing` wrapper (wiping it)
        // and leaves `None`.
        let cleared = cfg.with_encryption_key(None);
        assert!(cleared.master_key().is_none(), "key cleared");
    }

    /// `wrap_master_key` produces a wrapper that derefs back to the
    /// original bytes — sanity check the wrap/unwrap symmetry the
    /// crypto hot path relies on.
    #[test]
    fn wrap_master_key_preserves_bytes() {
        let raw = [0xC3u8; 32];
        let wrapped: MasterKeyBytes = wrap_master_key(raw);
        let view: &[u8; 32] = &wrapped;
        assert_eq!(view, &raw);
    }
}