//! Ports of log subsystem tests to noxu-log.
//!
//! Covers correctness properties from the following test files:
//!
//! - `LogEntryTest.java` — entry type lookup and type_num round-trips
//! - `LoggableTest.java` — write_to_log/read_from_log round-trips; log_size() matches actual bytes
//! - `FSyncManagerTest.java` — coalescing, grouping, fsync vs flush-only
//! - `LogManagerTest.java` — write → flush → read; LSN ordering; bad checksum detected on read
//! - `LastFileReaderTest.java` / `FileReaderTest.java` — forward scan, end-of-log, trailing junk
//! - `INFileReaderTest.java` / `LNFileReaderTest.java` — entry-type filtering via LogFileReader

// ============================================================================
// Shared imports
// ============================================================================

use bytes::BytesMut;
use hashbrown::HashMap;
use noxu_log::{
    FileManager, LogFileReader, LogManager,
    checksum::ChecksumValidator,
    entry::commit_abort_entry::TxnEndEntry,
    entry::empty_log_entry::EmptyLogEntry,
    entry::file_header_entry::{FileHeader, FileHeaderEntry},
    entry::in_log_entry::InLogEntry,
    entry::ln_log_entry::LnLogEntry,
    entry::restore_required::{FailureType, RestoreRequired},
    entry::trace_log_entry::TraceLogEntry,
    entry_header::{
        CHECKSUM_BYTES, LogEntryHeader, MAX_HEADER_SIZE, MIN_HEADER_SIZE,
    },
    entry_type::LogEntryType,
    provisional::Provisional,
};
use noxu_util::lsn::{Lsn, NULL_LSN};
use noxu_util::vlsn::{NULL_VLSN, Vlsn};
use std::sync::Arc;
use tempfile::TempDir;

// ============================================================================
// Helpers
// ============================================================================

/// Create a FileManager + LogManager pair in a fresh temp directory.
fn make_managers(dir: &TempDir) -> (Arc<FileManager>, LogManager) {
    let fm =
        Arc::new(FileManager::new(dir.path(), false, 10_000_000, 100).unwrap());
    let lm = LogManager::new(Arc::clone(&fm), 3, 1_048_576, 4096);
    (fm, lm)
}

/// Create managers with a very small max file size to trigger flips quickly.
fn make_managers_small_file(
    dir: &TempDir,
    max_file_size: u64,
) -> (Arc<FileManager>, LogManager) {
    let fm = Arc::new(
        FileManager::new(dir.path(), false, max_file_size, 100).unwrap(),
    );
    let lm = LogManager::new(Arc::clone(&fm), 3, 2_097_152, 4096);
    (fm, lm)
}

// ============================================================================
// LogEntryTest — entry type catalog
// ============================================================================

///
/// Every defined LogEntryType must have a stable type_num, and
/// from_type_num(type_num) must recover the original variant.
#[test]
fn test_entry_type_num_roundtrip() {
    // Probe the full 0..=255 range; for every number that maps to a valid
    // LogEntryType the round-trip must hold.
    let mut found = 0usize;
    for n in 0u8..=255 {
        if let Some(entry_type) = LogEntryType::from_type_num(n) {
            assert_eq!(
                entry_type.type_num(),
                n,
                "type_num round-trip failed for type_num {}",
                n
            );
            found += 1;
        } else {
            assert!(
                !LogEntryType::is_valid(n),
                "is_valid({}) disagrees with from_type_num({})",
                n,
                n
            );
        }
    }
    // Ensure we have a reasonable number of entry types.
    assert!(
        found > 10,
        "Expected at least 10 log entry types; found {}",
        found
    );
}

/// Spot-check key types.
#[test]
fn test_entry_type_spot_checks() {
    // File metadata
    assert_eq!(LogEntryType::FileHeader.type_num(), 1);
    assert_eq!(LogEntryType::from_type_num(1), Some(LogEntryType::FileHeader));

    // Tree nodes
    assert_eq!(LogEntryType::IN.type_num(), 2);
    assert_eq!(LogEntryType::BIN.type_num(), 3);
    assert_eq!(LogEntryType::BINDelta.type_num(), 4);

    // Transaction
    assert_eq!(LogEntryType::TxnCommit.type_num(), 30);
    assert_eq!(LogEntryType::TxnAbort.type_num(), 31);

    // Checkpoint
    assert_eq!(LogEntryType::CkptStart.type_num(), 40);
    assert_eq!(LogEntryType::CkptEnd.type_num(), 41);

    // Trace
    assert_eq!(LogEntryType::Trace.type_num(), 60);
}

/// FindType returns correct variant.
#[test]
fn test_entry_type_find_type() {
    let in_type = LogEntryType::from_type_num(LogEntryType::IN.type_num());
    assert_eq!(in_type, Some(LogEntryType::IN));

    let trace_type =
        LogEntryType::from_type_num(LogEntryType::Trace.type_num());
    assert_eq!(trace_type, Some(LogEntryType::Trace));

    // Unknown type
    assert_eq!(LogEntryType::from_type_num(255), None);
    assert_eq!(LogEntryType::from_type_num(0), None);
}

/// Flags (transactional, replication, sync point).
#[test]
fn test_entry_type_flags() {
    // Transactional
    assert!(LogEntryType::TxnCommit.is_transactional());
    assert!(LogEntryType::TxnAbort.is_transactional());
    assert!(LogEntryType::InsertLNTxn.is_transactional());
    assert!(!LogEntryType::InsertLN.is_transactional());
    assert!(!LogEntryType::BIN.is_transactional());

    // Replication
    assert!(LogEntryType::InsertLNTxn.is_replication_possible());
    assert!(LogEntryType::TxnCommit.is_replication_possible());
    assert!(!LogEntryType::IN.is_replication_possible());

    // Sync point
    assert!(LogEntryType::TxnCommit.is_sync_point());
    assert!(LogEntryType::TxnAbort.is_sync_point());
    assert!(LogEntryType::Matchpoint.is_sync_point());
    assert!(!LogEntryType::BIN.is_sync_point());

    // Marshall inside latch
    assert!(LogEntryType::MapLN.marshall_inside_latch());
    assert!(LogEntryType::TxnCommit.marshall_inside_latch());
    assert!(!LogEntryType::InsertLNTxn.marshall_inside_latch());
    assert!(!LogEntryType::IN.marshall_inside_latch());

    // User LN
    assert!(LogEntryType::InsertLN.is_user_ln_type());
    assert!(LogEntryType::DeleteLNTxn.is_user_ln_type());
    assert!(!LogEntryType::MapLN.is_user_ln_type());
    assert!(!LogEntryType::IN.is_user_ln_type());
}

