coordinode-lsm-tree 6.0.0

use super::*;
// `AbstractTree` looks unused at a glance but the test bodies below
// call `.insert()`, `.flush_active_memtable()`, and
// `.current_version()` on `AnyTree` values — those are trait
// methods, not inherent ones, so the trait MUST be in scope for
// method resolution. Removing the import is a compile error, not
// a clippy nit.
use crate::{
    AbstractTree, Config, SequenceNumberCounter, compression::CompressionType,
    config::CompressionPolicy,
};
use std::io::{Read, Seek, SeekFrom, Write};
// Shadows the built-in `#[test]` so `#[test]`-annotated functions
// below resolve to `test_log::test` (which wires up logging for
// failing tests). This matches every other test module in the
// crate — the import looks unused at a glance but the proc-macro
// attribute name resolution consumes it.
use test_log::test;

fn populate_tree(dir: &std::path::Path, items: usize) {
    let cfg = Config::new(
        dir,
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
    let tree = cfg.open().unwrap();
    for i in 0u64..items as u64 {
        let key = format!("k{i:08}");
        let val = format!("v{i:08}");
        tree.insert(key.as_bytes(), val.as_bytes(), 1 + i);
    }
    tree.flush_active_memtable(1 + items as u64).unwrap();
    // Drop the tree so all files are closed before the test that
    // mutates SST bytes on disk reopens them via Verify.
    drop(tree);
}

fn reopen_tree(dir: &std::path::Path) -> crate::AnyTree {
    Config::new(
        dir,
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
    .open()
    .unwrap()
}

/// Populates a tree with per-KV checksums enabled (`AllLevels`) so the
/// flushed SST carries data blocks with the `KV_CHECKSUM_FOOTER` flag
/// set and a per-entry checksum footer.
fn populate_tree_kv_checked(dir: &std::path::Path, items: usize) {
    use crate::AbstractTree;
    use crate::runtime_config::KvChecksumPolicy;

    let cfg = Config::new(
        dir,
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
    let any = cfg.open().unwrap();
    let crate::AnyTree::Standard(tree) = any else {
        panic!("expected Standard tree");
    };
    tree.update_runtime_config(|c| {
        c.kv_checksums = KvChecksumPolicy::AllLevels;
    })
    .unwrap();
    for i in 0u64..items as u64 {
        let key = format!("k{i:08}");
        let val = format!("v{i:08}");
        tree.insert(key.as_bytes(), val.as_bytes(), 1 + i);
    }
    tree.flush_active_memtable(1 + items as u64).unwrap();
    drop(tree);
}

#[test]
fn verify_block_checksums_clean_tree_has_no_errors() {
    let dir = tempfile::tempdir().unwrap();
    populate_tree(dir.path(), 1_000);

    let tree = reopen_tree(dir.path());
    let report = verify_block_checksums(&tree);
    assert!(
        report.is_ok(),
        "expected clean tree to verify with zero errors, got {:?}",
        report.errors
    );
    assert!(
        report.blocks_scanned > 0,
        "expected at least one block scanned",
    );
    assert!(
        report.sst_files_scanned >= 1,
        "expected at least one SST scanned",
    );
}

#[cfg(feature = "page_ecc")]
#[test]
fn verify_block_checksums_clean_page_ecc_tree_has_no_errors() {
    // Regression: with page_ecc on, every SST data / index / filter block
    // carries a Reed-Solomon parity trailer after its payload. SST blocks
    // omit the block_flags byte, so the scrub learns parity presence from
    // the per-SST descriptor and must skip `expected_parity_len(data_length)`
    // bytes per block. Without that skip the walk mis-reads parity as the
    // next block's header and reports spurious HeaderCorrupted. Enough items
    // to spill multiple data blocks so cross-block alignment is exercised.
    use crate::AbstractTree;

    let dir = tempfile::tempdir().unwrap();
    {
        let any = Config::new(
            dir.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
        .page_ecc(true)
        .ecc_scheme(crate::runtime_config::EccScheme::ReedSolomon {
            data_shards: 4,
            parity_shards: 2,
        })
        .open()
        .unwrap();
        for i in 0u64..2_000 {
            let key = format!("k{i:08}");
            let val = format!("v{i:08}");
            any.insert(key.as_bytes(), val.as_bytes(), 1 + i);
        }
        any.flush_active_memtable(2_001).unwrap();
        drop(any);
    }

    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
    .page_ecc(true)
    .ecc_scheme(crate::runtime_config::EccScheme::ReedSolomon {
        data_shards: 4,
        parity_shards: 2,
    })
    .open()
    .unwrap();
    let report = verify_block_checksums(&tree);
    assert!(
        report.is_ok(),
        "page_ecc tree must verify with zero errors (parity trailers skipped \
         per block), got {:?}",
        report.errors,
    );
    assert!(
        report.blocks_scanned > 1,
        "expected multiple blocks scanned to exercise cross-block alignment",
    );
}

#[cfg(feature = "page_ecc")]
#[test]
fn verify_block_checksums_clean_nondefault_ecc_tree_has_no_errors() {
    // Regression: the scrub must size each SST's parity trailer from the
    // per-SST descriptor scheme, NOT a hardcoded RS(4,2). A table written
    // with a non-default scheme (RS(8,2), different shard size → different
    // trailer length) is mis-walked if the scrub assumes RS(4,2): the
    // wrong trailer length mis-aligns the next block and reports spurious
    // corruption. With descriptor-driven sizing the walk stays aligned.
    use crate::AbstractTree;

    let dir = tempfile::tempdir().unwrap();
    {
        let any = Config::new(
            dir.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
        .page_ecc(true)
        .ecc_scheme(crate::runtime_config::EccScheme::ReedSolomon {
            data_shards: 8,
            parity_shards: 2,
        })
        .open()
        .unwrap();
        for i in 0u64..2_000 {
            let key = format!("k{i:08}");
            let val = format!("v{i:08}");
            any.insert(key.as_bytes(), val.as_bytes(), 1 + i);
        }
        any.flush_active_memtable(2_001).unwrap();
        drop(any);
    }

    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
    .page_ecc(true)
    .ecc_scheme(crate::runtime_config::EccScheme::ReedSolomon {
        data_shards: 8,
        parity_shards: 2,
    })
    .open()
    .unwrap();
    let report = verify_block_checksums(&tree);
    assert!(
        report.is_ok(),
        "non-default-scheme ECC tree must verify with zero errors \
         (parity sized from the descriptor, not RS(4,2)), got {:?}",
        report.errors,
    );
    assert!(
        report.blocks_scanned > 1,
        "expected multiple blocks scanned to exercise cross-block alignment",
    );
}

/// Rot confined to a Page-ECC parity trailer is invisible to the payload
/// checksum (the checksum covers the payload only; parity is consulted just
/// for recovery on a mismatch), so a walk that merely SKIPS the trailer
/// reports the SST as healthy while its ECC is dead — a later payload fault
/// on that block would no longer be recoverable. The out-of-band walk must
/// compare each clean block's trailer against freshly computed parity and
/// flag a mismatch.
#[cfg(feature = "page_ecc")]
#[test]
fn verify_sst_file_detects_a_rotted_parity_trailer() {
    use crate::coding::Decode;
    use crate::table::block::Header;

    let dir = tempfile::tempdir().unwrap();
    {
        let any = Config::new(
            dir.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
        .page_ecc(true)
        .ecc_scheme(crate::runtime_config::EccScheme::ReedSolomon {
            data_shards: 4,
            parity_shards: 2,
        })
        .open()
        .unwrap();
        for i in 0u64..2_000 {
            let key = format!("k{i:08}");
            let val = format!("v{i:08}");
            any.insert(key.as_bytes(), val.as_bytes(), 1 + i);
        }
        any.flush_active_memtable(2_001).unwrap();
        drop(any);
    }
    let sst_path = pick_first_sst_path(dir.path());

    // The pristine SST must verify clean BEFORE the flip, so the mismatch
    // asserted below is provably caused by the corruption, not a pre-existing
    // problem in the freshly written file.
    assert!(
        verify_sst_file(&sst_path).is_ok(),
        "the freshly written SST must verify clean before corruption",
    );

    // The first data block sits at file offset 0 (the writer opens the file
    // with the `data` section). Its parity trailer follows the payload:
    // header_len + data_length.
    let mut bytes = std::fs::read(&sst_path).unwrap();
    let mut cursor = bytes.as_slice();
    let header = Header::decode_from(&mut cursor).unwrap();
    let trailer_pos = Header::header_len(header.block_type) + header.data_length as usize;
    let slot = bytes
        .get_mut(trailer_pos)
        .expect("parity trailer within the file");
    *slot ^= 0xFF;
    std::fs::write(&sst_path, &bytes).unwrap();

    let report = verify_sst_file(&sst_path);
    assert!(
        !report.is_ok(),
        "a rotted parity trailer under a clean payload checksum must be \
         flagged (dead ECC), got {report:?}",
    );
    let mismatch = report
        .errors
        .iter()
        .find(|e| matches!(e, BlockVerifyError::EccParityMismatch { .. }))
        .unwrap_or_else(|| panic!("expected an EccParityMismatch error, got {report:?}"));
    // The rendered finding names the block and the dead-ECC condition.
    let rendered = mismatch.to_string();
    assert!(
        rendered.contains("parity trailer") && rendered.contains("offset 0"),
        "display names the block and the condition: {rendered}",
    );
}

/// Returns the on-disk path of the first SST registered with the
/// tree's current version. Drops the tree before returning so the
/// caller can mutate the file safely (no descriptor cache, no
/// file lock). Going through `current_version().iter_tables()`
/// instead of a filesystem walk keeps the test coupled to the
/// verifier's actual input set — a new on-disk file under the
/// tree directory cannot accidentally become the corruption
/// target.
fn pick_first_sst_path(dir: &std::path::Path) -> std::path::PathBuf {
    let tree = reopen_tree(dir);
    let path = tree
        .current_version()
        .iter_tables()
        .next()
        .map(|table| (*table.path).clone())
        .expect("at least one populated SST file");
    drop(tree);
    path
}

#[test]
fn verify_block_checksums_detects_flipped_byte_in_data_block() {
    use crate::table::block::Header;
    let dir = tempfile::tempdir().unwrap();
    populate_tree(dir.path(), 1_000);

    let sst_path = pick_first_sst_path(dir.path());

    // The flip target is the first byte AFTER the first block's
    // Header — that lands squarely inside the data segment of the
    // first data block, so the header's own XXH3 stays valid (no
    // HeaderCorrupted) but the data XXH3 will now mismatch.
    let flip_offset = Header::MIN_LEN as u64;
    {
        let mut f = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&sst_path)
            .unwrap();
        f.seek(SeekFrom::Start(flip_offset)).unwrap();
        let mut byte = [0u8; 1];
        f.read_exact(&mut byte).unwrap();
        byte[0] ^= 0xFF;
        f.seek(SeekFrom::Start(flip_offset)).unwrap();
        f.write_all(&byte).unwrap();
        f.sync_all().unwrap();
    }

    let tree = reopen_tree(dir.path());
    let report = verify_block_checksums(&tree);
    assert!(
        !report.is_ok(),
        "expected corruption to surface as report errors, got {report:?}",
    );
    let has_data_corruption = report.errors.iter().any(|e| {
        matches!(
            e,
            BlockVerifyError::DataCorrupted { path, .. } if path == &sst_path,
        )
    });
    assert!(
        has_data_corruption,
        "expected a DataCorrupted error for {}, got {:?}",
        sst_path.display(),
        report.errors,
    );
}

#[test]
fn verify_kv_checksums_clean_kv_checked_tree_passes() {
    // A tree written with per-KV checksums enabled must pass the
    // per-KV scrub with no error.
    let dir = tempfile::tempdir().unwrap();
    populate_tree_kv_checked(dir.path(), 500);

    let tree = reopen_tree(dir.path());
    let crate::AnyTree::Standard(tree) = tree else {
        panic!("expected Standard tree");
    };
    verify_kv_checksums(&tree).expect("clean kv-checked tree must pass per-KV scrub");
}

#[test]
fn verify_kv_checked_detects_corrupted_digest_under_valid_block_checksum() {
    // The per-KV verifier must catch a divergence that the block-level
    // XXH3 does NOT: corrupt a stored digest, then write the block so its
    // block-level checksum is computed over the already-corrupted bytes.
    // The block therefore loads cleanly (block checksum valid) and only
    // the per-KV recompute disagrees. (Flipping a payload byte in the
    // file without recomputing the block checksum would be caught at
    // load, never reaching the per-KV path — which would prove nothing.)
    use crate::InternalValue;
    use crate::ValueType::Value;
    use crate::comparator::default_comparator;
    use crate::runtime_config::ChecksumAlgorithm;
    use crate::table::block::header::block_flags;
    use crate::table::block::{Block, BlockIdentity, BlockTransform, BlockType, kv_checksum};
    use crate::table::data_block::DataBlock;

    let algo = ChecksumAlgorithm::Xxh3_64;
    let items = [
        InternalValue::from_components(b"alpha".to_vec(), b"one".to_vec(), 3, Value),
        InternalValue::from_components(b"bravo".to_vec(), b"two".to_vec(), 2, Value),
    ];
    let digests: Vec<u64> = items
        .iter()
        .map(|it| kv_checksum::kv_digest(it, algo).expect("xxh3 always available"))
        .collect();

    let mut payload = Vec::new();
    DataBlock::encode_kv_checked_into(&mut payload, &items, &digests, algo, 2, 0.0).unwrap();

    // Corrupt the first stored digest (its first byte sits right after the
    // inner payload, where the digest array begins).
    let inner_len = kv_checksum::split_inner(&payload).unwrap().len();
    *payload.get_mut(inner_len).expect("digest array byte") ^= 0xFF;

    // Write with a VALID block-level checksum over the corrupted payload.
    // Data blocks omit the `block_flags` byte (footer presence is a per-SST
    // descriptor property), so the KV_CHECKSUM_FOOTER flag passed here is
    // dropped on encode — the footer rides inside the payload structurally,
    // and `verify_kv_checked` splits it without consulting the header bit.
    let id = BlockIdentity::for_test(0, BlockType::Data);
    let at = crate::table::block::ChecksumAt::table(0, 0);
    let mut buf = Vec::new();
    Block::write_into_with_flags(
        &mut buf,
        &payload,
        id,
        &BlockTransform::PLAIN,
        block_flags::KV_CHECKSUM_FOOTER,
        at,
    )
    .unwrap();

    // Block loads fine (block-level checksum matches the corrupted bytes).
    let block = Block::from_reader(&mut &buf[..], id, &BlockTransform::PLAIN, at).unwrap();

    // Only the per-KV verifier catches the bad digest. `None` skips the
    // algorithm cross-check — this test exercises the digest-mismatch path.
    let err = DataBlock::verify_kv_checked(&block.data, block.header, default_comparator(), None)
        .expect_err("corrupted stored digest must fail the per-KV verifier");
    assert!(
        matches!(err, crate::Error::ChecksumMismatch { .. }),
        "expected ChecksumMismatch, got {err:?}"
    );
}

#[test]
fn verify_kv_checked_rejects_non_data_block_type() {
    // The scrub verifies only Data blocks. A header whose block_type is
    // not Data (corruption or a caller bug) must be rejected with
    // InvalidTag, not silently coerced to Data and verified as if it were
    // a data block — coercion would defeat the scrub's purpose.
    use crate::InternalValue;
    use crate::ValueType::Value;
    use crate::comparator::default_comparator;
    use crate::runtime_config::ChecksumAlgorithm;
    use crate::table::block::header::block_flags;
    use crate::table::block::{Block, BlockIdentity, BlockTransform, BlockType, kv_checksum};
    use crate::table::data_block::DataBlock;

    let algo = ChecksumAlgorithm::Xxh3_64;
    let items = [
        InternalValue::from_components(b"alpha".to_vec(), b"one".to_vec(), 3, Value),
        InternalValue::from_components(b"bravo".to_vec(), b"two".to_vec(), 2, Value),
    ];
    let digests: Vec<u64> = items
        .iter()
        .map(|it| kv_checksum::kv_digest(it, algo).expect("xxh3 always available"))
        .collect();

    let mut payload = Vec::new();
    DataBlock::encode_kv_checked_into(&mut payload, &items, &digests, algo, 2, 0.0).unwrap();

    let id = BlockIdentity::for_test(0, BlockType::Data);
    let at = crate::table::block::ChecksumAt::table(0, 0);
    let mut buf = Vec::new();
    Block::write_into_with_flags(
        &mut buf,
        &payload,
        id,
        &BlockTransform::PLAIN,
        block_flags::KV_CHECKSUM_FOOTER,
        at,
    )
    .unwrap();
    let block = Block::from_reader(&mut &buf[..], id, &BlockTransform::PLAIN, at).unwrap();

    // The footer + inner bytes form a valid data block, so only the
    // block_type gate can catch a tampered type: flip it to a non-Data
    // variant and require InvalidTag.
    let mut bad_header = block.header;
    bad_header.block_type = BlockType::Index;
    let err = DataBlock::verify_kv_checked(&block.data, bad_header, default_comparator(), None)
        .expect_err("non-Data block_type must be rejected, not coerced");
    assert!(
        matches!(err, crate::Error::InvalidTag(("BlockType", _))),
        "expected InvalidTag(BlockType), got {err:?}"
    );
}

/// Exercises the out-of-band wrapper on a real clean SST file.
/// `verify_sst_file` is the entry point sst-dump calls; this pins
/// that it stamps `sst_files_scanned = 1`, reports no errors on a
/// healthy file, and propagates the block count through the
/// `StdFs` -> `scan_sst_blocks` -> `BlockVerifyReport` path.
#[test]
fn verify_sst_file_clean_file_has_no_errors() {
    let dir = tempfile::tempdir().unwrap();
    populate_tree(dir.path(), 1_000);
    let sst_path = pick_first_sst_path(dir.path());

    let report = verify_sst_file(&sst_path);
    assert!(
        report.is_ok(),
        "expected clean SST to verify with zero errors, got {:?}",
        report.errors,
    );
    assert_eq!(
        report.sst_files_scanned, 1,
        "wrapper must always stamp sst_files_scanned = 1",
    );
    assert!(
        report.blocks_scanned > 0,
        "expected at least one block scanned in a populated SST",
    );
}

/// A re-stamped TOC that OMITS a correctness-bearing entry entirely (the SFA
/// trailer checksum is unkeyed) must fail the walk closed: `delete_bitmap` /
/// `range_tombstones` are optional at parse time, so a vanished section
/// resurrects deleted rows while every remaining block still passes its
/// byte-level checks. The writer emits sections strictly back-to-back, so the
/// gap the omission leaves in the section tiling is the only out-of-band
/// trace.
#[test]
fn verify_sst_file_flags_an_omitted_toc_section() {
    let dir = tempfile::tempdir().unwrap();
    {
        let cfg = Config::new(
            dir.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
        let tree = cfg.open().unwrap();
        for i in 0u64..100 {
            let key = format!("k{i:08}");
            tree.insert(key.as_bytes(), b"v", 1 + i);
        }
        // A range tombstone gives the SST the optional `range_tombstones`
        // section whose omission the walk must catch.
        tree.remove_range("k00000010", "k00000020", 200);
        tree.flush_active_memtable(300).unwrap();
        drop(tree);
    }
    let sst_path = pick_first_sst_path(dir.path());

    // Sanity: intact file verifies clean.
    let report = verify_sst_file(&sst_path);
    assert!(
        report.is_ok(),
        "intact SST must be clean: {:?}",
        report.errors
    );

    crate::test_forge::forge_section_omitted(&sst_path, b"range_tombstones").unwrap();

    let report = verify_sst_file(&sst_path);
    assert!(
        report.errors.iter().any(|e| matches!(
            e,
            BlockVerifyError::TocCorrupted { reason, .. }
                if reason.contains("the gap hides an omitted TOC entry")
        )),
        "an omitted TOC entry leaves a tiling gap the walk must flag with the \
         gap-specific TocCorrupted reason, got {:?}",
        report.errors,
    );
}

/// A tail meta mirror re-stamped with an UNRECOGNIZED ECC descriptor must still
/// be compared against `meta_mid` for its non-ECC fields. Excluding it from the
/// full mirror comparison lets a forge change a correctness field (here
/// `created_at`) behind the unknown descriptor and evade the divergence check:
/// a backdated `created_at`, selected by tail-first recovery, can make FIFO /
/// TTL compaction discard live data. Both copies decode, so the comparison must
/// see the tail's changed field even though its ECC descriptor is unknown.
#[test]
fn verify_sst_file_flags_diverging_mirrors_behind_an_unrecognized_ecc() {
    let dir = tempfile::tempdir().unwrap();
    populate_tree(dir.path(), 200);
    let sst_path = pick_first_sst_path(dir.path());

    // Sanity: intact file verifies clean.
    let report = verify_sst_file(&sst_path);
    assert!(
        report.is_ok(),
        "intact SST must be clean: {:?}",
        report.errors
    );

    // Forge ONLY the tail mirror: an UNRECOGNIZED ECC descriptor (`[9,_,_,_]` =
    // unknown kind, which would exclude the mirror from the comparison) AND a
    // changed `created_at` (the correctness field the forge hides behind it).
    // meta_mid stays intact, so the two now decode to different metadata.
    crate::test_forge::forge_tail_meta_value(&sst_path, b"descriptor#page_ecc", &[9, 0, 0, 0])
        .unwrap();
    crate::test_forge::forge_tail_meta_value(&sst_path, b"created_at", &[0xFF; 16]).unwrap();

    let report = verify_sst_file(&sst_path);
    assert!(
        report.errors.iter().any(|e| matches!(
            e,
            BlockVerifyError::TocCorrupted { reason, .. }
                if reason.contains("mirrors decode to different metadata")
        )),
        "a tail mirror that changed created_at behind an unrecognized ECC \
         descriptor must still diverge from meta_mid, got {:?}",
        report.errors,
    );
}

/// A table that REALLY carries parity and whose both meta mirrors advertise an
/// unrecognized ECC descriptor (they agree, so no divergence) cannot have its
/// ECC-bearing sections walked: the trailer length is underivable, and the one
/// remaining candidate — `Off` — cannot frame a file whose blocks each carry a
/// trailer. The walk SKIPS those sections, so the report must be INCOMPLETE and
/// `is_ok()` false; a clean verdict would falsely claim the data verified.
///
/// The parity is what makes this table unreadable rather than the descriptors:
/// on a parity-LESS table the same forge is survivable, because `Off` frames it
/// end to end.
#[cfg(feature = "page_ecc")]
#[test]
fn verify_sst_file_reports_incomplete_when_ecc_is_unrecognized_in_both_mirrors() -> crate::Result<()>
{
    use crate::table::Writer;
    use crate::table::block::EccParams;

    let dir = tempfile::tempdir()?;
    let sst_path = dir.path().join("t");

    let mut writer = Writer::new(
        sst_path.clone(),
        0,
        0,
        std::sync::Arc::new(crate::fs::StdFs),
    )?
    .use_ecc(Some(EccParams::RS_4_2));
    for i in 0u64..200 {
        writer.write(crate::InternalValue::from_components(
            format!("key-{i:05}").into_bytes(),
            format!("value-{i:05}").into_bytes(),
            i + 1,
            crate::ValueType::Value,
        ))?;
    }
    assert!(writer.finish()?.is_some(), "the fixture is non-empty");

    // Sanity: intact file verifies clean.
    let report = verify_sst_file(&sst_path);
    assert!(
        report.is_ok(),
        "intact SST must be clean: {:?}",
        report.errors
    );

    // Forge BOTH mirrors' ECC descriptor to an unknown kind: the two agree (no
    // mirror divergence), but neither can be applied, so the walk skips the SST
    // block sections.
    crate::test_forge::forge_meta_value_both_mirrors(
        &sst_path,
        b"descriptor#page_ecc",
        &[9, 0, 0, 0],
    )?;

    let report = verify_sst_file(&sst_path);
    assert!(
        report.errors.is_empty(),
        "no corruption, only an unwalkable ECC scheme: {:?}",
        report.errors,
    );
    assert!(
        report.incomplete,
        "the walk skipped the data blocks, so the scan is incomplete",
    );
    assert!(
        !report.is_ok(),
        "an incomplete scan (data blocks never verified) must not report OK",
    );
    assert!(
        report
            .warnings
            .iter()
            .any(|w| matches!(w, crate::verify::BlockVerifyWarning::UnrecognizedEcc { .. })),
        "the unrecognized-ECC warning must still be recorded: {:?}",
        report.warnings,
    );
    Ok(())
}

/// Tree-level regression for the merged report: `verify_block_checksums` folds
/// each per-SST partial report through `merge_report`. An SST with an
/// unrecognized ECC descriptor produces an INCOMPLETE partial (its data blocks
/// were skipped unwalked), so the merged whole-tree report must stay incomplete
/// and `is_ok()` false. If the merge drops the `incomplete` flag, the final
/// report reverts to `false` and falsely reports OK with no other errors.
/// SEVERAL SSTs, only one forged: the clean partials folded after the
/// incomplete one would launder the flag under an OVERWRITING (rather than
/// OR-ing) merge, so the multi-SST shape pins the accumulation itself.
#[cfg(feature = "std")]
#[test]
fn verify_block_checksums_stays_incomplete_when_one_sst_has_unrecognized_ecc() {
    let dir = tempfile::tempdir().unwrap();
    populate_multi_sst(dir.path(), 3, 200);

    // Sanity: intact tree verifies clean.
    let tree = reopen_tree(dir.path());
    let report = verify_block_checksums(&tree);
    assert!(
        report.is_ok(),
        "intact tree must be clean: {:?}",
        report.errors
    );
    drop(tree);

    // Forge the first SST's ECC descriptor (both mirrors) to an unknown kind:
    // its data-block sections become unwalkable, so its per-SST scan is
    // incomplete. The whole-tree fold must PRESERVE that incompleteness.
    let sst_path = pick_first_sst_path(dir.path());
    crate::test_forge::forge_meta_value_both_mirrors(
        &sst_path,
        b"descriptor#page_ecc",
        &[9, 0, 0, 0],
    )
    .unwrap();

    let tree = reopen_tree(dir.path());
    let report = verify_block_checksums(&tree);
    assert!(
        report.errors.is_empty(),
        "no corruption, only an unwalkable ECC scheme: {:?}",
        report.errors,
    );
    assert!(
        report.incomplete,
        "the merged report must inherit the incomplete flag from the skipped SST",
    );
    assert!(
        !report.is_ok(),
        "a merged report that skipped a whole SST's data blocks must not report OK",
    );
}

/// A TOC entry RENAMED to a duplicate recognized name — `range_tombstones`
/// renamed to a second `data`, its block header re-stamped with the `Data`
/// role — preserves the tiling AND passes the recognized-role walk, yet the
/// reader's name lookup (`Toc::section`) returns the first match, hiding the
/// renamed section so the deleted range silently resurrects. The duplicate
/// name is the only trace; the walk must reject it.
#[test]
fn verify_sst_file_flags_a_duplicate_toc_section_name() {
    let dir = tempfile::tempdir().unwrap();
    {
        let cfg = Config::new(
            dir.path(),
            SequenceNumberCounter::default(),
            SequenceNumberCounter::default(),
        )
        .data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
        let tree = cfg.open().unwrap();
        for i in 0u64..100 {
            let key = format!("k{i:08}");
            tree.insert(key.as_bytes(), b"v", 1 + i);
        }
        tree.remove_range("k00000010", "k00000020", 200);
        tree.flush_active_memtable(300).unwrap();
        drop(tree);
    }
    let sst_path = pick_first_sst_path(dir.path());

    // Sanity: intact file verifies clean.
    let report = verify_sst_file(&sst_path);
    assert!(
        report.is_ok(),
        "intact SST must be clean: {:?}",
        report.errors
    );

    crate::test_forge::forge_duplicate_section_name(
        &sst_path,
        b"range_tombstones",
        b"data",
        crate::table::block::BlockType::Data,
    )
    .unwrap();

    let report = verify_sst_file(&sst_path);
    // Match the duplicate-name finding specifically: the forge rewrites the
    // whole TOC, so an unrelated TocCorrupted (tiling gap, seek failure,
    // unrecognized name) would keep this green without proving the
    // duplicate-name detection ran.
    assert!(
        report.errors.iter().any(|e| matches!(
            e,
            BlockVerifyError::TocCorrupted { section_name, reason, .. }
                if section_name == b"data" && reason.contains("duplicate TOC section name")
        )),
        "a duplicate recognized section name must be flagged as TocCorrupted, \
         got {:?}",
        report.errors,
    );
}

/// A healthy SST with a PARTITIONED Bloom filter must verify clean: the
/// writer emits the `filter_tli` block with the Index role (it is a
/// top-level index over filter partitions, same encoding as the data
/// TLI), so a role map expecting Filter there flags a role mismatch on an
/// intact file — and `repair_with_salvage` would grade the healthy table
/// corrupt and drop or re-salvage it.
#[test]
fn verify_sst_file_accepts_a_partitioned_filter() {
    use crate::InternalValue;
    use crate::ValueType::Value;
    use crate::table::Writer;
    use std::sync::Arc;

    let dir = tempfile::tempdir().unwrap();
    let sst_path = dir.path().join("partitioned");
    let mut writer = Writer::new(sst_path.clone(), 0, 0, Arc::new(crate::fs::StdFs))
        .unwrap()
        .use_partitioned_filter()
        // A tiny partition budget so several filter partitions spill and
        // the writer emits the `filter_tli` top-level index over them.
        .use_meta_partition_size(3);
    for i in 0u64..64 {
        writer
            .write(InternalValue::from_components(
                format!("key-{i:03}").into_bytes(),
                format!("val-{i:03}").into_bytes(),
                i + 1,
                Value,
            ))
            .unwrap();
    }
    assert!(writer.finish().unwrap().is_some(), "SST is non-empty");

    // Sanity: the layout actually carries the partitioned filter TLI.
    {
        let mut f = std::fs::File::open(&sst_path).unwrap();
        let reader = crate::sfa::Reader::from_reader(&mut f).unwrap();
        assert!(
            reader.toc().iter().any(|e| e.name() == b"filter_tli"),
            "the fixture must produce a filter_tli section",
        );
    }

    let report = verify_sst_file(&sst_path);
    assert!(
        report.is_ok(),
        "a healthy partitioned-filter SST must verify clean: {:?}",
        report.errors,
    );
}

/// Exercises the file-open failure branch (the only path through
/// `verify_sst_file` that converts an underlying `io::Error` into
/// a `BlockVerifyError::SstFileUnreadable`). A missing file is the
/// simplest trigger; an unreadable-due-to-permissions trigger
/// would require root or chmod-induced state and is overkill for
/// pinning the variant routing.
#[test]
fn verify_sst_file_missing_file_reports_unreadable() {
    // Build the missing-file path under a fresh tempdir so it
    // resolves the same way on Linux / macOS / Windows runners.
    // A hardcoded Unix-style absolute path would either skip the
    // test on Windows (no `/this/...` semantics) or risk a flaky
    // pass if the path happened to exist.
    let dir = tempfile::tempdir().unwrap();
    let missing_path = dir.path().join("does-not-exist-sst-12345.sst");
    // Sanity: tempdir() guarantees the directory is empty.
    assert!(
        !missing_path.exists(),
        "tempdir entry must be absent for this test to exercise the missing-file branch",
    );

    let report = verify_sst_file(&missing_path);
    assert_eq!(
        report.sst_files_scanned, 1,
        "wrapper stamps sst_files_scanned = 1 even on file-open failure \
         so callers see the attempt was made",
    );
    assert_eq!(
        report.blocks_scanned, 0,
        "no blocks could be walked because the file couldn't be opened",
    );
    assert_eq!(
        report.errors.len(),
        1,
        "expected exactly one error, got {:?}",
        report.errors,
    );
    let err = report.errors.first().unwrap();
    assert!(
        matches!(
            err,
            BlockVerifyError::SstFileUnreadable { table_id: 0, path, .. }
                if path == &missing_path,
        ),
        "expected SstFileUnreadable for {}, got {err:?}",
        missing_path.display(),
    );
}

/// Pins the routing of post-header short-read failures to
/// `BlockVerifyError::DataReadError`. Regression guard for #315:
/// a refactor that collapses the `read_exact` failure branch back
/// into `HeaderCorrupted` (which is what a naive "any read error
/// inside the walker is a header problem" cleanup would do) loses
/// the distinction between "the file's TOC lies about where the
/// section ends" and "the header itself fails its own XXH3", and
/// downstream tooling (`sst-dump`, `repair_db`, lazy block repair)
/// pattern-matches on the variant to decide whether the block is
/// recoverable. Demoting truncated-data to `HeaderCorrupted` would
/// make those tools fall back to whole-section discard instead of
/// per-block surgery.
///
/// Setup forges an SFA archive whose `data` TOC entry claims a
/// section length large enough for one full block (header + N
/// bytes), but the underlying file contains only the header.
/// Result: `Header::decode_from` succeeds (the header's XXH3
/// matches its own bytes), the bounds check passes (`data_length`
/// fits within the lied section length), and the data-segment
/// `read_exact` hits EOF after consuming a handful of trailing
/// TOC + trailer bytes. The only valid landing variant is
/// `DataReadError`.
#[test]
#[expect(
    clippy::indexing_slicing,
    clippy::cast_possible_truncation,
    reason = "synthetic SFA forgery — offsets are all in-bounds by \
              construction (we just wrote the bytes ourselves), and \
              the u64 -> usize cast cannot overflow on any target \
              the test runs on (the forged archive is < 1 KiB)"
)]
fn walk_block_region_reports_data_read_error_on_truncated_data_segment() -> crate::Result<()> {
    use crate::coding::Encode;
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::{BlockType, Header};

    // Trailer layout (38 bytes at the tail of an SFA archive):
    //   MAGIC(4) | version(1) | csum_type(1) | toc_checksum(16) | toc_pos(8) | toc_len(8)
    const TRAILER_LEN: usize = 4 + 1 + 1 + 16 + 8 + 8;
    const DATA_LENGTH: u32 = 4096;
    const HEADER_LEN: u64 = Header::MIN_LEN as u64;

    let header = Header {
        // Arbitrary sentinel; the walker reaches `read_exact` and
        // bails BEFORE any data-segment XXH3 comparison, so this
        // value is never checked.
        stored_checksum: Checksum::from_raw(0xDEAD_BEEF_DEAD_BEEF),
        data_length: DATA_LENGTH,
        uncompressed_length: DATA_LENGTH,
        ..Header::test_dummy(BlockType::Data)
    };

    // Build a minimal SFA archive: one section "data" containing
    // exactly one Header (33 bytes) and zero following data bytes.
    let mut archive_bytes: Vec<u8> = Vec::new();
    {
        let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
        writer.start("data").unwrap();
        writer.write_all(&header.encode_into_vec()).unwrap();
        writer.finish().unwrap();
    }

    // Parse the trailer at the file tail.
    let trailer_start = archive_bytes.len() - TRAILER_LEN;
    let toc_pos_bytes: [u8; 8] = archive_bytes[trailer_start + 22..trailer_start + 30]
        .try_into()
        .unwrap();
    let toc_len_bytes: [u8; 8] = archive_bytes[trailer_start + 30..trailer_start + 38]
        .try_into()
        .unwrap();
    let toc_pos = u64::from_le_bytes(toc_pos_bytes) as usize;
    let toc_len = u64::from_le_bytes(toc_len_bytes) as usize;

    // TOC payload layout: `TOC!`(4) | entry_count(4 LE) | entries.
    // Each entry: pos(8 LE) | len(8 LE) | name_len(2 LE) | name.
    // The first (only) entry begins at toc_pos + 8.
    let first_entry_offset = toc_pos + 4 + 4;
    let len_field_offset = first_entry_offset + 8;

    // Inflate the section length so end_offset = HEADER_LEN +
    // DATA_LENGTH. The walker then computes remaining = DATA_LENGTH
    // (passes the bounds check), tries to `read_exact(DATA_LENGTH)`,
    // and hits EOF after the few trailing TOC + trailer bytes.
    let lied_len: u64 = HEADER_LEN + u64::from(DATA_LENGTH);
    archive_bytes[len_field_offset..len_field_offset + 8].copy_from_slice(&lied_len.to_le_bytes());

    // Recompute the TOC checksum (xxh3_128 over the TOC payload)
    // and patch the trailer's stored checksum so crate::sfa::Reader still
    // accepts the file.
    let new_toc_checksum = crate::hash::hash128(&archive_bytes[toc_pos..toc_pos + toc_len]);
    let csum_field_offset = trailer_start + 4 + 1 + 1;
    archive_bytes[csum_field_offset..csum_field_offset + 16]
        .copy_from_slice(&new_toc_checksum.to_le_bytes());

    // Materialize the forged archive on a real temp file and run the scanner.
    let dir = tempfile::tempdir()?;
    let fs = StdFs;
    let forged = dir.path().join("forged.sst");
    let path = forged.as_path();
    {
        let mut f = fs.open(
            path,
            &FsOpenOptions::new().write(true).create(true).truncate(true),
        )?;
        f.write_all(&archive_bytes)?;
    }

    let table_id: TableId = 42;
    let scan = scan_sst_blocks(&fs, path, table_id, SstLayout::PLAIN, None)?;
    // The inflated section length ALSO breaks the TOC tiling invariant
    // (the declared section end runs past where the TOC begins), so the
    // walk reports the tiling finding alongside the read error.
    assert_eq!(
        scan.errors.len(),
        2,
        "expected the tiling finding plus the read error, got {:?}",
        scan.errors,
    );
    assert!(
        scan.errors
            .iter()
            .any(|e| matches!(e, BlockVerifyError::TocCorrupted { .. })),
        "the inflated section length must break the TOC tiling: {:?}",
        scan.errors,
    );
    assert!(
        scan.errors.iter().any(|err| matches!(
            err,
            BlockVerifyError::DataReadError {
                table_id: t,
                offset: 0,
                data_length: d,
                ..
            } if *t == table_id && *d == DATA_LENGTH,
        )),
        "expected DataReadError {{ table_id: {table_id}, offset: 0, \
         data_length: {DATA_LENGTH}, .. }}; got {:?}",
        scan.errors,
    );
    assert_eq!(
        scan.blocks_scanned, 1,
        "header decoded successfully, so blocks_scanned must count this block \
         even though the data segment read failed",
    );
    Ok(())
}

/// A rated scan is charged the bytes it reads, never a length the TOC
/// declares: a re-stamped TOC that gives a raw section a terabyte must not
/// make the scan wait out that terabyte before it reports the TOC corrupt.
#[test]
#[expect(
    clippy::indexing_slicing,
    clippy::cast_possible_truncation,
    reason = "synthetic SFA forgery: the offsets are in bounds by construction and the \
              archive is under 1 KiB"
)]
fn a_rated_scan_charges_the_bytes_read_not_a_forged_section_length() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, StdFs};

    const TRAILER_LEN: usize = 4 + 1 + 1 + 16 + 8 + 8;
    let mut archive_bytes: Vec<u8> = Vec::new();
    {
        let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
        writer.start("meta_separator").unwrap();
        writer.write_all(&[0u8; 16]).unwrap();
        writer.finish().unwrap();
    }
    let trailer_start = archive_bytes.len() - TRAILER_LEN;
    let field = |at: usize| -> usize {
        u64::from_le_bytes(archive_bytes[at..at + 8].try_into().unwrap()) as usize
    };
    let (toc_pos, toc_len) = (field(trailer_start + 22), field(trailer_start + 30));
    // The only entry's length, past `TOC!`, the entry count and its position.
    let len_at = toc_pos + 4 + 4 + 8;
    archive_bytes[len_at..len_at + 8].copy_from_slice(&(1u64 << 40).to_le_bytes());
    let checksum = crate::hash::hash128(&archive_bytes[toc_pos..toc_pos + toc_len]);
    let checksum_at = trailer_start + 4 + 1 + 1;
    archive_bytes[checksum_at..checksum_at + 16].copy_from_slice(&checksum.to_le_bytes());

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("forged.sst");
    {
        let mut f = StdFs.open(
            &path,
            &FsOpenOptions::new().write(true).create(true).truncate(true),
        )?;
        f.write_all(&archive_bytes)?;
    }

    let limiter = crate::rate_limiter::RateLimiter::new(1 << 20);
    let start = std::time::Instant::now();
    let scan = scan_sst_blocks(&StdFs, &path, 9, SstLayout::PLAIN, Some(&limiter))?;
    let elapsed = start.elapsed();
    assert!(
        elapsed < std::time::Duration::from_secs(2),
        "the scan waited {elapsed:?} on a forged section length"
    );
    assert!(
        scan.errors
            .iter()
            .any(|e| matches!(e, BlockVerifyError::TocCorrupted { .. })),
        "the forged length is still reported: {:?}",
        scan.errors,
    );
    Ok(())
}

