coordinode-lsm-tree 5.8.6

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 mut buf = Vec::new();
    Block::write_into_with_flags(
        &mut buf,
        &payload,
        id,
        &BlockTransform::PLAIN,
        block_flags::KV_CHECKSUM_FOOTER,
    )
    .unwrap();

    // Block loads fine (block-level checksum matches the corrupted bytes).
    let block = Block::from_reader(&mut &buf[..], id, &BlockTransform::PLAIN).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 mut buf = Vec::new();
    Block::write_into_with_flags(
        &mut buf,
        &payload,
        id,
        &BlockTransform::PLAIN,
        block_flags::KV_CHECKSUM_FOOTER,
    )
    .unwrap();
    let block = Block::from_reader(&mut &buf[..], id, &BlockTransform::PLAIN).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.
        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, 0, None, false, 0)?;
    // 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(())
}

/// 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 header = Header {
        checksum: Checksum::from_raw(crate::hash::hash128(&data)),
        data_length: DATA_LENGTH,
        uncompressed_length: DATA_LENGTH,
        ..Header::test_dummy(BlockType::Data)
    };

    // 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 scan = scan_sst_blocks(&fs, path, table_id, 0, Some(EccParams::RS_4_2), false, 0)?;
    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 header = Header {
        checksum: Checksum::from_raw(crate::hash::hash128(&data)),
        data_length: DATA_LENGTH,
        uncompressed_length: DATA_LENGTH,
        ..Header::test_dummy(BlockType::Data)
    };

    // 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 scan = scan_sst_blocks(&fs, path, table_id, 0, Some(params), false, 0)?;
    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 {
        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, 0, None, false, 0)?;
    // 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 payload length no longer
    // matches `4 + count * 32`.
    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:?}",
    );
}

#[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 header = Header {
        checksum: Checksum::from_raw(crate::hash::hash128(&payload)),
        data_length: DATA_LENGTH,
        uncompressed_length: DATA_LENGTH,
        ..Header::test_dummy(BlockType::Data)
    };

    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(), toc, ScrubEcc::Scheme(real), 0, cap),
        "the real scheme reproduces the trailer, so nothing is suspect",
    );
    assert!(
        codec_disagrees_everywhere(probe.as_ref(), 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 header = Header {
                    checksum: Checksum::from_raw(crate::hash::hash128(payload)),
                    data_length: payload.len() as u32,
                    uncompressed_length: payload.len() as u32,
                    ..Header::test_dummy(BlockType::Data)
                };
                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(), ScrubEcc::Scheme(real), start, end, cap),
        CodecVerdict::Confirmed,
        "the real codec reproduces both trailers",
    );
    assert_eq!(
        codec_confirms_region(probe.as_ref(), 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(), 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(), 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(), 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(),
            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(), 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(),
            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(),
            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(), 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(),
            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(), 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(),
            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(), 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(), ScrubEcc::Scheme(real), start, end, cap),
        CodecVerdict::Confirmed,
        "the real codec reproduces every trailer",
    );
    assert_eq!(
        codec_confirms_region(probe.as_ref(), 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(),
            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(),
            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(), 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(),
            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(), 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(),
            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(),
            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(), toc, ScrubEcc::Scheme(impostor), 0, cap),
        "a section the descriptor sizes is consulted wherever it sits",
    );
    Ok(())
}