coordinode-lsm-tree 5.8.6

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024-present, fjall-rs
// Copyright (c) 2026-present, Dmitry Prudnikov

#![allow(
    clippy::doc_markdown,
    clippy::default_trait_access,
    reason = "test code"
)]
#![expect(clippy::expect_used, reason = "test code")]

use super::*;
use crate::{config::BloomConstructionPolicy, fs::StdFs, hash::hash64};
use tempfile::tempdir;
use test_log::test;

/// Recover params for the common test shape: table id 0, `StdFs`, the default
/// comparator, a small cache, and a pooled descriptor table. Tests tweak the
/// returned params for whatever they exercise (pinning, encryption, a custom
/// fs or dictionary).
fn test_recover_params(file_path: PathBuf, checksum: Checksum) -> RecoverParams {
    let mut params = RecoverParams::new(
        file_path,
        checksum,
        0,
        Arc::new(StdFs),
        crate::comparator::default_comparator(),
        Arc::new(Cache::with_capacity_bytes(1_000_000)),
    );
    params.descriptor_table = Some(Arc::new(DescriptorTable::new(10)));
    params
}

fn test_with_table(
    items: &[InternalValue],
    f: impl Fn(Table) -> crate::Result<()>,
    rotate_every: Option<usize>,
    config_writer: Option<impl Fn(Writer) -> Writer>,
) -> crate::Result<()> {
    test_with_table_impl(
        items,
        f,
        rotate_every,
        config_writer,
        #[cfg(zstd_any)]
        None,
    )
}

#[expect(
    clippy::too_many_lines,
    clippy::cognitive_complexity,
    clippy::cast_possible_truncation,
    clippy::unwrap_used
)]
fn test_with_table_impl(
    items: &[InternalValue],
    f: impl Fn(Table) -> crate::Result<()>,
    rotate_every: Option<usize>,
    config_writer: Option<impl Fn(Writer) -> Writer>,
    #[cfg(zstd_any)] zstd_dictionary: Option<Arc<crate::compression::ZstdDictionary>>,
) -> crate::Result<()> {
    let dir = tempdir()?;
    let file = dir.path().join("table");

    {
        let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;

        #[cfg(zstd_any)]
        if zstd_dictionary.is_some() {
            writer = writer.use_zstd_dictionary(zstd_dictionary.clone());
        }

        if let Some(f) = &config_writer {
            writer = f(writer);
        }

        for (idx, item) in items.iter().enumerate() {
            if let Some(rotate) = rotate_every
                && idx % rotate == 0
            {
                writer.spill_block()?;
            }
            writer.write(item.clone())?;
        }
        let (_, checksum) = writer.finish()?.unwrap();

        {
            #[cfg(feature = "metrics")]
            let metrics = Arc::new(Metrics::default());

            let table = {
                #[cfg_attr(not(zstd_any), expect(unused_mut))]
                let mut params = test_recover_params(file.clone(), checksum);
                #[cfg(zstd_any)]
                {
                    params.zstd_dictionary.clone_from(&zstd_dictionary);
                }
                #[cfg(feature = "metrics")]
                {
                    params.metrics = metrics;
                }
                Table::recover(params)?
            };

            assert_eq!(0, table.id());
            assert_eq!(items.len(), table.metadata.item_count as usize);
            assert!(table.regions.index.is_none(), "should use full index");
            assert_eq!(0, table.pinned_block_index_size(), "should not pin index");
            assert_eq!(0, table.pinned_filter_size(), "should not pin filter");
            assert!(matches!(
                table.file_accessor,
                FileAccessor::DescriptorTable { .. }
            ));

            f(table)?;
        }

        {
            #[cfg(feature = "metrics")]
            let metrics = Arc::new(Metrics::default());

            let table = {
                let mut params = test_recover_params(file.clone(), checksum);
                params.pin_filter = true;
                #[cfg(zstd_any)]
                {
                    params.zstd_dictionary.clone_from(&zstd_dictionary);
                }
                #[cfg(feature = "metrics")]
                {
                    params.metrics = metrics;
                }
                Table::recover(params)?
            };

            assert_eq!(0, table.id());
            assert_eq!(items.len(), table.metadata.item_count as usize);
            assert!(table.regions.index.is_none(), "should use full index");
            assert_eq!(0, table.pinned_block_index_size(), "should not pin index");
            // assert!(table.pinned_filter_size() > 0, "should pin filter");
            assert!(matches!(
                table.file_accessor,
                FileAccessor::DescriptorTable { .. }
            ));

            f(table)?;
        }

        {
            #[cfg(feature = "metrics")]
            let metrics = Arc::new(Metrics::default());

            let table = {
                let mut params = test_recover_params(file.clone(), checksum);
                params.pin_index = true;
                #[cfg(zstd_any)]
                {
                    params.zstd_dictionary.clone_from(&zstd_dictionary);
                }
                #[cfg(feature = "metrics")]
                {
                    params.metrics = metrics;
                }
                Table::recover(params)?
            };

            assert_eq!(0, table.id());
            assert_eq!(items.len(), table.metadata.item_count as usize);
            assert!(table.regions.index.is_none(), "should use full index");
            assert!(table.pinned_block_index_size() > 0, "should pin index");
            assert_eq!(0, table.pinned_filter_size(), "should not pin filter");
            assert!(matches!(
                table.file_accessor,
                FileAccessor::DescriptorTable { .. }
            ));

            f(table)?;
        }

        {
            #[cfg(feature = "metrics")]
            let metrics = Arc::new(Metrics::default());

            let table = {
                let mut params = test_recover_params(file.clone(), checksum);
                params.pin_filter = true;
                params.pin_index = true;
                #[cfg(zstd_any)]
                {
                    params.zstd_dictionary.clone_from(&zstd_dictionary);
                }
                #[cfg(feature = "metrics")]
                {
                    params.metrics = metrics;
                }
                Table::recover(params)?
            };

            assert_eq!(0, table.id());
            assert_eq!(items.len(), table.metadata.item_count as usize);
            assert!(table.regions.index.is_none(), "should use full index");
            assert!(table.pinned_block_index_size() > 0, "should pin index");
            // assert!(table.pinned_filter_size() > 0, "should pin filter");
            assert!(matches!(
                table.file_accessor,
                FileAccessor::DescriptorTable { .. }
            ));

            f(table)?;
        }

        {
            #[cfg(feature = "metrics")]
            let metrics = Arc::new(Metrics::default());

            let table = {
                let mut params = test_recover_params(file.clone(), checksum);
                params.descriptor_table = None;
                params.pin_filter = true;
                params.pin_index = true;
                #[cfg(zstd_any)]
                {
                    params.zstd_dictionary.clone_from(&zstd_dictionary);
                }
                #[cfg(feature = "metrics")]
                {
                    params.metrics = metrics;
                }
                Table::recover(params)?
            };

            assert_eq!(0, table.id());
            assert_eq!(items.len(), table.metadata.item_count as usize);
            assert!(table.regions.index.is_none(), "should use full index");
            assert!(table.pinned_block_index_size() > 0, "should pin index");
            // assert!(table.pinned_filter_size() > 0, "should pin filter");
            assert!(matches!(table.file_accessor, FileAccessor::File(..)));

            f(table)?;
        }
    }

    std::fs::remove_file(&file)?;

    // Test with partitioned indexes
    {
        let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_partitioned_index();

        #[cfg(zstd_any)]
        if zstd_dictionary.is_some() {
            writer = writer.use_zstd_dictionary(zstd_dictionary.clone());
        }

        if let Some(f) = config_writer {
            writer = f(writer);
        }

        for (idx, item) in items.iter().enumerate() {
            if let Some(rotate) = rotate_every
                && idx % rotate == 0
            {
                writer.spill_block()?;
            }
            writer.write(item.clone())?;
        }
        let (_, checksum) = writer.finish()?.unwrap();

        {
            #[cfg(feature = "metrics")]
            let metrics = Arc::new(Metrics::default());

            let table = {
                #[cfg_attr(not(zstd_any), expect(unused_mut))]
                let mut params = test_recover_params(file.clone(), checksum);
                #[cfg(zstd_any)]
                {
                    params.zstd_dictionary.clone_from(&zstd_dictionary);
                }
                #[cfg(feature = "metrics")]
                {
                    params.metrics = metrics;
                }
                Table::recover(params)?
            };

            assert_eq!(0, table.id());
            assert_eq!(items.len(), table.metadata.item_count as usize);
            assert!(table.regions.index.is_some(), "should use two-level index",);
            assert_eq!(0, table.pinned_filter_size(), "should not pin filter");
            assert!(matches!(
                table.file_accessor,
                FileAccessor::DescriptorTable { .. }
            ));

            f(table)?;
        }

        {
            #[cfg(feature = "metrics")]
            let metrics = Arc::new(Metrics::default());

            let table = {
                let mut params = test_recover_params(file.clone(), checksum);
                params.pin_filter = true;
                #[cfg(zstd_any)]
                {
                    params.zstd_dictionary.clone_from(&zstd_dictionary);
                }
                #[cfg(feature = "metrics")]
                {
                    params.metrics = metrics;
                }
                Table::recover(params)?
            };

            assert_eq!(0, table.id());
            assert_eq!(items.len(), table.metadata.item_count as usize);
            assert!(table.regions.index.is_some(), "should use two-level index",);
            // assert!(table.pinned_filter_size() > 0, "should pin filter");
            assert!(matches!(
                table.file_accessor,
                FileAccessor::DescriptorTable { .. }
            ));

            f(table)?;
        }

        {
            #[cfg(feature = "metrics")]
            let metrics = Arc::new(Metrics::default());

            let table = {
                let mut params = test_recover_params(file.clone(), checksum);
                params.pin_index = true;
                #[cfg(zstd_any)]
                {
                    params.zstd_dictionary.clone_from(&zstd_dictionary);
                }
                #[cfg(feature = "metrics")]
                {
                    params.metrics = metrics;
                }
                Table::recover(params)?
            };

            assert_eq!(0, table.id());
            assert_eq!(items.len(), table.metadata.item_count as usize);
            assert!(table.regions.index.is_some(), "should use two-level index",);
            assert!(table.pinned_block_index_size() > 0, "should pin index");
            // assert_eq!(0, table.pinned_filter_size(), "should not pin filter");
            assert!(matches!(
                table.file_accessor,
                FileAccessor::DescriptorTable { .. }
            ));

            f(table)?;
        }

        {
            #[cfg(feature = "metrics")]
            let metrics = Arc::new(Metrics::default());

            let table = {
                let mut params = test_recover_params(file.clone(), checksum);
                params.pin_filter = true;
                params.pin_index = true;
                #[cfg(zstd_any)]
                {
                    params.zstd_dictionary.clone_from(&zstd_dictionary);
                }
                #[cfg(feature = "metrics")]
                {
                    params.metrics = metrics;
                }
                Table::recover(params)?
            };

            assert_eq!(0, table.id());
            assert_eq!(items.len(), table.metadata.item_count as usize);
            assert!(table.regions.index.is_some(), "should use two-level index",);
            assert!(table.pinned_block_index_size() > 0, "should pin index");
            // assert!(table.pinned_filter_size() > 0, "should pin filter");
            assert!(matches!(
                table.file_accessor,
                FileAccessor::DescriptorTable { .. }
            ));

            f(table)?;
        }

        {
            #[cfg(feature = "metrics")]
            let metrics = Arc::new(Metrics::default());

            let table = {
                let mut params = test_recover_params(file, checksum);
                params.descriptor_table = None;
                params.pin_filter = true;
                params.pin_index = true;
                #[cfg(zstd_any)]
                {
                    params.zstd_dictionary = zstd_dictionary;
                }
                #[cfg(feature = "metrics")]
                {
                    params.metrics = metrics;
                }
                Table::recover(params)?
            };

            assert_eq!(0, table.id());
            assert_eq!(items.len(), table.metadata.item_count as usize);
            assert!(table.regions.index.is_some(), "should use two-level index",);
            assert!(table.pinned_block_index_size() > 0, "should pin index");
            // assert!(table.pinned_filter_size() > 0, "should pin filter");
            assert!(matches!(table.file_accessor, FileAccessor::File(..)));

            f(table)?;
        }
    }

    Ok(())
}

#[cfg(feature = "zstd")]
fn test_with_table_and_zstd_dictionary(
    items: &[InternalValue],
    f: impl Fn(Table) -> crate::Result<()>,
    rotate_every: Option<usize>,
    config_writer: Option<impl Fn(Writer) -> Writer>,
    zstd_dictionary: Arc<crate::compression::ZstdDictionary>,
) -> crate::Result<()> {
    test_with_table_impl(items, f, rotate_every, config_writer, Some(zstd_dictionary))
}

#[cfg(feature = "zstd")]
fn make_test_dictionary() -> crate::compression::ZstdDictionary {
    let mut samples = Vec::new();
    for i in 0u32..500 {
        let key = format!("key-{i:05}");
        let val = format!("value-{i:05}-padding-to-make-it-longer");
        samples.extend_from_slice(key.as_bytes());
        samples.extend_from_slice(val.as_bytes());
    }
    crate::compression::ZstdDictionary::new(&samples)
}

/// A table with large data blocks compressed at a high zstd level splits each
/// block into several inner zstd blocks. The writer must persist their
/// cumulative decompressed-end layout in the `block_layout` section, and the
/// reader must reload it on open. A default small-block table must NOT carry
/// the section. This is the write → persist → reload contract for range-query
/// partial decode.
#[cfg(feature = "zstd")]
#[test]
#[expect(clippy::unwrap_used)]
fn block_layout_section_roundtrips_for_large_zstd_blocks() {
    use crate::cache::Cache;
    use crate::fs::StdFs;
    #[cfg(feature = "metrics")]
    use crate::metrics::Metrics;
    use crate::table::Writer;

    // 256 KiB blocks at L19 split into many inner zstd blocks (the cold-tier
    // shape); ~600 KiB of sorted KV yields at least one full multi-inner-block
    // data block.
    let items: Vec<crate::InternalValue> = (0u64..20_000)
        .map(|i| {
            crate::InternalValue::from_components(
                format!("key-{i:012}").into_bytes(),
                format!("value-{i:08}-payload").into_bytes(),
                1,
                crate::ValueType::Value,
            )
        })
        .collect();

    let dir = tempdir().unwrap();
    let file = dir.path().join("table");

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))
        .unwrap()
        .use_data_block_size(256 * 1024)
        .use_data_block_compression(crate::CompressionType::Zstd(19));
    for item in &items {
        writer.write(item.clone()).unwrap();
    }
    let (_, checksum) = writer.finish().unwrap().unwrap();

    #[cfg(feature = "metrics")]
    let metrics = Arc::new(Metrics::default());
    let table = {
        let mut params = test_recover_params(file, checksum);
        params.cache = Arc::new(Cache::with_capacity_bytes(4_000_000));
        #[cfg(feature = "metrics")]
        {
            params.metrics = metrics;
        }
        Table::recover(params).unwrap()
    };

    assert!(
        table.regions.block_layout.is_some(),
        "large multi-inner-block table must carry a block_layout section",
    );
    assert!(
        !table.block_layout.is_empty(),
        "at least one data block must have a recorded inner-block layout",
    );
    // Every recorded entry must have strictly increasing cumulative ends whose
    // last value is the block's uncompressed length (a non-trivial split).
    for offset in table.block_layout.offsets() {
        let ends = table
            .block_layout
            .ends_for(offset)
            .expect("offsets() entries must resolve via ends_for");
        assert!(
            ends.len() >= 2,
            "recorded block must have >= 2 inner blocks"
        );
        assert!(
            ends.windows(2).all(|w| w[0] < w[1]),
            "cumulative ends must be strictly increasing: {ends:?}",
        );
    }

    // Negative control: a default small-block (4 KiB) zstd table must NOT carry
    // the section — each tiny block compresses into a single inner zstd block,
    // so there is nothing to partial-decode and no layout is persisted.
    let small_file = dir.path().join("table-small");
    let mut small_writer = Writer::new(small_file.clone(), 0, 0, Arc::new(StdFs))
        .unwrap()
        .use_data_block_size(4 * 1024)
        .use_data_block_compression(crate::CompressionType::Zstd(19));
    for item in &items {
        small_writer.write(item.clone()).unwrap();
    }
    let (_, small_checksum) = small_writer.finish().unwrap().unwrap();

    #[cfg(feature = "metrics")]
    let small_metrics = Arc::new(Metrics::default());
    let small_table = {
        let mut params = test_recover_params(small_file, small_checksum);
        params.cache = Arc::new(Cache::with_capacity_bytes(4_000_000));
        #[cfg(feature = "metrics")]
        {
            params.metrics = small_metrics;
        }
        Table::recover(params).unwrap()
    };

    assert!(
        small_table.regions.block_layout.is_none(),
        "default small-block table must NOT carry a block_layout section",
    );
    assert_eq!(
        small_table.block_layout.len(),
        0,
        "small-block table's layout map must be empty",
    );
}

#[test]
#[expect(clippy::unwrap_used)]
fn table_point_read() -> crate::Result<()> {
    let items = [crate::InternalValue::from_components(
        b"abc",
        b"asdasdasd",
        3,
        crate::ValueType::Value,
    )];

    test_with_table(
        &items,
        |table| {
            assert_eq!(
                b"abc",
                &*table
                    .get(b"abc", SeqNo::MAX, hash64(b"abc"))?
                    .unwrap()
                    .key
                    .user_key,
            );
            assert_eq!(None, table.get(b"def", SeqNo::MAX, hash64(b"def"))?,);
            assert_eq!(None, table.get(b"____", SeqNo::MAX, hash64(b"____"))?,);

            assert_eq!(
                table.metadata.key_range,
                crate::KeyRange::new((b"abc".into(), b"abc".into())),
            );

            Ok(())
        },
        None,
        Some(|x| x),
    )
}

#[test]
#[expect(clippy::unwrap_used)]
fn restricted_view_clamps_point_and_range_reads() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"c", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"d", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"e", b"v", 0, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            // Restrict the view to keys >= "c" (the prefix a, b is punched out
            // and superseded by a merged output table in tight-space reclaim).
            let restricted = table.with_restriction(crate::UserKey::from(&b"c"[..]));

            // Point reads below the bound miss (so the read falls through to the
            // superseding output); at/above the bound they hit.
            assert_eq!(None, restricted.get(b"a", SeqNo::MAX, hash64(b"a"))?);
            assert_eq!(None, restricted.get(b"b", SeqNo::MAX, hash64(b"b"))?);
            assert!(restricted.get(b"c", SeqNo::MAX, hash64(b"c"))?.is_some());
            assert!(restricted.get(b"d", SeqNo::MAX, hash64(b"d"))?.is_some());

            // The unrestricted view of the same physical SST is unaffected.
            assert!(table.get(b"a", SeqNo::MAX, hash64(b"a"))?.is_some());

            // A full scan yields only keys >= the bound, in order — the iterator
            // never walks into the punched prefix.
            let keys: Vec<_> = restricted
                .range(..)
                .map(|r| r.unwrap().key.user_key)
                .collect();
            assert_eq!(
                keys,
                vec![
                    crate::UserKey::from(&b"c"[..]),
                    crate::UserKey::from(&b"d"[..]),
                    crate::UserKey::from(&b"e"[..]),
                ],
            );

            let cmp = crate::comparator::default_comparator();
            // A query entirely below the bound does not overlap the live range.
            assert!(!restricted.check_key_range_overlap_cmp(
                &(
                    core::ops::Bound::Unbounded,
                    core::ops::Bound::Excluded(&b"c"[..]),
                ),
                cmp.as_ref(),
            ));
            // A query reaching into [bound, hi] does overlap.
            assert!(restricted.check_key_range_overlap_cmp(
                &(
                    core::ops::Bound::Included(&b"d"[..]),
                    core::ops::Bound::Unbounded,
                ),
                cmp.as_ref(),
            ));

            Ok(())
        },
        None,
        Some(|x| x),
    )
}

#[test]
#[expect(clippy::unwrap_used)]
fn reopen_restricted_yields_a_distinct_clamped_view() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"c", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"d", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"e", b"v", 0, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            // Re-open as a distinct Inner over the same file, clamped to >= "c".
            let restricted = table.reopen_restricted(crate::UserKey::from(&b"c"[..]))?;

            assert_eq!(None, restricted.get(b"a", SeqNo::MAX, hash64(b"a"))?);
            assert_eq!(None, restricted.get(b"b", SeqNo::MAX, hash64(b"b"))?);
            assert!(restricted.get(b"c", SeqNo::MAX, hash64(b"c"))?.is_some());
            assert!(restricted.get(b"e", SeqNo::MAX, hash64(b"e"))?.is_some());

            // The original view of the same file is unaffected.
            assert!(table.get(b"a", SeqNo::MAX, hash64(b"a"))?.is_some());

            // A full scan of the re-opened view yields only keys >= the bound.
            let keys: Vec<_> = restricted
                .range(..)
                .map(|r| r.unwrap().key.user_key)
                .collect();
            assert_eq!(
                keys,
                vec![
                    crate::UserKey::from(&b"c"[..]),
                    crate::UserKey::from(&b"d"[..]),
                    crate::UserKey::from(&b"e"[..]),
                ],
            );

            Ok(())
        },
        None,
        Some(|x| x),
    )
}

/// A LEGACY table (written before `descriptor#delete_bitmap_hash` existed)
/// carrying a delete bitmap must still reconcile on the directly attributable
/// heal path: there the pre-heal digest matched the manifest, authenticating
/// the bitmap bytes, so the missing hash proves nothing. Rejecting it strips
/// the heal attestation and strands the healed table under a stale digest
/// forever. Repair (no matching digest) must keep failing closed on the same
/// file.
///
/// Columnar-gated like the delete-bitmap authentication gate itself: a
/// non-columnar build has no positional-delete masking, so the gate (and
/// this fixture's bitmap) does not exist there.
#[test]
#[cfg(feature = "columnar")]
fn metadata_bounds_accept_a_legacy_bitmap_when_digest_authenticated() -> crate::Result<()> {
    use crate::config::DeleteStrategy;

    let dir = tempdir()?;
    let file = dir.path().join("legacy");
    let fs: Arc<dyn Fs> = Arc::new(StdFs);

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::clone(&fs))?
        .use_columnar(true)
        .use_zone_map(true)
        .delete_strategy(DeleteStrategy::MergeOnRead);
    writer.omit_delete_bitmap_hash_for_test = true;
    for i in 0..16u32 {
        writer.write(crate::InternalValue::from_components(
            format!("key{i:03}").into_bytes(),
            b"v".as_slice(),
            u64::from(i) + 1,
            crate::ValueType::Value,
        ))?;
    }
    writer.delete_bitmap_mut().insert(3);
    assert!(writer.finish()?.is_some(), "legacy SST is non-empty");

    let table = {
        let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
        params.fs = Arc::clone(&fs);
        Table::recover(params)?
    };
    assert!(
        matches!(
            table.verify_reconcile_gates(None, false),
            Err((crate::table::ReconcileGate::MetadataBounds, _))
        ),
        "repair (no matching digest) keeps failing closed on the \
         unauthenticatable legacy bitmap",
    );
    if let Err((gate, e)) = table.verify_reconcile_gates(None, true) {
        panic!("an authenticated digest accepts the legacy bitmap, {gate:?} refused it: {e}");
    }
    Ok(())
}

/// A restricted view's compaction scanner must start at the restriction:
/// the punched prefix reads as zeros (a raw scan aborts on the first punched
/// block), and even before the punch runs, the sub-bound rows' authoritative
/// copies live in the superseding slice output — a serial compaction reading
/// them through the restricted input would merge every prefix row twice.
/// The unrestricted view keeps scanning the whole file.
#[test]
#[expect(clippy::unwrap_used)]
fn restricted_view_scan_starts_at_the_bound() -> crate::Result<()> {
    let items: Vec<_> = (0..40u32)
        .map(|i| {
            crate::InternalValue::from_components(
                format!("key{i:03}").into_bytes(),
                b"v".as_slice(),
                0,
                crate::ValueType::Value,
            )
        })
        .collect();
    test_with_table(
        &items,
        |table| {
            let restricted = table.reopen_restricted(crate::UserKey::from(&b"key020"[..]))?;
            let keys: Vec<_> = restricted
                .scan()?
                .map(|r| r.unwrap().key.user_key)
                .collect();
            let expected: Vec<_> = (20..40u32)
                .map(|i| crate::UserKey::from(format!("key{i:03}").into_bytes()))
                .collect();
            assert_eq!(
                keys, expected,
                "the restricted scan must yield only keys at or past the bound",
            );

            let full = table.scan()?.count();
            assert_eq!(full, 40, "the unrestricted view scans the whole file");
            Ok(())
        },
        None,
        // Tiny blocks so the table spans several data blocks and the bound
        // lands mid-file (skipped whole blocks + a straddling block).
        Some(|w: Writer| w.use_data_block_size(64)),
    )
}