/// A file that hands back at most 512 bytes a read, as a backend may.
struct ShortReads(Box<dyn crate::fs::FsFile>);

impl std::io::Read for ShortReads {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let cap = buf.len().min(512);
        match buf.get_mut(..cap) {
            Some(head) => self.0.read(head),
            None => Ok(0),
        }
    }
}

impl std::io::Write for ShortReads {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.0.write(buf)
    }
    fn flush(&mut self) -> std::io::Result<()> {
        self.0.flush()
    }
}

impl std::io::Seek for ShortReads {
    fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
        self.0.seek(to)
    }
}

impl crate::fs::FsFile for ShortReads {
    fn sync_all(&self) -> crate::io::Result<()> {
        self.0.sync_all()
    }
    fn sync_data(&self) -> crate::io::Result<()> {
        self.0.sync_data()
    }
    fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
        self.0.metadata()
    }
    fn set_len(&self, size: u64) -> crate::io::Result<()> {
        self.0.set_len(size)
    }
    fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
        self.0.read_at(buf, offset)
    }
    fn lock_exclusive(&self) -> crate::io::Result<()> {
        self.0.lock_exclusive()
    }
}

/// A backend that returns fewer bytes than asked costs the budget the bytes it
/// reads: the buffered walk asks again for the rest, and a read already
/// charged is not charged a second time.
#[test]
fn a_paced_file_charges_short_reads_once() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, MemFs};

    let fs = MemFs::new();
    let path = std::path::Path::new("/f");
    let bytes: Vec<u8> = (0..64 * 1024u32).map(|i| (i % 251) as u8).collect();
    {
        let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
        f.write_all(&bytes)?;
    }
    // A second of rate is the whole file: charged once, the read never waits.
    let limiter = crate::rate_limiter::RateLimiter::new(bytes.len() as u64);
    let file = ShortReads(fs.open(path, &FsOpenOptions::new().read(true))?);
    let mut reader = std::io::BufReader::with_capacity(
        64 * 1024,
        PacedFile::new(Box::new(file), Some(&limiter))?,
    );
    let mut read = vec![0u8; bytes.len()];
    let start = std::time::Instant::now();
    reader.read_exact(&mut read[..])?;
    let elapsed = start.elapsed();
    assert_eq!(read, bytes, "the file reads back whole");
    assert!(
        elapsed < std::time::Duration::from_millis(500),
        "short reads were charged more than once: the read waited {elapsed:?}"
    );
    Ok(())
}