// ============================================================================
// LoggableTest — serialization round-trips for all entry types
//
// Key invariant (from `LoggableTest.writeAndRead`):
//   1. write_to_log() produces exactly log_size() bytes.
//   2. read_from_log(bytes) returns an equivalent object.
//   3. The re-read object also reports the same log_size().
// ============================================================================

/// TraceLogEntry round-trip.
#[test]
fn test_loggable_trace_log_entry_roundtrip() {
    let owned: Vec<String> = vec![
        "Hello there".to_string(),
        "".to_string(),
        "a".repeat(200),
        "Unicode: \u{00e9}\u{00e0}\u{00fc}".to_string(),
    ];

    for msg in &owned {
        let orig = TraceLogEntry::with_timestamp(12345678, msg.clone());

        let mut buf = BytesMut::new();
        orig.write_to_log(&mut buf);

        // Invariant 1: bytes written == log_size()
        assert_eq!(
            buf.len(),
            orig.log_size(),
            "log_size mismatch for trace message {:?}",
            msg
        );

        let decoded = TraceLogEntry::read_from_log(&buf).unwrap();

        // Invariant 2: decoded is equivalent
        assert_eq!(orig.message, decoded.message);
        assert_eq!(orig.timestamp, decoded.timestamp);

        // Invariant 3: decoded reports the same log_size
        assert_eq!(orig.log_size(), decoded.log_size());
    }
}

/// FileHeader round-trip.
#[test]
fn test_loggable_file_header_roundtrip() {
    let cases = [
        (0u64, NULL_LSN, 1u32),
        (42, Lsn::new(10, 5000), 1),
        (u32::MAX as u64, Lsn::new(0xFFFF, 0xFFFF_FFFF), 1),
    ];

    for (file_num, last_lsn, log_ver) in &cases {
        let orig = FileHeader::new(*file_num, *last_lsn, *log_ver);

        let mut buf = BytesMut::new();
        orig.write_to_log(&mut buf);

        // Invariant 1
        assert_eq!(buf.len(), FileHeader::log_size());

        let decoded = FileHeader::read_from_log(&buf).unwrap();

        // Invariant 2
        assert_eq!(orig.file_num, decoded.file_num);
        assert_eq!(
            orig.last_entry_in_prev_file,
            decoded.last_entry_in_prev_file
        );
        assert_eq!(orig.log_version, decoded.log_version);
        assert_eq!(orig.timestamp, decoded.timestamp);

        // Invariant 3
        assert_eq!(FileHeader::log_size(), FileHeader::log_size());
    }
}

/// FileHeaderEntry round-trip.
#[test]
fn test_loggable_file_header_entry_roundtrip() {
    let entry = FileHeaderEntry::new(7, Lsn::new(2, 128), 1);

    let mut buf = BytesMut::new();
    entry.write_to_log(&mut buf);

    assert_eq!(buf.len(), entry.log_size());

    let decoded = FileHeaderEntry::read_from_log(&buf).unwrap();
    assert_eq!(entry.header.file_num, decoded.header.file_num);
    assert_eq!(entry.header.log_version, decoded.header.log_version);
    assert_eq!(entry.log_size(), decoded.log_size());
}

/// EmptyLogEntry round-trip.
///
/// Used for CkptStart, CkptEnd.
#[test]
fn test_loggable_empty_log_entry_roundtrip() {
    let orig = EmptyLogEntry::new();

    let mut buf = BytesMut::new();
    orig.write_to_log(&mut buf);

    // Invariant 1
    assert_eq!(buf.len(), EmptyLogEntry::log_size());
    assert_eq!(buf.len(), 1);

    let decoded = EmptyLogEntry::read_from_log(&buf).unwrap();
    assert_eq!(orig, decoded);

    // Invariant 3
    assert_eq!(EmptyLogEntry::log_size(), EmptyLogEntry::log_size());
}

/// TxnEndEntry (commit) round-trip.
#[test]
fn test_loggable_txn_commit_roundtrip() {
    let orig = TxnEndEntry::new_commit(
        111,
        Lsn::new(10, 10),
        999_000,
        179,
        Vlsn::new(1),
    );

    let mut buf = BytesMut::new();
    orig.write_to_log(&mut buf);

    // Invariant 1
    assert_eq!(buf.len(), orig.log_size());

    let decoded = TxnEndEntry::read_from_log(&buf).unwrap();

    // Invariant 2
    assert_eq!(orig, decoded);
    assert!(decoded.is_commit());
    assert!(!decoded.is_abort());

    // Invariant 3
    assert_eq!(orig.log_size(), decoded.log_size());
}

/// TxnEndEntry (abort) round-trip.
#[test]
fn test_loggable_txn_abort_roundtrip() {
    let orig = TxnEndEntry::new_abort(
        111,
        Lsn::new(11, 11),
        999_001,
        7_654_321,
        Vlsn::new(1),
    );

    let mut buf = BytesMut::new();
    orig.write_to_log(&mut buf);

    assert_eq!(buf.len(), orig.log_size());

    let decoded = TxnEndEntry::read_from_log(&buf).unwrap();
    assert_eq!(orig, decoded);
    assert!(decoded.is_abort());
    assert!(!decoded.is_commit());
    assert_eq!(orig.log_size(), decoded.log_size());
}

/// InLogEntry round-trip.
#[test]
fn test_loggable_in_log_entry_roundtrip() {
    let node_data: Vec<u8> = (0u8..=15).collect();
    let orig = InLogEntry::new(42, Lsn::new(5, 100), NULL_LSN, node_data);

    let mut buf = BytesMut::new();
    orig.write_to_log(&mut buf);

    assert_eq!(buf.len(), orig.log_size(), "InLogEntry log_size mismatch");

    let decoded = InLogEntry::read_from_log(&buf).unwrap();

    assert_eq!(orig.db_id, decoded.db_id);
    assert_eq!(orig.prev_full_lsn, decoded.prev_full_lsn);
    assert_eq!(orig.prev_delta_lsn, decoded.prev_delta_lsn);
    assert_eq!(orig.node_data, decoded.node_data);
    assert_eq!(orig.log_size(), decoded.log_size());
}

/// InLogEntry with various data sizes.
#[test]
fn test_loggable_in_log_entry_various_sizes() {
    for size in [0usize, 1, 64, 256, 1024] {
        let node_data = vec![0xABu8; size];
        let orig = InLogEntry::new(1, NULL_LSN, NULL_LSN, node_data.clone());

        let mut buf = BytesMut::new();
        orig.write_to_log(&mut buf);

        assert_eq!(
            buf.len(),
            orig.log_size(),
            "log_size mismatch for InLogEntry with {} byte node_data",
            size
        );

        let decoded = InLogEntry::read_from_log(&buf).unwrap();
        assert_eq!(decoded.node_data.len(), size);
        assert_eq!(decoded.node_data, node_data);
        assert_eq!(orig.log_size(), decoded.log_size());
    }
}

