coordinode-lsm-tree 7.0.0

use super::*;
use crate::io::Cursor;
use crate::io::{ByteOrder, LittleEndian, ReadBytesExt};
use crate::{
    coding::Decode,
    comparator::default_comparator,
    table::{
        Block, BlockHandle, BlockOffset, IndexBlock, KeyedBlockHandle,
        block::{BlockType, Header, ParsedItem, Trailer},
    },
};
use varint_rs::VarintReader;

fn make_handles(count: usize) -> Vec<KeyedBlockHandle> {
    (0..count)
        .map(|i| {
            let key = format!("adj:out:vertex-0001:edge-{i:04}");
            KeyedBlockHandle::new(
                key.into(),
                i as u64,
                BlockHandle::new(BlockOffset((i as u64) * 4096), 4096),
            )
        })
        .collect()
}

fn make_index_block(restart_interval: u8) -> IndexBlock {
    let handles = make_handles(16);
    let bytes =
        IndexBlock::encode_into_vec_with_restart_interval(&handles, restart_interval).unwrap();
    IndexBlock::new(Block {
        data: bytes.into(),
        header: Header::test_dummy(BlockType::Index),
    })
}

fn make_corrupted_index_block_with_invalid_first_restart_head() -> IndexBlock {
    let handles = make_handles(16);
    let mut bytes = IndexBlock::encode_into_vec_with_restart_interval(&handles, 8).unwrap();
    let key_len_offset = first_restart_head_key_len_offset(&bytes);
    bytes[key_len_offset] = 0xFF;
    bytes[key_len_offset + 1] = 0xFF;
    bytes[key_len_offset + 2] = 0x03;
    IndexBlock::new(Block {
        data: bytes.into(),
        header: Header::test_dummy(BlockType::Index),
    })
}

fn first_restart_head_key_len_offset(bytes: &[u8]) -> usize {
    let mut cursor = Cursor::new(bytes);
    let marker = cursor.read_u8().unwrap();
    assert_eq!(marker, 0);
    let _ = BlockHandle::decode_from(&mut cursor).unwrap();
    let _ = cursor.read_u64_varint().unwrap();
    usize::try_from(cursor.position()).unwrap()
}

fn make_corrupted_index_block_with_invalid_binary_index_offset() -> IndexBlock {
    let handles = make_handles(16);
    let mut bytes = IndexBlock::encode_into_vec_with_restart_interval(&handles, 8).unwrap();

    let trailer_probe = IndexBlock::new(Block {
        data: bytes.clone().into(),
        header: Header::test_dummy(BlockType::Index),
    });
    let trailer_offset = Trailer::new(&trailer_probe.inner).trailer_offset();
    let binary_index_offset_pos = trailer_offset + 1 + 1 + core::mem::size_of::<u32>();
    #[expect(
        clippy::cast_possible_truncation,
        reason = "test block sizes stay well below u32::MAX"
    )]
    let invalid_binary_index_offset = (bytes.len() as u32).saturating_add(1);
    LittleEndian::write_u32(
        &mut bytes[binary_index_offset_pos..binary_index_offset_pos + core::mem::size_of::<u32>()],
        invalid_binary_index_offset,
    );

    IndexBlock::new(Block {
        data: bytes.into(),
        header: Header::test_dummy(BlockType::Index),
    })
}

/// Every entry names a row group by its tag, directory length and head zone
/// block length; the block mixes restart heads (full entries) with truncated
/// ones.
fn make_tagged_index_block(restart_interval: u8) -> (Vec<KeyedBlockHandle>, IndexBlock) {
    let handles: Vec<_> = make_handles(16)
        .into_iter()
        .zip(1_u32..)
        .map(|(h, n)| {
            let row_group = crate::table::index_block::RowGroupRef {
                tag: core::num::NonZeroU64::new(u64::from(n) << 40).unwrap(),
                // Two varint bytes each, as wide as a group size of up to
                // 16 KiB.
                directory_len: core::num::NonZeroU32::new(1_000 + n).unwrap(),
                head_zones_len: 200 + n,
            };
            KeyedBlockHandle::new(
                h.end_key().clone(),
                h.seqno(),
                h.into_inner().with_row_group(Some(row_group)),
            )
        })
        .collect();
    let bytes =
        IndexBlock::encode_into_vec_with_restart_interval(&handles, restart_interval).unwrap();
    let index = IndexBlock::new(Block {
        data: bytes.into(),
        header: Header::test_dummy(BlockType::Index),
    });
    (handles, index)
}