/// A file that hands back at most 512 bytes a read and, every other read,
/// `Interrupted` instead, as a backend may when a signal lands mid-read.
struct InterruptedReads {
    inner: Box<dyn crate::fs::FsFile>,
    calls: usize,
}

impl std::io::Read for InterruptedReads {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.calls += 1;
        if self.calls.is_multiple_of(2) {
            return Err(std::io::ErrorKind::Interrupted.into());
        }
        let cap = buf.len().min(512);
        match buf.get_mut(..cap) {
            Some(head) => self.inner.read(head),
            None => Ok(0),
        }
    }
}

impl std::io::Write for InterruptedReads {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.inner.write(buf)
    }
    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

impl std::io::Seek for InterruptedReads {
    fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
        self.inner.seek(to)
    }
}

impl crate::fs::FsFile for InterruptedReads {
    fn sync_all(&self) -> crate::io::Result<()> {
        self.inner.sync_all()
    }
    fn sync_data(&self) -> crate::io::Result<()> {
        self.inner.sync_data()
    }
    fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
        self.inner.metadata()
    }
    fn set_len(&self, size: u64) -> crate::io::Result<()> {
        self.inner.set_len(size)
    }
    fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
        self.inner.read_at(buf, offset)
    }
    fn lock_exclusive(&self) -> crate::io::Result<()> {
        self.inner.lock_exclusive()
    }
}

/// A file that hands back at most `cap` bytes a read, records the largest read
/// asked of it, and fails the read numbered `fail_at` once.
struct ScriptedReads {
    inner: Box<dyn crate::fs::FsFile>,
    cap: usize,
    fail_at: Option<usize>,
    calls: usize,
    largest: std::sync::Arc<std::sync::atomic::AtomicUsize>,
}

impl std::io::Read for ScriptedReads {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.calls += 1;
        self.largest
            .fetch_max(buf.len(), std::sync::atomic::Ordering::Relaxed);
        if self.fail_at == Some(self.calls) {
            return Err(std::io::ErrorKind::Other.into());
        }
        let cap = buf.len().min(self.cap);
        match buf.get_mut(..cap) {
            Some(head) => self.inner.read(head),
            None => Ok(0),
        }
    }
}

impl std::io::Write for ScriptedReads {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.inner.write(buf)
    }
    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

impl std::io::Seek for ScriptedReads {
    fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
        self.inner.seek(to)
    }
}

impl crate::fs::FsFile for ScriptedReads {
    fn sync_all(&self) -> crate::io::Result<()> {
        self.inner.sync_all()
    }
    fn sync_data(&self) -> crate::io::Result<()> {
        self.inner.sync_data()
    }
    fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
        self.inner.metadata()
    }
    fn set_len(&self, size: u64) -> crate::io::Result<()> {
        self.inner.set_len(size)
    }
    fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
        self.inner.read_at(buf, offset)
    }
    fn lock_exclusive(&self) -> crate::io::Result<()> {
        self.inner.lock_exclusive()
    }
}

/// A file that logs every positioned read, `('r', offset, len)`, into a log
/// a pacer can share.
struct LoggedReadAt {
    inner: Box<dyn crate::fs::FsFile>,
    log: std::sync::Arc<std::sync::Mutex<Vec<(char, u64, u64)>>>,
}

impl std::io::Read for LoggedReadAt {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.inner.read(buf)
    }
}

impl std::io::Write for LoggedReadAt {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.inner.write(buf)
    }
    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

impl std::io::Seek for LoggedReadAt {
    fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
        self.inner.seek(to)
    }
}

impl crate::fs::FsFile for LoggedReadAt {
    fn sync_all(&self) -> crate::io::Result<()> {
        self.inner.sync_all()
    }
    fn sync_data(&self) -> crate::io::Result<()> {
        self.inner.sync_data()
    }
    fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
        self.inner.metadata()
    }
    fn set_len(&self, size: u64) -> crate::io::Result<()> {
        self.inner.set_len(size)
    }
    fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
        self.log
            .lock()
            .expect("lock is not poisoned")
            .push(('r', offset, buf.len() as u64));
        self.inner.read_at(buf, offset)
    }
    fn lock_exclusive(&self) -> crate::io::Result<()> {
        self.inner.lock_exclusive()
    }
}

/// A paced read larger than a portion alternates its waits and its reads: each
/// portion is paced and then read, so a limiter's wait never comes all at
/// once before one read of the whole. An unpaced read stays one read.
#[test]
fn a_paced_read_alternates_each_portion_with_its_read() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, MemFs};

    let portion = PACE_PORTION as u64;
    let size = 2 * PACE_PORTION + 100;
    let fs = MemFs::new();
    let path = std::path::Path::new("/f");
    {
        let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
        f.write_all(&vec![7u8; size])?;
    }
    let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
    let file = LoggedReadAt {
        inner: fs.open(path, &FsOpenOptions::new().read(true))?,
        log: std::sync::Arc::clone(&log),
    };
    let pace = LogPacer {
        active: true,
        log: std::sync::Arc::clone(&log),
    };

    let read = crate::file::read_exact_paced(&file, 0, size, Some(&pace))?;
    assert_eq!(read.len(), size);
    let tail = size as u64 - 2 * portion;
    assert_eq!(
        *log.lock().expect("lock is not poisoned"),
        vec![
            ('p', 0, portion),
            ('r', 0, portion),
            ('p', portion, portion),
            ('r', portion, portion),
            ('p', 2 * portion, tail),
            ('r', 2 * portion, tail),
        ]
    );

    log.lock().expect("lock is not poisoned").clear();
    crate::file::read_exact(&file, 0, size)?;
    assert_eq!(
        *log.lock().expect("lock is not poisoned"),
        vec![('r', 0, size as u64)],
        "an unpaced read is one read"
    );

    // A pacer not active now, as a limiter at rate zero, leaves the read
    // whole and uncharged.
    log.lock().expect("lock is not poisoned").clear();
    let idle = LogPacer {
        active: false,
        log: std::sync::Arc::clone(&log),
    };
    crate::file::read_exact_paced(&file, 0, size, Some(&idle))?;
    assert_eq!(
        *log.lock().expect("lock is not poisoned"),
        vec![('r', 0, size as u64)],
        "a read under an idle pacer is one read, not charged"
    );
    Ok(())
}

/// A pacer that logs each portion it paces, `('p', offset, len)`, into a log
/// a file can share; `active` says whether it paces at all.
struct LogPacer {
    active: bool,
    log: std::sync::Arc<std::sync::Mutex<Vec<(char, u64, u64)>>>,
}

impl crate::table::util::ReadPacer for LogPacer {
    fn active(&self) -> bool {
        self.active
    }
    fn pace(&self, offset: u64, len: u64) {
        self.log
            .lock()
            .expect("lock is not poisoned")
            .push(('p', offset, len));
    }
}

/// A `ScriptedReads` over a fresh file holding `bytes`.
fn scripted_file(
    bytes: &[u8],
    cap: usize,
    fail_at: Option<usize>,
) -> crate::Result<(
    ScriptedReads,
    std::sync::Arc<std::sync::atomic::AtomicUsize>,
)> {
    use crate::fs::{Fs, FsOpenOptions, MemFs};

    let fs = MemFs::new();
    let path = std::path::Path::new("/f");
    {
        let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
        f.write_all(bytes)?;
    }
    let largest = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let file = ScriptedReads {
        inner: fs.open(path, &FsOpenOptions::new().read(true))?,
        cap,
        fail_at,
        calls: 0,
        largest: std::sync::Arc::clone(&largest),
    };
    Ok((file, largest))
}

/// Under a limiter, a read larger than a portion reaches the file a portion at
/// a time, each just after it is charged, never as one read after the whole
/// of it was waited for; without one, it reaches the file whole.
#[test]
fn a_paced_read_reaches_the_file_a_portion_at_a_time() -> crate::Result<()> {
    let bytes: Vec<u8> = (0..1024 * 1024u32).map(|i| (i % 251) as u8).collect();
    let limiter = crate::rate_limiter::RateLimiter::new(1 << 30);

    let (file, largest) = scripted_file(&bytes, usize::MAX, None)?;
    let mut paced = PacedFile::new(Box::new(file), Some(&limiter))?;
    let mut read = vec![0u8; bytes.len()];
    std::io::Read::read_exact(&mut paced, &mut read)?;
    assert_eq!(read, bytes, "the file reads back whole");
    let largest = largest.load(std::sync::atomic::Ordering::Relaxed);
    assert!(
        largest <= PACE_PORTION,
        "a read of {largest} bytes reached the file at once"
    );

    let (file, largest) = scripted_file(&bytes, usize::MAX, None)?;
    let mut unpaced = PacedFile::new(Box::new(file), None)?;
    std::io::Read::read_exact(&mut unpaced, &mut read)?;
    assert_eq!(read, bytes, "the file reads back whole");
    assert_eq!(
        largest.load(std::sync::atomic::Ordering::Relaxed),
        bytes.len(),
        "an unrated read is not split"
    );
    Ok(())
}

/// An error the file returns after some bytes of a read is not lost: the
/// bytes come back first and the error on the next read, though the file
/// would not fail again.
#[test]
fn a_paced_file_reports_an_error_that_followed_a_partial_read() -> crate::Result<()> {
    let bytes = vec![7u8; 16 * 1024];
    let limiter = crate::rate_limiter::RateLimiter::new(1 << 30);
    let (file, _) = scripted_file(&bytes, 512, Some(2))?;
    let mut paced = PacedFile::new(Box::new(file), Some(&limiter))?;

    let mut buf = vec![0u8; 4096];
    assert_eq!(std::io::Read::read(&mut paced, &mut buf)?, 512);
    let next = std::io::Read::read(&mut paced, &mut buf);
    assert!(
        matches!(&next, Err(e) if e.kind() == std::io::ErrorKind::Other),
        "the failure reached the caller: {next:?}"
    );
    Ok(())
}

/// An error held back after a partial read belongs to where it happened: a
/// seek away from there drops it, so the next read, of other bytes, is not
/// failed by it. Bytes it may concern are read again from wherever the scan
/// goes back to them, and a fault that persists fails that read.
#[test]
fn a_seek_drops_an_error_held_back_from_a_partial_read() -> crate::Result<()> {
    let bytes: Vec<u8> = (0..16 * 1024u32).map(|i| (i % 251) as u8).collect();
    let limiter = crate::rate_limiter::RateLimiter::new(1 << 30);
    let (file, _) = scripted_file(&bytes, 512, Some(2))?;
    let mut paced = PacedFile::new(Box::new(file), Some(&limiter))?;

    let mut buf = vec![0u8; 4096];
    assert_eq!(std::io::Read::read(&mut paced, &mut buf)?, 512);
    std::io::Seek::seek(&mut paced, SeekFrom::Start(8192))?;
    let read = std::io::Read::read(&mut paced, &mut buf)?;
    assert!(read > 0, "the read after the seek returns bytes");
    assert_eq!(
        buf.get(..read),
        bytes.get(8192..8192 + read),
        "the bytes at the seek"
    );
    Ok(())
}

/// A limiter at rate zero reads at full speed, like no limiter: a large read
/// reaches the file whole, not in portions.
#[test]
fn a_zero_rate_limiter_does_not_split_reads() -> crate::Result<()> {
    let bytes: Vec<u8> = (0..1024 * 1024u32).map(|i| (i % 251) as u8).collect();
    let limiter = crate::rate_limiter::RateLimiter::new(0);
    let (file, largest) = scripted_file(&bytes, usize::MAX, None)?;
    let mut paced = PacedFile::new(Box::new(file), Some(&limiter))?;
    let mut read = vec![0u8; bytes.len()];
    std::io::Read::read_exact(&mut paced, &mut read)?;
    assert_eq!(read, bytes, "the file reads back whole");
    assert_eq!(
        largest.load(std::sync::atomic::Ordering::Relaxed),
        bytes.len(),
        "an unrated read is not split"
    );
    Ok(())
}

/// A charge larger than a portion, as an index read up to a late bound, goes
/// to the limiter a portion at a time and adds up to the whole.
#[test]
fn a_large_charge_is_made_a_portion_at_a_time() {
    let portion = PACE_PORTION as u64;
    let parts: Vec<u64> = pace_portions(3 * portion + 5).collect();
    assert_eq!(parts, vec![portion, portion, portion, 5]);
    assert_eq!(pace_portions(0).count(), 0, "nothing to charge, no request");
    assert_eq!(pace_portions(portion).collect::<Vec<_>>(), vec![portion]);
}

/// A read that failed partway was charged for bytes it never read. They stay
/// to the file's credit across a seek, so the next read is not charged again
/// for budget already spent, and the limiter is left with what was read.
#[test]
fn bytes_charged_but_not_read_are_credited_to_the_next_read() -> crate::Result<()> {
    let bytes: Vec<u8> = (0..16 * 1024u32).map(|i| (i % 251) as u8).collect();
    // One second of rate holds 8 KiB: two 4 KiB charges empty it.
    let limiter = crate::rate_limiter::RateLimiter::new(8 * 1024);
    let (file, _) = scripted_file(&bytes, 512, Some(2))?;
    let mut paced = PacedFile::new(Box::new(file), Some(&limiter))?;

    let mut buf = vec![0u8; 4096];
    // Charged 4 KiB, read 512 bytes, then the read failed.
    assert_eq!(std::io::Read::read(&mut paced, &mut buf)?, 512);
    std::io::Seek::seek(&mut paced, SeekFrom::Start(8192))?;
    // Charged only what the credit does not cover.
    assert!(std::io::Read::read(&mut paced, &mut buf)? > 0);

    // About 4.5 KiB of budget was spent, not 8 KiB, so 3 KiB more is admitted
    // at once instead of after a third of a second.
    let started = std::time::Instant::now();
    assert!(!limiter.request_interruptible(3 * 1024, || false));
    assert!(
        started.elapsed() < std::time::Duration::from_millis(150),
        "the next request waited {:?} for bytes never read",
        started.elapsed()
    );
    Ok(())
}

/// A paced portion is charged the bytes it takes from the file: all of them
/// inside it, the prefix the file holds when it runs past the end, and none
/// at or past the end.
#[test]
fn a_paced_portion_is_charged_only_the_bytes_inside_the_file() {
    assert_eq!(bytes_in_file(0, 4096, 8192), 4096);
    assert_eq!(bytes_in_file(8092, 4096, 8192), 100);
    assert_eq!(bytes_in_file(8192, 4096, 8192), 0);
    assert_eq!(bytes_in_file(u64::MAX, 4096, 8192), 0);
}