/// LnLogEntry round-trip.
#[test]
fn test_loggable_ln_log_entry_roundtrip() {
    let data = b"abcdef";
    let key = b"mykey";

    let orig = LnLogEntry::new(
        1001,                // db_id
        Some(42i64),         // txn_id
        Lsn::new(3, 50),     // abort_lsn
        false,               // abort_known_deleted
        None,                // abort_key
        None,                // abort_data
        NULL_VLSN,           // abort_vlsn
        0,                   // abort_expiration
        false,               // embedded_ln
        key.to_vec(),        // key
        Some(data.to_vec()), // data
        0,                   // expiration
        NULL_VLSN,           // vlsn
    );

    let mut buf = BytesMut::new();
    orig.write_to_log(&mut buf);

    assert_eq!(buf.len(), orig.log_size(), "LnLogEntry log_size mismatch");

    let decoded = LnLogEntry::read_from_log(&buf, true).unwrap();
    assert_eq!(orig.txn_id, decoded.txn_id);
    assert_eq!(orig.key, decoded.key);
    assert_eq!(orig.data, decoded.data);
    assert_eq!(orig.abort_lsn, decoded.abort_lsn);
    assert_eq!(orig.log_size(), decoded.log_size());
}

/// LnLogEntry with empty data (delete).
#[test]
fn test_loggable_ln_log_entry_delete_roundtrip() {
    let orig = LnLogEntry::new(
        500,                      // db_id
        None,                     // txn_id (non-transactional)
        NULL_LSN,                 // abort_lsn
        false,                    // abort_known_deleted
        None,                     // abort_key
        None,                     // abort_data
        NULL_VLSN,                // abort_vlsn
        0,                        // abort_expiration
        false,                    // embedded_ln
        b"keyForDelete".to_vec(), // key
        None,                     // data = None means deletion
        0,                        // expiration
        NULL_VLSN,                // vlsn
    );

    let mut buf = BytesMut::new();
    orig.write_to_log(&mut buf);

    assert_eq!(buf.len(), orig.log_size());

    let decoded = LnLogEntry::read_from_log(&buf, false).unwrap();
    assert!(decoded.data.is_none(), "deleted entry must have None data");
    assert!(decoded.is_deleted());
    assert_eq!(orig.key, decoded.key);
    assert_eq!(orig.log_size(), decoded.log_size());
}

/// RestoreRequired round-trip.
#[test]
fn test_loggable_restore_required_roundtrip() {
    let mut props = HashMap::new();
    props.insert("foo".to_string(), "bar".to_string());
    props.insert("apple".to_string(), "tree".to_string());

    let orig = RestoreRequired::new(FailureType::NetworkRestore, props.clone());

    let mut buf = BytesMut::new();
    orig.write_to_log(&mut buf);

    assert_eq!(buf.len(), orig.log_size(), "RestoreRequired log_size mismatch");

    let decoded = RestoreRequired::read_from_log(&buf).unwrap();
    assert_eq!(orig.failure_type, decoded.failure_type);
    assert_eq!(orig.timestamp, decoded.timestamp);
    // All properties must round-trip
    for (k, v) in &props {
        assert_eq!(
            decoded.properties.get(k).map(String::as_str),
            Some(v.as_str()),
            "property {:?} missing or changed",
            k
        );
    }
    assert_eq!(orig.log_size(), decoded.log_size());
}

/// RestoreRequired with all failure types.
#[test]
fn test_loggable_restore_required_all_failure_types() {
    for failure_type in [
        FailureType::NetworkRestore,
        FailureType::LogChecksum,
        FailureType::BtreeCorruption,
    ] {
        let orig = RestoreRequired::new(failure_type, HashMap::new());

        let mut buf = BytesMut::new();
        orig.write_to_log(&mut buf);
        assert_eq!(buf.len(), orig.log_size());

        let decoded = RestoreRequired::read_from_log(&buf).unwrap();
        assert_eq!(decoded.failure_type, failure_type);
        assert_eq!(orig.log_size(), decoded.log_size());
    }
}

/// FailureType string round-trip.
#[test]
fn test_failure_type_parse_roundtrip() {
    for ft in [
        FailureType::NetworkRestore,
        FailureType::LogChecksum,
        FailureType::BtreeCorruption,
    ] {
        let s = ft.as_str();
        let parsed = FailureType::parse(s).unwrap();
        assert_eq!(ft, parsed, "FailureType parse round-trip failed for {}", s);
    }

    // Unknown string should return error
    assert!(FailureType::parse("UNKNOWN_TYPE").is_err());
}

// ============================================================================
// LogEntryHeader — serialization round-trips
//
// LoggableTest invariants applied to the header struct.
// ============================================================================

/// Applied to LogEntryHeader (no VLSN).
#[test]
fn test_header_loggable_no_vlsn_roundtrip() {
    let provisionals =
        [Provisional::No, Provisional::Yes, Provisional::BeforeCkptEnd];
    for prov in &provisionals {
        let header =
            LogEntryHeader::new(LogEntryType::BIN, 1024, *prov, false, None);

        let mut buf = Vec::new();
        header.write_to_log(&mut buf).unwrap();

        // Invariant 1: bytes written == header.size()
        assert_eq!(buf.len(), MIN_HEADER_SIZE);
        assert_eq!(buf.len(), header.size());

        let lsn = Lsn::new(1, 0);
        let decoded = LogEntryHeader::read_from_log(&buf, lsn).unwrap();

        // Invariant 2: round-trip
        assert_eq!(header.entry_type(), decoded.entry_type());
        assert_eq!(header.item_size(), decoded.item_size());
        assert_eq!(header.provisional(), decoded.provisional());
        assert_eq!(header.replicated(), decoded.replicated());

        // Invariant 3: decoded size matches
        assert_eq!(header.size(), decoded.size());
    }
}

/// Applied to LogEntryHeader (with VLSN).
#[test]
fn test_header_loggable_with_vlsn_roundtrip() {
    let vlsn = Some(Vlsn::new(99));
    let header = LogEntryHeader::new(
        LogEntryType::InsertLNTxn,
        512,
        Provisional::No,
        true,
        vlsn,
    );

    let mut buf = Vec::new();
    header.write_to_log(&mut buf).unwrap();

    assert_eq!(buf.len(), MAX_HEADER_SIZE);
    assert_eq!(buf.len(), header.size());

    let lsn = Lsn::new(2, 200);
    let decoded = LogEntryHeader::read_from_log(&buf, lsn).unwrap();

    assert_eq!(header.entry_type(), decoded.entry_type());
    assert_eq!(header.vlsn(), decoded.vlsn());
    assert!(decoded.vlsn_present());
    assert!(decoded.replicated());
    assert_eq!(header.size(), decoded.size());
}