/// The end keys a walk of `index` yields after `seek(lo)` then
/// `seek_upper(hi)`, with what `seek_upper` returned.
fn walk_between(index: &IndexBlock, lo: &[u8], hi: &[u8]) -> (bool, Vec<Vec<u8>>) {
    let mut iter = index.iter(default_comparator());
    assert!(iter.seek(lo, SeqNo::MAX), "seek lands");
    let landed = iter.seek_upper(hi, SeqNo::MAX);
    let keys = iter
        .map(|item| item.materialize(index.as_slice()).end_key().to_vec())
        .collect();
    (landed, keys)
}

/// `seek_upper` after `seek` keeps the entry `seek` landed on, when both land
/// on the same entry: alone in its block, or inside one restart interval with
/// others.
#[test]
fn seek_upper_keeps_the_entry_seek_landed_on() {
    for restart_interval in [1, 2, 4, 16, 232] {
        let single = make_handles(1);
        let bytes =
            IndexBlock::encode_into_vec_with_restart_interval(&single, restart_interval).unwrap();
        let index = IndexBlock::new(Block {
            data: bytes.into(),
            header: Header::test_dummy(BlockType::Index),
        });
        let key = single[0].end_key().to_vec();
        assert_eq!(
            (true, vec![key.clone()]),
            walk_between(&index, &key, &key),
            "restart interval {restart_interval}: single entry"
        );

        let index = make_index_block(restart_interval);
        let handles = make_handles(16);
        let key = |i: usize| handles[i].end_key().to_vec();
        // The bound keeps the entry after the needle too.
        assert_eq!(
            (true, vec![key(5), key(6)]),
            walk_between(&index, &key(5), &key(5)),
            "restart interval {restart_interval}: one entry of many"
        );
        assert_eq!(
            (true, (5..=8).map(key).collect::<Vec<_>>()),
            walk_between(&index, &key(5), &key(7)),
            "restart interval {restart_interval}: three neighbours"
        );
        assert_eq!(
            (true, vec![key(15)]),
            walk_between(&index, &key(15), &key(15)),
            "restart interval {restart_interval}: the last entry"
        );
    }
}

/// An entry naming a row group reads back with its tag and directory length,
/// through a full walk and through a seek that probes the restart heads; an
/// entry naming none reads back with none.
#[test]
fn a_tagged_entry_reads_back_its_row_group() {
    for restart_interval in [1, 4] {
        let (handles, index) = make_tagged_index_block(restart_interval);
        let read: Vec<_> = index
            .iter(default_comparator())
            .map(|item| item.materialize(index.as_slice()))
            .collect();
        assert_eq!(read.len(), handles.len());
        for (read, written) in read.iter().zip(&handles) {
            assert_eq!(read.end_key(), written.end_key());
            assert_eq!(read.offset(), written.offset());
            assert_eq!(read.size(), written.size());
            assert_eq!(read.as_ref().row_group(), written.as_ref().row_group());
        }

        let mut iter = index.iter(default_comparator());
        assert!(iter.seek(b"adj:out:vertex-0001:edge-0011", SeqNo::MAX));
        let Some(found) = iter.next().map(|item| item.materialize(index.as_slice())) else {
            panic!("the seek lands on an entry");
        };
        assert_eq!(found.end_key().as_ref(), b"adj:out:vertex-0001:edge-0011");
        assert_eq!(
            found.as_ref().row_group(),
            Some(crate::table::index_block::RowGroupRef {
                tag: core::num::NonZeroU64::new(12 << 40).unwrap(),
                directory_len: core::num::NonZeroU32::new(1_012).unwrap(),
                head_zones_len: 212,
            })
        );
    }

    let plain = make_index_block(4);
    assert!(
        plain.iter(default_comparator()).all(|item| item
            .materialize(plain.as_slice())
            .as_ref()
            .row_group()
            .is_none()),
        "an entry naming no row group carries none",
    );
}