/// A limiter at rate zero reads at full speed, so the punch-offset walk under
/// it must not depend on anything a walk without a limiter does not: a file
/// whose metadata cannot be read still has its offset found.
#[test]
fn a_restricted_lookup_at_rate_zero_reads_no_metadata() -> crate::Result<()> {
    use crate::fs::{Fault, FaultFs, FaultOp, FaultRule, StdFs};

    let dir = tempfile::tempdir()?;
    let fs = FaultFs::new(StdFs);
    let injector = fs.injector();
    let tree = crate::Config::new(
        dir.path(),
        crate::SequenceNumberCounter::default(),
        crate::SequenceNumberCounter::default(),
    )
    .with_shared_fs(std::sync::Arc::new(fs))
    .open()?;
    for i in 0..100u32 {
        tree.insert(format!("k{i:05}"), "v", u64::from(i));
    }
    tree.flush_active_memtable(0)?;
    let version = tree.current_version();
    let table = version.iter_tables().next().expect("one table");
    let expected = table.punch_offset_for(b"k00050")?;

    injector.arm(FaultRule::new(
        FaultOp::Metadata,
        Fault::Error(crate::io::ErrorKind::Other),
    ));
    let idle = std::sync::Arc::new(crate::rate_limiter::RateLimiter::new(0));
    let found = super::paced_punch_offset(table, b"k00050", Some(&idle));
    injector.clear();
    assert_eq!(found?, expected, "an idle limiter walks as no limiter does");
    Ok(())
}

/// A walk that starts under an idle limiter still carries its pacer: a rate set
/// while it runs paces the reads after it. Pacing reads the file's length on
/// the first paced portion, and a length that cannot be read charges the
/// portion uncut rather than failing the walk.
#[test]
fn a_walk_started_idle_is_paced_once_a_rate_is_set() -> crate::Result<()> {
    use crate::fs::{Fault, FaultFs, FaultOp, FaultRule, StdFs};
    use crate::table::util::ReadPacer;

    let dir = tempfile::tempdir()?;
    let fs = FaultFs::new(StdFs);
    let injector = fs.injector();
    let tree = crate::Config::new(
        dir.path(),
        crate::SequenceNumberCounter::default(),
        crate::SequenceNumberCounter::default(),
    )
    .with_shared_fs(std::sync::Arc::new(fs))
    .open()?;
    tree.insert("k", "v", 0);
    tree.flush_active_memtable(0)?;
    let version = tree.current_version();
    let table = version.iter_tables().next().expect("one table");

    let limiter = std::sync::Arc::new(crate::rate_limiter::RateLimiter::new(0));
    let pacer = super::walk_pacer(table, Some(&limiter)).expect("a limiter attaches a pacer");
    assert!(!pacer.active(), "an idle limiter reads at full speed");
    limiter.set_rate(1 << 30);
    assert!(
        pacer.active(),
        "a rate set mid-walk paces the reads after it"
    );

    injector.arm(FaultRule::new(
        FaultOp::Metadata,
        Fault::Error(crate::io::ErrorKind::Other),
    ));
    pacer.pace(0, 4096);
    injector.clear();
    assert!(
        super::walk_pacer(table, None).is_none(),
        "no limiter, no pacer"
    );
    Ok(())
}

/// The punch-offset walk is charged what it reads, as the loader reads it:
/// nothing for an index held in memory, the index block for one loaded per
/// read, and for a partitioned index only the partitions the walk loads up to
/// the bound, all of them for a bound past the last key.
#[test]
fn a_restricted_index_lookup_is_charged_what_it_reads() -> crate::Result<()> {
    use crate::config::{BlockSizePolicy, PinningPolicy};
    use crate::table::block::ParsedItem;
    use crate::table::block_index::BlockIndexImpl;

    let keys: Vec<String> = (0..2_000).map(|i| format!("k{i:05}")).collect();
    for shape in ["pinned", "volatile", "partitioned"] {
        let dir = tempfile::tempdir()?;
        // A cache too small to keep any block: a block the cache serves is
        // not read, so not charged, and every walk here reads what it walks.
        let config = crate::Config::new(
            dir.path(),
            crate::SequenceNumberCounter::default(),
            crate::SequenceNumberCounter::default(),
        )
        .use_cache(std::sync::Arc::new(crate::Cache::with_capacity_bytes(1)));
        // The writer keeps a small table's index in one block until it outgrows
        // the spill threshold, so the partitioned shape spills at once, with
        // small blocks and partitions to give it several.
        let config = match shape {
            "pinned" => config
                .index_block_partitioning_policy(PinningPolicy::all(false))
                .index_block_pinning_policy(PinningPolicy::all(true)),
            "volatile" => config
                .index_block_partitioning_policy(PinningPolicy::all(false))
                .index_block_pinning_policy(PinningPolicy::all(false)),
            _ => {
                let runtime = crate::runtime_config::RuntimeConfig {
                    index_partition_spill_threshold: 0,
                    ..Default::default()
                };
                config
                    .data_block_size_policy(BlockSizePolicy::all(256))
                    .index_block_partitioning_policy(PinningPolicy::all(true))
                    .index_block_partition_size_policy(BlockSizePolicy::all(256))
                    .with_runtime_config(runtime)
            }
        };
        let tree = config.open()?;
        for (seqno, key) in keys.iter().enumerate() {
            tree.insert(key, "v", seqno as u64);
        }
        tree.flush_active_memtable(0)?;
        let version = tree.current_version();
        let table = version.iter_tables().next().expect("one table");

        let file_size = table.fs.metadata(&table.path)?.len;
        // What one walk for `bound` is charged. Untraced, so no walk leaves a
        // cached partition for the next to skip.
        let charged = |bound: &[u8]| -> crate::Result<u64> {
            let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
            table.punch_offset_paced(
                bound,
                crate::table::util::ReadCharge::Untraced,
                std::sync::Arc::new(LogPacer {
                    active: true,
                    log: std::sync::Arc::clone(&log),
                }),
            )?;
            let paced = log.lock().expect("lock is not poisoned");
            Ok(paced.iter().map(|&(_, _, len)| len).sum())
        };
        let first = charged(b"k00000")?;
        let middle = charged(b"k01000")?;
        let past = charged(b"z")?;
        assert!(past <= file_size, "never more than the file holds");
        match &*table.block_index {
            BlockIndexImpl::Full(_) => {
                assert_eq!(shape, "pinned");
                assert_eq!((0, 0, 0), (first, middle, past));
            }
            BlockIndexImpl::VolatileFull(index) => {
                assert_eq!(shape, "volatile");
                let size = u64::from(index.handle.size());
                assert!(size > 0);
                assert_eq!((size, size, size), (first, middle, past));
            }
            BlockIndexImpl::TwoLevel(index) => {
                assert_eq!(shape, "partitioned");
                let tli = &index.top_level_index;
                let all: u64 = tli
                    .iter(table.comparator.clone())
                    .map(|item| u64::from(item.materialize(tli.as_slice()).size()))
                    .sum();
                assert!(tli.len() > 2, "several partitions");
                assert!(0 < first && first < middle && middle < past);
                assert_eq!(all, past, "a bound past the end walks every partition");
            }
            BlockIndexImpl::Closed => panic!("a live table"),
        }
    }
    Ok(())
}

/// A read interrupted after the backend already returned bytes keeps those
/// bytes: the file reads back whole, nothing skipped and nothing overwritten.
#[test]
fn a_paced_file_keeps_the_bytes_read_before_an_interruption() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, MemFs};

    let fs = MemFs::new();
    let path = std::path::Path::new("/f");
    let bytes: Vec<u8> = (0..16 * 1024u32).map(|i| (i % 251) as u8).collect();
    {
        let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
        f.write_all(&bytes)?;
    }
    let limiter = crate::rate_limiter::RateLimiter::new(1 << 30);
    let file = InterruptedReads {
        inner: fs.open(path, &FsOpenOptions::new().read(true))?,
        calls: 0,
    };
    let mut reader = std::io::BufReader::with_capacity(
        4 * 1024,
        PacedFile::new(Box::new(file), Some(&limiter))?,
    );
    let mut read = vec![0u8; bytes.len()];
    reader.read_exact(&mut read[..])?;
    assert_eq!(read, bytes, "the file reads back whole");
    Ok(())
}

/// A file that takes `delay` to answer every read, as a congested device does,
/// and answers with an error instead of bytes when `fails`.
struct SlowReads {
    inner: Box<dyn crate::fs::FsFile>,
    delay: std::time::Duration,
    fails: bool,
}

impl std::io::Read for SlowReads {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        std::thread::sleep(self.delay);
        if self.fails {
            return Err(std::io::ErrorKind::Other.into());
        }
        self.inner.read(buf)
    }
}

impl std::io::Write for SlowReads {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.inner.write(buf)
    }
    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

impl std::io::Seek for SlowReads {
    fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
        self.inner.seek(to)
    }
}

impl crate::fs::FsFile for SlowReads {
    fn sync_all(&self) -> crate::io::Result<()> {
        self.inner.sync_all()
    }
    fn sync_data(&self) -> crate::io::Result<()> {
        self.inner.sync_data()
    }
    fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
        self.inner.metadata()
    }
    fn set_len(&self, size: u64) -> crate::io::Result<()> {
        self.inner.set_len(size)
    }
    fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
        self.inner.read_at(buf, offset)
    }
    fn lock_exclusive(&self) -> crate::io::Result<()> {
        self.inner.lock_exclusive()
    }
}

/// The reads a paced scan charges feed the limiter's latency backoff: a
/// device slower than the ceiling lowers the rate the limiter grants, down to
/// no lower than its floor, while a limiter without a backoff keeps its rate.
#[test]
fn a_paced_file_reports_its_read_latency_to_the_limiter() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, MemFs};
    use crate::rate_limiter::{LatencyBackoff, RateLimiter};
    use std::time::Duration;

    let fs = MemFs::new();
    let path = std::path::Path::new("/f");
    let bytes = vec![7u8; 16 * 1024];
    {
        let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
        f.write_all(&bytes)?;
    }
    let slow = || -> crate::Result<SlowReads> {
        Ok(SlowReads {
            inner: fs.open(path, &FsOpenOptions::new().read(true))?,
            delay: Duration::from_millis(5),
            fails: false,
        })
    };

    // Enough rate that no read waits for the budget.
    let configured = 1 << 30;
    let limiter = RateLimiter::new(configured);
    limiter.set_latency_backoff(Some(
        LatencyBackoff::new(Duration::from_millis(1)).with_period(Duration::ZERO),
    ));
    let mut reader =
        std::io::BufReader::with_capacity(1024, PacedFile::new(Box::new(slow()?), Some(&limiter))?);
    let mut read = vec![0u8; bytes.len()];
    reader.read_exact(&mut read[..])?;
    assert_eq!(read, bytes);
    let granted = limiter.effective_rate();
    assert!(granted < configured, "a slow device did not lower the rate");
    assert!(
        granted >= configured / 20,
        "the rate fell below the 5% floor: {granted}"
    );

    let unmanaged = RateLimiter::new(configured);
    let mut reader = std::io::BufReader::with_capacity(
        1024,
        PacedFile::new(Box::new(slow()?), Some(&unmanaged))?,
    );
    reader.read_exact(&mut read[..])?;
    assert_eq!(configured, unmanaged.effective_rate());
    Ok(())
}

/// A read the device answers slowly with an error, before any byte, still
/// counts toward the backoff: a scan that goes on past failed sections must
/// not keep hammering a congested device at the full rate.
#[test]
fn a_slow_failed_read_lowers_the_granted_rate() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, MemFs};
    use crate::rate_limiter::{LatencyBackoff, RateLimiter};
    use std::time::Duration;

    let fs = MemFs::new();
    let path = std::path::Path::new("/f");
    {
        let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
        f.write_all(&[7u8; 4096])?;
    }
    let configured = 1 << 30;
    let limiter = RateLimiter::new(configured);
    limiter.set_latency_backoff(Some(
        LatencyBackoff::new(Duration::from_millis(1)).with_period(Duration::ZERO),
    ));
    let failing = SlowReads {
        inner: fs.open(path, &FsOpenOptions::new().read(true))?,
        delay: Duration::from_millis(5),
        fails: true,
    };
    let mut paced = PacedFile::new(Box::new(failing), Some(&limiter))?;
    let mut buf = [0u8; 1024];
    for _ in 0..4 {
        assert!(std::io::Read::read(&mut paced, &mut buf).is_err());
    }
    assert!(
        limiter.effective_rate() < configured,
        "slow failures did not lower the rate"
    );
    Ok(())
}

/// A failed read hands back what it was charged afresh, so the bytes are not
/// paid twice when read again, but not the credit it was paid from: handed
/// back each time, one credit would pay every later failed read and none
/// would wait, so a lowered rate would never hold back a run of failing
/// sections. Every other failure is charged.
#[test]
fn a_failed_read_does_not_hand_back_the_credit_it_spent() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, MemFs};
    use crate::rate_limiter::RateLimiter;
    use std::time::Duration;

    let fs = MemFs::new();
    let path = std::path::Path::new("/f");
    {
        let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
        f.write_all(&[7u8; 4096])?;
    }
    let limiter = RateLimiter::new(1 << 30);
    let failing = SlowReads {
        inner: fs.open(path, &FsOpenOptions::new().read(true))?,
        delay: Duration::ZERO,
        fails: true,
    };
    let mut paced = PacedFile::new(Box::new(failing), Some(&limiter))?;
    let mut buf = [0u8; 1024];
    let mut credits = Vec::new();
    for _ in 0..4 {
        assert!(std::io::Read::read(&mut paced, &mut buf).is_err());
        credits.push(paced.credit);
    }
    assert_eq!(credits, vec![1024, 0, 1024, 0]);
    Ok(())
}

/// A file whose read switches its limiter's rate on, as a live retune landing
/// while the read is in flight does, and takes `delay` to answer.
struct RetunedDuringRead {
    inner: Box<dyn crate::fs::FsFile>,
    limiter: std::sync::Arc<crate::rate_limiter::RateLimiter>,
    delay: std::time::Duration,
}

impl std::io::Read for RetunedDuringRead {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.limiter.set_rate(1 << 30);
        std::thread::sleep(self.delay);
        self.inner.read(buf)
    }
}

impl std::io::Write for RetunedDuringRead {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.inner.write(buf)
    }
    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

impl std::io::Seek for RetunedDuringRead {
    fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
        self.inner.seek(to)
    }
}

impl crate::fs::FsFile for RetunedDuringRead {
    fn sync_all(&self) -> crate::io::Result<()> {
        self.inner.sync_all()
    }
    fn sync_data(&self) -> crate::io::Result<()> {
        self.inner.sync_data()
    }
    fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
        self.inner.metadata()
    }
    fn set_len(&self, size: u64) -> crate::io::Result<()> {
        self.inner.set_len(size)
    }
    fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
        self.inner.read_at(buf, offset)
    }
    fn lock_exclusive(&self) -> crate::io::Result<()> {
        self.inner.lock_exclusive()
    }
}

/// A read decided at rate zero goes whole and uncharged; a rate switched on
/// while it is in flight must not turn it into a latency sample, which could
/// lower the newly granted rate on a read the limiter never admitted.
#[test]
fn a_read_decided_without_a_rate_reports_no_latency() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, MemFs};
    use crate::rate_limiter::{LatencyBackoff, RateLimiter};
    use std::time::Duration;

    let fs = MemFs::new();
    let path = std::path::Path::new("/f");
    {
        let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
        f.write_all(&[7u8; 4096])?;
    }
    let limiter = std::sync::Arc::new(RateLimiter::new(0));
    limiter.set_latency_backoff(Some(
        LatencyBackoff::new(Duration::from_millis(1)).with_period(Duration::ZERO),
    ));
    let retuned = RetunedDuringRead {
        inner: fs.open(path, &FsOpenOptions::new().read(true))?,
        limiter: std::sync::Arc::clone(&limiter),
        delay: Duration::from_millis(5),
    };
    let mut paced = PacedFile::new(Box::new(retuned), Some(&limiter))?;
    let mut buf = [0u8; 4096];
    std::io::Read::read(&mut paced, &mut buf)?;
    assert_eq!(
        limiter.effective_rate(),
        limiter.rate(),
        "an uncharged read lowered the rate"
    );
    Ok(())
}

/// The index reads a restricted table's punch-offset walk makes through the
/// block loader report how long each portion took, and a limiter's pacer
/// hands that to the latency backoff.
#[test]
fn a_paced_index_read_reports_its_latency() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, MemFs};
    use crate::rate_limiter::{LatencyBackoff, RateLimiter};
    use crate::table::util::ReadPacer;
    use std::time::Duration;

    /// Counts the portions it is told the read time of.
    struct Timed(std::sync::atomic::AtomicUsize);
    impl ReadPacer for Timed {
        fn active(&self) -> bool {
            true
        }
        fn pace(&self, _offset: u64, _len: u64) {}
        fn read_took(&self, _elapsed: Duration) {
            self.0.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        }
    }

    let fs = MemFs::new();
    let path = std::path::Path::new("/f");
    let size = 2 * PACE_PORTION + 100;
    {
        let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
        f.write_all(&vec![7u8; size])?;
    }
    let file = fs.open(path, &FsOpenOptions::new().read(true))?;
    let timed = Timed(std::sync::atomic::AtomicUsize::new(0));
    crate::file::read_exact_paced(file.as_ref(), 0, size, Some(&timed))?;
    assert_eq!(
        3,
        timed.0.load(std::sync::atomic::Ordering::Relaxed),
        "every portion read reports its time"
    );

    // The walk's pacer passes the time on to the limiter.
    let dir = tempfile::tempdir()?;
    let tree = crate::Config::new(
        dir.path(),
        crate::SequenceNumberCounter::default(),
        crate::SequenceNumberCounter::default(),
    )
    .open()?;
    tree.insert("k", "v", 0);
    tree.flush_active_memtable(0)?;
    let version = tree.current_version();
    let table = version.iter_tables().next().expect("one table");
    let configured = 1 << 30;
    let limiter = std::sync::Arc::new(RateLimiter::new(configured));
    limiter.set_latency_backoff(Some(
        LatencyBackoff::new(Duration::from_millis(1)).with_period(Duration::ZERO),
    ));
    let pacer = super::walk_pacer(table, Some(&limiter)).expect("a limiter attaches a pacer");
    pacer.read_took(Duration::from_millis(10));
    assert!(
        limiter.effective_rate() < configured,
        "a slow index read did not lower the rate"
    );
    Ok(())
}