/// A restricted view exposes range tombstones CLAMPED to the live suffix:
/// a tombstone wholly below the bound is the punched prefix's deletion (the
/// slice output that superseded the prefix carries its clipped copy), and a
/// straddling tombstone starts at the bound. Unclamped, `scan_since` would
/// emit the same deletion twice and cover keys this view no longer owns.
/// The unrestricted view keeps the full list.
#[test]
fn restricted_view_clamps_visible_range_tombstones() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"z", b"v", 0, crate::ValueType::Value),
    ];
    let rt = |s: &[u8], e: &[u8], seqno| {
        crate::range_tombstone::RangeTombstone::new(
            crate::UserKey::from(s),
            crate::UserKey::from(e),
            seqno,
        )
    };
    test_with_table(
        &items,
        |table| {
            let unrestricted: Vec<_> = table.visible_range_tombstones().collect();
            assert_eq!(
                unrestricted.len(),
                3,
                "the unrestricted view keeps the full list: {unrestricted:?}",
            );

            let restricted = table.reopen_restricted(crate::UserKey::from(&b"g"[..]))?;
            let visible: Vec<_> = restricted.visible_range_tombstones().collect();
            assert_eq!(
                visible,
                vec![rt(b"g", b"m", 5), rt(b"p", b"r", 6)],
                "wholly-below dropped, straddling clamped to the bound, \
                 above-bound untouched",
            );
            Ok(())
        },
        None,
        Some(|mut w: Writer| {
            w.write_range_tombstone(rt(b"a", b"c", 4)); // wholly below "g"
            w.write_range_tombstone(rt(b"a", b"m", 5)); // straddles "g"
            w.write_range_tombstone(rt(b"p", b"r", 6)); // above "g"
            w
        }),
    )
}

/// A restricted view must still cross-check its `linked_blob_files` section: the
/// section carries no checksum, so a same-size rot that under-counts (or drops) a
/// blob id the READABLE SUFFIX still references passes the block walk, and blob GC
/// could then retire a file the suffix addresses. The whole-table aggregate can't
/// be matched exactly once the prefix is punched, but every id/count the suffix
/// derives must be COVERED BY the recorded aggregate. Here id 9 (referenced by the
/// suffix key `key00009`) is recorded with a too-small byte total, so both the
/// unrestricted exact check and the restricted containment check must reject it.
#[cfg(feature = "std")]
#[test]
fn verify_blob_links_rejects_an_undercounted_suffix_id_on_a_restricted_view() -> crate::Result<()> {
    use crate::blob_tree::handle::BlobIndirection;

    use crate::coding::Encode;
    use crate::table::Writer;
    use crate::vlog::ValueHandle;
    use crate::{InternalValue, ValueType};

    let dir = tempdir()?;
    let file = dir.path().join("0");

    // Ten indirections (ids 0..10), each 1000 logical / 500 on-disk bytes. The
    // recorded section matches every id EXCEPT id 9, whose byte total is forged
    // small — the bit-flip a checksum-less section cannot otherwise catch.
    let checksum = {
        let mut w = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_data_block_size(128);
        for i in 0u64..10 {
            let value = BlobIndirection {
                size: 1000,
                vhandle: ValueHandle {
                    blob_file_id: i,
                    on_disk_size: 500,
                    offset: 0,
                },
            }
            .encode_into_vec();
            w.write(InternalValue::from_components(
                format!("key{i:05}").into_bytes(),
                value,
                i + 1,
                ValueType::Indirection,
            ))?;
        }
        for i in 0u64..10 {
            let bytes = if i == 9 { 1 } else { 1000 };
            w.link_blob_file(i, 1, bytes, 500);
        }
        w.finish()?.expect("the SST is non-empty").1
    };

    let recover =
        || -> crate::Result<Table> { Table::recover(test_recover_params(file.clone(), checksum)) };

    // Baseline: the unrestricted exact check already rejects the bad section.
    assert!(
        recover()?.verify_blob_links().is_err(),
        "the unrestricted exact check must reject the under-counted id",
    );

    // The restricted view clamped to key00005 still references id 9 in its live
    // suffix, so the containment check must reject the under-count too.
    let restricted = recover()?.reopen_restricted(crate::UserKey::from(&b"key00005"[..]))?;
    assert!(
        restricted.verify_blob_links().is_err(),
        "the restricted containment check must reject a suffix id the section under-counts",
    );
    Ok(())
}

/// An `Fs` that forwards to `MemFs` but reports the CONFIGURED hard-link count
/// for every file, so the punch path's shared-inode guard can be exercised —
/// including a checkpoint's link later disappearing (MemFs copies on
/// `hard_link`, so its real count is always 1, and `StdFs` cannot punch on
/// non-Linux hosts).
#[cfg(feature = "std")]
#[derive(Clone)]
struct SharedInodeFs(crate::fs::MemFs, Arc<core::sync::atomic::AtomicU64>);

#[cfg(feature = "std")]
impl crate::fs::Fs for SharedInodeFs {
    fn open(
        &self,
        path: &std::path::Path,
        options: &crate::fs::FsOpenOptions,
    ) -> crate::io::Result<Box<dyn crate::fs::FsFile>> {
        self.0.open(path, options)
    }
    fn remove_file(&self, path: &std::path::Path) -> crate::io::Result<()> {
        self.0.remove_file(path)
    }
    fn rename(&self, from: &std::path::Path, to: &std::path::Path) -> crate::io::Result<()> {
        self.0.rename(from, to)
    }
    fn create_dir_all(&self, path: &std::path::Path) -> crate::io::Result<()> {
        self.0.create_dir_all(path)
    }
    fn remove_dir_all(&self, path: &std::path::Path) -> crate::io::Result<()> {
        self.0.remove_dir_all(path)
    }
    fn sync_directory(&self, path: &std::path::Path) -> crate::io::Result<()> {
        self.0.sync_directory(path)
    }
    fn read_dir(&self, path: &std::path::Path) -> crate::io::Result<Vec<crate::fs::FsDirEntry>> {
        self.0.read_dir(path)
    }
    fn metadata(&self, path: &std::path::Path) -> crate::io::Result<crate::fs::FsMetadata> {
        self.0.metadata(path)
    }
    fn exists(&self, path: &std::path::Path) -> crate::io::Result<bool> {
        self.0.exists(path)
    }
    fn capabilities(&self, path: &std::path::Path) -> crate::fs::FsCapabilities {
        self.0.capabilities(path)
    }
    fn punch_hole(&self, path: &std::path::Path, offset: u64, len: u64) -> crate::io::Result<()> {
        self.0.punch_hole(path, offset, len)
    }
    /// The whole point: every file reports the configured link count.
    fn hard_link_count(&self, _path: &std::path::Path) -> crate::io::Result<u64> {
        Ok(self.1.load(core::sync::atomic::Ordering::Acquire))
    }
}

/// The tight-space prefix punch must FAIL CLOSED on a shared inode: a completed
/// checkpoint hard-links the SST, so punching the retired view's data blocks
/// would zero the SAME blocks inside the immutable checkpoint, whose manifest
/// still records the unrestricted file and its original digest. The delete path
/// already probes the link count before truncating; the punch must too.
#[cfg(feature = "std")]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn punch_on_drop_refuses_a_hard_linked_table() -> crate::Result<()> {
    use crate::fs::Fs;

    let memfs = crate::fs::MemFs::new();
    let shared: Arc<dyn Fs> = Arc::new(SharedInodeFs(
        memfs.clone(),
        Arc::new(core::sync::atomic::AtomicU64::new(2)),
    ));
    let plain: Arc<dyn Fs> = Arc::new(memfs);
    let root = std::path::absolute("/db")?;
    shared.create_dir_all(&root)?;

    let build = |fs: &Arc<dyn Fs>, name: &str| -> crate::Result<(std::path::PathBuf, Checksum)> {
        let path = root.join(name);
        let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(fs))?.use_data_block_size(256);
        for i in 0..256u32 {
            writer.write(crate::InternalValue::from_components(
                format!("k{i:04}").into_bytes(),
                b"v",
                1,
                crate::ValueType::Value,
            ))?;
        }
        let (_, checksum) = writer.finish()?.expect("table written");
        Ok((path, checksum))
    };
    let recover = |fs: &Arc<dyn Fs>, path: &std::path::Path, checksum| -> crate::Result<Table> {
        #[cfg(feature = "metrics")]
        let metrics = Arc::new(Metrics::default());
        let mut params = test_recover_params(path.to_path_buf(), checksum);
        params.descriptor_table = None;
        params.fs = Arc::clone(fs);
        #[cfg(feature = "metrics")]
        {
            params.metrics = metrics;
        }
        Table::recover(params)
    };
    let read_all = |fs: &Arc<dyn Fs>, path: &std::path::Path| -> crate::Result<Vec<u8>> {
        let file = fs.open(path, &crate::fs::FsOpenOptions::new().read(true))?;
        let len = crate::fs::FsFile::metadata(&*file)?.len;
        Ok(crate::file::read_exact(&*file, 0, usize::try_from(len).unwrap_or(0))?.to_vec())
    };

    // SHARED inode (a checkpoint hard-linked this SST): the punch must not run.
    let (shared_path, shared_checksum) = build(&shared, "0")?;
    let before = read_all(&shared, &shared_path)?;
    let table = recover(&shared, &shared_path, shared_checksum)?;
    let punch = table.punch_offset_for(b"k0128")?;
    assert!(punch > 0, "the fixture has a punchable prefix");
    table.mark_punch_on_drop(punch);
    drop(table);
    assert_eq!(
        read_all(&shared, &shared_path)?,
        before,
        "a hard-linked SST must not be punched: the shared inode is the \
         checkpoint's data too",
    );

    // EXCLUSIVE inode: the punch still reclaims, so the guard did not disable
    // the reclaim path itself.
    let (plain_path, plain_checksum) = build(&plain, "1")?;
    let table = recover(&plain, &plain_path, plain_checksum)?;
    let punch = table.punch_offset_for(b"k0128")?;
    table.mark_punch_on_drop(punch);
    drop(table);
    let after = read_all(&plain, &plain_path)?;
    assert!(
        after
            .get(..64)
            .is_some_and(|head| head.iter().all(|&b| b == 0)),
        "an exclusively-owned SST is still reclaimed",
    );

    // EXCLUSIVE inode but an ACTIVE checkpoint pause: the punch stands down —
    // the pause covers the checkpoint's whole copy/link pass, so deferring
    // removes the probe-then-punch window in which the checkpoint could link
    // the inode after the count read 1. Mirrors the blob-prefix reclaim.
    let (paused_path, paused_checksum) = build(&plain, "2")?;
    let before = read_all(&plain, &paused_path)?;
    let table = recover(&plain, &paused_path, paused_checksum)?;
    let pause = crate::deletion_pause::DeletionPause::new_shared();
    table.install_deletion_pause(std::sync::Arc::clone(&pause));
    let guard = pause.acquire();
    let punch = table.punch_offset_for(b"k0128")?;
    table.mark_punch_on_drop(punch);
    drop(table);
    assert_eq!(
        read_all(&plain, &paused_path)?,
        before,
        "an active checkpoint pause must defer the SST prefix reclaim",
    );

    // DEFERRED, not dropped: the view that carried the intent is gone, so the
    // release must run the reclaim. Losing it would strand the prefix forever
    // — exactly the space a tight-space compaction was reclaiming.
    drop(guard);
    let after = read_all(&plain, &paused_path)?;
    assert!(
        after
            .get(..64)
            .is_some_and(|head| head.iter().all(|&b| b == 0)),
        "releasing the pause must run the deferred reclaim",
    );
    assert_eq!(
        after.len(),
        before.len(),
        "the deferred reclaim punches, it does not truncate",
    );
    Ok(())
}

/// Proving a punch must ask whether the zeroed run CONTAINS a hole, not
/// whether each block's own extent wholly IS one: `punch_hole` on an
/// unaligned block extent zero-fills the edge pages and deallocates only the
/// wholly-contained ones, so on a real filesystem every punched block keeps
/// allocated zeros at its boundaries. A whole-extent probe then rejects the
/// genuine punch, and a manifest-loss repair misclassifies the reclaimed SST
/// as damage — discarding its intact live suffix or salvaging it without the
/// required bound.
#[cfg(feature = "std")]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn punched_run_with_allocated_edges_still_proves_the_punch() -> crate::Result<()> {
    use crate::fs::Fs;
    use std::io::{Seek, SeekFrom, Write};

    let memfs = crate::fs::MemFs::new();
    let fs: Arc<dyn Fs> = Arc::new(memfs.clone());
    let root = std::path::absolute("/db")?;
    fs.create_dir_all(&root)?;

    let path = root.join("0");
    let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(&fs))?.use_data_block_size(256);
    for i in 0..256u32 {
        writer.write(crate::InternalValue::from_components(
            format!("k{i:04}").into_bytes(),
            b"v",
            1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = {
        #[cfg(feature = "metrics")]
        let metrics = Arc::new(Metrics::default());
        let mut params = test_recover_params(path.clone(), checksum);
        params.descriptor_table = None;
        params.fs = Arc::clone(&fs);
        #[cfg(feature = "metrics")]
        {
            params.metrics = metrics;
        }
        Table::recover(params)?
    };
    let punch_off = table.punch_offset_for(b"k0128")?;
    assert!(punch_off > 64, "the fixture has a punchable prefix");

    // The reclaim's read-back shape: the whole prefix reads as zeros...
    {
        let mut file = fs.open(
            &path,
            &crate::fs::FsOpenOptions::new().read(true).write(true),
        )?;
        file.seek(SeekFrom::Start(0))?;
        file.write_all(&vec![
            0u8;
            usize::try_from(punch_off).expect("small fixture")
        ])?;
    }
    // ...but zeros alone are damage, not proof (the probe answers, so the
    // verdict is a definite Unpunched, not an unattributable Unproven).
    assert!(
        matches!(
            table.punch_geometry()?.verdict,
            crate::table::PunchProbe::Unpunched
        ),
        "zeros without a hole are corruption, never a punch",
    );

    // The filesystem's unaligned-punch behavior: only a small interior range
    // of the zeroed run is actually deallocated — no data block's own extent
    // is wholly a hole (its boundary pages stay allocated, zero-filled).
    let mid = punch_off / 2;
    memfs.punch_hole(&path, mid - 8, 16)?;
    assert!(
        matches!(
            table.punch_geometry()?.verdict,
            crate::table::PunchProbe::Punched
        ),
        "a hole contained in the zeroed run proves the punch even though no \
         block's whole extent is one",
    );
    Ok(())
}

/// A reclaim blocked by a COMPLETED checkpoint's surviving hard link must be
/// RETAINED, not discarded. The pause is no longer active (so the deferred
/// queue is not an option), yet the dropping view holds the only record of the
/// reclaim: deleting the checkpoint later frees the link, and only
/// `retry_pending_reclaims` can finish the punch then — nothing else would
/// ever free the consumed prefix, which is exactly the space the tight-space
/// path was reclaiming.
#[cfg(feature = "std")]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn punch_on_drop_retains_the_reclaim_while_a_checkpoint_link_survives() -> crate::Result<()> {
    use crate::fs::Fs;

    let links = Arc::new(core::sync::atomic::AtomicU64::new(2));
    let fs: Arc<dyn Fs> = Arc::new(SharedInodeFs(crate::fs::MemFs::new(), Arc::clone(&links)));
    let root = std::path::absolute("/db")?;
    fs.create_dir_all(&root)?;

    let path = root.join("0");
    let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(&fs))?.use_data_block_size(256);
    for i in 0..256u32 {
        writer.write(crate::InternalValue::from_components(
            format!("k{i:04}").into_bytes(),
            b"v",
            1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = {
        #[cfg(feature = "metrics")]
        let metrics = Arc::new(Metrics::default());
        let mut params = test_recover_params(path.clone(), checksum);
        params.descriptor_table = None;
        params.fs = Arc::clone(&fs);
        #[cfg(feature = "metrics")]
        {
            params.metrics = metrics;
        }
        Table::recover(params)?
    };
    let pause = crate::deletion_pause::DeletionPause::new_shared();
    table.install_deletion_pause(Arc::clone(&pause));
    let punch = table.punch_offset_for(b"k0128")?;
    assert!(punch > 0, "the fixture has a punchable prefix");
    table.mark_punch_on_drop(punch);
    drop(table);

    // The checkpoint's link survives, so no byte may be punched yet — but the
    // intent must be retained rather than discarded.
    assert!(
        pause.has_pending_reclaims(),
        "a reclaim blocked by a completed checkpoint's link must be retained \
         for a retry, not discarded",
    );

    // The checkpoint is deleted: its link disappears, and the retry finishes
    // the reclaim.
    links.store(1, core::sync::atomic::Ordering::Release);
    pause.retry_pending_reclaims();
    assert!(
        !pause.has_pending_reclaims(),
        "an exclusively-owned file's retained reclaim is finished by the retry",
    );
    let after = {
        let file = fs.open(&path, &crate::fs::FsOpenOptions::new().read(true))?;
        let len = crate::fs::FsFile::metadata(&*file)?.len;
        crate::file::read_exact(&*file, 0, usize::try_from(len).unwrap_or(0))?.to_vec()
    };
    assert!(
        after
            .get(..64)
            .is_some_and(|head| head.iter().all(|&b| b == 0)),
        "the retried reclaim punches the consumed prefix",
    );
    Ok(())
}

/// `reopen_restricted` creates a DISTINCT `Inner` for the same table, so it must
/// PROPAGATE the tree-installed shared gates onto it: the checkpoint deletion
/// pause and the heal lock. Without them a restricted view skips the checkpoint
/// mutation window (a checkpoint could link healed bytes under a stale digest)
/// and serializes heals against a different lock (two patrols could heal +
/// reconcile the same SST concurrently and leave a clean file mismatched with
/// the manifest).
#[cfg(all(feature = "std", feature = "page_ecc"))]
#[test]
fn reopen_restricted_propagates_the_shared_heal_and_deletion_gates() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"c", b"v", 0, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            let pause = crate::deletion_pause::DeletionPause::new_shared();
            table.install_deletion_pause(std::sync::Arc::clone(&pause));
            let lock = table.heal_lock_arc();

            let restricted = table.reopen_restricted(crate::UserKey::from(&b"b"[..]))?;

            let restricted_pause = restricted
                .0
                .deletion_pause
                .get()
                .expect("the deletion pause is propagated");
            assert!(
                std::sync::Arc::ptr_eq(restricted_pause, &pause),
                "the restricted view shares the ORIGINAL deletion pause",
            );
            assert!(
                std::sync::Arc::ptr_eq(&restricted.heal_lock_arc(), &lock),
                "the restricted view shares the ORIGINAL heal lock",
            );
            Ok(())
        },
        None,
        Some(|x| x),
    )
}

/// `raw_block_parity_delta` must reject a frame whose on-disk trailer length
/// differs from the freshly computed parity: returning `Ok(Some(fresh))` for
/// a short trailer would make the in-place heal write MORE bytes than the
/// frame holds at that offset — past the block's end, into the next block's
/// bytes — violating the size-preserving heal contract. A length mismatch is
/// an unverifiable frame, not a healable one.
#[cfg(feature = "page_ecc")]
#[test]
fn raw_block_parity_delta_rejects_a_trailer_length_mismatch() -> crate::Result<()> {
    use crate::coding::Decode;

    let items = [
        crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            let keyed = table
                .data_block_handles()
                .next()
                .expect("a data block")
                .expect("index entry decodes");
            let handle = keyed.as_ref();
            let file = std::fs::read(&*table.path)?;
            let start = usize::try_from(handle.offset().0).unwrap_or(usize::MAX);
            let Some(raw) = file.get(start..start + handle.size() as usize) else {
                panic!("block frame within the file");
            };
            let header = crate::table::block::Header::decode_from(&mut &raw[..])?;

            // Sanity: the intact frame's trailer verifies (no delta).
            assert!(
                matches!(table.raw_block_parity_delta(raw, &header), Ok(None)),
                "the intact frame's parity trailer matches",
            );

            // A frame one byte SHORT of its trailer must be unverifiable —
            // not a mismatch whose full-length rebuild the heal would write
            // past the frame's end.
            let Some(short) = raw.get(..raw.len() - 1) else {
                panic!("frame is non-empty");
            };
            assert!(
                table.raw_block_parity_delta(short, &header).is_err(),
                "a trailer-length mismatch must be unverifiable, not healable",
            );

            Ok(())
        },
        None,
        Some(|w: Writer| {
            let Ok(params) = crate::table::block::EccParams::try_new(8, 2) else {
                panic!("RS(8,2) params are valid");
            };
            w.use_ecc(Some(params))
        }),
    )
}

/// `scrub_block` must reject a block whose decoded ROLE differs from the
/// caller's expected type, mirroring `load_block`'s swap-defence: an index
/// entry misdirected at another (checksum-valid) block of a different role
/// passes its payload checksum, so without the role check the scrub reports
/// "clean" for a handle that no longer points at a data block at all.
#[test]
fn scrub_block_rejects_a_block_of_the_wrong_role() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            // The TLI block is a valid, checksum-clean block of role Index —
            // exactly what a misdirected data-block handle would land on.
            let outcome = crate::table::util::scrub_block(
                table.global_id(),
                &table.path,
                &table.file_accessor,
                &table.regions.tli,
                crate::table::block::BlockType::Data,
                table.metadata.data_block_compression,
                table.encryption.as_deref(),
                table.metadata.ecc_params,
                #[cfg(zstd_any)]
                table.zstd_dictionary.as_deref(),
                None,
                #[cfg(feature = "metrics")]
                &table.metrics,
            );
            assert!(
                matches!(outcome, Err(crate::Error::InvalidTag(("BlockType", _))),),
                "a checksum-clean block of the WRONG role must fail the \
                 role check specifically (not scrub clean, and not fail for \
                 an unrelated reason): {outcome:?}",
            );
            Ok(())
        },
        None,
        Some(|x| x),
    )
}

/// A frame read through an OVER-SIZED (forged) index handle decodes cleanly —
/// the payload checksum covers only `data_length` bytes, and the trailing
/// garbage classifies as an unrecognized ECC trailer (`EccStatus::Unrecognized`)
/// — but its raw bytes must NOT be marked verbatim-copy-safe: the salvage
/// writer rejects the overlong frame against the header's on-disk size and the
/// walk would drop a block whose payload actually verified. The load must fall
/// back to the re-encode path (`verbatim = None`) instead.
#[test]
fn salvage_load_block_reencodes_an_over_read_frame() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            let keyed = table
                .data_block_handles()
                .next()
                .expect("a data block")
                .expect("index entry decodes");
            let handle = keyed.as_ref();

            // Sanity: through the TRUE handle the block is verbatim-copy-safe.
            let clean = table.salvage_load_block(handle, crate::table::block::BlockType::Data)?;
            assert!(clean.verbatim.is_some(), "a clean read is verbatim-safe");

            // Forged handle: 8 bytes of the following section leak into the
            // frame (the index section follows the data blocks, so the file
            // has bytes there).
            let over = crate::table::BlockHandle::new(handle.offset(), handle.size() + 8);
            let sb = table.salvage_load_block(&over, crate::table::block::BlockType::Data)?;
            assert!(
                sb.verbatim.is_none(),
                "an over-read frame with an opaque trailer must fall back to \
                 the re-encode path, not offer its overlong raw bytes for a \
                 verbatim copy",
            );

            Ok(())
        },
        None,
        Some(|x| x),
    )
}

/// `reopen_restricted` must carry a REFRESHED full-file checksum (set by an
/// in-place heal after recovery) into the reopened view: recovering the fresh
/// `Inner` with the stale pre-heal digest would reinstall that stale digest
/// into the manifest when a tight-space compaction swaps the restricted view
/// in, making later integrity scans flag the healed file as corrupt again.
#[test]
fn reopen_restricted_carries_the_live_suffix_digest() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            // The restricted view's manifest digest must cover only its LIVE
            // SUFFIX `[punch_offset, end)`, not the whole file: the prefix is
            // hole-punched after install, so a whole-file digest would never
            // match the punched file. It is computed fresh from the current
            // bytes (so any heal is folded in).
            let restricted = table.reopen_restricted(crate::UserKey::from(&b"b"[..]))?;
            let punch = table.punch_offset_for(b"b")?;
            let expected = crate::Checksum::from_raw(crate::repair::compute_table_checksum_from(
                &crate::fs::StdFs,
                &table.path,
                punch,
            )?);
            assert_eq!(
                restricted.checksum(),
                expected,
                "the restricted view reports its live-suffix digest",
            );
            // `rotate_every == Some(1)` puts "b" in a later data block, so the
            // punch offset is strictly positive and the suffix digest is
            // genuinely exercised (not the degenerate whole-file case).
            assert!(
                punch > 0,
                "reopening at a later block punches a real prefix"
            );
            assert_ne!(
                restricted.checksum(),
                table.checksum(),
                "a non-zero punch offset makes the suffix digest differ from \
                 the whole-file digest",
            );
            Ok(())
        },
        Some(1),
        Some(|x| x),
    )
}