/// The first entry of a tagged block, its varint field `field` (0 the tag, 1
/// the directory length, 2 the head zone block length) replaced by `value` at
/// the same width, so every later byte stays where it was; `None` when `value`
/// does not fit that width.
fn with_first_row_group_field(field: usize, value: u64) -> Option<IndexBlock> {
    let (_, index) = make_tagged_index_block(1);
    let mut bytes = index.as_slice().to_vec();
    // The first entry: marker, offset (0, one byte), size (4096, two bytes),
    // seqno (0, one byte), then the tag and the directory length.
    assert_eq!(bytes.first(), Some(&4), "a tagged full entry");
    let mut at = 1 + 1 + 2 + 1;
    let varint_len = |bytes: &[u8]| {
        let mut cursor = Cursor::new(bytes);
        let _ = cursor.read_u64_varint().unwrap();
        usize::try_from(cursor.position()).unwrap()
    };
    for _ in 0..field {
        at += varint_len(&bytes[at..]);
    }
    let width = varint_len(&bytes[at..]);
    let mut rest = value;
    for (i, byte) in bytes[at..at + width].iter_mut().enumerate() {
        let low = u8::try_from(rest & 0x7f).unwrap();
        rest >>= 7;
        *byte = if i + 1 < width { low | 0x80 } else { low };
    }
    (rest == 0).then(|| {
        IndexBlock::new(Block {
            data: bytes.into(),
            header: Header::test_dummy(BlockType::Index),
        })
    })
}

/// A tagged entry whose tag is zero, which no writer emits, ends the walk
/// rather than reading back as an entry naming no group.
#[test]
fn a_tagged_entry_with_a_zero_tag_is_refused() {
    let corrupt = with_first_row_group_field(0, 0).unwrap();
    let first = corrupt
        .iter(default_comparator())
        .next()
        .map(|item| item.materialize(corrupt.as_slice()));
    assert!(
        first.is_none(),
        "a zero tag must not read back as an entry, got {first:?}"
    );
}

/// A tagged entry recording an empty directory, or a directory and head zone
/// block longer together than the group it names, which no writer emits, ends
/// the walk and fails a seek's probe of it as a restart head: a reader would
/// read that many bytes as the group's directory and head zone block.
#[test]
fn a_tagged_entry_with_an_impossible_directory_length_is_refused() {
    let (handles, _) = make_tagged_index_block(1);
    let size = u64::from(handles[0].size());
    let Some(named) = handles[0].as_ref().row_group() else {
        panic!("the fixture's entries name row groups");
    };
    let directory_len = u64::from(named.directory_len.get());
    let head_len = u64::from(named.head_zones_len);
    let cases = [
        ("an empty directory", 1, 0),
        ("a directory past the group", 1, size + 1),
        (
            "a directory running the head past the group",
            1,
            size - head_len + 1,
        ),
        (
            "a head zone block past the group",
            2,
            size - directory_len + 1,
        ),
    ];
    for (what, field, value) in cases {
        let corrupt = with_first_row_group_field(field, value).unwrap();
        let first = corrupt
            .iter(default_comparator())
            .next()
            .map(|item| item.materialize(corrupt.as_slice()));
        assert!(first.is_none(), "{what} must not read back, got {first:?}");
        let mut iter = corrupt.iter(default_comparator());
        let landed = iter.seek(handles[0].end_key(), SeqNo::MAX)
            && iter
                .next()
                .map(|item| item.materialize(corrupt.as_slice()))
                .is_some_and(|found| found.end_key() == handles[0].end_key());
        assert!(!landed, "a seek must not land on an entry recording {what}");
    }
}

/// An upper bound below the entry `seek` landed on walks nothing, unless that
/// entry is the first past the bound, which the bound keeps: the entry `seek`
/// holds is kept only inside the bound.
#[test]
fn seek_upper_below_the_entry_seek_landed_on_keeps_only_what_the_bound_covers() {
    for restart_interval in [1, 2, 4, 16, 232] {
        let index = make_index_block(restart_interval);
        let handles = make_handles(16);
        let key = |i: usize| handles[i].end_key().to_vec();
        assert_eq!(
            (false, Vec::<Vec<u8>>::new()),
            walk_between(&index, &key(11), &key(3)),
            "restart interval {restart_interval}: a bound far below"
        );
        assert_eq!(
            (false, Vec::<Vec<u8>>::new()),
            walk_between(&index, &key(6), &key(4)),
            "restart interval {restart_interval}: a bound two entries below"
        );
        assert_eq!(
            (true, vec![key(6)]),
            walk_between(&index, &key(6), &key(5)),
            "restart interval {restart_interval}: the entry is the first past the bound"
        );
    }
}