/// A large read on a limiter shared with other work is charged in bounded
/// portions, so a small request arriving meanwhile waits one portion, not the
/// whole large read.
#[test]
fn a_large_paced_read_lets_a_shared_limiter_serve_others_between_portions() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, MemFs};
    use std::time::{Duration, Instant};

    let fs = MemFs::new();
    let path = std::path::Path::new("/f");
    // Eight seconds of rate in one read.
    let rate: u64 = 256 * 1024;
    let bytes = vec![3u8; 8 * 256 * 1024];
    {
        let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
        f.write_all(&bytes)?;
    }
    let limiter = std::sync::Arc::new(crate::rate_limiter::RateLimiter::new(rate));
    // Drain the starting burst so every byte below waits for the rate.
    assert!(!limiter.request_interruptible(rate, || false));

    let scan = {
        let limiter = std::sync::Arc::clone(&limiter);
        let file = fs.open(path, &FsOpenOptions::new().read(true))?;
        std::thread::spawn(move || -> crate::Result<()> {
            let mut paced = PacedFile::new(file, Some(&limiter))?;
            let mut buf = vec![0u8; 8 * 256 * 1024];
            // One read call, as BufReader makes for a destination larger than
            // its buffer.
            std::io::Read::read_exact(&mut paced, &mut buf)?;
            Ok(())
        })
    };
    std::thread::sleep(Duration::from_millis(300));
    let started = Instant::now();
    assert!(!limiter.request_interruptible(1024, || false));
    let waited = started.elapsed();
    scan.join().expect("scan thread")?;
    assert!(
        waited < Duration::from_secs(2),
        "a small request waited {waited:?} behind one large read"
    );
    Ok(())
}

/// The parity-trailer drain reports a truncated read when an SST whose ECC
/// descriptor claims per-block parity is missing those trailer bytes. Forges
/// a `data` section of header + its full payload (so the data read and its
/// checksum both pass), then scans it as an RS(4,2) table: the walk drains
/// `expected_parity_len` bytes, hits EOF after the short SFA tail, and
/// surfaces a `DataReadError` for the short parity read rather than
/// mis-reading the tail as the next block.
#[test]
#[expect(
    clippy::indexing_slicing,
    clippy::cast_possible_truncation,
    reason = "synthetic SFA forgery — offsets are in-bounds by construction (we wrote the \
              bytes ourselves) and the archive is < 8 KiB, so the casts cannot overflow"
)]
fn walk_block_region_reports_data_read_error_on_truncated_parity_trailer() -> crate::Result<()> {
    use crate::coding::Encode;
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::{BlockType, EccParams, Header, expected_parity_len};

    // Trailer layout (38 bytes at the tail of an SFA archive):
    //   MAGIC(4) | version(1) | csum_type(1) | toc_checksum(16) | toc_pos(8) | toc_len(8)
    const TRAILER_LEN: usize = 4 + 1 + 1 + 16 + 8 + 8;
    const DATA_LENGTH: u32 = 4096;
    const HEADER_LEN: u64 = Header::MIN_LEN as u64;

    let data = vec![0xABu8; DATA_LENGTH as usize];
    let mut header = Header {
        data_length: DATA_LENGTH,
        uncompressed_length: DATA_LENGTH,
        ..Header::test_dummy(BlockType::Data)
    };
    header.bind_checksum(
        Checksum::from_raw(crate::hash::hash128(&data)),
        crate::table::block::ChecksumAt::table(7, 0),
    );

    // One `data` section: header + full payload, but no parity trailer.
    let mut archive_bytes: Vec<u8> = Vec::new();
    {
        let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
        writer.start("data").unwrap();
        writer.write_all(&header.encode_into_vec()).unwrap();
        writer.write_all(&data).unwrap();
        writer.finish().unwrap();
    }

    // Inflate the section length to header + payload + parity so the walker's
    // `data_length + parity_len <= remaining` bounds check passes; the parity
    // bytes were never written, so the drain hits EOF instead. Recompute the
    // TOC checksum afterwards so crate::sfa::Reader still accepts the file.
    let parity_len = u64::from(expected_parity_len(DATA_LENGTH, EccParams::RS_4_2));
    let trailer_start = archive_bytes.len() - TRAILER_LEN;
    let toc_pos = u64::from_le_bytes(
        archive_bytes[trailer_start + 22..trailer_start + 30]
            .try_into()
            .unwrap(),
    ) as usize;
    let toc_len = u64::from_le_bytes(
        archive_bytes[trailer_start + 30..trailer_start + 38]
            .try_into()
            .unwrap(),
    ) as usize;
    let len_field_offset = toc_pos + 4 + 4 + 8;
    let lied_len: u64 = HEADER_LEN + u64::from(DATA_LENGTH) + parity_len;
    archive_bytes[len_field_offset..len_field_offset + 8].copy_from_slice(&lied_len.to_le_bytes());
    let new_toc_checksum = crate::hash::hash128(&archive_bytes[toc_pos..toc_pos + toc_len]);
    let csum_field_offset = trailer_start + 4 + 1 + 1;
    archive_bytes[csum_field_offset..csum_field_offset + 16]
        .copy_from_slice(&new_toc_checksum.to_le_bytes());

    let dir = tempfile::tempdir()?;
    let fs = StdFs;
    let forged = dir.path().join("forged-parity.sst");
    let path = forged.as_path();
    {
        let mut f = fs.open(
            path,
            &FsOpenOptions::new().write(true).create(true).truncate(true),
        )?;
        f.write_all(&archive_bytes)?;
    }

    // Scan as an RS(4,2) table: a non-zero parity_len is drained after the
    // (clean) payload, hitting EOF in the short SFA tail.
    let table_id: TableId = 7;
    let layout = SstLayout {
        ecc: Some(EccParams::RS_4_2),
        ..SstLayout::PLAIN
    };
    let scan = scan_sst_blocks(&fs, path, table_id, layout, None)?;
    assert!(
        scan.errors.iter().any(|e| matches!(
            e,
            BlockVerifyError::DataReadError { table_id: t, offset: 0, error, .. }
                if *t == table_id && error.kind() == crate::io::ErrorKind::UnexpectedEof
        )),
        "expected a truncated-parity DataReadError, got {:?}",
        scan.errors,
    );
    Ok(())
}

/// A syntactically valid but absurd shard layout (RS(1,255): every payload
/// byte amplified 255x into parity) drives `expected_parity_len` toward
/// `u32::MAX`. Combined with a lying TOC length (a forged or sparse SST), the
/// walk would reserve the whole multi-GB trailer buffer BEFORE reporting any
/// corruption. The trailer length must be capped like `data_length` is: over
/// the cap it is `HeaderCorrupted`, reported without reserving anything.
#[test]
#[expect(
    clippy::indexing_slicing,
    clippy::cast_possible_truncation,
    reason = "synthetic SFA forgery — offsets are in-bounds by construction (we wrote the \
              bytes ourselves) and the archive is small, so the casts cannot overflow"
)]
fn walk_block_region_caps_an_absurd_parity_trailer_length() -> crate::Result<()> {
    use crate::coding::Encode;
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::{BlockType, EccParams, Header, expected_parity_len};

    // Trailer layout (38 bytes at the tail of an SFA archive):
    //   MAGIC(4) | version(1) | csum_type(1) | toc_checksum(16) | toc_pos(8) | toc_len(8)
    const TRAILER_LEN: usize = 4 + 1 + 1 + 16 + 8 + 8;
    // 2 MiB of real payload: RS(1,255) turns that into a ~510 MiB claimed
    // parity trailer — well over the 256 MiB hard cap.
    const DATA_LENGTH: u32 = 2 * 1024 * 1024;
    const HEADER_LEN: u64 = Header::MIN_LEN as u64;

    let params = EccParams::try_new(1, 255).expect("a 1/255 shard layout parses");
    let parity_len = u64::from(expected_parity_len(DATA_LENGTH, params));
    assert!(
        parity_len > u64::from(DATA_LENGTH) * 200,
        "the forged scheme must amplify parity far past the payload",
    );

    let data = vec![0xABu8; DATA_LENGTH as usize];
    let mut header = Header {
        data_length: DATA_LENGTH,
        uncompressed_length: DATA_LENGTH,
        ..Header::test_dummy(BlockType::Data)
    };
    header.bind_checksum(
        Checksum::from_raw(crate::hash::hash128(&data)),
        crate::table::block::ChecksumAt::table(7, 0),
    );

    // One `data` section: header + full payload, no parity bytes on disk.
    let mut archive_bytes: Vec<u8> = Vec::new();
    {
        let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
        writer.start("data").unwrap();
        writer.write_all(&header.encode_into_vec()).unwrap();
        writer.write_all(&data).unwrap();
        writer.finish().unwrap();
    }

    // Inflate the TOC section length to cover the claimed parity so the
    // bounds check passes (the sparse-file / forged-TOC shape), recomputing
    // the TOC checksum so crate::sfa::Reader still accepts the file.
    let trailer_start = archive_bytes.len() - TRAILER_LEN;
    let toc_pos = u64::from_le_bytes(
        archive_bytes[trailer_start + 22..trailer_start + 30]
            .try_into()
            .unwrap(),
    ) as usize;
    let toc_len = u64::from_le_bytes(
        archive_bytes[trailer_start + 30..trailer_start + 38]
            .try_into()
            .unwrap(),
    ) as usize;
    let len_field_offset = toc_pos + 4 + 4 + 8;
    let lied_len: u64 = HEADER_LEN + u64::from(DATA_LENGTH) + parity_len;
    archive_bytes[len_field_offset..len_field_offset + 8].copy_from_slice(&lied_len.to_le_bytes());
    let new_toc_checksum = crate::hash::hash128(&archive_bytes[toc_pos..toc_pos + toc_len]);
    let csum_field_offset = trailer_start + 4 + 1 + 1;
    archive_bytes[csum_field_offset..csum_field_offset + 16]
        .copy_from_slice(&new_toc_checksum.to_le_bytes());

    let dir = tempfile::tempdir()?;
    let fs = StdFs;
    let forged = dir.path().join("forged-parity-cap.sst");
    let path = forged.as_path();
    {
        let mut f = fs.open(
            path,
            &FsOpenOptions::new().write(true).create(true).truncate(true),
        )?;
        f.write_all(&archive_bytes)?;
    }

    let table_id: TableId = 7;
    let layout = SstLayout {
        ecc: Some(params),
        ..SstLayout::PLAIN
    };
    let scan = scan_sst_blocks(&fs, path, table_id, layout, None)?;
    assert!(
        scan.errors.iter().any(|e| matches!(
            e,
            BlockVerifyError::HeaderCorrupted { table_id: t, offset: 0, reason, .. }
                if *t == table_id && reason.contains("parity trailer length")
        )),
        "an over-cap parity trailer must be HeaderCorrupted without reserving \
         the buffer, got {:?}",
        scan.errors,
    );
    Ok(())
}

/// A block header whose own bytes extend past the section boundary must be
/// reported as `HeaderCorrupted`, not slip through with a clamped-to-zero
/// remaining payload.
///
/// Setup forges a section whose lied length is exactly `Header::MIN_LEN`
/// (so the `< MIN_LEN` guard passes) and stores a `Meta` block, whose
/// `header_len` is `MIN_LEN + 1`. `Header::decode_from` reads the full
/// `MIN_LEN + 1` header bytes from the file (they are physically present,
/// followed by the TOC), but those bytes cross the section boundary, so the
/// boundary guard fires.
#[test]
#[expect(
    clippy::indexing_slicing,
    clippy::cast_possible_truncation,
    reason = "synthetic SFA forgery — offsets are in-bounds by construction \
              and the forged archive is < 1 KiB"
)]
fn walk_block_region_reports_header_crossing_section_boundary() -> crate::Result<()> {
    use crate::coding::Encode;
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::{BlockType, Header};

    const TRAILER_LEN: usize = 4 + 1 + 1 + 16 + 8 + 8;

    // `Meta` blocks carry the block_flags byte, so header_len == MIN_LEN + 1.
    let header = Header {
        stored_checksum: Checksum::from_raw(0xDEAD_BEEF_DEAD_BEEF),
        data_length: 0,
        uncompressed_length: 0,
        ..Header::test_dummy(BlockType::Meta)
    };
    assert_eq!(
        Header::header_len(BlockType::Meta) as u64,
        Header::MIN_LEN as u64 + 1,
    );

    let mut archive_bytes: Vec<u8> = Vec::new();
    {
        let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
        // "meta", not "data": the section-vs-role cross-check would report
        // a Meta block inside a data section as a SECOND error, and this
        // test pins down exactly one (the boundary violation).
        writer.start("meta").unwrap();
        writer.write_all(&header.encode_into_vec()).unwrap();
        writer.finish().unwrap();
    }

    let trailer_start = archive_bytes.len() - TRAILER_LEN;
    let toc_pos_bytes: [u8; 8] = archive_bytes[trailer_start + 22..trailer_start + 30]
        .try_into()
        .unwrap();
    let toc_len_bytes: [u8; 8] = archive_bytes[trailer_start + 30..trailer_start + 38]
        .try_into()
        .unwrap();
    let toc_pos = u64::from_le_bytes(toc_pos_bytes) as usize;
    let toc_len = u64::from_le_bytes(toc_len_bytes) as usize;

    let first_entry_offset = toc_pos + 4 + 4;
    let len_field_offset = first_entry_offset + 8;

    // Lie that the section is exactly MIN_LEN bytes: one byte short of the
    // Meta header, so the header decode crosses the section boundary.
    let lied_len: u64 = Header::MIN_LEN as u64;
    archive_bytes[len_field_offset..len_field_offset + 8].copy_from_slice(&lied_len.to_le_bytes());

    let new_toc_checksum = crate::hash::hash128(&archive_bytes[toc_pos..toc_pos + toc_len]);
    let csum_field_offset = trailer_start + 4 + 1 + 1;
    archive_bytes[csum_field_offset..csum_field_offset + 16]
        .copy_from_slice(&new_toc_checksum.to_le_bytes());

    let dir = tempfile::tempdir()?;
    let fs = StdFs;
    let forged = dir.path().join("forged-boundary.sst");
    let path = forged.as_path();
    {
        let mut f = fs.open(
            path,
            &FsOpenOptions::new().write(true).create(true).truncate(true),
        )?;
        f.write_all(&archive_bytes)?;
    }

    let table_id: TableId = 7;
    let scan = scan_sst_blocks(&fs, path, table_id, SstLayout::PLAIN, None)?;
    // The shrunken section length ALSO breaks the TOC tiling invariant
    // (the sections no longer reach the TOC start), so the walk reports
    // the tiling finding alongside the boundary violation.
    assert_eq!(
        scan.errors.len(),
        2,
        "expected the tiling finding plus the boundary violation, got {:?}",
        scan.errors,
    );
    assert!(
        scan.errors
            .iter()
            .any(|e| matches!(e, BlockVerifyError::TocCorrupted { .. })),
        "the shrunken section length must break the TOC tiling: {:?}",
        scan.errors,
    );
    assert!(
        scan.errors.iter().any(|err| matches!(
            err,
            BlockVerifyError::HeaderCorrupted { table_id: t, offset: 0, reason, .. }
                if *t == table_id && reason.contains("extends past the section end"),
        )),
        "expected a section-boundary HeaderCorrupted; got {:?}",
        scan.errors,
    );
    Ok(())
}

/// Builds a tree with `batches` separate L0 SSTs (one flush per batch) so
/// the parallel scrubber actually has multiple files to fan out over.
fn populate_multi_sst(dir: &std::path::Path, batches: usize, per_batch: usize) {
    let cfg = Config::new(
        dir,
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
    let tree = cfg.open().unwrap();
    let mut seqno = 1u64;
    for b in 0..batches {
        for i in 0..per_batch {
            let key = format!("b{b:03}k{i:08}");
            tree.insert(key.as_bytes(), b"v".as_slice(), seqno);
            seqno += 1;
        }
        tree.flush_active_memtable(seqno).unwrap();
        seqno += 1;
    }
    drop(tree);
}

#[test]
fn verify_checksum_method_on_clean_tree_is_ok() {
    let dir = tempfile::tempdir().unwrap();
    populate_tree(dir.path(), 500);
    let tree = reopen_tree(dir.path());
    let report = tree.verify_checksum();
    assert!(report.is_ok(), "clean tree must verify clean: {report:?}");
    assert!(report.sst_files_scanned >= 1);
    assert!(report.blocks_scanned >= 1);
}

#[test]
fn verify_checksum_with_parallel_matches_sequential() {
    let dir = tempfile::tempdir().unwrap();
    populate_multi_sst(dir.path(), 5, 300);
    let tree = reopen_tree(dir.path());

    let seq = tree.verify_checksum_with(&VerifyOptions::default());
    let par = tree.verify_checksum_with(&VerifyOptions::default().parallelism(4));

    assert!(
        seq.sst_files_scanned >= 2,
        "need >1 SST to exercise parallelism, got {}",
        seq.sst_files_scanned,
    );
    // Parallel run reports the SAME findings as sequential — only order may
    // differ. Counts must match exactly.
    assert_eq!(seq.sst_files_scanned, par.sst_files_scanned);
    assert_eq!(seq.blocks_scanned, par.blocks_scanned);
    assert_eq!(seq.errors.len(), par.errors.len());
    assert!(
        seq.is_ok() && par.is_ok(),
        "clean tree: seq={seq:?} par={par:?}"
    );
}

#[test]
fn verify_checksum_with_throttle_runs_inter_sst_pause() {
    // A non-zero throttle on the default (serial) path exercises the
    // inter-SST pause between tables. The smallest possible delay keeps the
    // test fast while still hitting the sleep branch.
    let dir = tempfile::tempdir().unwrap();
    populate_multi_sst(dir.path(), 3, 300);
    let tree = reopen_tree(dir.path());

    let report = tree.verify_checksum_with(
        &VerifyOptions::default().throttle(std::time::Duration::from_nanos(1)),
    );
    assert!(
        report.sst_files_scanned >= 2,
        "need >1 SST to exercise the inter-SST throttle, got {}",
        report.sst_files_scanned,
    );
    assert!(report.is_ok(), "clean tree must verify clean: {report:?}");
}

#[test]
fn verify_checksum_with_parallel_detects_corruption() {
    use crate::table::block::Header;
    let dir = tempfile::tempdir().unwrap();
    populate_multi_sst(dir.path(), 4, 300);

    let sst_path = pick_first_sst_path(dir.path());
    let flip_offset = Header::MIN_LEN as u64;
    {
        let mut f = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&sst_path)
            .unwrap();
        f.seek(SeekFrom::Start(flip_offset)).unwrap();
        let mut byte = [0u8; 1];
        f.read_exact(&mut byte).unwrap();
        byte[0] ^= 0xFF;
        f.seek(SeekFrom::Start(flip_offset)).unwrap();
        f.write_all(&byte).unwrap();
        f.sync_all().unwrap();
    }

    let tree = reopen_tree(dir.path());
    let report = tree.verify_checksum_with(&VerifyOptions::default().parallelism(4));
    assert!(
        !report.is_ok(),
        "parallel scrub must surface the flipped byte: {report:?}",
    );
}

#[test]
fn verify_checksum_with_throttle_completes_clean() {
    let dir = tempfile::tempdir().unwrap();
    populate_multi_sst(dir.path(), 3, 200);
    let tree = reopen_tree(dir.path());
    let opts = VerifyOptions::default()
        .parallelism(2)
        .throttle(std::time::Duration::from_millis(1));
    let report = tree.verify_checksum_with(&opts);
    assert!(
        report.is_ok(),
        "throttled scrub must still verify clean: {report:?}"
    );
    assert!(report.sst_files_scanned >= 2);
}