/// The twelve single-letter keys the restriction-accounting tests write, four
/// per data block.
fn twelve_letter_items() -> Vec<crate::InternalValue> {
    (b'a'..=b'l')
        .map(|c| crate::InternalValue::from_components([c], b"v", 0, crate::ValueType::Value))
        .collect()
}

/// Reads a whole reconcile costs on a table of `blocks` four-entry data blocks,
/// counted at the filesystem. The table carries a zone map so the section-level
/// gates take part too. On a REAL on-disk file: the claim is about what a
/// reconcile costs in I/O, so the medium it is measured on should be the one it
/// runs against.
fn reconcile_read_count(blocks: usize) -> crate::Result<(usize, usize)> {
    use crate::fs::{FaultFs, StdFs};

    let dir = tempfile::tempdir()?;
    let faulty = std::sync::Arc::new(FaultFs::new(StdFs));
    let injector = faulty.injector();
    let path = dir.path().join("table");

    let mut writer = Writer::new(path.clone(), 0, 0, faulty.clone())?.use_zone_map(true);
    for idx in 0..blocks * 4 {
        if idx % 4 == 0 {
            writer.spill_block()?;
        }
        writer.write(crate::InternalValue::from_components(
            alloc::format!("key{idx:04}").into_bytes(),
            b"v".as_slice(),
            0,
            crate::ValueType::Value,
        ))?;
    }
    let Some((_, checksum)) = writer.finish()? else {
        panic!("the fixture writes entries");
    };
    let mut params = test_recover_params(path, checksum);
    params.fs = faulty;
    let table = Table::recover(params)?;
    assert_eq!(
        table.block_index.iter().count(),
        blocks,
        "the fixture must produce one data block per four entries",
    );

    injector.clear();
    if let Err((gate, e)) = table.verify_reconcile_gates(None, false) {
        panic!("a healthy table must pass every gate, {gate:?} refused it: {e}");
    }
    Ok((injector.read_count(), injector.open_count()))
}

/// The reconcile gates must share ONE read of each live block. Each of them
/// used to walk the table itself, so a reconcile re-read and re-decoded every
/// block once per gate and repair time grew with the gate count.
///
/// Asserted as a SLOPE: doubling the blocks must cost exactly one extra read
/// per added block. A per-gate walk would charge one per gate per block, so
/// the old shape cannot satisfy this no matter what the constant term is.
#[test]
fn reconcile_gates_read_each_block_once() -> crate::Result<()> {
    let (small, small_opens) = reconcile_read_count(3)?;
    let (large, large_opens) = reconcile_read_count(6)?;
    assert_eq!(
        large - small,
        3,
        "three more blocks must cost three more reads, got {small} then {large}",
    );
    // The sections are read through ONE lent handle, so the whole pass opens
    // the file once however many sections it consults and however many blocks
    // it walks. A reader that opens for itself makes this the section count.
    assert_eq!(
        (small_opens, large_opens),
        (1, 1),
        "a reconcile must open the table once",
    );
    Ok(())
}

/// A restriction landing on a block's LAST key leaves nearly that whole block
/// dead: the view serves only its keys `>= bound`, and the rows below belong to
/// the output table that superseded the punched prefix. Counting the straddling
/// block whole reports those rows twice while both live in one version.
#[test]
fn live_item_count_drops_the_straddling_block_rows_below_the_bound() -> crate::Result<()> {
    // `rotate_every == Some(4)` gives blocks [a..d] [e..h] [i..l]; "h" is the
    // LAST key of the middle one, so the view serves h..l = 5 of its 12 entries
    // and only ONE of the straddling block's four.
    test_with_table(
        &twelve_letter_items(),
        |table| {
            let restricted = table.with_restriction(crate::UserKey::from(&b"h"[..]));
            assert!(
                restricted.punch_offset_for(b"h")? > 0,
                "the fixture must punch a real prefix, not the degenerate whole file",
            );
            assert_eq!(
                5,
                restricted.live_item_count()?,
                "a zone-mapped view counts the straddling block's live suffix, \
                 not its whole row count",
            );
            Ok(())
        },
        Some(4),
        Some(|w: Writer| w.use_zone_map(true)),
    )
}

/// Without a zone map the count is apportioned over data bytes, but the
/// straddling block is still counted exactly: apportioning it whole credited
/// the view with every row below the bound.
#[test]
fn live_item_count_apportions_only_the_blocks_above_the_straddling_one() -> crate::Result<()> {
    test_with_table(
        &twelve_letter_items(),
        |table| {
            let restricted = table.with_restriction(crate::UserKey::from(&b"h"[..]));
            let live = restricted.live_item_count()?;
            // Exact for the straddling block (1 entry), apportioned by bytes
            // above it (~4 entries) — so within a block's granularity of the
            // true 5, and well below the 8 that counting the straddling block
            // whole reports.
            assert!(
                (4..=6).contains(&live),
                "apportioning above the straddling block must land near the \
                 5 live entries, got {live}",
            );
            Ok(())
        },
        Some(4),
        Some(|x| x),
    )
}

#[test]
fn punch_offset_for_locates_the_first_block_reaching_a_key() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"c", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"d", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"e", b"v", 0, crate::ValueType::Value),
    ];

    // rotate_every Some(1) spills a block before every item, so each key lands
    // in its own data block with a strictly increasing offset.
    test_with_table(
        &items,
        |table| {
            // "a" is in the first block at offset 0 (nothing below it to punch).
            assert_eq!(0, table.punch_offset_for(b"a")?);

            let pb = table.punch_offset_for(b"b")?;
            let pc = table.punch_offset_for(b"c")?;
            let pe = table.punch_offset_for(b"e")?;
            assert!(pb > 0, "punching up to b reclaims a's block");
            assert!(pc > pb, "offsets advance with the key");
            assert!(pe > pc);

            // A key past the last block reports the end of the data region, so
            // the whole data area is punchable.
            let beyond = table.punch_offset_for(b"zzz")?;
            assert!(
                beyond >= pe,
                "a key beyond the last block punches every data block",
            );

            Ok(())
        },
        Some(1),
        Some(|x| x),
    )
}

/// Writes `items` through an adaptive-index writer with the given spill
/// threshold and recovers the resulting [`Table`]. Returns the table plus
/// the backing temp dir (kept alive by the caller).
#[cfg(test)]
#[expect(clippy::unwrap_used)]
fn recover_adaptive_table(
    items: &[crate::InternalValue],
    spill_threshold: u64,
) -> crate::Result<(Table, tempfile::TempDir)> {
    // Writer::new opens the file exclusively, so the path must not exist yet.
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("table");

    let mut writer =
        Writer::new(path.clone(), 0, 0, Arc::new(StdFs))?.use_adaptive_index(spill_threshold);
    for item in items {
        writer.write(item.clone())?;
    }
    let (_, checksum) = writer.finish()?.unwrap();

    #[cfg(feature = "metrics")]
    let metrics = Arc::new(Metrics::default());
    let table = {
        #[cfg_attr(not(feature = "metrics"), expect(unused_mut))]
        let mut params = test_recover_params(path, checksum);
        #[cfg(feature = "metrics")]
        {
            params.metrics = metrics;
        }
        Table::recover(params)?
    };
    Ok((table, dir))
}

/// Adaptive index, small SST: a high spill threshold keeps the index
/// single-level (no separate index region), and every key reads back.
#[test]
#[expect(clippy::unwrap_used)]
fn adaptive_index_small_sst_is_single_level() -> crate::Result<()> {
    let items: Vec<_> = (0u32..500)
        .map(|i| {
            let key = format!("key-{i:08}");
            crate::InternalValue::from_components(
                key.as_bytes(),
                b"some-value-payload",
                0,
                crate::ValueType::Value,
            )
        })
        .collect();

    // Threshold far above any plausible index size for 500 keys.
    let (table, _dir) = recover_adaptive_table(&items, u64::MAX)?;

    assert!(
        table.regions.index.is_none(),
        "small index must stay single-level (Full), got a two-level index region",
    );
    assert_eq!(
        items.len(),
        usize::try_from(table.metadata.item_count).unwrap()
    );

    for i in 0u32..500 {
        let key = format!("key-{i:08}");
        let got = table.get(key.as_bytes(), SeqNo::MAX, hash64(key.as_bytes()))?;
        assert_eq!(
            b"some-value-payload",
            &*got.unwrap().value,
            "single-level read mismatch for {key}",
        );
    }
    Ok(())
}

/// Adaptive index, forced spill: a zero spill threshold forces the
/// two-level (partitioned) layout, and the same keys still read back —
/// proving both layouts round-trip identically.
#[test]
#[expect(clippy::unwrap_used)]
fn adaptive_index_zero_threshold_spills_to_two_level() -> crate::Result<()> {
    let items: Vec<_> = (0u32..500)
        .map(|i| {
            let key = format!("key-{i:08}");
            crate::InternalValue::from_components(
                key.as_bytes(),
                b"some-value-payload",
                0,
                crate::ValueType::Value,
            )
        })
        .collect();

    // Threshold 0 → spill on the first index entry → partitioned.
    let (table, _dir) = recover_adaptive_table(&items, 0)?;

    assert!(
        table.regions.index.is_some(),
        "zero threshold must spill to a two-level (partitioned) index",
    );
    assert_eq!(
        items.len(),
        usize::try_from(table.metadata.item_count).unwrap()
    );

    for i in 0u32..500 {
        let key = format!("key-{i:08}");
        let got = table.get(key.as_bytes(), SeqNo::MAX, hash64(key.as_bytes()))?;
        assert_eq!(
            b"some-value-payload",
            &*got.unwrap().value,
            "two-level read mismatch for {key}",
        );
    }
    Ok(())
}

#[test]
fn table_point_read_index_block_restart_interval() -> crate::Result<()> {
    let items: Vec<_> = (0u32..24)
        .map(|i| {
            let key = format!("adj:out:vertex-0001:edge-{i:04}");
            let value = format!("value-{i:04}");
            crate::InternalValue::from_components(
                key.as_bytes(),
                value.as_bytes(),
                u64::from(i),
                crate::ValueType::Value,
            )
        })
        .collect();

    test_with_table(
        &items,
        |table| {
            assert_eq!(
                b"value-0011",
                &*table
                    .get(
                        b"adj:out:vertex-0001:edge-0011",
                        SeqNo::MAX,
                        hash64(b"adj:out:vertex-0001:edge-0011"),
                    )?
                    .expect("test assertion: expected value for edge-0011")
                    .value,
            );

            let range = table
                .range(
                    UserKey::from("adj:out:vertex-0001:edge-0008")
                        ..=UserKey::from("adj:out:vertex-0001:edge-0012"),
                )
                .flatten()
                .collect::<Vec<_>>();

            assert_eq!(items[8..=12], range);

            Ok(())
        },
        Some(1),
        Some(|writer: Writer| {
            writer
                .use_data_block_size(128)
                .use_index_block_restart_interval(4)
        }),
    )
}

#[test]
#[cfg(feature = "zstd")]
fn table_point_read_zstd_dictionary() -> crate::Result<()> {
    let dict = Arc::new(make_test_dictionary());
    let expected_dict_id = dict.id();
    let compression = crate::CompressionType::zstd_dict(3, expected_dict_id)?;
    let items = [
        crate::InternalValue::from_components(
            b"key-00001",
            b"value-00001-padding-to-make-it-longer",
            3,
            crate::ValueType::Value,
        ),
        crate::InternalValue::from_components(
            b"key-00002",
            b"value-00002-padding-to-make-it-longer",
            2,
            crate::ValueType::Value,
        ),
    ];

    test_with_table_and_zstd_dictionary(
        &items,
        |table| {
            assert!(matches!(
                table.metadata.data_block_compression,
                crate::CompressionType::ZstdDict { dict_id, .. } if dict_id == expected_dict_id
            ));
            assert_eq!(items, &*table.iter().flatten().collect::<Vec<_>>());
            assert_eq!(
                b"value-00001-padding-to-make-it-longer",
                &*table
                    .get(b"key-00001", SeqNo::MAX, hash64(b"key-00001"),)?
                    .expect("test assertion: expected value for key-00001")
                    .value,
            );
            Ok(())
        },
        None,
        Some(|writer: Writer| writer.use_data_block_compression(compression)),
        dict,
    )
}

#[test]
fn table_range_exclusive_bounds() -> crate::Result<()> {
    use core::ops::Bound::{Excluded, Included};

    let items = [
        crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"c", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"d", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"e", b"v", 0, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            let res = table
                .range((Excluded(UserKey::from("b")), Included(UserKey::from("d"))))
                .flatten()
                .collect::<Vec<_>>();
            assert_eq!(
                items.iter().skip(2).take(2).cloned().collect::<Vec<_>>(),
                &*res,
            );

            let res = table
                .range((Excluded(UserKey::from("b")), Included(UserKey::from("d"))))
                .rev()
                .flatten()
                .collect::<Vec<_>>();
            assert_eq!(
                items
                    .iter()
                    .skip(2)
                    .take(2)
                    .rev()
                    .cloned()
                    .collect::<Vec<_>>(),
                &*res,
            );

            let res = table
                .range((Excluded(UserKey::from("b")), Excluded(UserKey::from("d"))))
                .flatten()
                .collect::<Vec<_>>();
            assert_eq!(
                items.iter().skip(2).take(1).cloned().collect::<Vec<_>>(),
                &*res,
            );

            let res = table
                .range((Excluded(UserKey::from("b")), Excluded(UserKey::from("d"))))
                .rev()
                .flatten()
                .collect::<Vec<_>>();
            assert_eq!(
                items
                    .iter()
                    .skip(2)
                    .take(1)
                    .rev()
                    .cloned()
                    .collect::<Vec<_>>(),
                &*res,
            );

            Ok(())
        },
        None,
        Some(|x: Writer| x.use_data_block_size(1)),
    )
}

#[test]
fn writer_records_effective_page_ecc_descriptor() -> crate::Result<()> {
    // descriptor#page_ecc must record the EFFECTIVE (compiled) setting, not
    // the requested flag. Without the `page_ecc` cargo feature,
    // use_page_ecc(true) is a no-op (with_ecc() is identity, no parity is
    // emitted and no ECC_PARITY bit is set), so the persisted descriptor
    // must read false to stay consistent with the actual on-disk blocks.
    // With the feature it reads true. `cfg!(feature = "page_ecc")` is the
    // effective value either way.
    let items = [crate::InternalValue::from_components(
        b"a",
        b"v",
        0,
        crate::ValueType::Value,
    )];
    test_with_table(
        &items,
        |table| {
            assert_eq!(
                table.metadata.page_ecc,
                cfg!(feature = "page_ecc"),
                "descriptor#page_ecc must reflect the effective (compiled) page_ecc setting",
            );
            Ok(())
        },
        None,
        Some(|w: Writer| {
            w.use_page_ecc(
                true,
                crate::runtime_config::EccScheme::ReedSolomon {
                    data_shards: 4,
                    parity_shards: 2,
                },
            )
        }),
    )
}

#[test]
#[expect(clippy::unwrap_used)]
fn table_point_read_mvcc_block_boundary() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"a", b"5", 5, crate::ValueType::Value),
        crate::InternalValue::from_components(b"a", b"4", 4, crate::ValueType::Value),
        crate::InternalValue::from_components(b"a", b"3", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"a", b"2", 2, crate::ValueType::Value),
        crate::InternalValue::from_components(b"a", b"1", 1, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            assert_eq!(2, table.metadata.data_block_count);

            let key_hash = hash64(b"a");

            assert_eq!(
                b"5",
                &*table.get(b"a", SeqNo::MAX, key_hash)?.unwrap().value
            );
            assert_eq!(b"4", &*table.get(b"a", 5, key_hash)?.unwrap().value);
            assert_eq!(b"3", &*table.get(b"a", 4, key_hash)?.unwrap().value);
            assert_eq!(b"2", &*table.get(b"a", 3, key_hash)?.unwrap().value);
            assert_eq!(b"1", &*table.get(b"a", 2, key_hash)?.unwrap().value);

            Ok(())
        },
        Some(3),
        Some(|x| x),
    )
}

#[test]
fn table_scan() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"abc", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"def", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"xyz", b"asdasdasd", 3, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            assert_eq!(items, &*table.scan()?.flatten().collect::<Vec<_>>());

            assert_eq!(
                table.metadata.key_range,
                crate::KeyRange::new((b"abc".into(), b"xyz".into())),
            );

            Ok(())
        },
        None,
        Some(|x| x),
    )
}

#[test]
fn table_iter_simple() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"abc", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"def", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"xyz", b"asdasdasd", 3, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            assert_eq!(items, &*table.iter().flatten().collect::<Vec<_>>());
            assert_eq!(
                items.iter().rev().cloned().collect::<Vec<_>>(),
                &*table.iter().rev().flatten().collect::<Vec<_>>(),
            );

            Ok(())
        },
        None,
        Some(|x| x),
    )
}

#[test]
fn table_range_simple() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"abc", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"def", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"xyz", b"asdasdasd", 3, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            assert_eq!(
                items.iter().skip(1).cloned().collect::<Vec<_>>(),
                &*table
                    .range(UserKey::from("b")..)
                    .flatten()
                    .collect::<Vec<_>>()
            );

            assert_eq!(
                items.iter().skip(1).rev().cloned().collect::<Vec<_>>(),
                &*table
                    .range(UserKey::from("b")..)
                    .rev()
                    .flatten()
                    .collect::<Vec<_>>(),
            );

            Ok(())
        },
        None,
        Some(|x| x),
    )
}

#[test]
fn table_range_ping_pong() -> crate::Result<()> {
    let items = (0u64..10)
        .map(|i| InternalValue::from_components(i.to_be_bytes(), "", 0, crate::ValueType::Value))
        .collect::<Vec<_>>();

    test_with_table(
        &items,
        |table| {
            let mut iter =
                table.range(UserKey::from(5u64.to_be_bytes())..UserKey::from(10u64.to_be_bytes()));

            let mut count = 0;

            for x in 0.. {
                if x % 2 == 0 {
                    let Some(_) = iter.next() else {
                        break;
                    };

                    count += 1;
                } else {
                    let Some(_) = iter.next_back() else {
                        break;
                    };

                    count += 1;
                }
            }

            assert_eq!(5, count);

            Ok(())
        },
        None,
        Some(|x| x),
    )
}

#[test]
fn table_range_multiple_data_blocks() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"a", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"b", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"c", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"d", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"e", b"asdasdasd", 3, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            assert_eq!(5, table.metadata.data_block_count);

            assert_eq!(
                items.iter().skip(1).take(3).cloned().collect::<Vec<_>>(),
                &*table
                    .range(UserKey::from("b")..=UserKey::from("d"))
                    .flatten()
                    .collect::<Vec<_>>()
            );

            assert_eq!(
                items
                    .iter()
                    .skip(1)
                    .take(3)
                    .rev()
                    .cloned()
                    .collect::<Vec<_>>(),
                &*table
                    .range(UserKey::from("b")..=UserKey::from("d"))
                    .rev()
                    .flatten()
                    .collect::<Vec<_>>(),
            );

            Ok(())
        },
        None,
        Some(|x: Writer| x.use_data_block_size(1)),
    )
}

#[test]
#[expect(clippy::unwrap_used)]
fn table_point_read_partitioned_filter_smoke_test() -> crate::Result<()> {
    let items = [
        crate::InternalValue::from_components(b"a", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"b", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"c", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"d", b"asdasdasd", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"e", b"asdasdasd", 3, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            assert_eq!(1, table.metadata.data_block_count);

            for item in &items {
                let key_hash = hash64(&item.key.user_key);

                assert_eq!(
                    item.value,
                    table
                        .get(&item.key.user_key, SeqNo::MAX, key_hash)
                        .unwrap()
                        .unwrap()
                        .value,
                );
            }

            Ok(())
        },
        None,
        Some(|x: Writer| x.use_partitioned_filter()),
    )
}

#[test]
#[expect(clippy::unwrap_used)]
fn table_partitioned_filter() -> crate::Result<()> {
    use crate::ValueType::Value;

    let items = [
        InternalValue::from_components("a", "a7", 7, Value),
        InternalValue::from_components("a", "a6", 6, Value),
        InternalValue::from_components("a", "a5", 5, Value),
        InternalValue::from_components("a", "a4", 4, Value),
        InternalValue::from_components("a", "a3", 3, Value),
        InternalValue::from_components("b", "b5", 5, Value),
        InternalValue::from_components("c", "c8", 8, Value),
        InternalValue::from_components("d", "d10", 10, Value),
    ];

    test_with_table(
        &items,
        |table| {
            assert!(table.regions.filter.is_some(), "filter should exist");
            assert!(
                table.regions.filter_tli.is_some(),
                "filter TLI should exist"
            );

            assert_eq!(b"a7", &*table.get(b"a", 8, hash64(b"a"))?.unwrap().value,);
            assert_eq!(b"a6", &*table.get(b"a", 7, hash64(b"a"))?.unwrap().value,);
            assert_eq!(b"a5", &*table.get(b"a", 6, hash64(b"a"))?.unwrap().value,);
            assert_eq!(b"a4", &*table.get(b"a", 5, hash64(b"a"))?.unwrap().value,);
            assert_eq!(b"a3", &*table.get(b"a", 4, hash64(b"a"))?.unwrap().value,);
            assert_eq!(b"b5", &*table.get(b"b", 6, hash64(b"b"))?.unwrap().value,);
            assert_eq!(b"c8", &*table.get(b"c", 9, hash64(b"c"))?.unwrap().value,);
            assert_eq!(b"d10", &*table.get(b"d", 11, hash64(b"d"))?.unwrap().value,);
            Ok(())
        },
        None,
        Some(|x: Writer| x.use_partitioned_filter().use_meta_partition_size(3)),
    )
}

#[test]
#[expect(clippy::unwrap_used, reason = "test code")]
fn plan_block_tasks_propagates_a_faulted_bloom_probe() -> crate::Result<()> {
    use crate::fs::{Fault, FaultFs, FaultInjector, FaultOp, FaultRule};
    use crate::io::ErrorKind;

    // A partitioned, unpinned filter makes `check_bloom` read a filter partition
    // block from disk on lookup; that read is the planning I/O that can fail.
    let items: Vec<InternalValue> = (0..64u32)
        .map(|i| {
            InternalValue::from_components(
                format!("key{i:04}").into_bytes(),
                b"v".to_vec(),
                1,
                crate::ValueType::Value,
            )
        })
        .collect();

    let dir = tempdir()?;
    let file = dir.path().join("table");
    let injector = Arc::new(FaultInjector::new());
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(FaultFs::with_injector(StdFs, Arc::clone(&injector)));

    let checksum = {
        let mut writer = Writer::new(file.clone(), 0, 0, Arc::clone(&fs))?
            .use_partitioned_filter()
            .use_meta_partition_size(8);
        for item in &items {
            writer.write(item.clone())?;
        }
        writer.finish()?.unwrap().1
    };

    let table = {
        let mut params = test_recover_params(file, checksum);
        // Do not pin: partition blocks read lazily.
        params.fs = Arc::clone(&fs);
        Table::recover(params)?
    };

    // Recovery is done (its reads passed cleanly); now fail the NEXT positional
    // read of the table file: the filter partition block `check_bloom` loads.
    injector.arm(FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Other)).on_path("table"));

    let key = b"key0000".as_slice();
    let sorted = [(key, hash64(key))];
    let result = table.plan_block_tasks(&sorted, SeqNo::MAX);

    // The serial path propagates a bloom-probe error via `?`; the chunked planner
    // must too, so a faulted probe surfaces as Err instead of a swallowed miss
    // that would let a stale lower level answer.
    assert!(
        result.is_err(),
        "a faulted bloom-probe read must surface as Err, not a swallowed miss"
    );
    Ok(())
}

#[test]
#[expect(clippy::unwrap_used, reason = "test code")]
fn plan_block_tasks_propagates_a_faulted_index_read() -> crate::Result<()> {
    use crate::fs::{Fault, FaultFs, FaultInjector, FaultOp, FaultRule};
    use crate::io::ErrorKind;

    // 500 keys + adaptive-index threshold 0 spill to a two-level (partitioned)
    // block index whose partition blocks are read lazily, so `block_iter.next()`
    // does a disk read that can fail. A pinned filter keeps check_bloom off disk,
    // so the planner's first faulted positional read is the index read.
    let items: Vec<InternalValue> = (0u32..500)
        .map(|i| {
            InternalValue::from_components(
                format!("key{i:06}").into_bytes(),
                b"v".to_vec(),
                1,
                crate::ValueType::Value,
            )
        })
        .collect();

    let dir = tempdir()?;
    let file = dir.path().join("table");
    let injector = Arc::new(FaultInjector::new());
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(FaultFs::with_injector(StdFs, Arc::clone(&injector)));

    let checksum = {
        let mut writer = Writer::new(file.clone(), 0, 0, Arc::clone(&fs))?.use_adaptive_index(0);
        for item in &items {
            writer.write(item.clone())?;
        }
        writer.finish()?.unwrap().1
    };

    let table = {
        let mut params = test_recover_params(file, checksum);
        params.fs = Arc::clone(&fs);
        // Pin the filter (keeps check_bloom off disk) but not the index:
        // partition blocks read lazily.
        params.pin_filter = true;
        Table::recover(params)?
    };

    // Recovery is done; fail the next positional read of the table file, which
    // the block-index walk performs.
    injector.arm(FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Other)).on_path("table"));

    let key = b"key000250".as_slice();
    let sorted = [(key, hash64(key))];
    let result = table.plan_block_tasks(&sorted, SeqNo::MAX);

    // batch_get propagates the same iterator error via `?`; the chunked planner
    // must too, so a faulted index read surfaces as Err instead of a swallowed
    // end-of-index that would let a stale lower level answer.
    assert!(
        result.is_err(),
        "a faulted block-index read must surface as Err, not a swallowed end-of-index"
    );
    Ok(())
}