/// Invisible flag survives round-trip.
#[test]
fn test_header_invisible_flag_roundtrip() {
    let mut header = LogEntryHeader::new(
        LogEntryType::BIN,
        50,
        Provisional::No,
        false,
        None,
    );
    header.set_invisible(true);

    let mut buf = Vec::new();
    header.write_to_log(&mut buf).unwrap();

    let lsn = Lsn::new(1, 0);
    let decoded = LogEntryHeader::read_from_log(&buf, lsn).unwrap();
    assert!(decoded.invisible());
    assert_eq!(header.size(), decoded.size());
}

/// Post-marshalling fields survive read-back.
#[test]
fn test_header_post_marshalling_roundtrip() {
    let mut header = LogEntryHeader::new(
        LogEntryType::Trace,
        64,
        Provisional::No,
        false,
        None,
    );

    let mut buf = Vec::new();
    header.write_to_log(&mut buf).unwrap();

    let prev_offset = 88u32;
    let checksum = 0xCAFE_BABEu32;
    header
        .add_post_marshalling_info(&mut buf, prev_offset, None, checksum)
        .unwrap();

    let lsn = Lsn::new(1, 0);
    let decoded = LogEntryHeader::read_from_log(&buf, lsn).unwrap();
    assert_eq!(decoded.prev_offset(), prev_offset);
    assert_eq!(decoded.checksum(), checksum);
}

// ============================================================================
// FSyncManagerTest — coalescing behavior
// ============================================================================

use noxu_log::fsync_manager::FsyncManager;
use std::sync::atomic::{AtomicUsize, Ordering};

/// Multiple threads requesting fsync
/// should result in fewer actual fsyncs than threads (coalescing).
#[test]
fn test_fsync_manager_grouping_reduces_fsyncs() {
    let manager = Arc::new(FsyncManager::new(0, 0));
    let fsync_count = Arc::new(AtomicUsize::new(0));

    let n_threads = 8usize;
    let barrier = Arc::new(std::sync::Barrier::new(n_threads));
    let mut handles = Vec::with_capacity(n_threads);

    for _ in 0..n_threads {
        let mgr = Arc::clone(&manager);
        let sc = Arc::clone(&fsync_count);
        let b = Arc::clone(&barrier);

        let handle = std::thread::spawn(move || {
            b.wait();
            mgr.fsync(|| {
                sc.fetch_add(1, Ordering::SeqCst);
                std::thread::sleep(std::time::Duration::from_millis(5));
                Ok(())
            })
            .unwrap();
        });

        handles.push(handle);
    }

    for h in handles {
        h.join().unwrap();
    }

    let fsyncs = fsync_count.load(Ordering::SeqCst);
    // Every thread must have completed (proven by join() above).
    // Coalescing means fsyncs < threads.
    assert!(
        fsyncs < n_threads,
        "fsync_count ({}) should be < n_threads ({}) due to coalescing",
        fsyncs,
        n_threads
    );
}

/// Single-thread fsync: closure called once.
#[test]
fn test_fsync_manager_flush_only_no_fsync() {
    let manager = FsyncManager::new(0, 0);
    let call_count = Arc::new(AtomicUsize::new(0));

    let c = Arc::clone(&call_count);
    manager
        .fsync(|| {
            c.fetch_add(1, Ordering::SeqCst);
            Ok(())
        })
        .unwrap();

    assert_eq!(
        call_count.load(Ordering::SeqCst),
        1,
        "closure must be called once"
    );
}

/// Waiter is notified after leader completes.
#[test]
fn test_fsync_manager_waiter_notified() {
    use std::sync::Barrier;
    let manager = Arc::new(FsyncManager::new(0, 0));
    let barrier = Arc::new(Barrier::new(2));

    let mgr1 = Arc::clone(&manager);
    let mgr2 = Arc::clone(&manager);
    let bar1 = Arc::clone(&barrier);
    let bar2 = Arc::clone(&barrier);

    let fsync_done = Arc::new(AtomicUsize::new(0));
    let done1 = Arc::clone(&fsync_done);
    let done2 = Arc::clone(&fsync_done);

    let t1 = std::thread::spawn(move || {
        bar1.wait();
        mgr1.fsync(|| {
            std::thread::sleep(std::time::Duration::from_millis(20));
            Ok(())
        })
        .unwrap();
        done1.fetch_add(1, Ordering::SeqCst);
    });

    let t2 = std::thread::spawn(move || {
        bar2.wait();
        mgr2.fsync(|| Ok(())).unwrap();
        done2.fetch_add(1, Ordering::SeqCst);
    });

    t1.join().unwrap();
    t2.join().unwrap();

    // Both threads must have completed
    assert_eq!(
        fsync_done.load(Ordering::SeqCst),
        2,
        "both threads must complete"
    );
}

/// Error propagates to caller.
#[test]
fn test_fsync_manager_flush_error_propagates() {
    let manager = FsyncManager::new(0, 0);
    let result =
        manager.fsync(|| Err(std::io::Error::other("simulated flush error")));
    assert!(result.is_err(), "error must be propagated to caller");
}

/// Consecutive single-threaded calls each run the closure.
#[test]
fn test_fsync_manager_sequential_calls_each_flush() {
    let manager = FsyncManager::new(0, 0);
    let call_count = Arc::new(AtomicUsize::new(0));

    for _ in 0..5 {
        let c = Arc::clone(&call_count);
        manager
            .fsync(|| {
                c.fetch_add(1, Ordering::SeqCst);
                Ok(())
            })
            .unwrap();
    }

    assert_eq!(
        call_count.load(Ordering::SeqCst),
        5,
        "each sequential call must run the fsync closure"
    );
}

// ============================================================================
// LogManagerTest — write, flush, read, checksum
// ============================================================================