/// An upper seek below where forward iteration already got to finds nothing:
/// it reports the range empty, and neither end yields an entry.
#[test]
fn seek_upper_behind_the_consumed_front_reports_an_empty_range() {
    for restart_interval in [1, 2, 4, 16, 232] {
        let index = make_index_block(restart_interval);
        let handles = make_handles(16);
        let mut iter = index.iter(default_comparator());
        for _ in 0..8 {
            assert!(iter.next().is_some(), "the block holds 16 entries");
        }
        assert!(
            !iter.seek_upper(handles[3].end_key(), SeqNo::MAX),
            "restart interval {restart_interval}: the range is empty"
        );
        assert!(
            iter.next_back().is_none(),
            "restart interval {restart_interval}"
        );
        assert!(iter.next().is_none(), "restart interval {restart_interval}");
    }
}

#[test]
fn seek_clears_stale_front_cache_before_reposition() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());
    let mut fresh = index.iter(default_comparator());

    assert!(iter.seek(b"adj:out:vertex-0001:edge-0002", 0));
    let seek_after_reposition = iter.seek(b"adj:out:vertex-0001:edge-0011", 0);
    let item_after_reposition = iter.next().map(|item| item.materialize(index.as_slice()));

    let fresh_seek = fresh.seek(b"adj:out:vertex-0001:edge-0011", 0);
    let fresh_item = fresh.next().map(|item| item.materialize(index.as_slice()));

    assert_eq!(seek_after_reposition, fresh_seek);
    assert_eq!(item_after_reposition, fresh_item);
}

#[test]
fn seek_upper_clears_stale_back_cache_before_reposition() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());
    let mut fresh = index.iter(default_comparator());

    assert!(iter.seek_upper(b"adj:out:vertex-0001:edge-0011", SeqNo::MAX));
    let seek_after_reposition = iter.seek_upper(b"adj:out:vertex-0001:edge-0003", SeqNo::MAX);
    let item_after_reposition = iter
        .next_back()
        .map(|item| item.materialize(index.as_slice()));

    let fresh_seek = fresh.seek_upper(b"adj:out:vertex-0001:edge-0003", SeqNo::MAX);
    let fresh_item = fresh
        .next_back()
        .map(|item| item.materialize(index.as_slice()));

    assert_eq!(seek_after_reposition, fresh_seek);
    assert_eq!(item_after_reposition, fresh_item);
}

#[test]
fn seek_upper_keeps_first_covering_handle_in_compressed_interval() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());

    assert!(iter.seek_upper(b"adj:out:vertex-0001:edge-0003z", SeqNo::MAX));
    let selected = iter.next_back().unwrap().materialize(index.as_slice());
    assert_eq!(
        selected.end_key().as_ref(),
        b"adj:out:vertex-0001:edge-0004"
    );
}

/// An exact match also keeps the entry after it: the needle's older versions
/// may run on into that block.
#[test]
fn seek_upper_exact_match_keeps_the_next_item() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());

    assert!(iter.seek_upper(b"adj:out:vertex-0001:edge-0003", SeqNo::MAX));
    let selected = iter.next_back().unwrap().materialize(index.as_slice());
    assert_eq!(
        selected.end_key().as_ref(),
        b"adj:out:vertex-0001:edge-0004"
    );
}

/// An exact match on the last entry of a restart interval keeps the next
/// interval's head.
#[test]
fn seek_upper_exact_match_at_an_interval_end_keeps_the_next_head() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());

    assert!(iter.seek_upper(b"adj:out:vertex-0001:edge-0007", SeqNo::MAX));
    let selected = iter.next_back().unwrap().materialize(index.as_slice());
    assert_eq!(
        selected.end_key().as_ref(),
        b"adj:out:vertex-0001:edge-0008"
    );
}