#[test]
#[expect(clippy::unwrap_used, reason = "test code")]
fn plan_block_tasks_returns_none_for_a_table_above_the_snapshot() -> crate::Result<()> {
    use crate::fs::{Fault, FaultFs, FaultInjector, FaultOp, FaultRule};
    use crate::io::ErrorKind;

    // All items live at seqno 10; a read at seqno 5 sees the whole table above
    // the snapshot, so the planner contributes nothing and never touches disk.
    let items: Vec<InternalValue> = (0u32..16)
        .map(|i| {
            InternalValue::from_components(
                format!("k{i:04}").into_bytes(),
                b"v".to_vec(),
                10,
                crate::ValueType::Value,
            )
        })
        .collect();

    let dir = tempdir()?;
    let file = dir.path().join("table");
    let injector = Arc::new(FaultInjector::new());
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(FaultFs::with_injector(StdFs, Arc::clone(&injector)));

    let checksum = {
        let mut writer = Writer::new(file.clone(), 0, 0, Arc::clone(&fs))?;
        for item in &items {
            writer.write(item.clone())?;
        }
        writer.finish()?.unwrap().1
    };

    let table = {
        let mut params = test_recover_params(file, checksum);
        params.fs = Arc::clone(&fs);
        Table::recover(params)?
    };

    // Recovery is done; fail ANY further positional read of the table file. The
    // above-snapshot guard must short-circuit before any bloom or index read, so
    // the fault must never fire: the call returns Ok(None), not Err.
    injector.arm(FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Other)).on_path("table"));

    let key = b"k0000".as_slice();
    let sorted = [(key, hash64(key))];
    // Read seqno 5 is below the table's lowest seqno (10): entirely above the
    // snapshot. Ok(None) despite the armed read fault proves the no-read path.
    assert!(table.plan_block_tasks(&sorted, 5)?.is_none());
    Ok(())
}

#[test]
fn table_seqnos() -> crate::Result<()> {
    use crate::ValueType::Value;

    let items = [
        InternalValue::from_components("a", nanoid::nanoid!().as_bytes(), 7, Value),
        InternalValue::from_components("b", nanoid::nanoid!().as_bytes(), 5, Value),
        InternalValue::from_components("c", nanoid::nanoid!().as_bytes(), 8, Value),
        InternalValue::from_components("d", nanoid::nanoid!().as_bytes(), 10, Value),
    ];

    test_with_table(
        &items,
        |table| {
            assert_eq!(5, table.metadata.seqnos.0);
            assert_eq!(10, table.metadata.seqnos.1);
            Ok(())
        },
        None,
        Some(|x| x),
    )
}

#[test]
fn table_zero_bpk() -> crate::Result<()> {
    use crate::ValueType::Value;

    let items = [
        InternalValue::from_components("a", nanoid::nanoid!().as_bytes(), 7, Value),
        InternalValue::from_components("b", nanoid::nanoid!().as_bytes(), 5, Value),
        InternalValue::from_components("c", nanoid::nanoid!().as_bytes(), 8, Value),
        InternalValue::from_components("d", nanoid::nanoid!().as_bytes(), 10, Value),
    ];

    test_with_table(
        &items,
        |table| {
            assert!(table.regions.filter.is_none());
            Ok(())
        },
        None,
        Some(|x: Writer| x.use_bloom_policy(BloomConstructionPolicy::BitsPerKey(0.0))),
    )
}

#[test]
#[expect(
    clippy::unreadable_literal,
    clippy::unwrap_used,
    clippy::indexing_slicing,
    clippy::cast_possible_truncation
)]
#[cfg(not(feature = "metrics"))]
fn table_read_fuzz_1() -> crate::Result<()> {
    use crate::Slice;
    use crate::ValueType::{Tombstone, Value};

    let items = [
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            18340908174618760209,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            18054235897395861447,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([103]),
            17820711698989577060,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            17652351990810576660,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            17576667967203573449,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([30]),
            16889403751796995588,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([186]),
            15595956295177086731,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            15512796775024989213,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([188, 156, 59, 85, 13]),
            15149465603839159843,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([174, 71]),
            15102256701513339307,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([35, 148]),
            15091160407760527013,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            14675333203365509622,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([245]),
            14571905818510788533,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            14541113699969547298,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            14486387191240337417,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            14112006182482717758,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([159]),
            13992512869528291746,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            13915106262991388976,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            13597506620670366065,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            13064400463180401957,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            12969967266897711474,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            12508372658468564628,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([138]),
            11795269606598686255,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([18]),
            10730214428751858128,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([236]),
            10124645034840293700,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([216, 81]),
            9559308046784608794,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([79]),
            8607115510826103394,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            7963767336149785641,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            7882646634183551394,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            7719307175583565930,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([111]),
            7522791039398476411,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([227, 164, 129]),
            7410771579448817672,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            7003757491682295965,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            5723101273557106371,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            5581364419922287132,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([119, 29]),
            5541782075650463683,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            5136199042703471864,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            5051972816573966850,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([162]),
            5020119417385108821,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([69]),
            4325966282181409009,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            4238714774310338082,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            4200824275757201410,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([92, 145, 251, 240, 133]),
            3894954012280195585,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([14]),
            3814525464013269105,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            3766663710061910506,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([129]),
            3749655073597306832,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([231]),
            3319226033273656005,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            3274394613296787928,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            2045761581956846404,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([78]),
            1704041985603476880,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([]),
            1441130125005023946,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([164, 136]),
            1225420702887300153,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([55]),
            974698856173325051,
            Value,
        ),
        InternalValue::from_components(
            Slice::from([0]),
            Slice::from([238, 237]),
            47340610649818236,
            Value,
        ),
        InternalValue::from_components(Slice::from([0]), Slice::from([]), 0, Value),
        InternalValue::from_components(
            Slice::from([0, 161]),
            Slice::from([]),
            17872519117933825384,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([0, 161]),
            Slice::from([]),
            4494664966150999400,
            Tombstone,
        ),
        InternalValue::from_components(
            Slice::from([1]),
            Slice::from([]),
            15373275907316083975,
            Value,
        ),
    ];

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

    let data_block_size = 97;

    let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))
        .unwrap()
        .use_data_block_size(data_block_size);

    for item in items.iter().cloned() {
        writer.write(item).unwrap();
    }

    let _trailer = writer.finish().unwrap();

    let table = {
        let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
        params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
        params.pin_filter = true;
        params.pin_index = true;
        crate::Table::recover(params).unwrap()
    };

    let item_count_usize = table.metadata.item_count as usize;
    assert_eq!(item_count_usize, items.len());

    assert_eq!(items.len(), item_count_usize);
    let items = items.into_iter().collect::<Vec<_>>();

    assert_eq!(items, table.iter().collect::<Result<Vec<_>, _>>().unwrap());
    assert_eq!(
        items.iter().rev().cloned().collect::<Vec<_>>(),
        table.iter().rev().collect::<Result<Vec<_>, _>>().unwrap(),
    );

    {
        let lo = 0;
        let hi = 54;

        let lo_key = &items[lo].key.user_key;
        let hi_key = &items[hi].key.user_key;

        assert_eq!(lo_key, hi_key);

        let expected_range: Vec<_> = items[lo..=hi].to_vec();

        let iter = table.range(lo_key..=hi_key);

        assert_eq!(expected_range, iter.collect::<Result<Vec<_>, _>>().unwrap());
    }

    Ok(())
}

#[test]
#[expect(clippy::unwrap_used)]
fn table_partitioned_index() -> crate::Result<()> {
    use crate::ValueType::Value;

    let items = [
        InternalValue::from_components("a", "a7", 7, Value),
        InternalValue::from_components("a", "a6", 6, Value),
        InternalValue::from_components("a", "a5", 5, Value),
        InternalValue::from_components("a", "a4", 4, Value),
        InternalValue::from_components("a", "a3", 3, Value),
        InternalValue::from_components("b", "b5", 5, Value),
        InternalValue::from_components("c", "c8", 8, Value),
        InternalValue::from_components("d", "d10", 10, Value),
    ];

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

    let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))
        .unwrap()
        .use_partitioned_index()
        .use_data_block_size(5)
        .use_meta_partition_size(3);

    for item in items.iter().cloned() {
        writer.write(item).unwrap();
    }

    let _trailer = writer.finish().unwrap();

    let table = {
        let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
        params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
        params.pin_filter = true;
        params.pin_index = true;
        crate::Table::recover(params).unwrap()
    };

    assert!(
        table.regions.index.is_some(),
        "2nd-level index should exist",
    );

    assert!(
        table.metadata.index_block_count > 1,
        "should use partitioned index",
    );

    assert_eq!(b"a7", &*table.get(b"a", 8, hash64(b"a"))?.unwrap().value,);
    assert_eq!(b"a6", &*table.get(b"a", 7, hash64(b"a"))?.unwrap().value,);
    assert_eq!(b"a5", &*table.get(b"a", 6, hash64(b"a"))?.unwrap().value,);
    assert_eq!(b"a4", &*table.get(b"a", 5, hash64(b"a"))?.unwrap().value,);
    assert_eq!(b"a3", &*table.get(b"a", 4, hash64(b"a"))?.unwrap().value,);
    assert_eq!(b"b5", &*table.get(b"b", 6, hash64(b"b"))?.unwrap().value,);
    assert_eq!(b"c8", &*table.get(b"c", 9, hash64(b"c"))?.unwrap().value,);
    assert_eq!(b"d10", &*table.get(b"d", 11, hash64(b"d"))?.unwrap().value,);

    Ok(())
}

#[test]
#[expect(clippy::unwrap_used)]
fn table_global_seqno() -> crate::Result<()> {
    use crate::ValueType::Value;

    let items = [
        InternalValue::from_components("a0", "a0", 0, Value),
        InternalValue::from_components("a1", "a1", 1, Value),
        InternalValue::from_components("b", "b", 8, Value),
    ];

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

    let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))
        .unwrap()
        .use_partitioned_filter()
        .use_data_block_size(1)
        .use_meta_partition_size(1);

    for item in items.iter().cloned() {
        writer.write(item).unwrap();
    }

    let _trailer = writer.finish().unwrap();

    let table = {
        let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
        params.global_seqno = 7;
        params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
        params.pin_filter = true;
        params.pin_index = true;
        crate::Table::recover(params).unwrap()
    };

    // global seqno is 7, so a1 is = 8 -> can not be read by snapshot=8
    assert!(table.get(b"a1", 8, hash64(b"a1"))?.is_none());

    assert_eq!(b"a0", &*table.get(b"a0", 8, hash64(b"a0"))?.unwrap().value,);

    Ok(())
}

/// Pins `Table::get` returning items with **global** seqno coordinates
/// even when the on-disk block carries `seqno = 0`. Mirrors the upstream
/// regression test for the equivalent fix in fjall-rs/lsm-tree (commit
/// bad4fe0a). Our fork's structural fix lives at the `Table::get` /
/// `get_with_block` / `batch_get` boundary (each call site adds
/// `global_seqno` back via `saturating_add` after the table-local
/// `point_read`), rather than inside `point_read` itself, but the
/// caller-observable contract is the same: a recovered ingested item
/// is returned with its effective global seqno, not the on-disk
/// table-local seqno.
///
/// A regression that drops the `saturating_add(global_seqno)` step
/// (e.g. a refactor that flattens `point_read` directly into `get`
/// without re-applying the offset) would fail this test by returning
/// `seqno = 0` instead of `seqno = SEQNO`.
#[test]
#[expect(clippy::unwrap_used, reason = "test assertions")]
fn table_return_global_seqno() -> crate::Result<()> {
    use crate::ValueType::Value;
    use crate::fs::StdFs;

    const SEQNO: SeqNo = 15;

    let items = [InternalValue::from_components("abc", "abc", 0, Value)];

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

    let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;

    for item in items {
        writer.write(item)?;
    }

    let _trailer = writer.finish()?;

    let table = {
        let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
        params.global_seqno = SEQNO;
        params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
        params.pin_filter = true;
        params.pin_index = true;
        crate::Table::recover(params)?
    };

    // On disk: seqno = 0. Effective global seqno: 0 + SEQNO = SEQNO.
    // Snapshot = 2 * SEQNO is above the effective seqno, so the read sees the item.
    // Returned value MUST carry the effective global seqno (= SEQNO),
    // not the table-local seqno (= 0) it has on disk.
    assert_eq!(
        InternalValue::from_components("abc", "abc", SEQNO, Value),
        table.get(b"abc", 2 * SEQNO, hash64(b"abc"))?.unwrap(),
    );

    Ok(())
}

/// A bulk-ingested table's whole content sits at effective seqno
/// `global_seqno` (every row is stored at local 0). An upper bound BELOW that
/// base must therefore return nothing — but both translated bounds saturate
/// to local 0, and `[0, 0]` is a valid one-seqno window that matches every
/// stored row, so without an explicit base check the scan returned rows whose
/// translated-back seqno exceeds the caller's inclusive upper bound.
#[test]
fn scan_seqno_range_returns_nothing_below_a_bulk_ingest_base() -> crate::Result<()> {
    use crate::ValueType::Value;
    use crate::fs::StdFs;

    const BASE: SeqNo = 100;

    let dir = tempfile::tempdir()?;
    let file = dir.path().join("ingested");
    let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
    writer.write(InternalValue::from_components("abc", "abc", 0, Value))?;
    let _trailer = writer.finish()?;

    let table = {
        let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
        params.global_seqno = BASE;
        params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
        crate::Table::recover(params)?
    };

    // The window [0, 50] ends below the base: no effective seqno of this
    // table can fall inside it.
    assert_eq!(
        table.scan_seqno_range(0, BASE / 2, true)?,
        Vec::new(),
        "an upper bound below the ingest base must exclude the whole table",
    );
    // Sanity: a window that DOES reach the base still returns the row, at its
    // effective (translated) seqno.
    assert_eq!(
        table.scan_seqno_range(0, BASE, true)?,
        vec![InternalValue::from_components("abc", "abc", BASE, Value)],
        "a window covering the base returns the row at its effective seqno",
    );
    Ok(())
}

/// Build a [`Block`] from raw bytes for `decode_range_tombstones` tests.
#[expect(
    clippy::expect_used,
    reason = "test helper: data length is controlled and fits in u32"
)]
fn rt_block(data: Vec<u8>) -> Block {
    let data_length = u32::try_from(data.len()).expect("test buffer fits in u32");
    Block {
        header: block::Header {
            data_length,
            uncompressed_length: data_length,
            ..block::Header::test_dummy(block::BlockType::RangeTombstone)
        },
        data: data.into(),
    }
}

/// Assert `decode_range_tombstones` returns [`RangeTombstoneDecode`](crate::Error::RangeTombstoneDecode)
/// with the given field and expected byte offset.
fn assert_rt_decode_error(data: Vec<u8>, expected_field: &str, expected_offset: u64) {
    let block = rt_block(data);
    // Uses DefaultUserComparator: tests verify structural decode errors
    // (truncation, missing fields), not comparator-dependent ordering.
    match Table::decode_range_tombstones(&block, &crate::comparator::DefaultUserComparator) {
        Err(crate::Error::RangeTombstoneDecode { field, offset }) => {
            assert_eq!(
                field, expected_field,
                "expected field '{expected_field}', got '{field}'"
            );
            assert_eq!(
                offset, expected_offset,
                "expected offset {expected_offset}, got {offset}"
            );
        }
        other => panic!(
            "expected RangeTombstoneDecode {{ field: \"{expected_field}\" }}, got: {other:?}"
        ),
    }
}

#[test]
#[expect(clippy::unwrap_used)]
fn decode_range_tombstones_invalid_interval_returns_error() {
    use crate::io::{LE, WriteBytesExt};

    // Build a single tombstone where start ("z") >= end ("a")
    let mut buf = Vec::new();
    buf.write_u16::<LE>(1).unwrap(); // start_len
    buf.extend_from_slice(b"z");
    buf.write_u16::<LE>(1).unwrap(); // end_len
    buf.extend_from_slice(b"a");
    buf.write_u64::<LE>(1).unwrap(); // seqno

    assert_rt_decode_error(buf, "interval", 0);
}

#[test]
fn decode_range_tombstones_truncated_start_len_returns_error() {
    // Only 1 byte — not enough for u16 start_len; offset = 0 (entry start)
    assert_rt_decode_error(vec![0x01], "start_len", 0);
}

#[test]
fn decode_range_tombstones_empty_block_returns_error() {
    // Empty RT block payload is corruption — writer only creates an RT block
    // handle when at least one tombstone exists.
    assert_rt_decode_error(Vec::new(), "start_len", 0);
}

#[test]
#[expect(clippy::unwrap_used)]
fn decode_range_tombstones_start_len_exceeds_remaining_returns_error() {
    use crate::io::{LE, WriteBytesExt};

    // start_len = 100 but only 1 byte of data follows; offset = 0 (entry start)
    let mut buf = Vec::new();
    buf.write_u16::<LE>(100).unwrap();
    buf.push(0xFF);

    assert_rt_decode_error(buf, "start_len", 0);
}

#[test]
#[expect(clippy::unwrap_used)]
fn decode_range_tombstones_truncated_end_len_returns_error() {
    use crate::io::{LE, WriteBytesExt};

    // Valid start_len + start, then truncated before end_len completes
    // offset = 3 (after u16 start_len + 1-byte key)
    let mut buf = Vec::new();
    buf.write_u16::<LE>(1).unwrap(); // start_len = 1
    buf.push(b'a'); // start key
    buf.push(0x01); // only 1 byte of end_len (need 2)

    assert_rt_decode_error(buf, "end_len", 3);
}

#[test]
#[expect(clippy::unwrap_used)]
fn decode_range_tombstones_end_len_exceeds_remaining_returns_error() {
    use crate::io::{LE, WriteBytesExt};

    // Valid start, then end_len = 100 but only 1 byte follows
    // offset = 3 (after u16 start_len + 1-byte key)
    let mut buf = Vec::new();
    buf.write_u16::<LE>(1).unwrap(); // start_len
    buf.push(b'a'); // start key
    buf.write_u16::<LE>(100).unwrap(); // end_len = 100
    buf.push(0xFF); // only 1 byte

    assert_rt_decode_error(buf, "end_len", 3);
}

#[test]
#[expect(clippy::unwrap_used)]
fn decode_range_tombstones_truncated_seqno_returns_error() {
    use crate::io::{LE, WriteBytesExt};

    // Valid start + end, but seqno truncated (only 4 of 8 bytes)
    // offset = 6 (after u16+1+u16+1 = 6 bytes for start/end fields)
    let mut buf = Vec::new();
    buf.write_u16::<LE>(1).unwrap(); // start_len
    buf.push(b'a'); // start key
    buf.write_u16::<LE>(1).unwrap(); // end_len
    buf.push(b'z'); // end key
    buf.extend_from_slice(&[0x01, 0x00, 0x00, 0x00]); // 4 bytes of seqno (need 8)

    assert_rt_decode_error(buf, "seqno", 6);
}

/// Exercises the `load_block` cache-miss and cache-hit paths for
/// `BlockType::RangeTombstone`, verifying that the dedicated RT metrics
/// counters are incremented instead of the data-block counters.
#[test]
#[cfg(feature = "metrics")]
fn load_block_range_tombstone_metrics() -> crate::Result<()> {
    use crate::{
        CompressionType,
        cache::Cache,
        range_tombstone::RangeTombstone,
        table::{block::BlockType, util::load_block},
    };
    use core::sync::atomic::Ordering::Relaxed;

    let dir = tempdir()?;
    let file = dir.path().join("table");

    // Build a table that contains a range tombstone block.
    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
    writer.write(InternalValue::from_components(
        b"a",
        b"v1",
        1,
        crate::ValueType::Value,
    ))?;
    writer.write(InternalValue::from_components(
        b"z",
        b"v2",
        2,
        crate::ValueType::Value,
    ))?;
    writer.write_range_tombstone(RangeTombstone::new(b"b".into(), b"y".into(), 3));
    #[expect(
        clippy::unwrap_used,
        reason = "finish() returns Some after writing data items"
    )]
    let (_, checksum) = writer.finish()?.unwrap();

    let metrics = Arc::new(crate::metrics::Metrics::default());

    let table = {
        // Recovery bypasses load_block() (reads via Block::from_file() directly),
        // so it intentionally does NOT increment block-load metrics — consistent
        // with how filter and index recovery reads are handled.
        let mut params = test_recover_params(file, checksum);
        params.cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
        #[cfg(feature = "metrics")]
        {
            params.metrics = metrics.clone();
        }
        Table::recover(params)?
    };

    let rt_handle = table
        .regions
        .range_tombstones
        .expect("table should have range tombstone block");

    let table_id = table.global_id();

    // Recovery does NOT increment block-load counters (bypasses load_block).
    assert_eq!(0, metrics.range_tombstone_block_load_io.load(Relaxed));

    // Use a fresh cache so the first load_block() call is a cache miss.
    let fresh_cache = Arc::new(Cache::with_capacity_bytes(10_000_000));

    // load_block cache miss → IO path
    let _block = load_block(
        table_id,
        &table.path,
        &table.file_accessor,
        &fresh_cache,
        &rt_handle,
        BlockType::RangeTombstone,
        CompressionType::None,
        None,
        None,
        #[cfg(zstd_any)]
        None,
        None,
        #[cfg(feature = "metrics")]
        &metrics,
    )?;

    assert_eq!(1, metrics.range_tombstone_block_load_io.load(Relaxed));
    assert_eq!(0, metrics.range_tombstone_block_load_cached.load(Relaxed));
    assert!(metrics.range_tombstone_block_io_requested.load(Relaxed) > 0);
    assert_eq!(0, metrics.data_block_load_io.load(Relaxed));

    // load_block cache hit (block was inserted into fresh_cache by previous call)
    let _block = load_block(
        table_id,
        &table.path,
        &table.file_accessor,
        &fresh_cache,
        &rt_handle,
        BlockType::RangeTombstone,
        CompressionType::None,
        None,
        None,
        #[cfg(zstd_any)]
        None,
        None,
        #[cfg(feature = "metrics")]
        &metrics,
    )?;

    assert_eq!(1, metrics.range_tombstone_block_load_io.load(Relaxed));
    assert_eq!(1, metrics.range_tombstone_block_load_cached.load(Relaxed));
    assert_eq!(0, metrics.data_block_load_cached.load(Relaxed));

    Ok(())
}

/// Regression test for <https://github.com/structured-world/coordinode-lsm-tree/issues/198>:
/// `load_block` must validate the cached block's `block_type` against the
/// caller's expected type.  Before the fix the cache-hit path returned the
/// block unconditionally, so a corrupted block handle pointing at a cached
/// block of the wrong type (e.g. a data block at an index block offset)
/// would slip through without an error.
#[test]
fn load_block_cache_hit_rejects_wrong_block_type() -> crate::Result<()> {
    use crate::{
        CompressionType,
        cache::Cache,
        table::{block::BlockType, util::load_block},
    };

    let dir = tempdir()?;
    let file = dir.path().join("table");

    // Build a minimal table with one data block.
    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
    writer.write(InternalValue::from_components(
        b"a",
        b"v1",
        1,
        crate::ValueType::Value,
    ))?;
    let (_, checksum) = writer
        .finish()?
        .expect("finish() returns Some after writing data items");

    #[cfg(feature = "metrics")]
    let metrics = Arc::new(crate::metrics::Metrics::default());

    let table = {
        let mut params = test_recover_params(file, checksum);
        params.cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
        #[cfg(feature = "metrics")]
        {
            params.metrics = metrics.clone();
        }
        Table::recover(params)?
    };

    let table_id = table.global_id();

    // The range-tombstone block handle is type-specific, but every table has a
    // TLI (top-level index) block whose handle we can reuse.  Load it first as
    // an Index block (correct type) so it lands in the cache.
    let tli_handle = table.regions.tli;
    let fresh_cache = Arc::new(Cache::with_capacity_bytes(10_000_000));

    let _block = load_block(
        table_id,
        &table.path,
        &table.file_accessor,
        &fresh_cache,
        &tli_handle,
        BlockType::Index,
        CompressionType::None,
        None,
        None,
        #[cfg(zstd_any)]
        None,
        None,
        #[cfg(feature = "metrics")]
        &metrics,
    )?;

    // Now request the same offset but claim it is a Data block.  The block is
    // already cached (as Index), so the cache-hit path must detect the type
    // mismatch and return `Error::InvalidTag`.
    let result = load_block(
        table_id,
        &table.path,
        &table.file_accessor,
        &fresh_cache,
        &tli_handle,
        BlockType::Data,
        CompressionType::None,
        None,
        None,
        #[cfg(zstd_any)]
        None,
        None,
        #[cfg(feature = "metrics")]
        &metrics,
    );

    assert!(
        matches!(&result, Err(crate::Error::InvalidTag(("BlockType", _)))),
        "expected InvalidTag for block type mismatch on cache hit, got Ok or wrong Err",
    );

    Ok(())
}