/// Raw sections carry NO per-section checksum (the SFA trailer checksum
/// covers only the TOC bytes), so the walk must at least STRUCTURALLY
/// validate `linked_blob_files`: a corrupted count prefix would otherwise
/// pass the walk clean, and a heal-enabled scrub could restamp the manifest
/// digest over a broken blob-link list that blob GC / relocation then
/// misreads.
#[test]
fn verify_sst_file_flags_a_corrupt_blob_link_count() {
    let dir = tempfile::tempdir().unwrap();

    // A KV-separated tree: large values go to a blob file, the SST carries
    // a linked_blob_files section.
    let crate::AnyTree::Blob(tree) = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .with_kv_separation(Some(crate::KvSeparationOptions::default()))
    .open()
    .unwrap() else {
        unreachable!("kv separation configured");
    };
    let big = |i: u32| format!("{i:08}").repeat(512);
    for i in 0u32..10 {
        tree.insert(format!("key{i:05}"), big(i), u64::from(i) + 1);
    }
    tree.flush_active_memtable(10).unwrap();
    let sst_path = {
        let binding = tree.index.version_history.read().latest_version();
        let table = binding
            .version
            .iter_tables()
            .next()
            .expect("flush produced one table");
        (*table.path).clone()
    };
    drop(tree);

    // Corrupt the section's u32 count prefix: the declared records no longer
    // end at the section's end.
    let pos = {
        let mut f = std::fs::File::open(&sst_path).unwrap();
        let reader = crate::sfa::Reader::from_reader(&mut f).expect("SFA trailer reads");
        let entry = reader
            .toc()
            .iter()
            .find(|e| e.name() == b"linked_blob_files")
            .expect("the SST carries a linked_blob_files section");
        usize::try_from(entry.pos()).expect("section offset fits usize")
    };
    let mut bytes = std::fs::read(&sst_path).unwrap();
    *bytes.get_mut(pos).expect("count prefix within the file") ^= 0xFF;
    std::fs::write(&sst_path, &bytes).unwrap();

    let fs: alloc::sync::Arc<dyn crate::fs::Fs> = alloc::sync::Arc::new(crate::fs::StdFs);
    let report = verify_sst_file_with_fs(&fs, &sst_path);
    assert!(
        report.errors.iter().any(|e| matches!(
            e,
            BlockVerifyError::TocCorrupted { section_name, reason, .. }
                if section_name == b"linked_blob_files" && reason.contains("blob-link count")
        )),
        "a corrupt blob-link count must fail the out-of-band walk, not be \
         skipped as an unchecked raw section: {report:?}",
    );
}

/// The shape walk over a `linked_blob_files` section holding `bytes`.
fn blob_link_shape(bytes: &[u8]) -> Option<String> {
    use crate::fs::{Fs as _, FsOpenOptions};

    let fs = crate::fs::MemFs::new();
    let path = std::path::Path::new("/linked_blob_files");
    {
        let mut file = fs
            .open(path, &FsOpenOptions::new().write(true).create(true))
            .expect("create");
        file.write_all(bytes).expect("write");
    }
    let file = fs
        .open(path, &FsOpenOptions::new().read(true))
        .expect("open");
    let mut reader = std::io::BufReader::new(file);
    let len = u64::try_from(bytes.len()).expect("test section fits u64");
    raw_section_shape_error(&mut reader, b"linked_blob_files", 0, len).expect("in-memory read")
}

/// One record: 32 fixed bytes, then the first and last key, each behind its
/// two-byte length.
fn blob_link_record(first: &[u8], last: &[u8]) -> Vec<u8> {
    let mut record = vec![0u8; crate::table::writer::LINKED_BLOB_FILE_FIXED_LEN];
    for key in [first, last] {
        let len = u16::try_from(key.len()).expect("test key fits u16");
        record.extend_from_slice(&len.to_le_bytes());
        record.extend_from_slice(key);
    }
    record
}

fn blob_link_section(count: u32, records: &[Vec<u8>]) -> Vec<u8> {
    let mut section = count.to_le_bytes().to_vec();
    for record in records {
        section.extend_from_slice(record);
    }
    section
}

/// Records whose keys fill the section exactly pass the shape walk.
#[test]
fn blob_link_shape_accepts_records_that_fill_the_section() {
    let records = [blob_link_record(b"a", b"m"), blob_link_record(b"", b"zz")];
    assert_eq!(blob_link_shape(&blob_link_section(2, &records)), None);
}

/// Each way a record can overrun its section is named with the record that
/// overran: the fixed fields, a key's length prefix, and the key bytes.
#[test]
fn blob_link_shape_names_the_record_that_runs_past_the_section() {
    let record = blob_link_record(b"a", b"b");

    // Fixed fields of the second record past the end.
    let fixed = blob_link_shape(&blob_link_section(2, core::slice::from_ref(&record)));
    assert!(
        fixed
            .as_deref()
            .is_some_and(|r| r.contains("record 1 runs past")),
        "{fixed:?}"
    );

    // The second record's fixed fields fit, the length of its first key does not.
    let mut prefix = blob_link_section(2, core::slice::from_ref(&record));
    prefix.extend_from_slice(&[0u8; crate::table::writer::LINKED_BLOB_FILE_FIXED_LEN + 1]);
    let short = blob_link_shape(&prefix);
    assert!(
        short
            .as_deref()
            .is_some_and(|r| r.contains("record 1 runs past")),
        "{short:?}"
    );

    // A key length that points past the section.
    let mut long_key = blob_link_record(b"a", b"b");
    let last_len_at = crate::table::writer::LINKED_BLOB_FILE_FIXED_LEN + 2 + 1;
    long_key
        .get_mut(last_len_at..last_len_at + 2)
        .expect("last key length")
        .copy_from_slice(&u16::MAX.to_le_bytes());
    let past = blob_link_shape(&blob_link_section(1, &[long_key]));
    assert!(
        past.as_deref()
            .is_some_and(|r| r.contains("record 0 runs past")),
        "{past:?}"
    );
}

/// Bytes left over after the declared records are a count that disagrees
/// with the section, reported with how much the records took.
#[test]
fn blob_link_shape_rejects_bytes_past_the_declared_records() {
    let record = blob_link_record(b"a", b"b");
    let mut section = blob_link_section(1, core::slice::from_ref(&record));
    section.push(0);
    let reason = blob_link_shape(&section);
    let taken = 4 + record.len();
    assert!(
        reason
            .as_deref()
            .is_some_and(|r| r.contains(&format!("({taken} bytes of records)"))),
        "{reason:?}"
    );
}

#[test]
fn verify_checksum_with_throttle_does_not_sleep_after_last_sst() {
    // Regression: the throttle is an INTER-SST pause and must not fire after
    // the final SST. A single-SST tree scrubbed with a large throttle must
    // return promptly; the bug slept one full throttle interval after the
    // only table, making a finished scrub look hung. Sequential path
    // (parallelism 1, one table) so this pins the single-worker loop.
    let dir = tempfile::tempdir().unwrap();
    populate_multi_sst(dir.path(), 1, 50);
    let tree = reopen_tree(dir.path());
    let throttle = std::time::Duration::from_millis(400);
    let opts = VerifyOptions::default().parallelism(1).throttle(throttle);
    let start = std::time::Instant::now();
    let report = tree.verify_checksum_with(&opts);
    let elapsed = start.elapsed();
    assert!(report.is_ok(), "clean single-SST scrub: {report:?}");
    assert_eq!(report.sst_files_scanned, 1, "test needs exactly one SST");
    assert!(
        elapsed < throttle / 2,
        "a single-SST scrub must not sleep the inter-SST throttle after the \
         last table: took {elapsed:?} with a {throttle:?} throttle",
    );
}

/// A table whose real parity is RS(4,2) but whose surviving RECOGNIZED mirror
/// says ECC is off must NOT be walked under that descriptor. One mirror
/// re-stamped to an unrecognized kind and the other to `Off` leaves exactly one
/// recognized copy, and trusting it sizes every frame without its parity
/// trailer: the walk then reads parity bytes as the next block header and
/// condemns a healthy table. The descriptor must be judged against the block
/// framing it implies, and a descriptor that does not frame the data must fail
/// safe to unrecognized (skip the ECC-dependent sections, report incomplete).
#[cfg(feature = "page_ecc")]
#[test]
fn verify_sst_file_lone_recognized_mirror_that_misframes_falls_back_to_unrecognized() {
    use crate::table::Writer;
    use crate::table::block::EccParams;

    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("t");

    let mut writer = Writer::new(path.clone(), 0, 0, std::sync::Arc::new(crate::fs::StdFs))
        .unwrap()
        .use_ecc(Some(EccParams::RS_4_2));
    for i in 0u64..200 {
        writer
            .write(crate::InternalValue::from_components(
                format!("key-{i:05}").into_bytes(),
                format!("value-{i:05}").into_bytes(),
                i + 1,
                crate::ValueType::Value,
            ))
            .unwrap();
    }
    assert!(
        writer.finish().unwrap().is_some(),
        "the fixture is non-empty"
    );

    // Sanity: the intact table verifies clean under its real descriptor.
    let report = verify_sst_file(&path);
    assert!(
        report.is_ok(),
        "an intact RS(4,2) table must verify clean: {:?}",
        report.errors,
    );

    // The MID mirror keeps a recognized descriptor, but a FORGED one: `Off`.
    // The tail goes to an unrecognized kind, so arbitration sees exactly one
    // recognized copy and no full-metadata divergence (the comparison masks the
    // ECC fields when a mirror is unrecognized).
    // `[kind, data_shards, parity_shards, granularity]`: kind 0 with the
    // reserved bytes zeroed is the canonical `Off`, kind 9 is unknown.
    crate::test_forge::forge_mid_meta_value(&path, b"descriptor#page_ecc", &[0, 0, 0, 0]).unwrap();
    crate::test_forge::forge_tail_meta_value(&path, b"descriptor#page_ecc", &[9, 0, 0, 0]).unwrap();

    let report = verify_sst_file(&path);
    assert!(
        report.errors.is_empty(),
        "the blocks are healthy — a mis-framing descriptor must not be trusted \
         into reporting them corrupt: {:?}",
        report.errors,
    );
    assert!(
        report
            .warnings
            .iter()
            .any(|w| matches!(w, crate::verify::BlockVerifyWarning::UnrecognizedEcc { .. })),
        "the descriptor that does not frame the data must fail safe to \
         unrecognized: {:?}",
        report.warnings,
    );
    assert!(
        report.incomplete,
        "the ECC-dependent sections were skipped, so the scan is incomplete",
    );
}

/// The other side of the same arbitration: a HEALTHY table whose one meta
/// mirror had its ECC descriptor re-stamped to an unknown kind must keep being
/// walked under the surviving valid descriptor. Condemning it would be
/// terminal for a range-tombstone SST, which salvage cannot re-emit — and the
/// framing check exists to separate this case from the mis-framing one, not to
/// fail both.
#[test]
fn verify_sst_file_lone_recognized_mirror_that_frames_stays_authoritative() {
    let dir = tempfile::tempdir().unwrap();
    populate_tree(dir.path(), 200);
    let sst_path = pick_first_sst_path(dir.path());

    let report = verify_sst_file(&sst_path);
    assert!(
        report.is_ok(),
        "intact SST must be clean: {:?}",
        report.errors,
    );

    // Only the TAIL descriptor is forged to an unknown kind; `meta_mid` keeps
    // the real one, which frames the blocks.
    crate::test_forge::forge_tail_meta_value(&sst_path, b"descriptor#page_ecc", &[9, 0, 0, 0])
        .unwrap();

    let report = verify_sst_file(&sst_path);
    assert!(
        report.errors.is_empty(),
        "the surviving descriptor frames the data, so the walk must proceed \
         under it: {:?}",
        report.errors,
    );
    assert!(
        !report.incomplete,
        "the data blocks were walked, so the scan is complete",
    );
    assert!(
        report.is_ok(),
        "a descriptor-only forge on one mirror must not condemn a healthy table",
    );
}

/// BOTH mirrors reading as unknown kinds does not make a table unreadable.
/// `Off` is a descriptor like any other — it frames the blocks or it does not —
/// and most tables carry no parity at all, so it is tried before giving up.
/// Without it such a table is skipped section by section and, if it carries
/// range tombstones salvage cannot re-emit, dropped outright: a total loss of
/// its key range over two bytes that say nothing about the data.
#[test]
fn verify_sst_file_both_mirrors_unrecognized_falls_back_to_off() -> crate::Result<()> {
    use crate::table::Writer;

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("t");

    let mut writer =
        Writer::new(path.clone(), 0, 0, std::sync::Arc::new(crate::fs::StdFs))?.use_ecc(None);
    for i in 0u64..200 {
        writer.write(crate::InternalValue::from_components(
            format!("key-{i:05}").into_bytes(),
            format!("value-{i:05}").into_bytes(),
            i + 1,
            crate::ValueType::Value,
        ))?;
    }
    assert!(writer.finish()?.is_some(), "the fixture is non-empty");

    let report = verify_sst_file(&path);
    assert!(
        report.is_ok(),
        "an intact parity-less table must verify clean: {:?}",
        report.errors,
    );

    // Both mirrors go to unknown kinds, so no recognized descriptor survives.
    crate::test_forge::forge_mid_meta_value(&path, b"descriptor#page_ecc", &[9, 0, 0, 0])?;
    crate::test_forge::forge_tail_meta_value(&path, b"descriptor#page_ecc", &[8, 0, 0, 0])?;

    let report = verify_sst_file(&path);
    assert!(
        report.errors.is_empty(),
        "the blocks are untouched: {:?}",
        report.errors,
    );
    assert!(
        !report
            .warnings
            .iter()
            .any(|w| matches!(w, crate::verify::BlockVerifyWarning::UnrecognizedEcc { .. })),
        "the file frames without a trailer, which answers the question the \
         descriptors no longer can: {:?}",
        report.warnings,
    );
    assert!(
        !report.incomplete,
        "every section was walked, so nothing is left unverified — the whole \
         point of trying `Off` rather than giving up",
    );
    // Framing answered how to READ the blocks. It did not make either
    // descriptor valid, and passing the table as clean would leave the next
    // reader to infer the layout all over again.
    assert!(
        report.warnings.iter().any(|w| matches!(
            w,
            crate::verify::BlockVerifyWarning::EccDescriptorsUnreadable { .. }
        )),
        "the inferred layout must not hide the malformed descriptors: {:?}",
        report.warnings,
    );
    Ok(())
}

/// Framing pins the trailer LENGTH, never the codec. RS(4,2) and XOR(2,1)
/// derive the SAME parity length for any payload whose `ceil(N/4)` and
/// `ceil(N/2)` are both even, so a mirror re-stamped from one to the other
/// frames perfectly and the descriptor is KEPT — the walk needs the length, and
/// the length is right.
///
/// What the codec question answers is only which explanation to print: the
/// impostor disagrees with the trailer bytes, the real scheme reproduces them.
#[cfg(feature = "page_ecc")]
#[test]
fn codec_disagrees_everywhere_same_length_scheme_is_flagged_but_not_refused() -> crate::Result<()> {
    use crate::coding::Encode;
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::{BlockType, EccParams, Header, expected_parity_len};

    let real = EccParams::RS_4_2;
    let impostor = EccParams::try_new(2, 1)?;
    // The collision this test exists for: pick a payload length the two schemes
    // size identically, so framing alone cannot separate them.
    const DATA_LENGTH: u32 = 4096;
    assert_eq!(
        expected_parity_len(DATA_LENGTH, real),
        expected_parity_len(DATA_LENGTH, impostor),
        "the fixture must exercise a length collision, or it proves nothing",
    );

    // Shards that differ from each other: a uniform payload makes every shard
    // identical, and BOTH codecs then emit all-zero parity — which would let the
    // impostor pass for the wrong reason.
    let payload = discriminating_payload(DATA_LENGTH);
    let parity = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes the fixture");
    let mut header = Header {
        data_length: DATA_LENGTH,
        uncompressed_length: DATA_LENGTH,
        ..Header::test_dummy(BlockType::Data)
    };
    header.bind_checksum(
        Checksum::from_raw(crate::hash::hash128(&payload)),
        crate::table::block::ChecksumAt::table(0, 0),
    );

    let mut archive_bytes: Vec<u8> = Vec::new();
    {
        let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
        writer.start("data").unwrap();
        writer.write_all(&header.encode_into_vec()).unwrap();
        writer.write_all(&payload).unwrap();
        writer.write_all(&parity).unwrap();
        writer.finish().unwrap();
    }

    let dir = tempfile::tempdir()?;
    let fs = StdFs;
    let path = dir.path().join("rs42.sst");
    {
        let mut f = fs.open(
            &path,
            &FsOpenOptions::new().write(true).create(true).truncate(true),
        )?;
        f.write_all(&archive_bytes)?;
    }

    let mut probe = fs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let toc = sfa_reader.toc();
    let entry = toc
        .section(b"data")
        .expect("the fixture has a data section");
    let (start, end) = (entry.pos(), entry.pos() + entry.len());

    // Both schemes frame the section: that is exactly the blind spot.
    assert_eq!(
        scheme_frames_region(probe.as_ref(), ScrubEcc::Scheme(real), start, end)?,
        Some(true),
        "the real scheme must frame its own block",
    );
    assert_eq!(
        scheme_frames_region(probe.as_ref(), ScrubEcc::Scheme(impostor), start, end)?,
        Some(true),
        "the same-length impostor frames identically — framing cannot separate them",
    );

    // BOTH are kept: the length is what the walk needs, and both get it right.
    // Refusing the impostor here would skip every ECC-bearing section, and a
    // table carrying range tombstones is then excluded outright.
    let cap = block_data_length_cap(0);
    assert_eq!(
        arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
        Some(true),
        "the real scheme sizes its own blocks",
    );
    assert_eq!(
        arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(impostor), 0)?,
        Some(true),
        "the impostor sizes them identically, so the walk can still proceed — \
         refusing it would cost the whole table for a parity question",
    );

    // The trailer bytes separate them, and that difference is reported.
    assert!(
        !codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(real), 0, cap),
        "the real scheme reproduces the trailer, so nothing is suspect",
    );
    assert!(
        codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(impostor), 0, cap),
        "the impostor reproduces no trailer — the signature of a mis-identified \
         scheme, which is what the operator is told",
    );
    Ok(())
}