/// / `testBasicOnDisk`.
///
/// Write several entries, flush, read them all back out of order and verify
/// the payloads match.
#[test]
fn test_log_manager_write_and_read_multiple_entries() {
    let dir = TempDir::new().unwrap();
    let (_fm, lm) = make_managers(&dir);

    let payloads: Vec<Vec<u8>> = (0..10)
        .map(|i| format!("Hello there, rec {}", i + 1).into_bytes())
        .collect();

    // Log the first 3 entries, remember their LSNs.
    let mut lsns: Vec<Lsn> = Vec::new();
    for payload in &payloads[..3] {
        let lsn = lm
            .log(LogEntryType::Trace, payload, Provisional::No, false, false)
            .unwrap();
        lsns.push(lsn);
    }

    // Flush to disk.
    lm.flush_no_sync().unwrap();

    // Read back out of order: 2, 0, 1
    let (_, p2) = lm.read_entry(lsns[2]).unwrap();
    assert_eq!(p2, payloads[2]);

    let (_, p0) = lm.read_entry(lsns[0]).unwrap();
    assert_eq!(p0, payloads[0]);

    let (_, p1) = lm.read_entry(lsns[1]).unwrap();
    assert_eq!(p1, payloads[1]);

    // Intersperse more logs and reads
    let lsn3 = lm
        .log(LogEntryType::Trace, &payloads[3], Provisional::No, false, false)
        .unwrap();
    let lsn4 = lm
        .log(LogEntryType::Trace, &payloads[4], Provisional::No, false, false)
        .unwrap();

    lm.flush_no_sync().unwrap();

    let (_, p2b) = lm.read_entry(lsns[2]).unwrap();
    assert_eq!(p2b, payloads[2]);

    let (_, p4) = lm.read_entry(lsn4).unwrap();
    assert_eq!(p4, payloads[4]);

    let (_, p3) = lm.read_entry(lsn3).unwrap();
    assert_eq!(p3, payloads[3]);
}

/// Returned LSNs are in strictly ascending order
/// within the same file.
#[test]
fn test_log_manager_lsns_are_ascending() {
    let dir = TempDir::new().unwrap();
    let (_fm, lm) = make_managers(&dir);

    let mut prev_lsn: Option<Lsn> = None;

    for i in 0u32..20 {
        let payload = format!("entry-{}", i).into_bytes();
        let lsn = lm
            .log(LogEntryType::Trace, &payload, Provisional::No, false, false)
            .unwrap();

        if let Some(prev) = prev_lsn {
            if prev.file_number() == lsn.file_number() {
                assert!(
                    lsn.file_offset() > prev.file_offset(),
                    "LSN offsets must be strictly increasing within a file: \
                     prev={:?} current={:?}",
                    prev,
                    lsn
                );
            } else {
                // File flipped — next file number must be exactly prev + 1
                assert_eq!(
                    lsn.file_number(),
                    prev.file_number() + 1,
                    "File number must increment by 1 on flip"
                );
            }
        }
        prev_lsn = Some(lsn);
    }
}

/// Corrupting any byte in the
/// on-disk entry must cause a checksum error when reading back from disk.
#[test]
fn test_log_manager_bad_checksum_detected() {
    use std::fs::OpenOptions;
    use std::io::{Seek, SeekFrom, Write};

    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    let payload = b"checksum test payload";
    let lsn = lm
        .log(LogEntryType::Trace, payload, Provisional::No, true, false)
        .unwrap();

    // Flush all buffers to disk.
    lm.flush_sync().unwrap();
    fm.clear_cache();

    // Corrupt a payload byte (not the checksum field itself).
    let payload_byte_offset = lsn.file_offset() as u64 + MIN_HEADER_SIZE as u64;
    let file_path = dir.path().join(format!("{:08x}.ndb", lsn.file_number()));

    {
        let mut f = OpenOptions::new().write(true).open(&file_path).unwrap();
        f.seek(SeekFrom::Start(payload_byte_offset)).unwrap();
        f.write_all(&[payload[0] ^ 0xFF]).unwrap();
        f.flush().unwrap();
    }

    fm.clear_cache();

    // Read via LogFileReader (always reads from disk).
    let mut reader =
        LogFileReader::open(Arc::clone(&fm), lsn.file_number()).unwrap();
    let result = reader.read_next_strict();

    assert!(result.is_err(), "Expected checksum error but read succeeded");
    match result.unwrap_err() {
        noxu_log::NoxuLogError::Checksum { .. } => {} // expected
        other => panic!("Expected Checksum error, got {:?}", other),
    }
}

/// Unmodified entry validates cleanly.
#[test]
fn test_log_manager_valid_checksum_reads_ok() {
    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    let payload = b"valid payload";
    let lsn = lm
        .log(LogEntryType::Trace, payload, Provisional::No, true, false)
        .unwrap();

    lm.flush_sync().unwrap();
    fm.clear_cache();

    let mut reader =
        LogFileReader::open(Arc::clone(&fm), lsn.file_number()).unwrap();
    let result = reader.read_next_strict();

    assert!(result.is_ok(), "Unmodified entry should read without error");
    let entry_opt = result.unwrap();
    assert!(entry_opt.is_some(), "entry must be present");
    let (_, _, read_payload) = entry_opt.unwrap();
    assert_eq!(read_payload.as_slice(), payload);
}

/// Entry type is preserved through the write/read cycle.
#[test]
fn test_log_manager_entry_type_preserved() {
    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    let entry_types = [
        LogEntryType::Trace,
        LogEntryType::TxnCommit,
        LogEntryType::TxnAbort,
        LogEntryType::CkptStart,
    ];

    let mut lsns = Vec::new();
    for &et in &entry_types {
        let lsn = lm.log(et, b"dummy", Provisional::No, false, false).unwrap();
        lsns.push((lsn, et));
    }

    lm.flush_no_sync().unwrap();
    fm.clear_cache();

    for (lsn, expected_type) in &lsns {
        let (actual_type, _) = lm.read_entry(*lsn).unwrap();
        assert_eq!(
            actual_type, *expected_type,
            "Entry type mismatch at LSN {:?}",
            lsn
        );
    }
}

/// Adjacent entries are offset by exactly
/// (header_size + payload_size) within the same file.
#[test]
fn test_log_manager_lsn_stride() {
    let dir = TempDir::new().unwrap();
    let (_fm, lm) = make_managers(&dir);

    let payload = b"stride-test-payload";
    let expected_stride = MIN_HEADER_SIZE + payload.len();

    let lsn0 = lm
        .log(LogEntryType::Trace, payload, Provisional::No, false, false)
        .unwrap();
    let lsn1 = lm
        .log(LogEntryType::Trace, payload, Provisional::No, false, false)
        .unwrap();

    if lsn0.file_number() == lsn1.file_number() {
        let stride = (lsn1.file_offset() - lsn0.file_offset()) as usize;
        assert_eq!(
            stride, expected_stride,
            "LSN stride must equal header_size + payload_size"
        );
    }
    // (If a file flip occurred, the stride check does not apply.)
}