/// A read that recovers a data block from its Page-ECC parity, and confirms the
/// on-disk fault persists across a cache-bypassing re-read, must record the SST
/// in the heal sink for a healing recompaction. A clean read records nothing.
#[cfg(feature = "page_ecc")]
#[test]
fn load_block_records_heal_hint_on_persistent_ecc_correction() -> crate::Result<()> {
    use crate::{
        Cache, InternalValue,
        fs::StdFs,
        heal_hints::HealHints,
        table::{
            BlockHandle,
            block::{BlockType, EccParams, Header},
            util::load_block,
        },
    };

    let dir = tempdir()?;
    let file = dir.path().join("table");

    // Build a table whose data blocks carry RS(4,2) parity.
    let scheme = EccParams::RS_4_2;
    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_ecc(Some(scheme));
    for i in 0..200u32 {
        let key = format!("key{i:05}");
        writer.write(InternalValue::from_components(
            key.as_bytes(),
            b"value-payload-bytes",
            u64::from(i) + 1,
            crate::ValueType::Value,
        ))?;
    }
    #[expect(
        clippy::unwrap_used,
        reason = "finish() returns Some after writing items"
    )]
    let (_, checksum) = writer.finish()?.unwrap();

    #[cfg(feature = "metrics")]
    let metrics = Arc::new(crate::metrics::Metrics::default());
    let table = {
        let mut params = test_recover_params(file.clone(), checksum);
        params.cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
        #[cfg(feature = "metrics")]
        {
            params.metrics = metrics.clone();
        }
        Table::recover(params)?
    };

    let table_id = table.global_id();
    let compression = table.metadata.data_block_compression;

    // First data block.
    #[expect(clippy::unwrap_used, reason = "table has at least one data block")]
    let keyed = table.block_index.iter().next().unwrap()?;
    let handle = BlockHandle::new(keyed.offset(), keyed.size());

    // Clean read with an enabled sink: a non-corrected read records nothing.
    {
        let clean_sink = HealHints::default();
        clean_sink.set_enabled(true);
        let fresh_cache = Cache::with_capacity_bytes(10_000_000);
        let _block = load_block(
            table_id,
            &table.path,
            &table.file_accessor,
            &fresh_cache,
            &handle,
            BlockType::Data,
            compression,
            None,
            table.metadata.ecc_params,
            #[cfg(zstd_any)]
            None,
            Some(&clean_sink),
            #[cfg(feature = "metrics")]
            &metrics,
        )?;
        assert!(
            clean_sink.snapshot().is_empty(),
            "a clean read must not record a heal hint",
        );
    }

    // Flip one payload byte of the first data block so the read must repair it
    // via RS parity, and drop any cached fd so the re-read re-opens the tampered
    // file from disk (confirming the fault is persistent).
    let mut bytes = std::fs::read(&file)?;
    // `as usize` is a target-conditional truncation (only narrows on 32-bit
    // pointer widths); `allow`, not `expect`, so it stays clean on the 64-bit
    // host where clippy frames it purely as a portability note.
    #[allow(
        clippy::cast_possible_truncation,
        reason = "in-file block offset fits usize; only narrows on 32-bit targets"
    )]
    let pos = handle.offset().0 as usize + Header::MIN_LEN + 3;
    bytes[pos] ^= 0x80;
    std::fs::write(&file, &bytes)?;
    table.file_accessor.remove_for_table(&table_id);

    let sink = HealHints::default();
    sink.set_enabled(true);
    let fresh_cache = Cache::with_capacity_bytes(10_000_000);
    let block = load_block(
        table_id,
        &table.path,
        &table.file_accessor,
        &fresh_cache,
        &handle,
        BlockType::Data,
        compression,
        None,
        table.metadata.ecc_params,
        #[cfg(zstd_any)]
        None,
        Some(&sink),
        #[cfg(feature = "metrics")]
        &metrics,
    )?;
    assert_eq!(
        block.header.block_type,
        BlockType::Data,
        "repaired read still yields a valid data block",
    );
    assert_eq!(
        sink.snapshot(),
        vec![table_id],
        "a persistent ECC correction must queue the SST for healing",
    );

    // A DISABLED sink (auto_heal off) corrects on read but records nothing.
    table.file_accessor.remove_for_table(&table_id);
    let off_sink = HealHints::default(); // enabled == false
    let fresh_cache = Cache::with_capacity_bytes(10_000_000);
    let block = load_block(
        table_id,
        &table.path,
        &table.file_accessor,
        &fresh_cache,
        &handle,
        BlockType::Data,
        compression,
        None,
        table.metadata.ecc_params,
        #[cfg(zstd_any)]
        None,
        Some(&off_sink),
        #[cfg(feature = "metrics")]
        &metrics,
    )?;
    assert_eq!(
        block.header.block_type,
        BlockType::Data,
        "disabled auto-heal still returns repaired data",
    );
    assert!(
        off_sink.snapshot().is_empty(),
        "auto_heal off must not schedule a rewrite",
    );

    Ok(())
}

/// Writes a small-block ECC SST of `n` entries under `scheme` through `fs`,
/// returning its checksum. Shared by the in-place heal tests.
#[cfg(feature = "page_ecc")]
fn build_ecc_sst_for_heal(
    file: &std::path::Path,
    fs: Arc<dyn crate::fs::Fs>,
    scheme: crate::table::block::EccParams,
    n: u32,
) -> crate::Checksum {
    #[expect(
        clippy::expect_used,
        reason = "test setup; a panic is the failure signal"
    )]
    let mut writer = Writer::new(file.to_path_buf(), 0, 0, fs)
        .expect("open writer")
        .use_data_block_size(256)
        .use_ecc(Some(scheme));
    for i in 0..n {
        #[expect(clippy::expect_used, reason = "test setup")]
        writer
            .write(InternalValue::from_components(
                format!("key{i:05}").into_bytes(),
                b"value-payload-bytes".to_vec(),
                u64::from(i) + 1,
                crate::ValueType::Value,
            ))
            .expect("write");
    }
    #[expect(clippy::expect_used, reason = "finish() returns Some after writes")]
    let (_, checksum) = writer.finish().expect("finish").expect("non-empty");
    checksum
}

/// Recovers a `Table` for `file` through `fs` with fresh caches.
#[cfg(feature = "page_ecc")]
#[expect(
    clippy::expect_used,
    reason = "test setup; a panic is the failure signal"
)]
fn recover_table_on(
    file: &std::path::Path,
    checksum: crate::Checksum,
    fs: Arc<dyn crate::fs::Fs>,
) -> Table {
    let mut params = test_recover_params(file.to_path_buf(), checksum);
    params.cache = Arc::new(crate::Cache::with_capacity_bytes(10_000_000));
    params.fs = fs;
    Table::recover(params).expect("recover table")
}

/// The first data block's file offset in `table`.
#[cfg(feature = "page_ecc")]
fn first_data_block_offset(table: &Table) -> u64 {
    use crate::table::block_index::BlockIndex as _;
    let Some(keyed) = table.block_index.iter().find_map(Result::ok) else {
        panic!("a non-empty SST has at least one data block");
    };
    keyed.offset().0
}

/// A single-bit fault in a Page-ECC data block is healed in place AND the file
/// is restored byte-for-byte (RS / SEC-DED reconstruct the original payload, the
/// parity is recomputed deterministically). Exercises the SEC-DED branch of the
/// heal primitive.
#[cfg(feature = "page_ecc")]
#[test]
#[allow(
    clippy::cast_possible_truncation,
    reason = "in-file block offset fits usize; only narrows on 32-bit targets"
)]
fn heal_data_blocks_in_place_restores_a_secded_block_byte_for_byte() -> crate::Result<()> {
    use crate::table::block::{EccParams, Header};

    let dir = tempdir()?;
    let file = dir.path().join("table");
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(crate::fs::StdFs);
    let checksum = build_ecc_sst_for_heal(&file, Arc::clone(&fs), EccParams::Secded, 200);

    let original = std::fs::read(&file)?;
    let first_off =
        first_data_block_offset(&recover_table_on(&file, checksum, Arc::clone(&fs))) as usize;

    // Flip a single bit in the first data block's payload — SEC-DED corrects one
    // bit flip per word.
    let pos = first_off + Header::MIN_LEN + 3;
    let mut bytes = original.clone();
    if let Some(b) = bytes.get_mut(pos) {
        *b ^= 0x01;
    }
    std::fs::write(&file, &bytes)?;
    assert_ne!(bytes, original, "the seeded fault changed the file");

    let table = recover_table_on(&file, checksum, Arc::clone(&fs));
    let (report, attributable) =
        table.heal_data_blocks_in_place(crate::fs::SyncMode::Full, table.checksum());
    assert_eq!(report.blocks_healed_in_place, 1, "{report:?}");
    assert_eq!(report.uncorrectable_blocks, 0, "{report:?}");
    // The manifest digest is the HEALTHY file's, but the seeded fault changed
    // the bytes before the heal ran: the pre-heal digest cannot match, so the
    // mismatch is NOT attributable to this pass's writes.
    assert!(
        !attributable,
        "a pre-heal digest differing from the manifest must not attribute",
    );

    let healed = std::fs::read(&file)?;
    assert_eq!(
        healed, original,
        "SEC-DED in-place heal restores the block byte-for-byte",
    );
    Ok(())
}

/// The heal reports ATTRIBUTION (`true`) when the file's digest right before
/// its first write-back matches the manifest digest: parity-trailer rot with
/// the manifest digest recomputed over the ROTTED bytes (the shape a manifest
/// rebuild leaves behind) makes the post-heal mismatch provably the heal's
/// own work — the flag that lets the digest reconciliation restamp tables
/// whose authoritative content (deletion metadata, footer-less values) has
/// no semantic cross-check.
#[cfg(feature = "page_ecc")]
#[test]
#[allow(
    clippy::cast_possible_truncation,
    reason = "in-file block offset fits usize; only narrows on 32-bit targets"
)]
fn heal_data_blocks_in_place_attributes_a_matching_pre_heal_digest() -> crate::Result<()> {
    use crate::coding::Decode;
    use crate::table::block::{EccParams, Header};

    let dir = tempdir()?;
    let file = dir.path().join("table");
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(crate::fs::StdFs);
    let healthy = build_ecc_sst_for_heal(&file, Arc::clone(&fs), EccParams::RS_4_2, 200);
    let first_off =
        first_data_block_offset(&recover_table_on(&file, healthy, Arc::clone(&fs))) as usize;

    // Rot one parity-trailer byte of the first data block: the payload stays
    // checksum-clean, only the heal's trailer verification notices.
    let mut bytes = std::fs::read(&file)?;
    let Some(mut cursor) = bytes.get(first_off..) else {
        panic!("first data block within the file");
    };
    let header = Header::decode_from(&mut cursor)?;
    let trailer_pos =
        first_off + Header::header_len(header.block_type) + header.data_length as usize;
    let Some(slot) = bytes.get_mut(trailer_pos) else {
        panic!("parity trailer within the file");
    };
    *slot ^= 0xFF;
    std::fs::write(&file, &bytes)?;

    // The manifest digest covers the ROTTED bytes (a manifest rebuild admits
    // the degraded-but-readable file as-is), so the pre-heal probe matches.
    let rotted = crate::Checksum::from_raw(crate::repair::compute_table_checksum(&*fs, &file)?);
    let table = recover_table_on(&file, rotted, Arc::clone(&fs));
    let (report, attributable) =
        table.heal_data_blocks_in_place(crate::fs::SyncMode::Full, table.checksum());
    assert_eq!(
        report.blocks_healed_in_place, 1,
        "the rotted trailer is rebuilt in place: {report:?}",
    );
    assert_eq!(report.uncorrectable_blocks, 0, "{report:?}");
    assert!(
        attributable,
        "a pre-heal digest matching the manifest attributes the mismatch to \
         this pass's own verified corrections",
    );
    Ok(())
}

/// The streaming heal-digest prediction must be byte-identical to materializing
/// every correction and splicing it through
/// `compute_table_checksum_with_overrides`. The streaming path exists to bound
/// heap on broadly damaged tables (it never holds more than one corrected frame
/// at a time); this guards that the memory win did not change the predicted
/// digest, which the crash-recovery marker binds. The OOM failure mode itself
/// is not directly testable (it needs a multi-gigabyte damaged table).
#[cfg(feature = "page_ecc")]
#[test]
#[allow(
    clippy::cast_possible_truncation,
    reason = "in-file block offset fits usize; only narrows on 32-bit targets"
)]
fn predict_heal_streams_the_same_digest_as_materializing_corrections() -> crate::Result<()> {
    use crate::coding::Decode;
    use crate::table::block::{EccParams, Header};

    let dir = tempdir()?;
    let file = dir.path().join("table");
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(StdFs);
    let healthy = build_ecc_sst_for_heal(&file, Arc::clone(&fs), EccParams::RS_4_2, 200);
    let first_off =
        first_data_block_offset(&recover_table_on(&file, healthy, Arc::clone(&fs))) as usize;

    // Rot the first block's parity trailer → exactly one correction to splice.
    let mut bytes = std::fs::read(&file)?;
    let Some(mut cursor) = bytes.get(first_off..) else {
        panic!("first data block within the file");
    };
    let header = Header::decode_from(&mut cursor)?;
    let trailer_pos =
        first_off + Header::header_len(header.block_type) + header.data_length as usize;
    let Some(slot) = bytes.get_mut(trailer_pos) else {
        panic!("parity trailer within the file");
    };
    *slot ^= 0xFF;
    std::fs::write(&file, &bytes)?;

    let rotted = crate::Checksum::from_raw(crate::repair::compute_table_checksum(&*fs, &file)?);
    let table = recover_table_on(&file, rotted, Arc::clone(&fs));

    let transform = crate::table::util::build_block_transform(
        table.metadata.data_block_compression,
        table.encryption.as_deref(),
        table.metadata.ecc_params,
        #[cfg(zstd_any)]
        table.zstd_dictionary.as_deref(),
    )?;
    let fh = fs.open(&file, &crate::fs::FsOpenOptions::new().read(true))?;

    // Streamed prediction (one frame of heap at a time).
    let (streamed, offsets) = table.predict_heal_digest_and_offsets(fh.as_ref(), &transform, 0)?;

    // Materialized reference: gather every correction and splice them all at once.
    let mut corrections: Vec<(u64, Vec<u8>)> = Vec::new();
    for entry in table.block_index.iter() {
        let keyed = entry?;
        if let Some(c) = table.heal_correction_for_block(fh.as_ref(), &keyed, &transform)? {
            corrections.push(c);
        }
    }
    let materialized =
        crate::repair::compute_table_checksum_with_overrides(&*fs, &file, 0, &corrections)?;

    assert_eq!(
        streamed, materialized,
        "the streamed digest must equal the materialized-overrides digest",
    );
    assert_eq!(
        offsets.len(),
        corrections.len(),
        "one predicted offset per correction: {corrections:?}",
    );
    for (off, _) in &corrections {
        assert!(
            offsets.contains(off),
            "every correction offset ({off}) is in the predicted set",
        );
    }
    // Sanity: the rot did produce exactly the one trailer correction.
    assert_eq!(corrections.len(), 1, "exactly one block was rotted");
    Ok(())
}

/// When the corrected block's write-back fails (a failing `Fs`), the heal does
/// not count it as healed and records it as an uncorrectable finding so it is
/// left for block salvage rather than silently lost.
#[cfg(feature = "page_ecc")]
#[test]
#[allow(
    clippy::cast_possible_truncation,
    reason = "in-file block offset fits usize; only narrows on 32-bit targets"
)]
fn heal_data_blocks_in_place_reports_a_block_whose_write_back_fails() -> crate::Result<()> {
    use crate::fs::{Fault, FaultFs, FaultOp, FaultRule, StdFs};
    use crate::io::ErrorKind;
    use crate::table::block::{EccParams, Header};

    let dir = tempdir()?;
    let file = dir.path().join("table");
    let fault = FaultFs::new(StdFs);
    let injector = fault.injector();
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(fault);

    let checksum = build_ecc_sst_for_heal(&file, Arc::clone(&fs), EccParams::RS_4_2, 200);

    let first_off =
        first_data_block_offset(&recover_table_on(&file, checksum, Arc::clone(&fs))) as usize;

    // RS-recoverable single-byte fault: the read recovers it (so heal returns a
    // corrected frame), then the write-back is what fails.
    let pos = first_off + Header::MIN_LEN + 3;
    let mut bytes = std::fs::read(&file)?;
    if let Some(b) = bytes.get_mut(pos) {
        *b ^= 0x80;
    }
    std::fs::write(&file, &bytes)?;

    // The rot leaves the file differing from the (healthy) manifest checksum, so
    // this is the restorative heal path: its FIRST write is the crash-recovery
    // marker sidecar. Skip that write so the marker lands, then fail every
    // subsequent write, so the heal reads + recovers the block and its write-back
    // is what errors.
    injector.arm(FaultRule::new(FaultOp::Write, Fault::Error(ErrorKind::Other)).skip(1));

    let table = recover_table_on(&file, checksum, fs);
    let (report, _) = table.heal_data_blocks_in_place(crate::fs::SyncMode::Full, table.checksum());
    assert_eq!(
        report.blocks_healed_in_place, 0,
        "a failed write-back heals nothing: {report:?}",
    );
    assert!(
        report.uncorrectable_blocks >= 1,
        "the failed write-back is reported, not silently dropped: {report:?}",
    );
    assert!(
        report
            .errors
            .iter()
            .any(|e| matches!(e, crate::scrub::ScrubError::UncorrectableBlock { .. })),
        "the finding is an UncorrectableBlock: {report:?}",
    );
    Ok(())
}

/// A non-ECC SST carries no parity, so an in-place heal finds nothing to
/// reconstruct: every block reads back as "no recognized parity", nothing is
/// written, and no block is reported as healed or uncorrectable.
#[cfg(feature = "page_ecc")]
#[test]
fn heal_data_blocks_in_place_is_a_noop_on_a_non_ecc_sst() -> crate::Result<()> {
    let dir = tempdir()?;
    let file = dir.path().join("table");
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(crate::fs::StdFs);
    // No ECC (no `use_ecc`): blocks carry no parity trailer.
    let mut writer = Writer::new(file.clone(), 0, 0, Arc::clone(&fs))?.use_data_block_size(256);
    for i in 0..200u32 {
        writer.write(InternalValue::from_components(
            format!("key{i:05}").into_bytes(),
            b"value-payload-bytes".to_vec(),
            u64::from(i) + 1,
            crate::ValueType::Value,
        ))?;
    }
    let Some((_, checksum)) = writer.finish()? else {
        panic!("non-empty SST");
    };

    let table = recover_table_on(&file, checksum, fs);
    let (report, _) = table.heal_data_blocks_in_place(crate::fs::SyncMode::Full, table.checksum());
    assert!(
        report.blocks_scanned > 0,
        "the walk inspected blocks: {report:?}"
    );
    assert_eq!(
        report.blocks_healed_in_place, 0,
        "no parity means nothing to heal: {report:?}",
    );
    assert_eq!(report.uncorrectable_blocks, 0, "{report:?}");
    assert!(report.errors.is_empty(), "{report:?}");
    Ok(())
}

/// If the read+write handle for the heal cannot even be opened (a read-only
/// replica, restrictive permissions, a failing `Fs`), the pass must NOT return
/// a silent healthy report with zero blocks scanned: it records the failed
/// open AND falls back to the read-only scrub, so the table's integrity is
/// still checked and real corruption still surfaces.
#[cfg(feature = "page_ecc")]
#[test]
fn heal_data_blocks_in_place_reports_when_the_file_cannot_be_opened() -> crate::Result<()> {
    use crate::fs::{Fault, FaultFs, FaultOp, FaultRule, StdFs};
    use crate::io::ErrorKind;
    use crate::table::block::{EccParams, Header};

    let dir = tempdir()?;
    let file = dir.path().join("table");
    let fault = FaultFs::new(StdFs);
    let injector = fault.injector();
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(fault);

    let checksum = build_ecc_sst_for_heal(&file, Arc::clone(&fs), EccParams::RS_4_2, 200);
    let table = recover_table_on(&file, checksum, Arc::clone(&fs));

    // Wreck the first data block's whole payload (header intact, far beyond
    // the RS(4,2) budget) so the read-only fallback has real corruption to
    // find and report.
    {
        use crate::table::block_index::BlockIndex as _;
        let Some(keyed) = table.block_index.iter().find_map(Result::ok) else {
            panic!("the SST has at least one data block");
        };
        let base = usize::try_from(keyed.offset().0).unwrap_or(usize::MAX);
        let mut bytes = std::fs::read(&file)?;
        let Some(payload) = bytes.get_mut(base + Header::MIN_LEN..base + keyed.size() as usize)
        else {
            panic!("block payload range within the file");
        };
        for b in payload {
            *b ^= 0xFF;
        }
        std::fs::write(&file, &bytes)?;
    }

    // Fail exactly the heal's read+write open (the first open after arming);
    // the read-only fallback may reopen freely afterwards.
    injector.arm(FaultRule::new(FaultOp::Open, Fault::Error(ErrorKind::Other)).once());

    let (report, _) = table.heal_data_blocks_in_place(crate::fs::SyncMode::Full, table.checksum());
    assert!(
        report.blocks_scanned >= 1,
        "the read-only fallback still scans the table: {report:?}",
    );
    assert_eq!(
        report.blocks_healed_in_place, 0,
        "nothing is healed without a writable file: {report:?}",
    );
    assert!(
        report.uncorrectable_blocks >= 1,
        "the fallback scrub reports the seeded corruption: {report:?}",
    );
    assert!(!report.is_ok(), "corruption fails the pass: {report:?}");
    assert!(
        report
            .errors
            .iter()
            .any(|e| matches!(e, crate::scrub::ScrubError::BlockIndexUnreadable { .. })),
        "the failed read+write open is still reported: {report:?}",
    );
    Ok(())
}