#[test]
fn seek_upper_with_small_needle_restores_first_item_when_interval_trim_empties_stack() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());

    assert!(iter.seek_upper(b"adj:out:vertex-0001:edge--001", SeqNo::MAX));
    let selected = iter.next_back().unwrap().materialize(index.as_slice());
    assert_eq!(
        selected.end_key().as_ref(),
        b"adj:out:vertex-0001:edge-0000"
    );
}

#[test]
fn seek_upper_exact_match_restart_interval_one_keeps_the_next_item() {
    let index = make_index_block(1);
    let mut iter = index.iter(default_comparator());

    assert!(iter.seek_upper(b"adj:out:vertex-0001:edge-0003", SeqNo::MAX));
    let selected = iter.next_back().unwrap().materialize(index.as_slice());
    assert_eq!(
        selected.end_key().as_ref(),
        b"adj:out:vertex-0001:edge-0004"
    );
}

/// Several entries ending at one key (its versions spanning blocks) are all
/// kept, with the entry after them, whatever the restart interval.
#[test]
fn seek_upper_keeps_every_entry_ending_at_the_needle_and_the_next() {
    let handles: Vec<KeyedBlockHandle> = [("a", 9), ("k", 8), ("k", 5), ("k", 2), ("z", 1)]
        .into_iter()
        .enumerate()
        .map(|(i, (key, seqno))| {
            KeyedBlockHandle::new(
                key.into(),
                seqno,
                BlockHandle::new(BlockOffset((i as u64) * 4096), 4096),
            )
        })
        .collect();
    for restart_interval in [1, 2, 4, 8] {
        let bytes =
            IndexBlock::encode_into_vec_with_restart_interval(&handles, restart_interval).unwrap();
        let index = IndexBlock::new(Block {
            data: bytes.into(),
            header: Header::test_dummy(BlockType::Index),
        });
        let mut iter = index.iter(default_comparator());
        assert!(iter.seek_upper(b"k", SeqNo::MAX));
        let walk: Vec<(Vec<u8>, SeqNo)> = iter
            .map(|item| {
                let handle = item.materialize(index.as_slice());
                (handle.end_key().to_vec(), handle.seqno())
            })
            .collect();
        assert_eq!(
            vec![
                (b"a".to_vec(), 9),
                (b"k".to_vec(), 8),
                (b"k".to_vec(), 5),
                (b"k".to_vec(), 2),
                (b"z".to_vec(), 1),
            ],
            walk,
            "restart interval {restart_interval}"
        );
    }
}

#[test]
fn seek_upper_with_large_needle_keeps_back_cursor_at_last_item() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());

    assert!(iter.seek_upper(b"adj:out:vertex-0001:edge-9999", SeqNo::MAX));
    let selected = iter.next_back().unwrap().materialize(index.as_slice());
    assert_eq!(
        selected.end_key().as_ref(),
        b"adj:out:vertex-0001:edge-0015"
    );
}

#[test]
fn seek_upper_between_intervals_keeps_next_restart_head() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());

    assert!(iter.seek_upper(b"adj:out:vertex-0001:edge-0007z", SeqNo::MAX));
    let selected = iter.next_back().unwrap().materialize(index.as_slice());
    assert_eq!(
        selected.end_key().as_ref(),
        b"adj:out:vertex-0001:edge-0008"
    );
}

#[test]
fn seek_upper_forward_iteration_stops_after_first_covering_item() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());

    assert!(iter.seek_upper(b"adj:out:vertex-0001:edge-0003z", SeqNo::MAX));

    let keys: Vec<Vec<u8>> = iter
        .map(|item| item.materialize(index.as_slice()).end_key().to_vec())
        .collect();

    assert_eq!(
        keys,
        vec![
            b"adj:out:vertex-0001:edge-0000".to_vec(),
            b"adj:out:vertex-0001:edge-0001".to_vec(),
            b"adj:out:vertex-0001:edge-0002".to_vec(),
            b"adj:out:vertex-0001:edge-0003".to_vec(),
            b"adj:out:vertex-0001:edge-0004".to_vec(),
        ]
    );
}