/// Get_end_of_log advances after each write.
#[test]
fn test_log_manager_end_of_log_advances() {
    let dir = TempDir::new().unwrap();
    let (_fm, lm) = make_managers(&dir);

    let eol_before = lm.get_end_of_log();

    lm.log(LogEntryType::Trace, b"advance-test", Provisional::No, false, false)
        .unwrap();

    let eol_after = lm.get_end_of_log();

    // end-of-log must have moved forward
    if eol_before.file_number() == eol_after.file_number() {
        assert!(
            eol_after.file_offset() > eol_before.file_offset(),
            "end_of_log must advance after a write"
        );
    } else {
        assert!(
            eol_after.file_number() > eol_before.file_number(),
            "end_of_log file_number must increase on file flip"
        );
    }
}

/// Flush_sync makes data readable from disk
/// even after clearing the buffer pool.
#[test]
fn test_log_manager_flush_sync_durability() {
    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    let payload = b"durable-after-flush_sync";
    let lsn = lm
        .log(LogEntryType::Trace, payload, Provisional::No, false, false)
        .unwrap();

    lm.flush_sync().unwrap();
    fm.clear_cache();

    // Read from disk (not from the buffer pool) via LogFileReader.
    let mut reader =
        LogFileReader::open(Arc::clone(&fm), lsn.file_number()).unwrap();
    let result = reader.read_next();
    assert!(result.is_some(), "data must be on disk after flush_sync");
    let (_, et, rp) = result.unwrap();
    assert_eq!(et, LogEntryType::Trace);
    assert_eq!(rp.as_slice(), payload);
}

// ============================================================================
// Checksum
// ============================================================================

/// Checksum covers bytes
/// [CHECKSUM_BYTES..entry_size], i.e. the first 4 bytes are the stored
/// checksum itself and are excluded from the computation.
#[test]
fn test_checksum_skips_first_four_bytes() {
    let data = b"some payload bytes for the checksum test";

    // Build a fake entry buffer: [checksum:4][rest]
    let mut buf = vec![0u8; CHECKSUM_BYTES + data.len()];
    buf[CHECKSUM_BYTES..].copy_from_slice(data);

    // Compute checksum over everything after the checksum field.
    let crc = ChecksumValidator::compute_range(
        &buf,
        CHECKSUM_BYTES,
        buf.len() - CHECKSUM_BYTES,
    );

    // Store it.
    buf[0..4].copy_from_slice(&crc.to_le_bytes());

    // Validate by re-reading.
    let stored = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
    assert_eq!(stored, crc, "stored checksum must equal computed checksum");

    // Corrupt one byte in the data portion and verify the checksum no longer
    // matches.
    buf[CHECKSUM_BYTES] ^= 0xFF;
    let bad_crc = ChecksumValidator::compute_range(
        &buf,
        CHECKSUM_BYTES,
        buf.len() - CHECKSUM_BYTES,
    );
    assert_ne!(stored, bad_crc, "checksum must differ after corruption");
}

/// Modifying any individual bit
/// of a committed entry invalidates the checksum.
#[test]
fn test_checksum_any_bit_flip_detected() {
    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    let payload = b"bit-flip-test";
    let lsn = lm
        .log(LogEntryType::Trace, payload, Provisional::No, true, false)
        .unwrap();

    lm.flush_sync().unwrap();
    fm.clear_cache();

    // Read the raw bytes on disk.
    let file_path = dir.path().join(format!("{:08x}.ndb", lsn.file_number()));
    let original = std::fs::read(&file_path).unwrap();

    // Find the start of the entry (after file header).
    let entry_start = lsn.file_offset() as usize;
    let entry_end = entry_start + MIN_HEADER_SIZE + payload.len();

    // Flip one bit in each byte of the entry (except the checksum bytes) and
    // verify that the checksum computed from the modified buffer differs from
    // the stored one.
    for byte_idx in (entry_start + CHECKSUM_BYTES)..entry_end {
        let mut modified = original.clone();
        modified[byte_idx] ^= 0xFF; // flip all bits in that byte

        let stored_crc = u32::from_le_bytes([
            modified[entry_start],
            modified[entry_start + 1],
            modified[entry_start + 2],
            modified[entry_start + 3],
        ]);

        let computed_crc = ChecksumValidator::compute_range(
            &modified[entry_start..],
            CHECKSUM_BYTES,
            (entry_end - entry_start) - CHECKSUM_BYTES,
        );

        assert_ne!(
            stored_crc, computed_crc,
            "checksum must differ after flipping byte at index {}",
            byte_idx
        );
    }
}

// ============================================================================
// LastFileReaderTest / FileReaderTest — forward scan and end-of-log detection
// ============================================================================

/// And
/// `LastFileReaderTest.testLastFileEmpty`.
///
/// The LogFileReader must handle being given an empty file (zero length or
/// length equals only the file header) without panicking.
#[test]
fn test_log_file_reader_empty_file_no_entries() {
    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    // Write one entry and flush to create file 0.
    lm.log(LogEntryType::Trace, b"seed", Provisional::No, true, false).unwrap();
    lm.flush_sync().unwrap();

    // Open file 0 and drain it.
    let file_num = 0u32;
    let mut reader = LogFileReader::open(Arc::clone(&fm), file_num).unwrap();

    let mut count = 0usize;
    while reader.read_next().is_some() {
        count += 1;
    }
    // There is 1 valid entry in the file.
    assert_eq!(count, 1, "Expected exactly 1 entry in file 0");

    // Now open a non-existent file — the reader must fail gracefully.
    let bad_result = LogFileReader::open(Arc::clone(&fm), 999);
    assert!(
        bad_result.is_err(),
        "Opening a non-existent file should return an error"
    );
}

/// / `LastFileReaderTest.testBasic`.
///
/// Write N entries, open a fresh reader, read them all forward — count must
/// match exactly.
#[test]
fn test_log_file_reader_sequential_forward_scan() {
    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    let n_entries = 50usize;

    for i in 0..n_entries {
        let payload = format!("Hello there, rec {}", i + 1);
        lm.log(
            LogEntryType::Trace,
            payload.as_bytes(),
            Provisional::No,
            false,
            false,
        )
        .unwrap();
    }
    lm.flush_no_sync().unwrap();

    // File 0 was never flipped so all entries live there.
    let mut reader = LogFileReader::open(Arc::clone(&fm), 0).unwrap();
    let mut count = 0usize;

    while let Some((lsn, et, payload)) = reader.read_next() {
        assert_eq!(et, LogEntryType::Trace, "entry type mismatch at {}", count);
        let expected = format!("Hello there, rec {}", count + 1);
        assert_eq!(
            payload,
            expected.as_bytes(),
            "payload mismatch at entry {}",
            count
        );
        let _ = lsn; // LSN is valid but we only check content here
        count += 1;
    }

    assert_eq!(
        count, n_entries,
        "reader must return all {} entries",
        n_entries
    );
}