/// End-to-end corruption test: tamper on-disk `seqno#kv_max` so it exceeds
/// `seqno#max`, then verify that `ParsedMeta::load_with_handle` rejects the
/// file with an `InvalidData` error.
///
/// Covers the validation path in `validated_kv_seqno` via the real on-disk
/// deserialization pipeline (not just the unit-level helper).
#[test]
#[expect(
    clippy::expect_used,
    reason = "test invariants: key and value patterns must exist in the meta block"
)]
#[expect(
    clippy::indexing_slicing,
    reason = "test fixture: deliberate slice operations on controlled meta block bytes"
)]
fn meta_seqno_kv_max_corruption_returns_invalid_data() -> crate::Result<()> {
    use super::block::Header;
    use super::meta::ParsedMeta;
    use super::regions::ParsedRegions;
    use crate::coding::{Decode, Encode};
    use std::io::{Seek, Write};

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

    // Write a valid table with KV entries at seqnos 1..=5.
    // Both seqno#max and seqno#kv_max will be 5.
    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
    for (i, key) in (b'a'..=b'e').enumerate() {
        writer.write(InternalValue::from_components(
            [key],
            b"val",
            (i as u64) + 1,
            crate::ValueType::Value,
        ))?;
    }
    #[expect(
        clippy::unwrap_used,
        reason = "finish() returns Some after writing data items"
    )]
    let _ = writer.finish()?.unwrap();

    // Find the meta block region, tamper the seqno#kv_max value in the
    // payload, recompute the block checksum so the corruption reaches
    // the metadata validation layer (not caught by block checksum).
    {
        let mut f = std::fs::File::open(&file)?;
        let trailer = crate::sfa::Reader::from_reader(&mut f)?;
        let regions = ParsedRegions::parse_from_toc(trailer.toc())?;
        let meta_handle = regions.metadata;

        let raw_block =
            crate::file::read_exact(&f, *meta_handle.offset(), meta_handle.size() as usize)?;

        // Meta blocks carry the block_flags byte, so their header is
        // header_len(Meta), not the SST MIN_LEN.
        let header_len = Header::header_len(crate::table::block::BlockType::Meta);
        let payload = &raw_block[header_len..];

        // Find the seqno#kv_max value bytes in the payload and replace
        // with u64::MAX (exceeds seqno#max = 5).
        let needle = b"seqno#kv_max";
        let key_pos = payload
            .windows(needle.len())
            .position(|w| w == needle)
            .expect("seqno#kv_max key must be present in the meta block payload");

        // Meta entries are stored in a DataBlock with restart_interval = 1, so
        // keys are written as full keys followed by an InternalValue payload
        // (value_type, seqno, value length, etc.) encoded using varints.  We do
        // not rely on the exact field layout here; instead, we scan forward from
        // the end of the key string to find the first occurrence of the LE-encoded
        // u64 value in the payload.
        let search_start = key_pos + needle.len();
        let original_le = 5u64.to_le_bytes();
        let val_rel = payload[search_start..]
            .windows(original_le.len())
            .position(|w| w == original_le)
            .expect("original LE value must appear after the key");
        let val_offset_in_payload = search_start + val_rel;

        let mut tampered_payload = payload.to_vec();
        tampered_payload[val_offset_in_payload..val_offset_in_payload + 8]
            .copy_from_slice(&u64::MAX.to_le_bytes());

        // Rebuild the header with the correct checksum over the
        // tampered payload so Block::from_file accepts the block.
        let mut orig_header = Header::decode_from(&mut &raw_block[..header_len])?;
        orig_header.checksum = crate::Checksum::from_raw(crate::hash::hash128(&tampered_payload));
        let new_header = orig_header.encode_into_vec();

        // Write the tampered block back into the file at the meta
        // block's original offset.
        let mut wf = std::fs::OpenOptions::new().write(true).open(&file)?;
        wf.seek(std::io::SeekFrom::Start(*meta_handle.offset()))?;
        wf.write_all(&new_header)?;
        wf.write_all(&tampered_payload)?;
        wf.sync_all()?;
    }

    // Re-open the (now corrupted) file and attempt to load metadata.
    {
        let mut f = std::fs::File::open(&file)?;
        let trailer = crate::sfa::Reader::from_reader(&mut f)?;
        let regions = ParsedRegions::parse_from_toc(trailer.toc())?;

        let result = ParsedMeta::load_with_handle(&f, &regions.metadata, None, None);

        let err = result.expect_err("corrupted seqno#kv_max should cause an error");
        assert!(
            matches!(&err, crate::Error::Io(e) if e.kind() == crate::io::ErrorKind::InvalidData),
            "expected InvalidData, got: {err:?}",
        );
    }

    Ok(())
}

/// `meta_mid` and `meta` (TAIL) must encode the SAME `created_at`. The
/// writer was generating `unix_timestamp()` independently inside each
/// `write_meta_section` call, so MID and TAIL would observe slightly
/// different wall-clock values. After recovery via MID the table would
/// report a different creation time than after recovery via TAIL,
/// which silently shifts TTL / FIFO ordering depending on which copy
/// the reader fell back to.
#[test]
fn meta_mid_and_tail_have_identical_created_at() -> crate::Result<()> {
    use super::meta::ParsedMeta;
    use super::regions::ParsedRegions;

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

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
    for (i, key) in (b'a'..=b'e').enumerate() {
        writer.write(InternalValue::from_components(
            [key],
            b"val",
            (i as u64) + 1,
            crate::ValueType::Value,
        ))?;
    }
    #[expect(
        clippy::unwrap_used,
        reason = "finish() returns Some after writing data items"
    )]
    let _ = writer.finish()?.unwrap();

    let mut f = std::fs::File::open(&file)?;
    let trailer = crate::sfa::Reader::from_reader(&mut f)?;
    let regions = ParsedRegions::parse_from_toc(trailer.toc())?;

    let tail = ParsedMeta::load_with_handle(&f, &regions.metadata, None, None)?;
    let mid_handle = regions
        .metadata_mid
        .expect("writer must emit meta_mid alongside meta");
    let mid = ParsedMeta::load_with_handle(&f, &mid_handle, None, None)?;

    assert_eq!(
        tail.created_at, mid.created_at,
        "MID and TAIL meta copies must share an identical created_at \
         (writer must snapshot the timestamp once and pass it to both \
         write_meta_section calls; observed tail={:?} mid={:?})",
        tail.created_at, mid.created_at,
    );

    Ok(())
}

/// `meta_mid` and `meta` (TAIL) must encode the SAME `file_size`. The
/// writer was stamping MID with a 0 sentinel and the reader was
/// patching the value with `std::fs::metadata(path).len()` on MID
/// fallback — that path (a) bypasses the pluggable `Fs` backend (so
/// `Table::recover` would fail on MemFs / io_uring trees the moment it
/// touched the MID fallback branch), and (b) reported the entire
/// physical file length (including TOC, trailer, and the TAIL meta
/// block itself), while TAIL stores `self.meta.file_pos` taken before
/// any of those tail sections were written. Recovered tables therefore
/// reported wildly different sizes depending on which meta copy survived.
///
/// `self.meta.file_pos` is only ever incremented inside `spill_block()`
/// (data-block writes). The index/tli/filter/range-tombstone writes,
/// the MID meta block itself, the linked_blob_files / table_version /
/// meta_separator raw sections, and the TAIL meta block all leave it
/// unchanged. So the value is identical at MID and TAIL write time —
/// MID can encode it directly, no recovery-time patching, no
/// `std::fs::metadata` call.
#[test]
fn meta_mid_and_tail_have_identical_file_size() -> crate::Result<()> {
    use super::meta::ParsedMeta;
    use super::regions::ParsedRegions;

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

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
    for (i, key) in (b'a'..=b'e').enumerate() {
        writer.write(InternalValue::from_components(
            [key],
            b"val",
            (i as u64) + 1,
            crate::ValueType::Value,
        ))?;
    }
    #[expect(
        clippy::unwrap_used,
        reason = "finish() returns Some after writing data items"
    )]
    let _ = writer.finish()?.unwrap();

    let mut f = std::fs::File::open(&file)?;
    let trailer = crate::sfa::Reader::from_reader(&mut f)?;
    let regions = ParsedRegions::parse_from_toc(trailer.toc())?;

    let tail = ParsedMeta::load_with_handle(&f, &regions.metadata, None, None)?;
    let mid_handle = regions
        .metadata_mid
        .expect("writer must emit meta_mid alongside meta");
    let mid = ParsedMeta::load_with_handle(&f, &mid_handle, None, None)?;

    assert_eq!(
        tail.file_size, mid.file_size,
        "MID and TAIL meta copies must store an identical file_size \
         (both observe the same `self.meta.file_pos` because no \
         post-data section bumps it); observed tail={} mid={}",
        tail.file_size, mid.file_size,
    );
    assert_ne!(
        mid.file_size, 0,
        "MID file_size must not be the legacy 0 sentinel — that pushed \
         the recovery path through std::fs::metadata, which bypasses \
         the pluggable Fs backend"
    );

    Ok(())
}

/// `bloom_may_contain_key` with full (non-partitioned) filter delegates to
/// `bloom_may_contain_hash`. Both methods agree for full filters.
#[test]
fn bloom_may_contain_key_full_filter() -> crate::Result<()> {
    let items: Vec<InternalValue> = ["a", "c", "e"]
        .iter()
        .enumerate()
        .map(|(i, &k)| {
            InternalValue::from_components(k, "v", i as u64 + 1, crate::ValueType::Value)
        })
        .collect();

    test_with_table(
        &items,
        |table| {
            let hash_a = hash64(b"a");
            let hash_b = hash64(b"b");

            // Existing key: both methods must accept
            assert!(
                table.bloom_may_contain_key(b"a", hash_a)?,
                "bloom_may_contain_key must not reject existing key"
            );
            assert!(
                table.bloom_may_contain_key_hash(hash_a)?,
                "bloom_may_contain_key_hash must not reject existing key"
            );

            // For full filters, bloom_may_contain_key delegates to the same
            // hash-only path, so both methods return the same result.
            let key_result = table.bloom_may_contain_key(b"b", hash_b)?;
            let hash_result = table.bloom_may_contain_key_hash(hash_b)?;
            assert_eq!(
                key_result, hash_result,
                "full filter: key-based and hash-only should agree"
            );

            Ok(())
        },
        None,
        Some(|w: Writer| w.use_bloom_policy(BloomConstructionPolicy::BitsPerKey(10.0))),
    )
}

/// `bloom_may_contain_key` with partitioned filter seeks the correct partition
/// and returns Ok(false) for a key beyond all partition boundaries.
///
/// Contrast: `bloom_may_contain_key_hash` returns Ok(true) conservatively
/// for the same key because it cannot seek partitions by hash alone.
/// This is the core behavioral improvement introduced by this PR.
#[test]
fn bloom_may_contain_key_partitioned_filter() -> crate::Result<()> {
    let items: Vec<InternalValue> = (0u64..100)
        .map(|i| {
            let key = format!("key_{i:04}");
            InternalValue::from_components(key, "v", i + 1, crate::ValueType::Value)
        })
        .collect();

    test_with_table(
        &items,
        |table| {
            // Key that exists: both methods must accept
            let hash_exist = hash64(b"key_0050");
            assert!(
                table.bloom_may_contain_key(b"key_0050", hash_exist)?,
                "bloom must not reject existing key in partitioned filter"
            );

            // Key beyond all partitions: with a pinned partition index, key-based
            // seek finds no ceiling and must return Ok(false).
            // Note: pinned_filter_index is always loaded when filter_tli exists
            // (unconditional in Table::recover), so this is always the partition-aware path.
            let hash_beyond = hash64(b"zzz_beyond");
            assert!(
                !table.bloom_may_contain_key(b"zzz_beyond", hash_beyond)?,
                "key beyond all partitions should be rejected when partition index is available"
            );

            // Hash-only path always returns Ok(true) conservatively for partitioned filters
            assert!(
                table.bloom_may_contain_key_hash(hash_beyond)?,
                "hash-only bloom check should remain conservative for partitioned filters"
            );

            Ok(())
        },
        None,
        Some(|w: Writer| {
            w.use_bloom_policy(BloomConstructionPolicy::BitsPerKey(10.0))
                .use_partitioned_filter()
        }),
    )
}

/// Regression test for #194: two-level index scan stops prematurely when
/// `from_block_with_bounds` returns `Ok(None)` for a child partition whose
/// entries are all outside the requested `[lo, hi]` window.
///
/// We build a table with a partitioned (two-level) index containing multiple
/// child partitions and then iterate through the block index with bounds that
/// span several partitions. Both forward (`next`) and reverse (`next_back`)
/// directions are verified to yield the correct block handle sequences.
///
/// NOTE: The `Ok(None)` child path cannot be triggered with well-formed
/// block data regardless of `restart_interval` — `trim_back_to_upper_bound`
/// always restores a covering entry when the stack empties, so
/// `seek_upper_bound_cursor` returns `true`. The `Ok(None)` branch fires
/// only when `fill_stack` or `advance_upper_restart_interval` encounters
/// a corrupt/malformed block (empty stack after decode failure). The fix
/// is therefore a defensive guard; this test validates overall iteration
/// correctness through the two-level path.
#[test]
fn two_level_index_scan_skips_empty_child_partition() -> crate::Result<()> {
    use crate::ValueType::Value;
    use crate::table::block_index::{BlockIndex, BlockIndexIter};

    // Eight distinct keys, each gets its own data block (block_size=1 byte).
    // meta_partition_size=3 is a very small byte budget for partitioned index
    // metadata, so the index writer splits child partitions aggressively
    // (effectively on or before the first handle), yielding multiple child
    // partitions for this test.
    let items: Vec<InternalValue> = ["a", "b", "c", "d", "e", "f", "g", "h"]
        .iter()
        .enumerate()
        .map(|(i, k)| InternalValue::from_components(*k, format!("v{i}"), (i + 1) as u64, Value))
        .collect();

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

    let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
        .use_partitioned_index()
        .use_data_block_size(1)
        .use_meta_partition_size(3);

    for item in items.iter().cloned() {
        writer.write(item)?;
    }
    writer.finish()?;

    let table = {
        let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
        params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
        params.pin_filter = true;
        crate::Table::recover(params)?
    };

    assert!(
        table.regions.index.is_some(),
        "table must use partitioned (two-level) index",
    );
    assert!(
        table.metadata.index_block_count > 1,
        "table must have >1 index partitions, got {}",
        table.metadata.index_block_count,
    );

    // --- full scan without bounds: collect all block handles ---
    let all_handles: Vec<_> = {
        let it = table.block_index.iter();
        it.collect::<Result<Vec<_>, _>>()?
    };
    assert_eq!(
        all_handles.len(),
        items.len(),
        "full scan should yield one block handle per data block",
    );

    // --- forward scan with lo bound ---
    // Seek past the first partition(s) to exercise the case where earlier
    // child partitions are empty after applying bounds.
    {
        let mut it = table.block_index.iter();
        assert!(it.seek_lower(b"d", u64::MAX));
        let forward_keys: Vec<_> = it
            .map(|r| r.map(|h| h.end_key().to_vec()))
            .collect::<Result<Vec<_>, _>>()?;
        assert_eq!(
            forward_keys,
            vec![
                b"d".to_vec(),
                b"e".to_vec(),
                b"f".to_vec(),
                b"g".to_vec(),
                b"h".to_vec(),
            ],
            "forward scan from 'd' should yield exactly d..h",
        );
    }

    // --- backward scan with hi bound ---
    // seek_upper("e", 0) positions the back cursor at the first handle
    // whose end_key > "e", which is "f". Reverse iteration starts from
    // "f" and works down to "a".
    {
        let mut it = table.block_index.iter();
        assert!(it.seek_upper(b"e", 0));
        let mut backward_keys = Vec::new();
        while let Some(res) = it.next_back() {
            backward_keys.push(res?.end_key().to_vec());
        }
        assert_eq!(
            backward_keys,
            vec![
                b"f".to_vec(),
                b"e".to_vec(),
                b"d".to_vec(),
                b"c".to_vec(),
                b"b".to_vec(),
                b"a".to_vec(),
            ],
            "backward scan up to 'e' should yield f..a in reverse",
        );
    }

    // --- mixed forward + backward with both bounds ---
    {
        let mut it = table.block_index.iter();
        assert!(it.seek_lower(b"c", u64::MAX));
        assert!(it.seek_upper(b"f", 0));

        let mut forward_keys = vec![];
        let mut backward_keys = vec![];

        // Consume two from front
        if let Some(res) = it.next() {
            forward_keys.push(res?.end_key().to_vec());
        }
        if let Some(res) = it.next() {
            forward_keys.push(res?.end_key().to_vec());
        }

        // Consume from back
        while let Some(res) = it.next_back() {
            backward_keys.push(res?.end_key().to_vec());
        }

        // The block index is a sparse index: seek_upper positions the back
        // cursor at the first block whose end_key > hi, so next_back()
        // starts from "g" (the first handle past "f"), then works down
        // through "f" and "e" until the cursors meet.
        assert_eq!(forward_keys, vec![b"c".to_vec(), b"d".to_vec()]);
        assert_eq!(
            backward_keys,
            vec![b"g".to_vec(), b"f".to_vec(), b"e".to_vec()]
        );
        assert!(it.next().is_none(), "iterator should be exhausted");
    }

    Ok(())
}

#[test]
fn batch_get_empty_input_returns_empty_results() -> crate::Result<()> {
    let items = [crate::InternalValue::from_components(
        b"a",
        b"v",
        0,
        crate::ValueType::Value,
    )];
    test_with_table(
        &items,
        |table| {
            let r = table.batch_get(&[], SeqNo::MAX)?;
            assert!(r.is_empty(), "empty input must yield empty result vec");
            Ok(())
        },
        None,
        Some(|x| x),
    )
}

#[test]
#[expect(clippy::unwrap_used)]
fn batch_get_single_block_multiple_keys_returns_in_input_order() -> crate::Result<()> {
    // Three keys, all fall in the same data block (default block
    // size is much larger than the few bytes here).
    let items: Vec<_> = ["a", "b", "c"]
        .iter()
        .enumerate()
        .map(|(i, k)| {
            crate::InternalValue::from_components(
                k.as_bytes(),
                format!("val-{k}").as_bytes(),
                u64::try_from(i).expect("test fixture index fits in u64"),
                crate::ValueType::Value,
            )
        })
        .collect();

    test_with_table(
        &items,
        |table| {
            let batch: Vec<(&[u8], u64)> = vec![
                (b"a", hash64(b"a")),
                (b"b", hash64(b"b")),
                (b"c", hash64(b"c")),
            ];
            let results = table.batch_get(&batch, SeqNo::MAX)?;
            assert_eq!(results.len(), 3, "one result slot per input key");
            assert_eq!(&*results[0].as_ref().unwrap().value, b"val-a");
            assert_eq!(&*results[1].as_ref().unwrap().value, b"val-b");
            assert_eq!(&*results[2].as_ref().unwrap().value, b"val-c");
            Ok(())
        },
        None,
        Some(|x| x),
    )
}

#[test]
#[expect(clippy::unwrap_used)]
fn batch_get_keys_spread_across_blocks_return_correct_values() -> crate::Result<()> {
    // Force one item per data block via tiny block size +
    // rotate_every=1. Then a batch covering keys from different
    // blocks must produce the correct value for each key. This
    // test asserts CORRECTNESS only — the "block loaded at most
    // once for the entire batch" perf claim is a property of the
    // implementation, verifiable through the block cache's
    // hit-rate counters under metrics instrumentation, but
    // deliberately not asserted here (the test would need to
    // hook the cache to count loads, which would couple to
    // internal cache mechanics).
    let items: Vec<_> = (0u32..8)
        .map(|i| {
            let key = format!("key-{i:04}");
            let value = format!("val-{i:04}");
            crate::InternalValue::from_components(
                key.as_bytes(),
                value.as_bytes(),
                u64::from(i),
                crate::ValueType::Value,
            )
        })
        .collect();

    test_with_table(
        &items,
        |table| {
            // Pick 4 keys spread across the 8 blocks.
            let queries: Vec<(&[u8], u64)> = vec![
                (b"key-0000" as &[u8], hash64(b"key-0000")),
                (b"key-0002" as &[u8], hash64(b"key-0002")),
                (b"key-0005" as &[u8], hash64(b"key-0005")),
                (b"key-0007" as &[u8], hash64(b"key-0007")),
            ];
            let results = table.batch_get(&queries, SeqNo::MAX)?;
            assert_eq!(results.len(), 4);
            assert_eq!(&*results[0].as_ref().unwrap().value, b"val-0000");
            assert_eq!(&*results[1].as_ref().unwrap().value, b"val-0002");
            assert_eq!(&*results[2].as_ref().unwrap().value, b"val-0005");
            assert_eq!(&*results[3].as_ref().unwrap().value, b"val-0007");
            Ok(())
        },
        Some(1),
        Some(|writer: Writer| writer.use_data_block_size(64)),
    )
}

#[test]
#[expect(clippy::unwrap_used)]
fn batch_get_missing_keys_return_none_present_keys_return_some() -> crate::Result<()> {
    let items: Vec<_> = ["b", "d", "f"]
        .iter()
        .enumerate()
        .map(|(i, k)| {
            crate::InternalValue::from_components(
                k.as_bytes(),
                format!("val-{k}").as_bytes(),
                u64::try_from(i).expect("test fixture index fits in u64"),
                crate::ValueType::Value,
            )
        })
        .collect();

    test_with_table(
        &items,
        |table| {
            // Mix present and absent keys, sorted ascending.
            let batch: Vec<(&[u8], u64)> = vec![
                (b"a" as &[u8], hash64(b"a")), // absent (before any key)
                (b"b" as &[u8], hash64(b"b")), // present
                (b"c" as &[u8], hash64(b"c")), // absent (between b and d)
                (b"d" as &[u8], hash64(b"d")), // present
                (b"f" as &[u8], hash64(b"f")), // present (last key)
                (b"g" as &[u8], hash64(b"g")), // absent (after last key)
            ];
            let results = table.batch_get(&batch, SeqNo::MAX)?;
            assert_eq!(results.len(), 6);
            assert!(results[0].is_none(), "key 'a' is absent");
            assert_eq!(&*results[1].as_ref().unwrap().value, b"val-b");
            assert!(results[2].is_none(), "key 'c' is absent");
            assert_eq!(&*results[3].as_ref().unwrap().value, b"val-d");
            assert_eq!(&*results[4].as_ref().unwrap().value, b"val-f");
            assert!(results[5].is_none(), "key 'g' is absent");
            Ok(())
        },
        None,
        Some(|x| x),
    )
}

#[test]
fn batch_get_matches_per_key_get() -> crate::Result<()> {
    // Cross-check: for every input key, `batch_get` and a per-key
    // `get` loop must produce identical results. This is the
    // regression guard against the batch path diverging from the
    // single-key path on any edge case (bloom misses, seqno
    // skew, block boundaries).
    let items: Vec<_> = (0u32..20)
        .map(|i| {
            let key = format!("k-{i:03}");
            let value = format!("v-{i:03}");
            crate::InternalValue::from_components(
                key.as_bytes(),
                value.as_bytes(),
                u64::from(i),
                crate::ValueType::Value,
            )
        })
        .collect();

    test_with_table(
        &items,
        |table| {
            // Build a query batch with a mix of present, absent,
            // and out-of-range keys.
            let keys: Vec<Vec<u8>> = (0..25).map(|i| format!("k-{i:03}").into_bytes()).collect();
            let batch: Vec<(&[u8], u64)> = keys.iter().map(|k| (k.as_slice(), hash64(k))).collect();

            let batch_results = table.batch_get(&batch, SeqNo::MAX)?;
            let single_results: Vec<_> = batch
                .iter()
                .map(|&(k, h)| table.get(k, SeqNo::MAX, h))
                .collect::<crate::Result<Vec<_>>>()?;

            assert_eq!(batch_results.len(), single_results.len());
            for (i, (b, s)) in batch_results.iter().zip(&single_results).enumerate() {
                assert_eq!(
                    b,
                    s,
                    "batch/single divergence at index {i} (key={})",
                    String::from_utf8_lossy(&keys[i]),
                );
            }
            Ok(())
        },
        Some(2),
        Some(|writer: Writer| writer.use_data_block_size(96)),
    )
}

#[test]
fn batch_get_same_user_key_across_block_boundary_finds_older_visible_version() -> crate::Result<()>
{
    // Regression for the multi-block MVCC walk bug in batch_get.
    //
    // The bug: when batch_get's inner loop hits a key with
    // `key == block.end_key` AND `point_read` returns None
    // (no visible entry in this block), the loop advanced `p`
    // unconditionally — so the walk skipped to the NEXT batch
    // key without checking whether the SAME user key continues
    // into the NEXT block. `Table::get` handles this case via
    // `point_read_inner`'s end-key boundary check; the batch
    // path must mirror it.
    //
    // To trigger the bug we need:
    //   1. `forward_reader` lands at a block whose end_key
    //      equals some batched key K, and
    //   2. that block has no visible version of K at the query
    //      seqno, and
    //   3. the next block contains the visible version of K.
    //
    // Single-key fixtures don't reproduce: `forward_reader` is
    // seqno-aware enough to seek past a block that has no
    // visible entries for the lone passing key, so the iter
    // lands at block 1 directly. We need a SECOND batched key
    // earlier in the order to force the seek to land at
    // block 0 (which IS the block for that earlier key), so
    // the later batched key then exercises the equal-end-key /
    // None-point_read / "look in next block" path.
    //
    // Fixture: user keys "0" (one version at seqno=1) +
    // five versions of "a" (seqno 5 → 1), `rotate_every=3`.
    // Internal-key sort puts "0" before any "a"; the resulting
    // blocks are:
    //   block 0: [0@1, a@5, a@4]   end_key="a"
    //   block 1: [a@3, a@2, a@1]   end_key="a"
    //
    // Query batch = [("0", h0), ("a", ha)] at snapshot seqno=3.
    // forward_reader seeks to block 0 to satisfy "0".
    // The "a" then sees end_key="a" with no visible version
    // in block 0 (all seqnos ≥ 3) — the fix must keep the
    // walk going into block 1 where a@2 is visible.
    let items = [
        crate::InternalValue::from_components(b"0", b"zero", 1, crate::ValueType::Value),
        crate::InternalValue::from_components(b"a", b"5", 5, crate::ValueType::Value),
        crate::InternalValue::from_components(b"a", b"4", 4, crate::ValueType::Value),
        crate::InternalValue::from_components(b"a", b"3", 3, crate::ValueType::Value),
        crate::InternalValue::from_components(b"a", b"2", 2, crate::ValueType::Value),
        crate::InternalValue::from_components(b"a", b"1", 1, crate::ValueType::Value),
    ];

    test_with_table(
        &items,
        |table| {
            assert_eq!(2, table.metadata.data_block_count);

            let batch: Vec<(&[u8], u64)> = vec![(b"0", hash64(b"0")), (b"a", hash64(b"a"))];

            // snapshot seqno=3: visible seqnos < 3.
            //   "0" → 0@1 (only version, visible)
            //   "a" → a@2 (largest visible; a@5/4/3 are not)
            let results = table.batch_get(&batch, 3)?;
            assert_eq!(results.len(), 2);
            assert_eq!(
                &*results[0]
                    .as_ref()
                    .expect("0@1 must be found in block 0")
                    .value,
                b"zero",
            );
            assert_eq!(
                &*results[1]
                    .as_ref()
                    .expect("a@2 must be found via block 1")
                    .value,
                b"2",
                "batch_get must walk past block 0 (end_key=a, but all a-seqnos ≥3) \
                 into block 1 (end_key=a, seqnos 2 and 1) to find the visible version \
                 at snapshot 3",
            );

            // Sanity: cross-check against Table::get for both keys.
            let single_zero = table.get(b"0", 3, hash64(b"0"))?;
            let single_a = table.get(b"a", 3, hash64(b"a"))?;
            assert_eq!(
                results[0], single_zero,
                "batch_get must match Table::get for '0'"
            );
            assert_eq!(
                results[1], single_a,
                "batch_get must match Table::get for 'a'"
            );
            Ok(())
        },
        Some(3),
        Some(|x| x),
    )
}