#[test]
fn seek_upper_bound_cursor_between_intervals_includes_next_restart_head() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());

    assert!(
        iter.seek_upper_bound_cursor(b"adj:out:vertex-0001:edge-0007z", SeqNo::MAX)
            .unwrap()
    );

    let keys: Vec<Vec<u8>> = iter
        .map(|item| item.materialize(index.as_slice()).end_key().to_vec())
        .collect();

    assert_eq!(
        keys,
        vec![
            b"adj:out:vertex-0001:edge-0000".to_vec(),
            b"adj:out:vertex-0001:edge-0001".to_vec(),
            b"adj:out:vertex-0001:edge-0002".to_vec(),
            b"adj:out:vertex-0001:edge-0003".to_vec(),
            b"adj:out:vertex-0001:edge-0004".to_vec(),
            b"adj:out:vertex-0001:edge-0005".to_vec(),
            b"adj:out:vertex-0001:edge-0006".to_vec(),
            b"adj:out:vertex-0001:edge-0007".to_vec(),
            b"adj:out:vertex-0001:edge-0008".to_vec(),
        ]
    );
}

#[test]
fn seek_reposition_clears_stale_back_cache() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());
    let mut control = index.iter(default_comparator());

    assert!(iter.seek_upper(b"adj:out:vertex-0001:edge-0011", SeqNo::MAX));
    assert_eq!(
        iter.decoder
            .peek_back()
            .unwrap()
            .materialize(index.as_slice())
            .end_key()
            .as_ref(),
        b"adj:out:vertex-0001:edge-0012"
    );

    assert!(iter.seek(b"adj:out:vertex-0001:edge-0003", 0));
    let item_after_reposition = iter
        .next_back()
        .map(|item| item.materialize(index.as_slice()));

    assert!(control.seek(b"adj:out:vertex-0001:edge-0003", 0));
    let control_item = control
        .next_back()
        .map(|item| item.materialize(index.as_slice()));

    assert_eq!(item_after_reposition, control_item);
    assert_eq!(
        item_after_reposition.unwrap().end_key().as_ref(),
        b"adj:out:vertex-0001:edge-0015"
    );
}

#[test]
fn seek_upper_reposition_clears_stale_front_cache() {
    let index = make_index_block(8);
    let mut iter = index.iter(default_comparator());
    let mut control = index.iter(default_comparator());

    assert!(iter.seek(b"adj:out:vertex-0001:edge-0002", 0));
    let stale_front_key = iter
        .decoder
        .peek()
        .unwrap()
        .materialize(index.as_slice())
        .end_key()
        .to_vec();

    assert!(iter.seek_upper(b"adj:out:vertex-0001:edge-0007", SeqNo::MAX));
    let item_after_reposition = iter
        .next_back()
        .map(|item| item.materialize(index.as_slice()));

    assert!(control.seek(b"adj:out:vertex-0001:edge-0002", 0));
    assert!(control.seek_upper(b"adj:out:vertex-0001:edge-0007", SeqNo::MAX));
    let control_item = control
        .next_back()
        .map(|item| item.materialize(index.as_slice()));

    assert_eq!(item_after_reposition, control_item);
    assert_ne!(
        item_after_reposition.unwrap().end_key().as_ref(),
        stale_front_key.as_slice()
    );
}

#[test]
fn next_stays_none_after_invalid_restart_head_parse() {
    let index = make_corrupted_index_block_with_invalid_first_restart_head();
    let mut iter = index.iter(default_comparator());

    assert!(iter.next().is_none());
    assert!(iter.next().is_none());
}

#[test]
fn try_iter_rejects_invalid_binary_index_offset() {
    let index = make_corrupted_index_block_with_invalid_binary_index_offset();
    // try_iter rejects the block eagerly due to invalid layout metadata
    assert!(
        matches!(
            index.try_iter(default_comparator()),
            Err(crate::Error::InvalidTrailer)
        ),
        "corrupt binary_index_offset must be rejected by try_iter",
    );
}