/// / `testMedBuffers`.
///
/// Read back all entries with small read buffers to exercise buffering code.
#[test]
fn test_log_file_reader_small_payloads_all_readable() {
    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    let payloads: Vec<Vec<u8>> = (1u8..=30)
        .map(|i| vec![i; i as usize]) // varying sizes 1..=30 bytes
        .collect();

    let mut expected_lsns = Vec::new();
    for payload in &payloads {
        let lsn = lm
            .log(LogEntryType::Trace, payload, Provisional::No, false, false)
            .unwrap();
        expected_lsns.push(lsn);
    }

    lm.flush_no_sync().unwrap();

    // Gather entries from the reader.
    let mut reader = LogFileReader::open(Arc::clone(&fm), 0).unwrap();
    let mut read_entries: Vec<(Lsn, Vec<u8>)> = Vec::new();
    while let Some((lsn, _et, payload)) = reader.read_next() {
        read_entries.push((lsn, payload));
    }

    assert_eq!(
        read_entries.len(),
        payloads.len(),
        "must read back all {} entries",
        payloads.len()
    );

    for (i, ((lsn, read_payload), (expected_lsn, expected_payload))) in
        read_entries
            .iter()
            .zip(expected_lsns.iter().zip(payloads.iter()))
            .enumerate()
    {
        assert_eq!(lsn, expected_lsn, "LSN mismatch at entry {}", i);
        assert_eq!(
            read_payload, expected_payload,
            "payload mismatch at entry {}",
            i
        );
    }
}

/// Trailing junk bytes at the end of
/// a file do not corrupt valid entries that precede them.
///
/// The reader must stop at the first bad (corrupt) entry and return all
/// valid entries before it.
#[test]
fn test_log_file_reader_junk_at_end_of_file() {
    use std::io::Write;

    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    let n_valid = 5usize;
    let payload = b"valid entry";

    for _ in 0..n_valid {
        lm.log(LogEntryType::Trace, payload, Provisional::No, false, false)
            .unwrap();
    }
    lm.flush_sync().unwrap();
    fm.clear_cache();

    // Append junk bytes to the end of file 0.
    let file_path = dir.path().join("00000000.ndb");
    {
        let mut f =
            std::fs::OpenOptions::new().append(true).open(&file_path).unwrap();
        f.write_all(b"hello, some junk").unwrap();
        f.flush().unwrap();
    }

    // The reader must still return all n_valid entries before stopping.
    fm.clear_cache();
    let mut reader = LogFileReader::open(Arc::clone(&fm), 0).unwrap();

    let mut count = 0usize;
    // read_next() returns None when it encounters bad data.
    while let Some((_, et, rp)) = reader.read_next() {
        assert_eq!(et, LogEntryType::Trace);
        assert_eq!(rp.as_slice(), payload);
        count += 1;
    }

    assert_eq!(count, n_valid, "all valid entries must be returned");
}

/// Reader correctly handles files spanning
/// multiple log files (file flip).
///
/// Uses a 4 KB payload with a 10 KB max file size — the same ratio used by
/// the existing `test_file_flip` integration test — to reliably trigger a
/// file flip within a small number of writes.
#[test]
fn test_log_file_reader_multi_file_scan() {
    let dir = TempDir::new().unwrap();

    // Each entry is ~4110 bytes (14-byte header + 4096-byte payload).
    // Two such entries (8220 bytes) plus the 20-byte file header exceed the
    // 10 240-byte max, so a flip happens after the 2nd entry.
    let max_size: u64 = 10_240;
    let (fm, lm) = make_managers_small_file(&dir, max_size);

    let payload = vec![0xABu8; 4096];
    let n_entries = 6usize; // enough to produce at least 2 files

    let mut all_lsns = Vec::new();
    for _ in 0..n_entries {
        let lsn = lm
            .log(LogEntryType::Trace, &payload, Provisional::No, true, false)
            .unwrap();
        all_lsns.push(lsn);
    }
    lm.flush_sync().unwrap();

    // Confirm we have multiple files.
    let file_nums = fm.list_file_numbers().unwrap();
    assert!(
        file_nums.len() >= 2,
        "expected at least 2 log files, got {:?}",
        file_nums
    );

    // Scan each file and count entries.
    let mut total = 0usize;
    for &file_num in &file_nums {
        if let Ok(mut reader) = LogFileReader::open(Arc::clone(&fm), file_num) {
            while let Some((_, et, _)) = reader.read_next() {
                assert_eq!(et, LogEntryType::Trace);
                total += 1;
            }
        }
    }

    assert_eq!(
        total, n_entries,
        "sum of entries across all files must equal n_entries"
    );
}

// ============================================================================
// INFileReaderTest / LNFileReaderTest — entry-type filtering
//
// The concrete INFileReader / LNFileReader stubs don't expose a read loop yet,
// but we can test the equivalent behaviour through LogFileReader: write mixed
// entry types and verify that only the expected types appear.
// ============================================================================

/// No crash when scanning an empty file.
#[test]
fn test_log_file_reader_empty_returns_no_entries() {
    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    // Force-create file 0 by logging one entry then flushing.
    lm.log(LogEntryType::Trace, b"x", Provisional::No, true, false).unwrap();
    lm.flush_sync().unwrap();
    fm.clear_cache();

    // Open the file and drain — there should be exactly 1 entry.
    let mut reader = LogFileReader::open(Arc::clone(&fm), 0).unwrap();
    let mut count = 0usize;
    while reader.read_next().is_some() {
        count += 1;
    }
    assert_eq!(count, 1);
}

/// / `LNFileReaderTest.testBasicRedo`.
///
/// Write a mix of IN, BIN and Trace entries; scan and count only IN/BIN types.
#[test]
fn test_log_file_reader_filter_by_entry_type() {
    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    let n_per_type = 5usize;

    // Write Trace, IN, and BIN entries interleaved.
    for i in 0..n_per_type {
        let trace_payload = format!("debug rec {}", i).into_bytes();
        lm.log(
            LogEntryType::Trace,
            &trace_payload,
            Provisional::No,
            false,
            false,
        )
        .unwrap();

        let in_payload = vec![0x01u8; 32];
        lm.log(LogEntryType::IN, &in_payload, Provisional::No, false, false)
            .unwrap();

        let bin_payload = vec![0x02u8; 32];
        lm.log(LogEntryType::BIN, &bin_payload, Provisional::No, false, false)
            .unwrap();
    }

    lm.flush_no_sync().unwrap();

    let mut reader = LogFileReader::open(Arc::clone(&fm), 0).unwrap();

    let mut trace_count = 0usize;
    let mut in_count = 0usize;
    let mut bin_count = 0usize;
    let mut total = 0usize;

    while let Some((_, et, _)) = reader.read_next() {
        match et {
            LogEntryType::Trace => trace_count += 1,
            LogEntryType::IN => in_count += 1,
            LogEntryType::BIN => bin_count += 1,
            _ => {}
        }
        total += 1;
    }

    assert_eq!(
        total,
        3 * n_per_type,
        "total entries must equal 3 × n_per_type"
    );
    assert_eq!(trace_count, n_per_type, "Trace count mismatch");
    assert_eq!(in_count, n_per_type, "IN count mismatch");
    assert_eq!(bin_count, n_per_type, "BIN count mismatch");
}