/// Builds an SST from `items`, optionally with parallel block compression
/// (`parallel_threads`), applying `config` to the writer, then recovers it.
/// Returns the table plus the temp dir (kept alive for the table's lifetime).
#[cfg(all(test, feature = "parallel"))]
#[expect(clippy::unwrap_used, reason = "test code")]
fn build_and_recover(
    items: &[crate::InternalValue],
    parallel_threads: Option<usize>,
    config: impl Fn(Writer) -> Writer,
) -> crate::Result<(Table, tempfile::TempDir)> {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("table");

    let mut writer = config(Writer::new(path.clone(), 0, 0, Arc::new(StdFs))?);
    if let Some(threads) = parallel_threads {
        let spawner = Arc::new(crate::table::writer::RayonSpawner::with_threads(threads)?);
        writer = writer.use_parallel_compression(spawner, threads);
    }
    for item in items {
        writer.write(item.clone())?;
    }
    let (_, checksum) = writer.finish()?.unwrap();

    #[cfg(feature = "metrics")]
    let metrics = Arc::new(Metrics::default());
    let table = {
        #[cfg_attr(not(feature = "metrics"), expect(unused_mut))]
        let mut params = test_recover_params(path, checksum);
        #[cfg(feature = "metrics")]
        {
            params.metrics = metrics;
        }
        Table::recover(params)?
    };
    Ok((table, dir))
}

/// The parallel block-compression pipeline must produce an SST functionally
/// identical to the serial path: workers compress out of order, but the writer
/// drains and frames blocks strictly in submission order, so block boundaries,
/// scan order, contents and index entries are unchanged. (The on-disk data
/// section is in fact byte-identical; only the `created_at` metadata timestamp
/// varies between builds, so we compare recovered content rather than raw
/// bytes.) Checked across the encode + transform variations that flow through
/// the pipeline.
#[cfg(feature = "parallel")]
#[test]
fn parallel_compression_matches_serial_output() -> crate::Result<()> {
    // Enough keys, with small blocks, to force many data-block spills so the
    // pipeline genuinely reorders work across its 4 workers.
    let items: Vec<_> = (0u32..4000)
        .map(|i| {
            crate::InternalValue::from_components(
                format!("key{i:08}").as_bytes(),
                format!("value-{i}-some-payload-bytes").as_bytes(),
                u64::from(i),
                crate::ValueType::Value,
            )
        })
        .collect();

    let check = |config: &dyn Fn(Writer) -> Writer, label: &str| -> crate::Result<()> {
        let (serial, _ds) = build_and_recover(&items, None, config)?;
        let (parallel, _dp) = build_and_recover(&items, Some(4), config)?;

        // Identical block boundaries and item count.
        assert_eq!(
            serial.metadata.data_block_count, parallel.metadata.data_block_count,
            "{label}: data_block_count must match"
        );
        assert_eq!(
            serial.metadata.item_count, parallel.metadata.item_count,
            "{label}: item_count must match"
        );

        // Identical scan content and order.
        let s: Vec<_> = serial.iter().collect::<crate::Result<_>>()?;
        let p: Vec<_> = parallel.iter().collect::<crate::Result<_>>()?;
        assert_eq!(s.len(), items.len(), "{label}: all items must scan back");
        assert_eq!(s, p, "{label}: scan content/order must match serial");

        // Index resolves point reads identically (sampled across the key space).
        for i in (0..items.len()).step_by(137) {
            let key = format!("key{i:08}");
            let hash = hash64(key.as_bytes());
            assert_eq!(
                serial.get(key.as_bytes(), crate::SeqNo::MAX, hash)?,
                parallel.get(key.as_bytes(), crate::SeqNo::MAX, hash)?,
                "{label}: point read for {key} must match"
            );
        }
        Ok(())
    };

    check(&|w| w.use_data_block_size(256), "plain")?;
    check(
        &|w| w.use_data_block_size(256).use_seqno_in_index(true),
        "seqno_in_index",
    )?;
    #[cfg(feature = "lz4")]
    check(
        &|w| {
            w.use_data_block_size(256)
                .use_data_block_compression(CompressionType::Lz4)
        },
        "lz4",
    )?;

    Ok(())
}

#[test]
fn zone_map_section_roundtrips_one_entry_per_block() -> crate::Result<()> {
    // 200 keys with frequent block rotation force many data blocks, so the
    // section must carry several entries that survive write + reopen.
    let items: Vec<crate::InternalValue> = (0..200u32)
        .map(|i| {
            crate::InternalValue::from_components(
                format!("k{i:05}").into_bytes(),
                format!("v{i}").into_bytes(),
                0,
                crate::ValueType::Value,
            )
        })
        .collect();

    test_with_table(
        &items,
        |table| {
            let zm = &table.zone_map;
            assert!(!zm.is_empty(), "zone map should be populated when enabled");
            assert!(
                zm.len() >= 2,
                "rotation should yield several blocks, got {}",
                zm.len()
            );
            Ok(())
        },
        Some(20),
        Some(|w: Writer| w.use_zone_map(true)),
    )
}

#[test]
fn zone_map_corrupt_section_falls_back_instead_of_failing_open() -> crate::Result<()> {
    let dir = tempdir()?;
    let file = dir.path().join("table");

    let checksum = {
        let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_zone_map(true);
        for i in 0..200u32 {
            if i % 20 == 0 {
                writer.spill_block()?;
            }
            writer.write(crate::InternalValue::from_components(
                format!("k{i:05}").into_bytes(),
                b"v".to_vec(),
                0,
                crate::ValueType::Value,
            ))?;
        }
        writer.finish()?.expect("table written").1
    };

    let recover = || -> crate::Result<Table> {
        #[cfg(feature = "metrics")]
        let metrics = Arc::new(Metrics::default());
        #[cfg_attr(not(feature = "metrics"), expect(unused_mut))]
        let mut params = test_recover_params(file.clone(), checksum);
        #[cfg(feature = "metrics")]
        {
            params.metrics = metrics;
        }
        Table::recover(params)
    };

    // First open: the zone-map section is present and populated.
    let zm_handle = {
        let table = recover()?;
        assert!(!table.zone_map.is_empty(), "zone map should be populated");
        table.regions.zone_map.expect("zone-map section present")
    };

    // Corrupt one byte inside the zone-map section block on disk so its
    // checksum / AEAD rejects it on the next open.
    let mut bytes = std::fs::read(&file)?;
    let corrupt_at = usize::try_from(zm_handle.offset().0).expect("offset fits usize") + 4;
    *bytes
        .get_mut(corrupt_at)
        .expect("corruption offset within file") ^= 0xFF;
    std::fs::write(&file, &bytes)?;

    // Second open: a corrupt OPTIONAL zone-map is derived, non-authoritative
    // metadata — it must NOT fail table open. It degrades to no block-skip (an
    // empty map), and the rest of the table still loads intact.
    let table = recover()?;
    assert!(
        table.zone_map.is_empty(),
        "corrupt zone-map section should disable block-skip, not fail open"
    );
    assert_eq!(
        table.metadata.item_count, 200,
        "the rest of the table must still load with a corrupt zone map"
    );

    Ok(())
}

#[test]
fn zone_map_absent_without_policy() -> crate::Result<()> {
    let items: Vec<crate::InternalValue> = (0..50u32)
        .map(|i| {
            crate::InternalValue::from_components(
                format!("k{i:05}").into_bytes(),
                b"v".to_vec(),
                0,
                crate::ValueType::Value,
            )
        })
        .collect();

    test_with_table(
        &items,
        |table| {
            assert!(
                table.zone_map.is_empty(),
                "no zone map should be loaded without the policy"
            );
            Ok(())
        },
        None,
        None::<fn(Writer) -> Writer>,
    )
}

/// Helper: recover a freshly written table from `file` with default test config.
fn recover_test_table(file: &std::path::Path, checksum: Checksum) -> crate::Result<Table> {
    recover_test_table_with_id(file, checksum, 0)
}

fn recover_test_table_with_id(
    file: &std::path::Path,
    checksum: Checksum,
    table_id: TableId,
) -> crate::Result<Table> {
    #[cfg(feature = "metrics")]
    let metrics = Arc::new(Metrics::default());
    let mut params = test_recover_params(file.to_path_buf(), checksum);
    params.table_id = table_id;
    #[cfg(feature = "metrics")]
    {
        params.metrics = metrics;
    }
    Table::recover(params)
}

/// A retired table reclaims its `.restrict-bound` sidecar on drop, so a
/// tight-space-restricted SST that is later compacted away does not leak an
/// orphan sidecar beside its deleted file (the recovery scan would eventually
/// sweep it, but leaving it is a leak until then).
#[test]
fn dropping_a_deleted_table_removes_its_restrict_bound_sidecar() -> crate::Result<()> {
    use crate::fs::Fs;

    let dir = tempdir()?;
    let file = dir.path().join("0");

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
    writer.write(crate::InternalValue::from_components(
        b"k",
        b"v",
        1,
        crate::ValueType::Value,
    ))?;
    let (_, checksum) = writer.finish()?.expect("table written");

    // Publish a restrict-bound sidecar beside the SST, as a tight-space slice would.
    let fs = StdFs;
    crate::restrict_bound::write(&fs, &file, None, 0, b"k", crate::fs::SyncMode::Normal)?;
    let sidecar = crate::restrict_bound::sidecar_path(&file);
    assert!(fs.exists(&sidecar)?, "sidecar present before retirement");

    // Retire the table and drop its last handle.
    let table = recover_test_table(&file, checksum)?;
    table.mark_as_deleted();
    drop(table);

    assert!(
        !fs.exists(&sidecar)?,
        "retiring the table must reclaim its restrict-bound sidecar, not leak it",
    );
    Ok(())
}

#[test]
fn delete_bitmap_section_round_trips() -> crate::Result<()> {
    let dir = tempdir()?;
    let file = dir.path().join("table");

    let keys: [&[u8]; 8] = [b"a", b"b", b"c", b"d", b"e", b"f", b"g", b"h"];
    // Row positions follow write order: rows 0, 2, 5 are keys a, c, f.
    let deleted_rows = [0u32, 2, 5];

    // A delete-bitmap SST must carry a zone map (per-block row counts power the
    // positional masking; recovery enforces this).
    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_zone_map(true);
    for key in keys {
        writer.write(crate::InternalValue::from_components(
            key,
            b"v",
            1,
            crate::ValueType::Value,
        ))?;
    }
    for &row in &deleted_rows {
        writer.delete_bitmap_mut().insert(row);
    }
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;
    assert!(
        table.regions.delete_bitmap.is_some(),
        "delete-bitmap section must be present when rows are deleted"
    );
    let dv = table.delete_bitmap();
    assert_eq!(dv.len(), deleted_rows.len() as u64);
    for row in 0..8u32 {
        assert_eq!(
            dv.contains(row),
            deleted_rows.contains(&row),
            "row {row} membership mismatch after reopen"
        );
    }
    Ok(())
}

#[test]
fn delete_bitmap_section_absent_when_no_deletes() -> crate::Result<()> {
    let dir = tempdir()?;
    let file = dir.path().join("table");

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
    writer.write(crate::InternalValue::from_components(
        b"a",
        b"v",
        1,
        crate::ValueType::Value,
    ))?;
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;
    assert!(
        table.regions.delete_bitmap.is_none(),
        "no delete-bitmap section when the segment has no deletes"
    );
    assert!(table.delete_bitmap().is_empty());
    Ok(())
}

/// A columnar table's zone map records one entry PER stored column (the
/// user-key column AND the value column), each with its own range, and the
/// verifier re-derives them from the decoded blocks so the honest table passes
/// its forgery cross-check. Regression: the writer once stamped a single
/// synthetic key-range column for every columnar block, hiding the non-key
/// columns from `can_skip_block` pushdown.
#[cfg(feature = "columnar")]
#[test]
fn columnar_zone_map_records_per_column_stats_and_round_trips() -> crate::Result<()> {
    use crate::table::columnar::{COL_USER_KEY, COL_VALUE};

    let dir = tempdir()?;
    let file = dir.path().join("table");

    // Distinct keys AND distinct values so the user-key and value columns have
    // genuinely different ranges: a single synthetic key-range column could not
    // describe the value column.
    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    for i in 0..32u32 {
        let key = format!("k{i:04}").into_bytes();
        let value = format!("v{i:04}").into_bytes();
        writer.write(crate::InternalValue::from_components(
            key,
            value,
            1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;
    assert!(table.metadata.columnar, "written as a columnar table");
    assert!(!table.zone_map.is_empty(), "zone map populated");

    // Every data block records BOTH the user-key column (0) and the value
    // column (3), each with its own range — not a single synthetic column.
    let mut saw_value_column = false;
    for handle in table.block_index.iter() {
        let handle = handle?;
        let stats = table
            .zone_map
            .columns_for(handle.offset().0)
            .expect("every data block has a zone-map entry");
        let ids: Vec<u32> = stats.iter().map(|s| s.column_id).collect();
        assert!(
            ids.contains(&u32::from(COL_USER_KEY)),
            "the user-key column is recorded, got ids {ids:?}",
        );
        if let Some(v) = stats.iter().find(|s| s.column_id == u32::from(COL_VALUE)) {
            saw_value_column = true;
            assert!(v.min <= v.max, "the value column's range is ordered");
            assert!(!v.min.is_empty(), "the value column carries a real range");
        }
    }
    assert!(
        saw_value_column,
        "at least one block records the non-key value column's stats",
    );

    // The writer's per-column map and the verifier's re-derivation agree, so the
    // forgery cross-check accepts the honest table.
    if let Err((gate, e)) = table.verify_reconcile_gates(None, false) {
        panic!("the honest columnar table must pass every gate, {gate:?} refused it: {e}");
    }
    Ok(())
}

#[cfg(feature = "columnar")]
#[test]
fn delete_bitmap_masks_rows_in_columnar_scan() -> crate::Result<()> {
    use crate::table::columnar::{
        COL_SEQNO, COL_USER_KEY, COL_VALUE, COL_VALUE_TYPE, column_batch_to_entries,
    };

    let dir = tempdir()?;
    let file = dir.path().join("table");

    let n = 64u32;
    // Row positions follow write (= key) order: these are keys k0003/k0010/k0050.
    let deleted = [3u32, 10, 50];

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    for i in 0..n {
        let key = format!("k{i:04}").into_bytes();
        writer.write(crate::InternalValue::from_components(
            key,
            b"v",
            1,
            crate::ValueType::Value,
        ))?;
    }
    for &row in &deleted {
        writer.delete_bitmap_mut().insert(row);
    }
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;
    let batches =
        table.columnar_scan(&[COL_USER_KEY, COL_SEQNO, COL_VALUE_TYPE, COL_VALUE], None)?;

    let mut got: Vec<Vec<u8>> = Vec::new();
    for batch in &batches {
        for entry in column_batch_to_entries(batch)? {
            got.push(entry.key.user_key.to_vec());
        }
    }

    let expected: Vec<Vec<u8>> = (0..n)
        .filter(|i| !deleted.contains(i))
        .map(|i| format!("k{i:04}").into_bytes())
        .collect();
    assert_eq!(
        got, expected,
        "deleted row positions must be masked out of the columnar scan"
    );
    Ok(())
}

/// A tight-space-restricted columnar SST has its consumed prefix hole-punched
/// (those data blocks read as zeros). The columnar scan must skip the punched
/// blocks instead of decoding them, mask the straddling block's sub-bound rows,
/// and keep positional delete-bitmap mapping intact across the skipped blocks.
#[cfg(feature = "columnar")]
#[test]
#[expect(clippy::unwrap_used)]
fn restricted_columnar_scan_skips_punched_prefix_and_masks_sub_bound_rows() -> crate::Result<()> {
    use crate::table::columnar::{
        COL_SEQNO, COL_USER_KEY, COL_VALUE, COL_VALUE_TYPE, column_batch_to_entries,
    };
    use std::io::{Seek, SeekFrom, Write as _};

    let dir = tempdir()?;
    let file = dir.path().join("table");

    let n = 256u32;
    let keys: Vec<Vec<u8>> = (0..n).map(|i| format!("k{i:04}").into_bytes()).collect();
    // One deleted row inside the (to-be-punched) prefix, one in the live
    // suffix: the live one only masks correctly if the scan advances the
    // positional row base across the skipped punched blocks.
    let deleted_punched = 1u32;
    let deleted_live = 250u32;

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true)
        .use_data_block_size(256);
    for key in &keys {
        writer.write(crate::InternalValue::from_components(
            key.as_slice(),
            b"v",
            1,
            crate::ValueType::Value,
        ))?;
    }
    writer.delete_bitmap_mut().insert(deleted_punched);
    writer.delete_bitmap_mut().insert(deleted_live);
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;
    let handles: Vec<_> = table
        .block_index
        .iter()
        .collect::<crate::Result<Vec<_>>>()?;
    assert!(handles.len() >= 3, "need several blocks to punch a prefix");

    // Bound = two keys past the first block's end, so exactly one block is
    // punched and the first live block STRADDLES the bound (one sub-bound row).
    let first_end = handles[0].end_key().to_vec();
    let j = keys.iter().position(|k| *k == first_end).unwrap();
    let bound_idx = j + 2;
    assert!(
        bound_idx < deleted_live as usize,
        "the live deleted row must stay above the bound"
    );
    let bound = crate::UserKey::from(keys[bound_idx].as_slice());

    let punch_off = table.punch_offset_for(&bound)?;
    assert_eq!(
        punch_off,
        handles[1].offset().0,
        "exactly the first block is below the bound"
    );

    let restricted = table.with_restriction(bound);

    // Simulate the tight-space hole punch: the consumed prefix reads as zeros.
    let mut f = std::fs::OpenOptions::new().write(true).open(&file)?;
    f.seek(SeekFrom::Start(0))?;
    f.write_all(&vec![0u8; usize::try_from(punch_off).unwrap()])?;
    f.sync_all()?;

    let batches =
        restricted.columnar_scan(&[COL_USER_KEY, COL_SEQNO, COL_VALUE_TYPE, COL_VALUE], None)?;
    let mut got: Vec<Vec<u8>> = Vec::new();
    for batch in &batches {
        for entry in column_batch_to_entries(batch)? {
            got.push(entry.key.user_key.to_vec());
        }
    }

    let expected: Vec<Vec<u8>> = (bound_idx..n as usize)
        .filter(|&i| i != deleted_live as usize)
        .map(|i| keys[i].clone())
        .collect();
    assert_eq!(
        got, expected,
        "scan must start at the bound, mask the straddling block's sub-bound \
         rows, and keep delete positions aligned across the punched prefix"
    );
    Ok(())
}

/// The heal unshare copy must reproduce the source's ACTUAL hole pattern, not
/// the logical restriction: a tight-space slice that committed but failed its
/// restriction-sidecar write deliberately leaves the restricted SST unpunched
/// (punching without the sidecar would force a lossy conservative bound on a
/// later manifest-loss repair). A heal detach of such a table must therefore
/// copy the intact prefix verbatim — introducing holes below the logical
/// bound would punch the file without its sidecar. Actually-punched extents
/// still stay holes (no re-allocation on the near-full tight-space disk).
#[cfg(feature = "page_ecc")]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn unshare_for_heal_preserves_unpunched_blocks_of_a_restricted_table() -> crate::Result<()> {
    use crate::fs::{Fs, MemFs};
    use crate::table::block_index::BlockIndex;
    use std::sync::Arc;

    let memfs = Arc::new(MemFs::new());
    let fs: Arc<dyn Fs> = memfs.clone();
    let root = std::path::absolute("/db")?;
    memfs.create_dir_all(&root)?;

    let build = |name: &str| -> crate::Result<Table> {
        let path = root.join(name);
        let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(&fs))?.use_data_block_size(256);
        for i in 0..256u32 {
            writer.write(crate::InternalValue::from_components(
                format!("k{i:04}").into_bytes(),
                b"v",
                1,
                crate::ValueType::Value,
            ))?;
        }
        let (_, checksum) = writer.finish()?.expect("table written");
        #[cfg(feature = "metrics")]
        let metrics = Arc::new(Metrics::default());
        let mut params = test_recover_params(path, checksum);
        params.descriptor_table = None;
        params.fs = Arc::clone(&fs);
        #[cfg(feature = "metrics")]
        {
            params.metrics = metrics;
        }
        Table::recover(params)
    };

    let all_zero = |path: &std::path::Path, off: u64, len: usize| -> crate::Result<bool> {
        let file = fs.open(path, &crate::fs::FsOpenOptions::new().read(true))?;
        let bytes = crate::file::read_exact(&*file, off, len)?;
        Ok(bytes.iter().all(|&b| b == 0))
    };

    // COMMITTED-BUT-UNPUNCHED: restricted view over an intact file. The heal
    // copy must keep every byte (no new holes).
    let table = build("0")?;
    let handles: Vec<_> = table
        .block_index
        .iter()
        .collect::<crate::Result<Vec<_>>>()?;
    assert!(handles.len() >= 3, "need several blocks to restrict over");
    let bound = handles.get(1).expect("second block").end_key().clone();
    let (b0_off, b0_len) = {
        let h = handles.first().expect("first block");
        (h.offset().0, h.size() as usize)
    };
    let restricted = table.with_restriction(bound.clone());
    let source = fs.open(
        &restricted.path,
        &crate::fs::FsOpenOptions::new().read(true),
    )?;
    restricted
        .unshare_for_heal(&*source, crate::fs::SyncMode::Normal)
        .expect("unshare succeeds");
    assert!(
        !all_zero(&restricted.path, b0_off, b0_len)?,
        "an unpunched restricted table's prefix blocks must be copied verbatim, \
         not turned into holes the (missing) sidecar does not cover"
    );

    // ACTUALLY PUNCHED: the same restriction whose prefix data blocks were
    // hole-punched. The heal copy must keep those extents as holes.
    let table = build("1")?;
    let handles: Vec<_> = table
        .block_index
        .iter()
        .collect::<crate::Result<Vec<_>>>()?;
    let punch = table.punch_offset_for(&bound)?;
    for h in &handles {
        if h.offset().0 < punch {
            memfs.punch_hole(&table.path, h.offset().0, u64::from(h.size()))?;
        }
    }
    let (p0_off, p0_len) = {
        let h = handles.first().expect("first block");
        (h.offset().0, h.size() as usize)
    };
    let restricted = table.with_restriction(bound);
    let source = fs.open(
        &restricted.path,
        &crate::fs::FsOpenOptions::new().read(true),
    )?;
    restricted
        .unshare_for_heal(&*source, crate::fs::SyncMode::Normal)
        .expect("unshare succeeds");
    assert!(
        all_zero(&restricted.path, p0_off, p0_len)?,
        "a genuinely punched extent stays a hole in the heal copy"
    );
    Ok(())
}

#[test]
fn copy_on_write_strategy_suppresses_the_delete_bitmap_section() -> crate::Result<()> {
    let dir = tempdir()?;
    let file = dir.path().join("table");

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
        .delete_strategy(crate::config::DeleteStrategy::CopyOnWrite);
    for key in [b"a".as_ref(), b"b", b"c"] {
        writer.write(crate::InternalValue::from_components(
            key,
            b"v",
            1,
            crate::ValueType::Value,
        ))?;
    }
    // Mark a row deleted; copy-on-write drops rows instead of masking, so it must
    // not persist a bitmap section even though a position was marked.
    writer.delete_bitmap_mut().insert(1);
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;
    assert!(
        table.regions.delete_bitmap.is_none(),
        "copy-on-write must not persist a delete-bitmap section"
    );
    assert!(table.delete_bitmap().is_empty());
    Ok(())
}