/// A region that cannot be framed refuses the descriptor even when the others
/// frame, and `Off` is no exception — it faces the same rule as every scheme.
///
/// The split is genuinely ambiguous: a corrupt header explains it if the
/// descriptor is right, and so does a wrong descriptor whose trailer lengths
/// coincide for one region's payload sizes and not the other's. Nothing here
/// tells those apart, so the table is refused rather than walked under a
/// descriptor that might mis-size every frame in a HEALTHY region.
///
/// The cost is the specific finding: the walk skips the ECC-dependent sections
/// and reports "unrecognized, incomplete" instead of naming the damaged header.
/// The table still fails verification and still routes to salvage.
#[test]
fn verify_sst_file_unframeable_data_region_refuses_the_descriptor() {
    use crate::fs::{Fs, FsOpenOptions, StdFs};

    let dir = tempfile::tempdir().unwrap();
    populate_tree(dir.path(), 200);
    let sst_path = pick_first_sst_path(dir.path());

    // One mirror's descriptor goes to an unknown kind, so the surviving
    // recognized copy has to be arbitrated rather than trusted outright.
    crate::test_forge::forge_tail_meta_value(&sst_path, b"descriptor#page_ecc", &[9, 0, 0, 0])
        .unwrap();

    // Corrupt the FIRST data block's header in place: the data region no longer
    // frames, while every other section still does.
    let fs = StdFs;
    let data_pos = {
        let mut probe = fs
            .open(&sst_path, &FsOpenOptions::new().read(true))
            .unwrap();
        let reader = crate::sfa::Reader::from_reader(&mut probe).unwrap();
        reader
            .toc()
            .section(b"data")
            .expect("the SST has a data section")
            .pos()
    };
    {
        let mut f = fs
            .open(&sst_path, &FsOpenOptions::new().write(true))
            .unwrap();
        f.seek(SeekFrom::Start(data_pos)).unwrap();
        f.write_all(&[0xFFu8; 16]).unwrap();
    }

    let report = verify_sst_file(&sst_path);
    assert!(
        report
            .warnings
            .iter()
            .any(|w| matches!(w, crate::verify::BlockVerifyWarning::UnrecognizedEcc { .. })),
        "a region that will not frame refuses the descriptor, whatever the \
         other regions did: {:?}",
        report.warnings,
    );
    assert!(
        report.incomplete,
        "the ECC-dependent sections were skipped, so the scan is incomplete",
    );
    assert!(
        !report.is_ok(),
        "the table must not verify clean: its data region holds a corrupt header",
    );
}

/// A region that could not be READ is not a region that agreed. Framing decides
/// which descriptor the walk uses, so silently dropping an unreadable region
/// lets the remaining ones carry the verdict — and if the dropped one is the
/// one that would have refused, a transient failure here plus a successful
/// retry in the walk reports corruption across a HEALTHY table.
///
/// The fixture puts the refusing region FIRST so the injected failure lands on
/// it: `data` carries no parity (RS(4,2) cannot frame it) while `tli` does.
#[cfg(feature = "page_ecc")]
#[test]
fn arbitrate_by_framing_propagates_an_unreadable_region() -> crate::Result<()> {
    use crate::fs::{Fault, FaultFs, FaultOp, FaultRule, Fs, FsOpenOptions, StdFs};
    use crate::io::ErrorKind;
    use crate::table::block::EccParams;

    const LEN: u32 = 4096;
    let real = EccParams::RS_4_2;

    let bare = discriminating_payload(LEN);
    let framed = discriminating_payload(LEN);
    let framed_parity = crate::ecc::encode_parity(&framed, 4, 2).expect("RS(4,2) encodes");

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("unreadable-region.sst");
    write_block_archive(
        &path,
        &[
            ("data", vec![(bare, Vec::new())]),
            ("tli", vec![(framed, framed_parity)]),
        ],
    )?;

    let mut plain = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut plain)?;
    let toc = sfa_reader.toc();

    // Readable, the first region refuses: the verdict the fault must not erase.
    assert_eq!(
        arbitrate_by_framing(plain.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
        Some(false),
        "the parity-less data region cannot frame under RS(4,2) — the premise \
         of this test",
    );

    // The first read of that region now fails. Dropping it would leave `tli`
    // framing alone and the descriptor accepted.
    let fault = FaultFs::new(StdFs);
    let injector = fault.injector();
    injector.arm(FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Other)).once());
    let faulted = fault.open(&path, &FsOpenOptions::new().read(true))?;
    let verdict = arbitrate_by_framing(faulted.as_ref(), toc, ScrubEcc::Scheme(real), 0);
    injector.clear();
    assert!(
        verdict.is_err(),
        "an unread region must surface as a read failure, not as a region that \
         had nothing to say: got {verdict:?}",
    );
    Ok(())
}

/// One synthetic block: its payload and the parity trailer stored after it.
/// Parity is supplied rather than derived so a test can store a DAMAGED
/// trailer, which is what separates a wrong codec from a rotted block.
#[cfg(feature = "page_ecc")]
type SyntheticBlock = (Vec<u8>, Vec<u8>);

/// Writes a synthetic SFA archive: for each named section, a run of data blocks
/// carrying `(payload, parity)` exactly as given.
#[cfg(feature = "page_ecc")]
fn write_block_archive(
    path: &std::path::Path,
    sections: &[(&str, Vec<SyntheticBlock>)],
) -> crate::Result<()> {
    use crate::coding::Encode;
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::{BlockType, Header};

    let mut archive_bytes: Vec<u8> = Vec::new();
    {
        let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
        for (name, blocks) in sections {
            writer.start(*name).unwrap();
            for (payload, parity) in blocks {
                #[expect(
                    clippy::cast_possible_truncation,
                    reason = "test payloads are kilobytes"
                )]
                let mut header = Header {
                    data_length: payload.len() as u32,
                    uncompressed_length: payload.len() as u32,
                    ..Header::test_dummy(BlockType::Data)
                };
                header.bind_checksum(
                    Checksum::from_raw(crate::hash::hash128(payload)),
                    crate::table::block::ChecksumAt::table(0, writer.get_ref().position()),
                );
                writer.write_all(&header.encode_into_vec()).unwrap();
                writer.write_all(payload).unwrap();
                writer.write_all(parity).unwrap();
            }
        }
        writer.finish().unwrap();
    }
    let mut f = StdFs.open(
        path,
        &FsOpenOptions::new().write(true).create(true).truncate(true),
    )?;
    f.write_all(&archive_bytes)?;
    Ok(())
}

/// A payload whose shards all differ, so the codecs cannot coincide on it.
/// Period 251 divides neither shard size under RS(4,2) nor under XOR(2,1).
#[cfg(feature = "page_ecc")]
fn discriminating_payload(len: u32) -> Vec<u8> {
    (0..len)
        .map(|i| u8::try_from(i % 251).expect("the modulus keeps every value below 256"))
        .collect()
}

/// A mismatch does not end the region. The report's claim is that the scheme
/// reproduces NO trailer, so one that it does reproduce refutes it — wherever
/// it sits. Stopping at the first mismatch would call ordinary scattered rot a
/// mis-identified scheme and send the operator recompacting a table whose
/// descriptor is right.
///
/// The fixture puts the mismatch FIRST: a block only the real codec reproduces,
/// then a uniform one whose shards are identical under either split, so both
/// codecs emit its all-zero parity.
#[cfg(feature = "page_ecc")]
#[test]
fn codec_confirms_region_keeps_scanning_past_a_mismatch() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::EccParams;

    const LEN: u32 = 4096;
    let real = EccParams::RS_4_2;
    let impostor = EccParams::try_new(2, 1)?;

    // Block 2: uniform, so its shards are identical under either split.
    let degenerate = vec![0xABu8; LEN as usize];
    let degenerate_parity = crate::ecc::encode_parity(&degenerate, 4, 2).expect("RS(4,2) encodes");
    assert_eq!(
        degenerate_parity,
        crate::ecc::encode_parity(&degenerate, 2, 1).expect("XOR(2,1) encodes"),
        "the fixture's second block must be one the two codecs agree on, or the \
         scan has no later match to find",
    );

    // Block 1: shards differ, so only the real codec reproduces its trailer.
    let discriminating = discriminating_payload(LEN);
    let discriminating_parity =
        crate::ecc::encode_parity(&discriminating, 4, 2).expect("RS(4,2) encodes");

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("two-blocks.sst");
    write_block_archive(
        &path,
        &[(
            "data",
            vec![
                (discriminating, discriminating_parity),
                (degenerate, degenerate_parity),
            ],
        )],
    )?;

    let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let toc = sfa_reader.toc();
    let cap = block_data_length_cap(0);

    let data = toc
        .section(b"data")
        .expect("the fixture has a data section");
    let (start, end) = (data.pos(), data.pos() + data.len());

    assert_eq!(
        codec_confirms_region((probe.as_ref(), 0), ScrubEcc::Scheme(real), start, end, cap),
        CodecVerdict::Confirmed,
        "the real codec reproduces both trailers",
    );
    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(impostor),
            start,
            end,
            cap
        ),
        CodecVerdict::Confirmed,
        "the second block's trailer IS reproduced, and a scan that stopped at \
         the first mismatch would never see it",
    );
    assert!(
        !codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(impostor), 0, cap),
        "one reproduced trailer refutes the report's claim, so a table whose \
         mismatches are ordinary rot is not blamed on its scheme",
    );
    Ok(())
}

/// A scan that STOPPED is not a scan that found nothing. "No trailer anywhere
/// is reproduced" can only be said about a region that was inspected to its
/// end, so a traversal cut short — an unreadable or undecodable header, a
/// length past the cap, frames that stop tiling — leaves the question
/// unanswered rather than answered in the negative. Otherwise a mismatch before
/// the cut plus a matching trailer beyond it reads as a mis-identified scheme,
/// and the operator recompacts a table whose descriptor is right.
#[cfg(feature = "page_ecc")]
#[test]
fn codec_confirms_region_scan_stops_early_reports_incomplete() -> crate::Result<()> {
    use crate::coding::Decode;
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::{EccParams, Header};

    const LEN: u32 = 4096;
    let impostor = EccParams::try_new(2, 1)?;

    // Block 1 mismatches under the impostor, block 2 (uniform) matches it.
    let discriminating = discriminating_payload(LEN);
    let discriminating_parity =
        crate::ecc::encode_parity(&discriminating, 4, 2).expect("RS(4,2) encodes");
    let degenerate = vec![0xABu8; LEN as usize];
    let degenerate_parity = crate::ecc::encode_parity(&degenerate, 4, 2).expect("RS(4,2) encodes");

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("truncated-scan.sst");
    write_block_archive(
        &path,
        &[(
            "data",
            vec![
                (discriminating, discriminating_parity.clone()),
                (degenerate, degenerate_parity),
            ],
        )],
    )?;

    let (start, end, second_at) = {
        let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
        let reader = crate::sfa::Reader::from_reader(&mut probe)?;
        let data = reader
            .toc()
            .section(b"data")
            .expect("the fixture has a data section");
        let head = crate::file::read_exact(probe.as_ref(), data.pos(), Header::MAX_LEN)?;
        let header = Header::decode_from(&mut &head[..])?;
        let frame = Header::header_len(header.block_type) as u64
            + u64::from(header.data_length)
            + discriminating_parity.len() as u64;
        (data.pos(), data.pos() + data.len(), data.pos() + frame)
    };

    // The second block's header no longer decodes, so the scan stops after the
    // first — the one that mismatches.
    {
        let mut f = StdFs.open(&path, &FsOpenOptions::new().write(true))?;
        f.seek(SeekFrom::Start(second_at))?;
        f.write_all(&[0xFFu8; 16])?;
    }

    let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let toc = sfa_reader.toc();
    let cap = block_data_length_cap(0);

    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(impostor),
            start,
            end,
            cap
        ),
        CodecVerdict::Incomplete,
        "the region was not inspected to its end, so it cannot say the scheme \
         reproduces nothing",
    );
    assert!(
        !codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(impostor), 0, cap),
        "and no diagnosis is reported off a truncated probe",
    );
    Ok(())
}

/// An EMPTY region was not cut short — there was nothing to cut. Reporting it
/// as unfinished would let one zero-length section silence the diagnosis for
/// the whole table, and such a section is not exotic: a restricted view whose
/// punch offset reaches the end of the data section produces exactly that.
#[cfg(feature = "page_ecc")]
#[test]
fn codec_confirms_region_empty_region_reports_no_evidence() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::EccParams;

    const LEN: u32 = 4096;
    let real = EccParams::RS_4_2;

    // `data` rejects: clean payload, rotted trailer.
    let payload = discriminating_payload(LEN);
    let mut rotted = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes");
    *rotted.first_mut().expect("the trailer is non-empty") ^= 0xFF;

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("empty-region.sst");
    write_block_archive(
        &path,
        &[("data", vec![(payload, rotted)]), ("filter", Vec::new())],
    )?;

    let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let toc = sfa_reader.toc();
    let cap = block_data_length_cap(0);

    let filter = toc
        .section(b"filter")
        .expect("the fixture has a filter section");
    assert_eq!(filter.len(), 0, "the fixture's filter section is empty");
    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(real),
            filter.pos(),
            filter.pos() + filter.len(),
            cap,
        ),
        CodecVerdict::NoEvidence,
        "an empty region holds nothing to judge, which is not the same as a \
         traversal that stopped",
    );
    assert!(
        codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(real), 0, cap),
        "so it leaves the answer to the regions that do hold blocks, instead of \
         silencing the whole table",
    );
    Ok(())
}

/// The table-wide claim needs EVERY region, so one region's rejection cannot
/// speak over another region the probe never finished reading. A rotted trailer
/// in one section beside a section whose scan stopped short is exactly the
/// mixture that would blame a correct codec: the trailer it reproduces may sit
/// behind the cut.
#[cfg(feature = "page_ecc")]
#[test]
fn codec_disagrees_everywhere_is_silenced_by_a_region_it_could_not_finish() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::EccParams;

    const LEN: u32 = 4096;
    let real = EccParams::RS_4_2;

    // `data`: clean payload, rotted trailer — a complete scan that rejects.
    let payload = discriminating_payload(LEN);
    let mut rotted = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes");
    *rotted.first_mut().expect("the trailer is non-empty") ^= 0xFF;

    // `tli`: its only header is corrupted below, so the scan stops at once.
    let other = discriminating_payload(LEN / 2);
    let other_parity = crate::ecc::encode_parity(&other, 4, 2).expect("RS(4,2) encodes");

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("mixed-regions.sst");
    write_block_archive(
        &path,
        &[
            ("data", vec![(payload, rotted)]),
            ("tli", vec![(other, other_parity)]),
        ],
    )?;

    let tli_pos = {
        let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
        let reader = crate::sfa::Reader::from_reader(&mut probe)?;
        reader
            .toc()
            .section(b"tli")
            .expect("the fixture has a tli section")
            .pos()
    };
    {
        let mut f = StdFs.open(&path, &FsOpenOptions::new().write(true))?;
        f.seek(SeekFrom::Start(tli_pos))?;
        f.write_all(&[0xFFu8; 16])?;
    }

    let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let toc = sfa_reader.toc();
    let cap = block_data_length_cap(0);

    let data = toc
        .section(b"data")
        .expect("the fixture has a data section");
    let tli = toc.section(b"tli").expect("the fixture has a tli section");
    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(real),
            data.pos(),
            data.pos() + data.len(),
            cap,
        ),
        CodecVerdict::Rejected,
        "the rotted trailer makes the fully-scanned region reject — one half of \
         the premise",
    );
    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(real),
            tli.pos(),
            tli.pos() + tli.len(),
            cap,
        ),
        CodecVerdict::Incomplete,
        "and the other region was never finished — the other half",
    );
    assert!(
        !codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(real), 0, cap),
        "an unfinished region leaves the table-wide claim unavailable, whatever \
         a finished one found",
    );
    Ok(())
}

/// A rotted trailer is damage, not a verdict on the descriptor. The table keeps
/// its scheme, the walk reads the section, and the damaged block is named as
/// `EccParityMismatch` — which parity may still repair. Refusing the descriptor
/// here would skip the section instead, and a table carrying range tombstones
/// would then be excluded outright over one damaged trailer.
///
/// It is not reported as suspect either: another region reproduces the trailer,
/// so the scheme is not the explanation for this one.
#[cfg(feature = "page_ecc")]
#[test]
fn rotted_trailer_keeps_the_descriptor_and_is_not_reported_suspect() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::EccParams;

    const LEN: u32 = 4096;
    let real = EccParams::RS_4_2;

    // The data block's payload is clean but its stored trailer is rotted.
    let payload = discriminating_payload(LEN);
    let mut rotted = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes");
    *rotted.first_mut().expect("the trailer is non-empty") ^= 0xFF;

    // A second region carries a healthy block under the same codec.
    let healthy = discriminating_payload(LEN / 2);
    let healthy_parity = crate::ecc::encode_parity(&healthy, 4, 2).expect("RS(4,2) encodes");

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("rotted-trailer.sst");
    write_block_archive(
        &path,
        &[
            ("data", vec![(payload, rotted)]),
            ("tli", vec![(healthy, healthy_parity)]),
        ],
    )?;

    let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let toc = sfa_reader.toc();
    let cap = block_data_length_cap(0);

    let data = toc
        .section(b"data")
        .expect("the fixture has a data section");
    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(real),
            data.pos(),
            data.pos() + data.len(),
            cap,
        ),
        CodecVerdict::Rejected,
        "the rotted trailer makes the data region reject on its own",
    );
    assert_eq!(
        arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
        Some(true),
        "damage does not change the trailer LENGTH, so the walk can still read \
         the section and name the damaged block",
    );
    assert!(
        !codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(real), 0, cap),
        "a region that reproduces the trailer rules the scheme out as the \
         explanation, leaving the mismatch reported as what it is: damage",
    );
    Ok(())
}

/// An untrusted `data_length` must not size an allocation. The header is not
/// verified when the arbitration reads it, so a forged one paired with a
/// re-stamped TOC could ask for gigabytes; past the cap the traversal stops
/// instead, which is one of the ways a region ends up only partly inspected.
/// Driven with a tiny cap so the bound itself is what is pinned, rather than
/// needing a multi-gigabyte fixture to reach the real one.
#[cfg(feature = "page_ecc")]
#[test]
fn codec_confirms_region_skips_a_block_whose_declared_length_exceeds_the_cap() -> crate::Result<()>
{
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::EccParams;

    const LEN: u32 = 4096;
    let real = EccParams::RS_4_2;
    let payload = discriminating_payload(LEN);
    let parity = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes");

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("capped.sst");
    write_block_archive(&path, &[("data", vec![(payload, parity)])])?;

    let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let data = sfa_reader
        .toc()
        .section(b"data")
        .expect("the fixture has a data section");
    let (start, end) = (data.pos(), data.pos() + data.len());

    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(real),
            start,
            end,
            cap_for_test()
        ),
        CodecVerdict::Confirmed,
        "under the real cap the block is read and confirms the codec",
    );
    assert_eq!(
        codec_confirms_region((probe.as_ref(), 0), ScrubEcc::Scheme(real), start, end, 16),
        CodecVerdict::Incomplete,
        "a declared length past the cap stops the traversal instead of sizing a \
         read, and a stopped traversal answers nothing about the region",
    );
    Ok(())
}

/// The production payload cap, for tests that contrast it with a tiny one.
#[cfg(feature = "page_ecc")]
fn cap_for_test() -> u64 {
    block_data_length_cap(0)
}