/// Reader on a freshly-written file
/// with only one entry of a specific type.
#[test]
fn test_log_file_reader_single_entry_type_filtering() {
    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    // Write transactional LN entries only.
    let n = 7usize;
    let mut expected_lsns = Vec::new();
    for i in 0..n {
        let payload = format!("txn-data-{}", i).into_bytes();
        let lsn = lm
            .log(
                LogEntryType::InsertLNTxn,
                &payload,
                Provisional::No,
                false,
                false,
            )
            .unwrap();
        expected_lsns.push(lsn);
    }

    lm.flush_no_sync().unwrap();

    let mut reader = LogFileReader::open(Arc::clone(&fm), 0).unwrap();
    let mut count = 0usize;

    while let Some((lsn, et, _payload)) = reader.read_next() {
        assert_eq!(
            et,
            LogEntryType::InsertLNTxn,
            "expected InsertLNTxn, got {:?}",
            et
        );
        assert_eq!(
            lsn, expected_lsns[count],
            "LSN mismatch at entry {}",
            count
        );
        count += 1;
    }

    assert_eq!(count, n, "must read all {} InsertLNTxn entries", n);
}

/// The reader can start from a
/// given LSN (mid-file).
///
/// We verify this by writing 10 entries, then scanning and verifying we can
/// locate a specific one by its known LSN via LogManager.read_entry.
#[test]
fn test_log_manager_random_access_mid_file() {
    let dir = TempDir::new().unwrap();
    let (_fm, lm) = make_managers(&dir);

    let payloads: Vec<Vec<u8>> =
        (0..10).map(|i| format!("mid-file-entry-{}", i).into_bytes()).collect();

    let mut lsns = Vec::new();
    for payload in &payloads {
        let lsn = lm
            .log(LogEntryType::Trace, payload, Provisional::No, false, false)
            .unwrap();
        lsns.push(lsn);
    }

    lm.flush_no_sync().unwrap();

    // Read from the middle entry (index 5) forward using LogManager.read_entry
    for i in 5..10 {
        let (et, rp) = lm.read_entry(lsns[i]).unwrap();
        assert_eq!(et, LogEntryType::Trace);
        assert_eq!(rp, payloads[i]);
    }
}

/// Entries are accessible in
/// reverse (random-access) order.
#[test]
fn test_log_manager_read_entries_in_reverse_order() {
    let dir = TempDir::new().unwrap();
    let (_fm, lm) = make_managers(&dir);

    let payloads: Vec<Vec<u8>> =
        (0..10).map(|i| format!("reverse-entry-{}", i).into_bytes()).collect();

    let mut lsns = Vec::new();
    for payload in &payloads {
        let lsn = lm
            .log(LogEntryType::Trace, payload, Provisional::No, false, false)
            .unwrap();
        lsns.push(lsn);
    }

    lm.flush_no_sync().unwrap();

    // Read backwards
    for i in (0..10).rev() {
        let (et, rp) = lm.read_entry(lsns[i]).unwrap();
        assert_eq!(et, LogEntryType::Trace);
        assert_eq!(rp, payloads[i]);
    }
}

// ============================================================================
// Provisional flag — LogManager round-trip
// ============================================================================

/// Provisional status is encoded in the flags byte
/// and preserved through write/read.
#[test]
fn test_log_manager_provisional_flag_preserved() {
    let dir = TempDir::new().unwrap();
    let (fm, lm) = make_managers(&dir);

    let provisionals =
        [Provisional::No, Provisional::Yes, Provisional::BeforeCkptEnd];

    let mut lsns = Vec::new();
    for prov in &provisionals {
        let lsn = lm
            .log(LogEntryType::IN, b"node-data", *prov, false, false)
            .unwrap();
        lsns.push(lsn);
    }

    lm.flush_no_sync().unwrap();

    // Scan via LogFileReader and verify we get one entry per provisional value.
    let mut reader = LogFileReader::open(Arc::clone(&fm), 0).unwrap();
    let mut count = 0;
    while reader.read_next().is_some() {
        count += 1;
    }
    assert_eq!(
        count,
        provisionals.len(),
        "all provisional variants must produce readable entries"
    );
}

// ============================================================================
// Checksum validator — unit tests for correctness properties
// ============================================================================

/// CRC32 is deterministic.
#[test]
fn test_checksum_deterministic() {
    let data = b"deterministic data";
    let c1 = ChecksumValidator::compute(data);
    let c2 = ChecksumValidator::compute(data);
    assert_eq!(c1, c2);
}

/// Incremental update equals one-shot computation.
#[test]
fn test_checksum_incremental_matches_oneshot() {
    let part1 = b"part one ";
    let part2 = b"part two";
    let combined: Vec<u8> = part1.iter().chain(part2.iter()).copied().collect();

    let oneshot = ChecksumValidator::compute(&combined);

    let mut v = ChecksumValidator::new();
    v.update_all(part1);
    v.update_all(part2);
    assert_eq!(v.value(), oneshot);
}

/// Any single-byte modification changes the checksum.
#[test]
fn test_checksum_sensitive_to_single_byte_change() {
    let original = b"original test data for checksum sensitivity";
    let base = ChecksumValidator::compute(original);

    for i in 0..original.len() {
        let mut modified = original.to_vec();
        modified[i] ^= 0x01; // flip one bit
        let altered = ChecksumValidator::compute(&modified);
        assert_ne!(
            base, altered,
            "checksum should differ after flipping bit in byte {}",
            i
        );
    }
}

/// Validation succeeds when the checksum matches.
#[test]
fn test_checksum_validate_success() {
    let data = b"validate me";
    let expected = ChecksumValidator::compute(data);

    let mut v = ChecksumValidator::new();
    v.update_all(data);
    assert!(v.validate(expected, NULL_LSN).is_ok());
}

/// Validation fails when the checksum does not match.
#[test]
fn test_checksum_validate_failure() {
    let data = b"validate me";
    let wrong = ChecksumValidator::compute(data).wrapping_add(1);

    let mut v = ChecksumValidator::new();
    v.update_all(data);
    assert!(v.validate(wrong, NULL_LSN).is_err());
}