#[cfg(feature = "columnar")]
#[test]
fn delete_bitmap_masks_rows_in_range_scan() -> crate::Result<()> {
    let dir = tempdir()?;
    let file = dir.path().join("table");

    let n = 64u32;
    // Row positions follow write (= key) order: keys k0003 / k0010 / k0050.
    let deleted = [3u32, 10, 50];

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    for i in 0..n {
        let key = format!("k{i:04}").into_bytes();
        writer.write(crate::InternalValue::from_components(
            key,
            b"v",
            1,
            crate::ValueType::Value,
        ))?;
    }
    for &row in &deleted {
        writer.delete_bitmap_mut().insert(row);
    }
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;
    // A full forward range scan goes through the columnar reconstruction +
    // positional mask; deleted positions must never be yielded.
    let got: Vec<Vec<u8>> = table
        .range_iter(..)
        .map(|r| r.map(|kv| kv.key.user_key.to_vec()))
        .collect::<crate::Result<Vec<_>>>()?;

    let expected: Vec<Vec<u8>> = (0..n)
        .filter(|i| !deleted.contains(i))
        .map(|i| format!("k{i:04}").into_bytes())
        .collect();
    assert_eq!(
        got, expected,
        "deleted row positions must be masked out of the range scan"
    );
    Ok(())
}

#[cfg(feature = "columnar")]
#[test]
fn delete_bitmap_masks_deleted_key_in_point_read() -> crate::Result<()> {
    let dir = tempdir()?;
    let file = dir.path().join("table");

    let n = 64u32;
    // Row positions follow write (= key) order: keys k0003 / k0010 / k0050.
    let deleted = [3u32, 10, 50];

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    for i in 0..n {
        let key = format!("k{i:04}").into_bytes();
        writer.write(crate::InternalValue::from_components(
            key,
            b"v",
            1,
            crate::ValueType::Value,
        ))?;
    }
    for &row in &deleted {
        writer.delete_bitmap_mut().insert(row);
    }
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;

    // A deleted key reads as absent; a live key is still found.
    for i in 0..n {
        let key = format!("k{i:04}").into_bytes();
        let got = table.get(&key, SeqNo::MAX, hash64(&key))?;
        if deleted.contains(&i) {
            assert!(got.is_none(), "deleted key {i} must read as absent");
        } else {
            assert!(got.is_some(), "live key {i} must be found");
        }
    }
    Ok(())
}

#[cfg(feature = "columnar")]
#[test]
fn write_columnar_batch_stores_value_subcolumns_and_round_trips() -> crate::Result<()> {
    use crate::table::columnar::{Column, TypeTag, entries_to_column_batch, unframe_value_cells};

    let dir = tempdir()?;
    let file = dir.path().join("table");

    // A consumer batch: the intrinsic columns for two sorted keys, with the
    // single value column replaced by two value sub-columns (a fixed-4 + a bytes).
    // Per-row seqnos are 0 (the ingest contract; the table assigns the seqno).
    let mut batch = entries_to_column_batch(&[
        crate::InternalValue::from_components(b"k0", b"ignored", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"k1", b"ignored", 0, crate::ValueType::Value),
    ])?;
    batch.columns.pop();
    batch.columns.push(Column {
        column_id: 3,
        type_tag: TypeTag::Fixed(4),
        validity: None,
        data: vec![1, 0, 0, 0, 2, 0, 0, 0].into(),
    });
    let mut bytes_data = Vec::new();
    for off in [0u32, 2, 5] {
        bytes_data.extend_from_slice(&off.to_le_bytes());
    }
    bytes_data.extend_from_slice(b"aabbb");
    batch.columns.push(Column {
        column_id: 4,
        type_tag: TypeTag::Bytes,
        validity: None,
        data: bytes_data.into(),
    });

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    writer.write_columnar_batch(&batch, &crate::comparator::default_comparator())?;
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;
    assert!(table.metadata.columnar, "segment must be columnar");

    // Point reads reconstruct the framed value, which unframes to the original
    // sub-cells.
    let tags = [TypeTag::Fixed(4), TypeTag::Bytes];
    let v0 = table
        .get(b"k0", SeqNo::MAX, hash64(b"k0"))?
        .expect("k0 present");
    assert_eq!(
        unframe_value_cells(v0.value.as_ref(), &tags)?,
        vec![&[1, 0, 0, 0][..], &b"aa"[..]],
    );
    let v1 = table
        .get(b"k1", SeqNo::MAX, hash64(b"k1"))?
        .expect("k1 present");
    assert_eq!(
        unframe_value_cells(v1.value.as_ref(), &tags)?,
        vec![&[2, 0, 0, 0][..], &b"bbb"[..]],
    );
    Ok(())
}

/// A positional delete-bitmap masks whole rows of a value-sub-column segment in
/// both the point and the projection read paths: the mask is value-agnostic, so
/// deleting by position hides each row's intrinsic key and every sub-column
/// while survivors keep their reconstructed sub-cells and projected bytes.
#[cfg(feature = "columnar")]
#[test]
fn delete_bitmap_masks_value_subcolumns_in_point_and_projection_reads() -> crate::Result<()> {
    use crate::fs::SyncMode;
    use crate::table::columnar::{Column, TypeTag, entries_to_column_batch, unframe_value_cells};
    use crate::table::delete_bitmap::DeleteBitmap;

    let dir = tempdir()?;
    let src = dir.path().join("src");
    let out = dir.path().join("out");

    // Five rows; the value is split into a fixed-4 (col 3) and a bytes (col 4)
    // sub-column. fixed values 10,20,30,40,50; bytes "a","bb","ccc","dddd","eeeee".
    let fixed: [u32; 5] = [10, 20, 30, 40, 50];
    let payloads: [&[u8]; 5] = [b"a", b"bb", b"ccc", b"dddd", b"eeeee"];
    let mut batch = entries_to_column_batch(
        &(0..5u32)
            .map(|i| {
                crate::InternalValue::from_components(
                    format!("k{i}").into_bytes(),
                    b"x",
                    0, // ingest contract: per-row seqno is 0
                    crate::ValueType::Value,
                )
            })
            .collect::<Vec<_>>(),
    )?;
    batch.columns.pop();
    batch.columns.push(Column {
        column_id: 3,
        type_tag: TypeTag::Fixed(4),
        validity: None,
        data: fixed.iter().flat_map(|v| v.to_le_bytes()).collect(),
    });
    let mut bytes_data = Vec::new();
    let mut acc = 0u32;
    bytes_data.extend_from_slice(&acc.to_le_bytes());
    for p in payloads {
        acc += u32::try_from(p.len()).unwrap();
        bytes_data.extend_from_slice(&acc.to_le_bytes());
    }
    for p in payloads {
        bytes_data.extend_from_slice(p);
    }
    batch.columns.push(Column {
        column_id: 4,
        type_tag: TypeTag::Bytes,
        validity: None,
        data: bytes_data.into(),
    });

    let mut writer = Writer::new(src.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    writer.write_columnar_batch(&batch, &crate::comparator::default_comparator())?;
    let (_, checksum) = writer.finish()?.expect("source written");
    let source = recover_test_table(&src, checksum)?;

    // Mask rows 1 and 3 by position (value-agnostic) and relocate.
    let mut bitmap = DeleteBitmap::new();
    bitmap.insert(1);
    bitmap.insert(3);
    let out_checksum =
        source.relocate_columnar_with_deletes(&out, &StdFs, 1, &bitmap, SyncMode::Normal)?;
    let relocated = recover_test_table_with_id(&out, out_checksum, 1)?;

    // Point path: masked rows read absent; survivors reconstruct their sub-cells.
    let tags = [TypeTag::Fixed(4), TypeTag::Bytes];
    for i in 0..5u32 {
        let key = format!("k{i}").into_bytes();
        let got = relocated.get(&key, SeqNo::MAX, hash64(&key))?;
        if i == 1 || i == 3 {
            assert!(got.is_none(), "masked row {i} must read absent");
        } else {
            let v = got.expect("survivor present");
            assert_eq!(
                unframe_value_cells(v.value.as_ref(), &tags)?,
                vec![&fixed[i as usize].to_le_bytes()[..], payloads[i as usize]],
                "survivor {i} sub-cells",
            );
        }
    }

    // Projection path: a scan over sub-column 3 yields the survivors only, with
    // their fixed bytes; the masked rows never appear.
    let batches = relocated.columnar_scan(&[3], None)?;
    let mut col3 = Vec::new();
    let mut rows = 0u32;
    for b in &batches {
        assert!(
            b.columns.iter().all(|c| c.column_id == 3),
            "projection decodes only sub-column 3",
        );
        rows += b.row_count;
        for c in b.columns.iter().filter(|c| c.column_id == 3) {
            col3.extend_from_slice(&c.data);
        }
    }
    assert_eq!(rows, 3, "two of five rows masked out of the projection");
    let want: Vec<u8> = [10u32, 30, 50]
        .iter()
        .flat_map(|v| v.to_le_bytes())
        .collect();
    assert_eq!(col3, want, "projected fixed bytes are the survivors");

    Ok(())
}

#[cfg(feature = "columnar")]
#[test]
fn write_columnar_batch_accounts_tombstones_seqno_bounds_and_restart_locator() -> crate::Result<()>
{
    use crate::config::{LocatorPolicyEntry, LocatorPrecision};
    use crate::table::columnar::{Column, TypeTag, entries_to_column_batch};

    let dir = tempdir()?;
    let file = dir.path().join("t");
    // Distinct keys with mixed value types plus one fixed value sub-column,
    // written with seqno-in-index and restart-precision locator enabled so the
    // tombstone / weak-tombstone accounting, the seqno-bounds, and the
    // restart-precision locator branches all run.
    let mut batch = entries_to_column_batch(&[
        crate::InternalValue::from_components(b"k0", b"v", 0, crate::ValueType::Value),
        crate::InternalValue::from_components(b"k1", b"", 0, crate::ValueType::Tombstone),
        crate::InternalValue::from_components(b"k2", b"", 0, crate::ValueType::WeakTombstone),
    ])?;
    batch.columns.pop();
    batch.columns.push(Column {
        column_id: 3,
        type_tag: TypeTag::Fixed(2),
        validity: None,
        data: vec![1, 1, 2, 2, 3, 3].into(),
    });

    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true)
        .use_seqno_in_index(true)
        .use_locator(LocatorPolicyEntry::Enabled {
            precision: LocatorPrecision::Restart,
            block_id_bits: None,
            slot_bits: None,
        });
    writer.write_columnar_batch(&batch, &crate::comparator::default_comparator())?;
    let (_, checksum) = writer.finish()?.expect("table written");
    let table = recover_test_table(&file, checksum)?;

    // is_tombstone covers Tombstone + WeakTombstone; weak count is just the latter.
    assert_eq!(table.metadata.tombstone_count, 2, "two tombstone-kind rows");
    assert_eq!(table.metadata.weak_tombstone_count, 1, "one weak tombstone");
    // The seqno-bounds section is written and loaded: every ingested row carries
    // seqno 0, so the recovered bounds are exactly (0, 0). This proves the
    // use_seqno_in_index path ran rather than relying on the point read alone
    // (which can succeed through the normal index path regardless).
    assert_eq!(
        table.metadata.seqnos,
        (0, 0),
        "columnar ingest writes local seqno bounds of (0, 0)",
    );
    assert!(
        table.get(b"k0", SeqNo::MAX, hash64(b"k0"))?.is_some(),
        "the live row reads back",
    );
    Ok(())
}

/// The index SEPARATOR cross-check is a gate of the reconcile pass, fed from
/// the walk's decode. A separator lowered in BOTH mirrors keeps the handle list
/// sorted, the mirrors equal and the section tiling intact — every byte-level
/// and structural check reads clean — yet the binary search then routes keys in
/// `(forged_separator, real_last_key]` to the wrong block. The pass must refuse
/// the table and name the separator gate, not merely fail somewhere.
#[test]
fn reconcile_gates_lowered_tli_separator_rejects_with_separators_gate() -> crate::Result<()> {
    let dir = tempdir()?;
    let file = dir.path().join("t");

    // Small blocks so the table has several data blocks (the forge lowers the
    // first block's separator and needs a next one to stay below).
    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_data_block_size(128);
    for i in 0u64..64 {
        writer.write(crate::InternalValue::from_components(
            alloc::format!("key-{i:04}").into_bytes(),
            alloc::format!("v{i:04}").into_bytes(),
            i + 1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;
    if let Err((gate, e)) = table.verify_reconcile_gates(None, false) {
        panic!("intact separators must pass every gate, {gate:?} refused it: {e}");
    }

    crate::test_forge::forge_tli_mirrors_lower_first_separator(&file, 0, None)?;

    let table = recover_test_table(&file, checksum)?;
    let result = table.verify_reconcile_gates(None, false);
    assert!(
        matches!(
            result,
            Err((
                crate::table::ReconcileGate::Separators,
                crate::Error::InvalidHeader(
                    "tli separator does not match the addressed block's decoded last key"
                )
            ))
        ),
        "a lowered separator must be rejected by the separator gate, got {:?}",
        result.map_err(|(gate, e)| (gate, e.to_string())),
    );
    Ok(())
}

/// The per-KV gate is fed from the entries the walk materialized rather than
/// decoding the block a second time for itself. It must still catch a footer
/// whose stored digest no longer matches the entry bytes behind a re-stamped
/// block checksum — the block-level walk reads that clean — and the pass must
/// name THAT gate, not merely fail somewhere.
#[test]
fn reconcile_gates_stale_kv_footer_rejects_with_kv_checksums_gate() -> crate::Result<()> {
    use crate::runtime_config::{ChecksumAlgorithm, KvChecksumPolicy};

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

    // Uncompressed: the forge patches the payload in place.
    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
        .use_kv_checksums(KvChecksumPolicy::AllLevels, ChecksumAlgorithm::Xxh3_64);
    for i in 0u64..40 {
        writer.write(crate::InternalValue::from_components(
            alloc::format!("key-{i:04}").into_bytes(),
            alloc::format!("v{i:04}").into_bytes(),
            i + 1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;
    if let Err((gate, e)) = table.verify_reconcile_gates(None, false) {
        panic!("an intact footer must pass every gate, {gate:?} refused it: {e}");
    }

    crate::test_forge::forge_stale_kv_footer(&file)?;

    let table = recover_test_table(&file, checksum)?;
    let result = table.verify_reconcile_gates(None, false);
    assert!(
        matches!(
            result,
            Err((
                crate::table::ReconcileGate::KvChecksums,
                crate::Error::ChecksumMismatch { .. }
            ))
        ),
        "a stale per-KV footer must be rejected by the per-KV gate, got {:?}",
        result.map_err(|(gate, e)| (gate, e.to_string())),
    );
    Ok(())
}

/// `verify_locator` must reject a locator re-stamped to resolve a key to a
/// block OTHER than the one holding its newest version: every byte-level
/// check reads clean, but `point_read_inner` trusts the answer and can return
/// a stale value from the mispointed block without falling back to the index.
/// An intact locator passes; a redirected one fails.
#[test]
fn verify_locator_rejects_a_redirected_key_mapping() -> crate::Result<()> {
    use crate::config::{LocatorPolicyEntry, LocatorPrecision};
    use crate::table::locator::{LocatorSpec, build_locator_section};

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

    // Small blocks so several data blocks (several block_ids) exist.
    let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
        .use_data_block_size(128)
        .use_locator(LocatorPolicyEntry::Enabled {
            precision: LocatorPrecision::Block,
            block_id_bits: None,
            slot_bits: None,
        });
    for i in 0u64..200 {
        writer.write(crate::InternalValue::from_components(
            format!("key-{i:04}").into_bytes(),
            format!("v{i:04}").into_bytes(),
            i + 1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, checksum) = writer.finish()?.expect("table written");

    let table = recover_test_table(&file, checksum)?;
    // Intact locator verifies clean.
    if let Err((gate, e)) = table.verify_reconcile_gates(None, false) {
        panic!("an intact locator must pass every gate, {gate:?} refused it: {e}");
    }

    // Rebuild the SAME ribbon (same key set, same widths → byte-identical
    // length) but redirect the FIRST key to a DIFFERENT block ordinal.
    let mut triples: Vec<(u64, u64, u64)> = Vec::new();
    let block_count = table.block_index.iter().count() as u64;
    assert!(block_count >= 2, "need multiple blocks to redirect between");
    let mut seen = crate::HashSet::default();
    for (ordinal, handle) in table.block_index.iter().enumerate() {
        let handle = handle?;
        let block_handle = crate::table::BlockHandle::new(handle.offset(), handle.size());
        let entries = table.decode_block_entries(&block_handle)?;
        for e in entries {
            let uk = e.key.user_key.to_vec();
            if seen.insert(uk.clone()) {
                triples.push((crate::hash::hash64(&uk), ordinal as u64, 0));
            }
        }
    }
    // Redirect the first key to a different existing block.
    let orig = triples[0].1;
    triples[0].1 = if orig == 0 { block_count - 1 } else { 0 };
    let spec = LocatorSpec {
        precision: LocatorPrecision::Block,
        block_id_bits: None,
        slot_bits: None,
    };
    let forged = build_locator_section(&triples, spec).expect("forged section builds");

    crate::test_forge::forge_replace_section_payload(&file, b"locator", &forged, None)?;

    let table = recover_test_table(&file, checksum)?;
    let result = table.verify_reconcile_gates(None, false);
    assert!(
        matches!(
            result,
            Err((
                crate::table::ReconcileGate::Locator,
                crate::Error::InvalidHeader(_)
            ))
        ),
        "a redirected locator must be rejected by the locator gate, got {:?}",
        result.map_err(|(gate, e)| (gate, e.to_string())),
    );
    Ok(())
}

#[cfg(feature = "columnar")]
#[test]
fn write_columnar_batch_on_an_empty_batch_writes_no_block() -> crate::Result<()> {
    use crate::table::columnar::{Column, TypeTag};

    let dir = tempdir()?;
    let file = dir.path().join("t");
    // An empty batch (zero rows) carrying the intrinsic columns plus a value
    // sub-column writes no block and returns no last key.
    let empty = crate::table::columnar::ColumnBatch {
        row_count: 0,
        columns: vec![
            Column {
                column_id: 0,
                type_tag: TypeTag::Bytes,
                validity: None,
                data: 0u32.to_le_bytes().to_vec().into(),
            },
            Column {
                column_id: 1,
                type_tag: TypeTag::Fixed(8),
                validity: None,
                data: Vec::new().into(),
            },
            Column {
                column_id: 2,
                type_tag: TypeTag::Fixed(1),
                validity: None,
                data: Vec::new().into(),
            },
            Column {
                column_id: 3,
                type_tag: TypeTag::Fixed(4),
                validity: None,
                data: Vec::new().into(),
            },
        ],
    };
    let mut writer = Writer::new(file, 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    assert!(
        writer
            .write_columnar_batch(&empty, &crate::comparator::default_comparator())?
            .is_none(),
        "an empty batch yields no last key",
    );
    // Finishing must also produce no SST: a None last key alone does not prove
    // the "writes no block" contract (a buggy writer could emit a table yet
    // still return None).
    assert!(
        writer.finish()?.is_none(),
        "an empty batch must not produce an SST",
    );
    Ok(())
}

/// A TRANSIENT locator-section read during a SALVAGE open must PROPAGATE, not
/// degrade the section to `None`. The degradation sets `rebuildable_section_degraded`,
/// which makes `salvage_attempt` read a delete-free table as possibly hiding
/// deletion metadata and fail the whole SST (`FeatureUnsupported`), dropping an
/// otherwise-salvageable table instead of retrying the retryable read. A non-salvage
/// open keeps the best-effort accelerator behavior (degrade to the sorted-index
/// path), so this only applies in salvage mode (#80).
#[test]
fn recover_salvage_propagates_a_transient_locator_read() -> crate::Result<()> {
    use crate::config::{LocatorPolicyEntry, LocatorPrecision};
    use crate::fs::{Fault, FaultFs, FaultOp, FaultRule};
    use crate::io::ErrorKind;

    let dir = tempdir()?;
    let sst = dir.path().join("0");
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(StdFs);
    {
        let mut w = Writer::new(sst.clone(), 0, 0, Arc::clone(&fs))?
            .use_data_block_size(128)
            .use_locator(LocatorPolicyEntry::Enabled {
                precision: LocatorPrecision::Entry,
                block_id_bits: None,
                slot_bits: None,
            });
        for i in 0..256u32 {
            w.write(crate::InternalValue::from_components(
                format!("k{i:05}").into_bytes(),
                format!("v{i}").into_bytes(),
                u64::from(i) + 1,
                crate::ValueType::Value,
            ))?;
        }
        assert!(w.finish()?.is_some(), "the SST is non-empty");
    }

    // Resolve the locator section's byte offset from a clean (Live) open.
    let checksum = crate::Checksum::from_raw(crate::repair::compute_table_checksum(&*fs, &sst)?);
    let loc_offset = {
        let live = {
            let mut params = test_recover_params(sst.clone(), checksum);
            params.fs = Arc::clone(&fs);
            Table::recover(params)?
        };
        live.regions
            .locator
            .expect("the multi-block SST carries a locator section")
            .offset()
            .0
    };

    // Fault ONLY the positioned read at the locator offset, once, with a genuine
    // transient kind.
    let fault = FaultFs::new(StdFs);
    let injector = fault.injector();
    injector.arm(
        FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Interrupted))
            .at_offset(loc_offset)
            .once(),
    );
    let faulted: Arc<dyn crate::fs::Fs> = Arc::new(fault);

    let result = {
        let mut params = test_recover_params(sst, checksum);
        params.fs = Arc::clone(&faulted);
        Table::recover_inner(
            params,
            RecoveryMode::Salvage {
                expected_id: None,
                prefer_mid_meta: false,
            },
        )
    };
    injector.clear();

    assert!(
        matches!(&result, Err(crate::Error::Io(e)) if e.kind() == ErrorKind::Interrupted),
        "a transient locator read in salvage mode must propagate (not degrade to a \
         section-degraded open that later fails as FeatureUnsupported): {result:?}",
    );
    Ok(())
}

/// A PERSISTENT (non-retryable) filter-index read during a SALVAGE open must
/// DEGRADE the rebuildable section, not propagate — the destination writer
/// re-derives the filter from the recovered keys, so a bad sector under the
/// partitioned filter's top-level index must not cost the whole recoverable
/// table. Only a TRANSIENT read propagates (repair retries). Mirrors the
/// seqno-bounds / zone-map / delete-bitmap / locator loaders.
#[test]
fn recover_salvage_degrades_a_persistent_filter_index_read() -> crate::Result<()> {
    use crate::fs::{Fault, FaultFs, FaultOp, FaultRule};
    use crate::io::ErrorKind;

    let dir = tempdir()?;
    let sst = dir.path().join("0");
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(StdFs);
    {
        let mut w = Writer::new(sst.clone(), 0, 0, Arc::clone(&fs))?
            .use_data_block_size(128)
            .use_partitioned_filter();
        for i in 0..256u32 {
            w.write(crate::InternalValue::from_components(
                format!("k{i:05}").into_bytes(),
                format!("v{i}").into_bytes(),
                u64::from(i) + 1,
                crate::ValueType::Value,
            ))?;
        }
        assert!(w.finish()?.is_some(), "the SST is non-empty");
    }

    // Resolve the partitioned filter's top-level-index (`filter_tli`) byte offset
    // from a clean (Live) open.
    let checksum = crate::Checksum::from_raw(crate::repair::compute_table_checksum(&*fs, &sst)?);
    let tli_offset = {
        let live = {
            let mut params = test_recover_params(sst.clone(), checksum);
            params.fs = Arc::clone(&fs);
            Table::recover(params)?
        };
        live.regions
            .filter_tli
            .expect("the partitioned-filter SST carries a filter_tli section")
            .offset()
            .0
    };

    // Fault the positioned read at the filter-index offset with a PERSISTENT
    // (non-retryable) kind.
    let fault = FaultFs::new(StdFs);
    let injector = fault.injector();
    injector.arm(
        FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Other))
            .at_offset(tli_offset)
            .once(),
    );
    let faulted: Arc<dyn crate::fs::Fs> = Arc::new(fault);

    let result = {
        let mut params = test_recover_params(sst, checksum);
        params.fs = Arc::clone(&faulted);
        Table::recover_inner(
            params,
            RecoveryMode::Salvage {
                expected_id: None,
                prefer_mid_meta: false,
            },
        )
    };
    injector.clear();

    let recovered = match result {
        Ok(table) => table,
        Err(e) => panic!(
            "a persistent filter-index read in salvage mode must degrade the rebuildable \
             section and recover the table, not propagate and drop it: {e:?}",
        ),
    };
    // A bare success would also pass if the injected fault never fired; the
    // degradation flag proves the faulted filter-index load was actually
    // routed through the salvage degrade arm.
    assert!(
        recovered.salvage_degraded_a_rebuildable_section(),
        "the recovered table must report the degraded rebuildable section",
    );
    Ok(())
}