/// Finds the byte offset of the second restart head by reading the binary
/// index from the block trailer.
fn second_restart_head_byte_offset(bytes: &[u8]) -> usize {
    let probe = IndexBlock::new(Block {
        data: bytes.to_vec().into(),
        header: Header::test_dummy(BlockType::Index),
    });
    let trailer_offset = Trailer::new(&probe.inner).trailer_offset();
    let trailer = &bytes[trailer_offset..];

    // Trailer layout: restart_interval(u8)[0], step_size(u8)[1],
    //   binary_index_len(u32LE)[2..6], binary_index_offset(u32LE)[6..10], ...
    let step_size = trailer[1] as usize;
    let binary_index_offset = LittleEndian::read_u32(&trailer[6..10]) as usize;

    // Entry 1 in the binary index is the second restart head offset
    let entry_pos = binary_index_offset + step_size;
    if step_size == 2 {
        LittleEndian::read_u16(&bytes[entry_pos..entry_pos + 2]) as usize
    } else {
        LittleEndian::read_u32(&bytes[entry_pos..entry_pos + 4]) as usize
    }
}

/// Returns the byte offset of the first truncated (non-head) item in the
/// restart interval starting at `restart_offset`, after skipping the full
/// restart head + its key bytes.
fn first_truncated_item_offset_in_interval(bytes: &[u8], restart_offset: usize) -> usize {
    let mut cursor = Cursor::new(&bytes[restart_offset..]);
    // Restart head: marker(0) + BlockHandle + seqno + key_len + key_bytes
    let marker = cursor.read_u8().unwrap();
    assert_eq!(marker, 0, "expected restart head marker");
    let _ = BlockHandle::decode_from(&mut cursor).unwrap();
    let _ = cursor.read_u64_varint().unwrap();
    let key_len: u64 = cursor.read_u16_varint().unwrap().into();
    cursor.set_position(cursor.position() + key_len);
    // Now at the first truncated item
    let truncated_marker = cursor.read_u8().unwrap();
    assert_eq!(truncated_marker, 1, "second entry should be truncated");
    let _ = cursor.read_u16_varint().unwrap(); // shared prefix len
    // cursor is now at rest_key_len — corrupt THIS to break fill_stack
    restart_offset + usize::try_from(cursor.position()).unwrap()
}

fn make_index_block_with_corrupt_second_interval_item() -> IndexBlock {
    let handles = make_handles(16);
    let mut bytes = IndexBlock::encode_into_vec_with_restart_interval(&handles, 8).unwrap();
    let second_restart = second_restart_head_byte_offset(&bytes);
    let rest_key_len_pos = first_truncated_item_offset_in_interval(&bytes, second_restart);
    // Overwrite rest_key_len with an impossibly large varint
    bytes[rest_key_len_pos] = 0xFF;
    bytes[rest_key_len_pos + 1] = 0xFF;
    bytes[rest_key_len_pos + 2] = 0x03;
    IndexBlock::new(Block {
        data: bytes.into(),
        header: Header::test_dummy(BlockType::Index),
    })
}

#[test]
fn seek_upper_bound_cursor_returns_err_on_poisoned_cursor() {
    // Block layout: 16 entries with restart_interval=8 → 2 restart intervals.
    // First interval (entries 0-7, keys edge-0000..edge-0007) is valid.
    // Second interval has a corrupted non-head item (entry 9): its
    // rest_key_len is overwritten so fill_stack poisons the back cursor.
    // The restart head (entry 8) is valid so binary search still works.
    //
    // Needle "edge-0007z" lands in the first interval via binary search.
    // trim_back_to_upper_bound doesn't pop anything (all first-interval
    // items ≤ needle), but the stack tail "edge-0007" < "edge-0007z" so
    // the advance loop fires.
    // advance_upper_restart_interval clears the stack and tries to fill from
    // the corrupt second interval → fill_stack fails → stack empty →
    // seek_upper_bound_cursor must return Err(InvalidTrailer), not Ok(false).
    let index = make_index_block_with_corrupt_second_interval_item();
    let mut iter = index.iter(default_comparator());

    let result = iter.seek_upper_bound_cursor(b"adj:out:vertex-0001:edge-0007z", SeqNo::MAX);
    assert!(
        matches!(result, Err(crate::Error::InvalidTrailer)),
        "poisoned upper cursor must return Err(InvalidTrailer), got {result:?}",
    );
}