/// The one trailer that refutes the report can sit at the END of a long run of
/// mismatches, so nothing short of the whole region will do. Nine blocks: eight
/// that only the real codec reproduces, and a ninth that is uniform, whose
/// identical shards make both codecs emit the same all-zero parity.
#[cfg(feature = "page_ecc")]
#[test]
fn codec_confirms_region_scans_past_a_run_of_mismatches() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::EccParams;

    const LEN: u32 = 4096;
    const MISMATCH_RUN: usize = 8;
    let real = EccParams::RS_4_2;
    let impostor = EccParams::try_new(2, 1)?;

    let degenerate = vec![0xABu8; LEN as usize];
    let degenerate_parity = crate::ecc::encode_parity(&degenerate, 4, 2).expect("RS(4,2) encodes");
    assert_eq!(
        degenerate_parity,
        crate::ecc::encode_parity(&degenerate, 2, 1).expect("XOR(2,1) encodes"),
        "the LAST block must be one the two codecs agree on, or there is no \
         match at the far end to find",
    );

    let discriminating = discriminating_payload(LEN);
    let discriminating_parity =
        crate::ecc::encode_parity(&discriminating, 4, 2).expect("RS(4,2) encodes");

    let mut blocks: Vec<SyntheticBlock> = (0..MISMATCH_RUN)
        .map(|_| (discriminating.clone(), discriminating_parity.clone()))
        .collect();
    blocks.push((degenerate, degenerate_parity));

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("nine-blocks.sst");
    write_block_archive(&path, &[("data", blocks)])?;

    let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let toc = sfa_reader.toc();
    let cap = block_data_length_cap(0);

    let data = toc
        .section(b"data")
        .expect("the fixture has a data section");
    let (start, end) = (data.pos(), data.pos() + data.len());

    assert_eq!(
        codec_confirms_region((probe.as_ref(), 0), ScrubEcc::Scheme(real), start, end, cap),
        CodecVerdict::Confirmed,
        "the real codec reproduces every trailer",
    );
    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(impostor),
            start,
            end,
            cap
        ),
        CodecVerdict::Confirmed,
        "the ninth block's trailer IS reproduced, and a scan that gave up over \
         the eight before it would never reach the evidence",
    );
    Ok(())
}

/// Framing carries the verdict, and it only carries it when EVERY judged region
/// framed. A split verdict says the descriptor sizes one region's blocks and
/// not another's, and a length that is wrong ANYWHERE is not a length the walk
/// can advance on: it consumes the wrong trailer across that region and reports
/// corruption that is not there. So a split fails safe to unrecognized.
#[cfg(feature = "page_ecc")]
#[test]
fn arbitrate_by_framing_rejects_a_split_verdict() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::{EccParams, Header};

    const LEN: u32 = 4096;
    let real = EccParams::RS_4_2;

    let payload = discriminating_payload(LEN);
    let parity = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes");
    // The second region's trailer is the WRONG length for this scheme, so its
    // frames cannot tile the section.
    let other = discriminating_payload(LEN / 2);
    let short_parity = vec![0u8; 8];

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("split.sst");
    write_block_archive(
        &path,
        &[
            ("data", vec![(payload, parity)]),
            ("tli", vec![(other, short_parity)]),
        ],
    )?;

    // Break the data block's payload checksum WITHOUT touching its framing, so
    // the region still frames but offers no clean block to judge the codec on.
    let data_pos = {
        let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
        let reader = crate::sfa::Reader::from_reader(&mut probe)?;
        reader
            .toc()
            .section(b"data")
            .expect("the fixture has a data section")
            .pos()
    };
    {
        let mut f = StdFs.open(&path, &FsOpenOptions::new().write(true))?;
        // 0xFF differs from this payload's first byte (the pattern starts at 0),
        // so the write actually changes the checksummed bytes.
        f.seek(SeekFrom::Start(data_pos + Header::MIN_LEN as u64))?;
        f.write_all(&[0xFF])?;
    }

    let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let toc = sfa_reader.toc();
    let cap = block_data_length_cap(0);

    let data = toc
        .section(b"data")
        .expect("the fixture has a data section");
    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(real),
            data.pos(),
            data.pos() + data.len(),
            cap,
        ),
        CodecVerdict::NoEvidence,
        "no clean block, so the codec has nothing to say here either",
    );
    assert_eq!(
        arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
        Some(false),
        "one region framed and the other did not — accepting here walks a \
         region with the wrong trailer length and reports corruption that is \
         not there",
    );
    Ok(())
}

/// A single agreement silences the report even when it proves nothing: an
/// impostor reproduces the all-zero parity of a uniform region, so such a
/// region "confirms" any codec. The claim the warning makes is "no trailer
/// anywhere is reproduced, which is what a mis-identified scheme looks like",
/// and one reproduced trailer, informative or not, is enough to make that claim
/// false. A mixed picture stays unreported rather than being narrated as a
/// scheme problem the operator would then chase.
#[cfg(feature = "page_ecc")]
#[test]
fn codec_disagrees_everywhere_is_silenced_by_any_agreement() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::EccParams;

    const LEN: u32 = 4096;
    let real = EccParams::RS_4_2;
    let impostor = EccParams::try_new(2, 1)?;

    // Healthy region: only the real codec reproduces this trailer.
    let discriminating = discriminating_payload(LEN);
    let discriminating_parity =
        crate::ecc::encode_parity(&discriminating, 4, 2).expect("RS(4,2) encodes");
    // Degenerate region: identical shards, so both codecs emit the same parity.
    let degenerate = vec![0xABu8; LEN as usize];
    let degenerate_parity = crate::ecc::encode_parity(&degenerate, 4, 2).expect("RS(4,2) encodes");
    assert!(
        degenerate_parity.iter().all(|b| *b == 0),
        "the degenerate region's trailer must be all-zero, which is what makes \
         its agreement uninformative",
    );

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("degenerate-vs-healthy.sst");
    write_block_archive(
        &path,
        &[
            ("data", vec![(discriminating, discriminating_parity)]),
            ("tli", vec![(degenerate, degenerate_parity)]),
        ],
    )?;

    let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let toc = sfa_reader.toc();
    let cap = block_data_length_cap(0);

    let tli = toc.section(b"tli").expect("the fixture has a tli section");
    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(impostor),
            tli.pos(),
            tli.pos() + tli.len(),
            cap,
        ),
        CodecVerdict::Confirmed,
        "the impostor agrees with the all-zero region — the premise of this test",
    );
    assert!(
        !codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(impostor), 0, cap),
        "one region disagreed and another agreed, so the report's claim does \
         not hold and it stays silent",
    );
    assert_eq!(
        arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
        Some(true),
        "the real scheme sizes both regions",
    );
    Ok(())
}

/// The index TAIL mirror is its own region, and it can be the ONLY one able to
/// speak. The writer emits it so a damaged index copy cannot take the other
/// down, and it is written under the same codec — so when neither the data
/// region nor the head offers a checksum-clean block, the tail is the only
/// place the codec question can be answered at all.
#[cfg(feature = "page_ecc")]
#[test]
fn codec_disagrees_everywhere_consults_the_tli_tail_mirror() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::{EccParams, Header};

    const LEN: u32 = 4096;
    let real = EccParams::RS_4_2;

    let data_payload = discriminating_payload(LEN);
    let data_parity = crate::ecc::encode_parity(&data_payload, 4, 2).expect("RS(4,2) encodes");
    // Head index copy: its payload checksum is broken below, so it offers no
    // evidence either way.
    let head = discriminating_payload(LEN / 2);
    let head_parity = crate::ecc::encode_parity(&head, 4, 2).expect("RS(4,2) encodes");
    // Tail mirror: payload intact, trailer rotted — the one region left with
    // something to say.
    let tail = discriminating_payload(LEN / 2);
    let mut tail_parity = crate::ecc::encode_parity(&tail, 4, 2).expect("RS(4,2) encodes");
    *tail_parity.first_mut().expect("the trailer is non-empty") ^= 0xFF;

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("tail-mirror.sst");
    write_block_archive(
        &path,
        &[
            ("data", vec![(data_payload, data_parity)]),
            ("tli", vec![(head, head_parity)]),
            ("tli_tail", vec![(tail, tail_parity)]),
        ],
    )?;

    // Break the DATA and HEAD payload checksums so neither offers evidence,
    // leaving the tail mirror as the only region that can speak.
    let (data_pos, tli_pos) = {
        let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
        let reader = crate::sfa::Reader::from_reader(&mut probe)?;
        let toc = reader.toc();
        (
            toc.section(b"data")
                .expect("the fixture has a data section")
                .pos(),
            toc.section(b"tli")
                .expect("the fixture has a tli section")
                .pos(),
        )
    };
    {
        let mut f = StdFs.open(&path, &FsOpenOptions::new().write(true))?;
        f.seek(SeekFrom::Start(data_pos + Header::MIN_LEN as u64))?;
        f.write_all(&[0xFF])?;
        f.seek(SeekFrom::Start(tli_pos + Header::MIN_LEN as u64))?;
        f.write_all(&[0xFF])?;
    }

    let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let toc = sfa_reader.toc();
    let cap = block_data_length_cap(0);

    let tli = toc.section(b"tli").expect("the fixture has a tli section");
    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(real),
            tli.pos(),
            tli.pos() + tli.len(),
            cap,
        ),
        CodecVerdict::NoEvidence,
        "the head offers nothing to judge on — the premise of this test",
    );
    assert_eq!(
        arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
        Some(true),
        "every region frames, so the descriptor stands whatever the trailers say",
    );
    assert!(
        codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(real), 0, cap),
        "only the tail mirror had anything to say, so a region set omitting it \
         would report nothing at all",
    );
    Ok(())
}

/// The descriptor sizes EVERY block-format section, not the three a hand-kept
/// list would have named. `range_tombstones` is written under the same
/// descriptor and sits outside the mirrors, so a region set that omits it
/// leaves the operator without the one hint that explains this table's
/// mismatches.
#[cfg(feature = "page_ecc")]
#[test]
fn codec_disagrees_everywhere_consults_sections_outside_the_mirrors() -> crate::Result<()> {
    use crate::fs::{Fs, FsOpenOptions, StdFs};
    use crate::table::block::{EccParams, Header};

    const LEN: u32 = 4096;
    let impostor = EccParams::try_new(2, 1)?;

    // The data region's payload checksum is broken below, so it says nothing
    // and cannot silence the report before the section under test is reached.
    let silent = discriminating_payload(LEN);
    let silent_parity = crate::ecc::encode_parity(&silent, 4, 2).expect("RS(4,2) encodes");

    // `range_tombstones` discriminates: only the real codec matches it.
    let discriminating = discriminating_payload(LEN);
    let discriminating_parity =
        crate::ecc::encode_parity(&discriminating, 4, 2).expect("RS(4,2) encodes");

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("outside-mirrors.sst");
    write_block_archive(
        &path,
        &[
            ("data", vec![(silent, silent_parity)]),
            (
                "range_tombstones",
                vec![(discriminating, discriminating_parity)],
            ),
        ],
    )?;

    let data_pos = {
        let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
        let reader = crate::sfa::Reader::from_reader(&mut probe)?;
        reader
            .toc()
            .section(b"data")
            .expect("the fixture has a data section")
            .pos()
    };
    {
        let mut f = StdFs.open(&path, &FsOpenOptions::new().write(true))?;
        f.seek(SeekFrom::Start(data_pos + Header::MIN_LEN as u64))?;
        f.write_all(&[0xFF])?;
    }

    let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
    let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
    let toc = sfa_reader.toc();
    let cap = block_data_length_cap(0);

    let rt = toc
        .section(b"range_tombstones")
        .expect("the fixture has a range_tombstones section");
    let data = toc
        .section(b"data")
        .expect("the fixture has a data section");
    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(impostor),
            rt.pos(),
            rt.pos() + rt.len(),
            cap,
        ),
        CodecVerdict::Rejected,
        "the section outside the mirrors is the only one that disagrees — the \
         premise of this test",
    );
    assert_eq!(
        codec_confirms_region(
            (probe.as_ref(), 0),
            ScrubEcc::Scheme(impostor),
            data.pos(),
            data.pos() + data.len(),
            cap,
        ),
        CodecVerdict::NoEvidence,
        "and the data region says nothing, so it cannot silence the report",
    );
    assert!(
        codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(impostor), 0, cap),
        "a section the descriptor sizes is consulted wherever it sits",
    );
    Ok(())
}

/// A healthy columnar SST verifies clean, and every block in its data section
/// is walked.
///
/// The block walk checks each block's role against the roles its section is
/// allowed to hold, and a columnar row group is not one block: it is a page
/// directory followed by its pages, all in the data section. A walk that
/// expected one role per row group would report every healthy columnar SST as
/// carrying misrouted blocks — and one that skipped them would verify a data
/// section it never looked at. Both are asserted against here: no finding,
/// and more blocks walked than the table has row groups.
#[cfg(feature = "columnar")]
#[test]
fn verify_sst_file_walks_a_healthy_columnar_sst_clean() -> crate::Result<()> {
    use crate::table::Writer;

    let dir = tempfile::tempdir()?;
    let sst = dir.path().join("table");
    let mut writer = Writer::new(sst.clone(), 0, 0, alloc::sync::Arc::new(crate::fs::StdFs))?
        .use_columnar(true)
        .use_zone_map(true)
        .use_row_group_size(256);
    for i in 0u32..200 {
        writer.write(crate::InternalValue::from_components(
            format!("key{i:05}").into_bytes(),
            format!("val{i:05}").into_bytes(),
            1,
            crate::ValueType::Value,
        ))?;
    }
    assert!(writer.finish()?.is_some(), "the SST is non-empty");

    let report = verify_sst_file(&sst);
    assert!(
        report.is_ok(),
        "a healthy columnar SST must verify clean: {:?}",
        report.errors,
    );
    // 256-byte row groups over 200 rows give several groups; each is at least
    // a directory and four pages, so a walk that visited only the entries
    // (one block per group) would report fewer blocks than this.
    let groups = {
        let checksum = crate::Checksum::from_raw(crate::repair::compute_table_checksum(
            &crate::fs::StdFs,
            &sst,
        )?);
        let mut params = crate::table::RecoverParams::new(
            sst,
            checksum,
            0,
            alloc::sync::Arc::new(crate::fs::StdFs),
            crate::comparator::default_comparator(),
            alloc::sync::Arc::new(crate::cache::Cache::with_capacity_bytes(1 << 20)),
        );
        params.descriptor_table = Some(alloc::sync::Arc::new(
            crate::descriptor_table::DescriptorTable::new(4),
        ));
        crate::table::Table::recover(params)?
            .data_block_handles()
            .filter_map(Result::ok)
            .count()
    };
    assert!(groups > 1, "the fixture must span several row groups");
    assert!(
        report.blocks_scanned >= groups * 5,
        "every directory and page must be walked: {} blocks for {groups} groups",
        report.blocks_scanned,
    );
    Ok(())
}

/// Four SSTs of incompressible 1 KiB values, about 1.2 MiB on disk in all.
fn populate_rated_fixture(dir: &std::path::Path) {
    let cfg = Config::new(
        dir,
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
    let tree = cfg.open().unwrap();
    let mut seqno = 1u64;
    for b in 0..4u32 {
        for i in 0..300u32 {
            let key = format!("b{b:03}k{i:08}");
            let value: Vec<u8> = (0..1024u32)
                .map(|j| {
                    let [low, ..] = (j.wrapping_mul(2_654_435_761) ^ i ^ (b << 16)).to_le_bytes();
                    low
                })
                .collect();
            tree.insert(key.as_bytes(), value, seqno);
            seqno += 1;
        }
        tree.flush_active_memtable(seqno).unwrap();
        seqno += 1;
    }
}

/// The on-disk bytes of the tree's tables.
fn table_bytes(tree: &impl AbstractTree) -> u64 {
    tree.current_version()
        .iter_tables()
        .map(|t| std::fs::metadata(&*t.path).unwrap().len())
        .sum()
}

/// A rated scan of `bytes` takes what the limiter owes past its one-second
/// burst: `(bytes - rate) / rate`, give or take the few bytes per table the
/// scan reads without charging (trailer and TOC) and scheduling slack.
fn assert_rated_scan(parallelism: usize) {
    let dir = tempfile::tempdir().unwrap();
    populate_rated_fixture(dir.path());
    let tree = reopen_tree(dir.path());
    let bytes = table_bytes(&tree);
    let rate = 400 * 1024u64;
    assert!(
        bytes > 2 * rate,
        "the fixture must outlast the burst: {bytes} bytes"
    );

    let opts = VerifyOptions::default()
        .parallelism(parallelism)
        .max_bytes_per_sec(rate);
    let start = std::time::Instant::now();
    let report = tree.verify_checksum_with(&opts);
    let elapsed = start.elapsed();

    assert!(report.is_ok(), "a rated scan verifies the same: {report:?}");
    assert_eq!(report.sst_files_scanned, 4);
    let owed = |bytes: u64| std::time::Duration::from_millis((bytes - rate) * 1000 / rate);
    // Trailer and TOC are read uncharged: well under 1 KiB per table, and a
    // twentieth off for the clock.
    let floor = owed(bytes - 4 * 1024) * 19 / 20;
    let ceiling = owed(bytes) * 3 / 2 + std::time::Duration::from_secs(1);
    assert!(
        elapsed >= floor,
        "{parallelism} worker(s): {bytes} bytes at {rate} B/s took {elapsed:?}, \
         under the {floor:?} the limiter owes",
    );
    assert!(
        elapsed <= ceiling,
        "{parallelism} worker(s): {bytes} bytes at {rate} B/s took {elapsed:?}, \
         past {ceiling:?}",
    );
}

/// One worker reads at the configured rate.
#[test]
fn verify_at_a_byte_rate_takes_its_bytes_over_the_rate_with_one_worker() {
    assert_rated_scan(1);
}

/// Four workers draw on the one limiter, so together they read at the rate,
/// not four times it.
#[test]
fn verify_at_a_byte_rate_takes_its_bytes_over_the_rate_with_four_workers() {
    assert_rated_scan(4);
}

/// Without a limiter, or with a rate of zero, the scan does not wait.
#[test]
fn verify_without_a_byte_rate_reads_at_full_speed() {
    let dir = tempfile::tempdir().unwrap();
    populate_rated_fixture(dir.path());
    let tree = reopen_tree(dir.path());
    for opts in [
        VerifyOptions::default(),
        VerifyOptions::default().max_bytes_per_sec(0),
    ] {
        let start = std::time::Instant::now();
        let report = tree.verify_checksum_with(&opts);
        let elapsed = start.elapsed();
        assert!(report.is_ok(), "{report:?}");
        assert!(
            elapsed < std::time::Duration::from_millis(900),
            "an unrated scan of {} bytes took {elapsed:?}",
            table_bytes(&tree),
        );
    }
}

/// A limit changes when blocks are read, not what is found in them.
#[test]
fn verify_at_a_byte_rate_still_reports_a_corrupt_block() {
    use crate::table::block::Header;
    let dir = tempfile::tempdir().unwrap();
    populate_rated_fixture(dir.path());
    let sst_path = pick_first_sst_path(dir.path());
    let flip_offset = Header::MIN_LEN as u64 + 8;
    {
        let mut f = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&sst_path)
            .unwrap();
        f.seek(SeekFrom::Start(flip_offset)).unwrap();
        let mut byte = [0u8; 1];
        f.read_exact(&mut byte).unwrap();
        byte[0] ^= 0xFF;
        f.seek(SeekFrom::Start(flip_offset)).unwrap();
        f.write_all(&byte).unwrap();
        f.sync_all().unwrap();
    }
    let tree = reopen_tree(dir.path());
    let rate = table_bytes(&tree);
    let report = tree.verify_checksum_with(&VerifyOptions::default().max_bytes_per_sec(rate));
    assert!(
        report
            .errors
            .iter()
            .any(|e| matches!(e, BlockVerifyError::DataCorrupted { .. })),
        "the flipped payload byte is reported under a limit: {report:?}",
    );
}