coordinode-lsm-tree 5.7.0

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2025-present, fjall-rs
// Copyright (c) 2026-present, Structured World Foundation

use super::{
    TRAILER_START_MARKER, binary_index::Reader as BinaryIndexReader,
    hash_index::Reader as HashIndexReader,
};
use crate::io::{LittleEndian, ReadBytesExt};
use crate::{
    SeqNo, Slice,
    table::{Block, block::Trailer},
};
#[cfg(not(feature = "std"))]
use alloc::vec::Vec;
use core::marker::PhantomData;

use crate::io::Cursor;

/// Decodes one LEB128 varint from `$buf` at byte offset `$pos`, evaluating to
/// `(u64, usize)`: the value and the position just past it.
///
/// A macro, not a function, so the whole decode (single-byte fast path **and**
/// the multi-byte continuation) expands inline at the call site. This is the
/// hottest cost on the point-read path's block-entry parse, where several
/// varints are decoded per entry; as a `#[inline]` function it still showed up
/// as a standalone hot symbol (LLVM declined to inline the body once the loop
/// made it large enough), so each varint paid a call. The macro removes that
/// call for every case and lets the decode fuse with the caller's bounds checks
/// and constants. Reads through a plain slice index (no `Cursor::read_u8`
/// `read_exact`-per-byte) and rejects an overlong (> 64-bit) encoding, matching
/// `VarintReader::read_u64_varint`.
///
/// On a truncated buffer (`?` on the bounds-checked load) or an overlong
/// encoding it returns `None` from the **enclosing** function, so every call
/// site must sit in a function returning `Option`.
macro_rules! read_leb128 {
    ($buf:expr, $pos:expr) => {{
        let buf: &[u8] = $buf;
        let start: usize = $pos;
        let first = *buf.get(start)?;
        if first < 0x80 {
            // Single-byte fast path (the common case: small keys, seqnos,
            // shared/rest lengths). One bounds-checked load, a compare, done.
            (u64::from(first), start + 1)
        } else {
            // Multi-byte continuation: fold in the low 7 bits already read,
            // then loop. Block offsets / sizes routinely land here, so this
            // path is inlined too rather than outlined.
            let mut result = u64::from(first & 0x7f);
            let mut shift = 7u32;
            let mut p = start + 1;
            loop {
                let byte = *buf.get(p)?;
                p += 1;
                // The 10th byte (shift == 63) can only carry bit 63; any higher
                // payload bit (mask 0x7e) would overflow u64. Reject before
                // folding it in so a corrupt overlong encoding fails the decode
                // instead of silently truncating to a wrapped value.
                if shift == 63 && (byte & 0x7e) != 0 {
                    return None;
                }
                result |= u64::from(byte & 0x7f) << shift;
                if byte & 0x80 == 0 {
                    break (result, p);
                }
                shift += 7;
                if shift >= 64 {
                    return None;
                }
            }
        }
    }};
}
pub(crate) use read_leb128;

/// Function form of [`read_leb128!`] for the unit tests, which assert on the
/// `Option` return directly. Returns the decoded `(value, position)` or `None`
/// on a truncated or overlong encoding. Production call sites use the macro so
/// the decode inlines; a non-test caller that prefers the function form can lift
/// this out of the `cfg(test)` gate.
#[cfg(test)]
fn read_leb128_fn(buf: &[u8], pos: usize) -> Option<(u64, usize)> {
    Some(read_leb128!(buf, pos))
}

/// Validates that `restart_interval` and `binary_index_step_size` read from a
/// block trailer are within their allowed ranges.
///
/// Returns `Err(crate::Error::InvalidTrailer)` on malformed values that would
/// otherwise trigger an assertion failure deeper in the decoder.
fn validate_trailer_fields(restart_interval: u8, binary_index_step_size: u8) -> crate::Result<()> {
    if restart_interval == 0 {
        return Err(crate::Error::InvalidTrailer);
    }
    if binary_index_step_size != 2 && binary_index_step_size != 4 {
        return Err(crate::Error::InvalidTrailer);
    }
    Ok(())
}

/// Represents an object that was parsed from a byte array
///
/// Parsed items only hold references to their keys and values, use `materialize` to create an owned value.
pub trait ParsedItem<M> {
    /// Compares this item's key with a needle using the given comparator.
    ///
    /// We can not access the key directly because it may be comprised of prefix + suffix.
    fn compare_key(
        &self,
        needle: &[u8],
        bytes: &[u8],
        cmp: &dyn crate::comparator::UserComparator,
    ) -> core::cmp::Ordering;

    /// Returns the item's seqno.
    fn seqno(&self) -> SeqNo;

    /// Returns the byte offset of the key's start position.
    fn key_offset(&self) -> usize;

    /// Returns one-past-the-end byte offset of the key.
    fn key_end_offset(&self) -> usize;

    /// Converts the parsed representation to an owned value.
    fn materialize(&self, bytes: &Slice) -> M;
}

/// Describes an object that can be parsed from a block, either a full item (restart head), or a truncated item
pub trait Decodable<ParsedItem> {
    /// Parses the key of the next restart head from a reader.
    ///
    /// This is the fast path for binary-search probes: it skips full-item
    /// decode and only extracts `(key, seqno)`. `entries_end` bounds the
    /// key span so malformed lengths cannot leak into trailer/index bytes.
    fn parse_restart_key<'a>(
        reader: &mut Cursor<&[u8]>,
        offset: usize,
        data: &'a [u8],
        entries_end: usize,
    ) -> Option<(&'a [u8], SeqNo)>;

    /// Parses a restart head from a reader.
    ///
    /// `offset` is the position of the item to read in the block's byte slice.
    fn parse_full(
        reader: &mut Cursor<&[u8]>,
        offset: usize,
        entries_end: usize,
    ) -> Option<ParsedItem>;

    /// Parses a (possibly) prefix truncated item from a reader.
    fn parse_truncated(
        reader: &mut Cursor<&[u8]>,
        offset: usize,
        base_key_offset: usize,
        base_key_end: usize,
        entries_end: usize,
    ) -> Option<ParsedItem>;
}

#[derive(Debug)]
struct LoScanner {
    offset: usize,
    remaining_in_interval: usize,
    base_key_offset: Option<usize>,
    base_key_end: Option<usize>,
}

#[derive(Debug)]
struct HiScanner {
    offset: usize,
    ptr_idx: usize,
    stack: Vec<usize>, // TODO: SmallVec?
    base_key_offset: Option<usize>,
    base_key_end: Option<usize>,
}

/// Generic block decoder for RocksDB-style blocks
///
/// Supports prefix truncation and binary search index (through restart intervals).
pub struct Decoder<'a, Item: Decodable<Parsed>, Parsed: ParsedItem<Item>> {
    block: &'a Block,
    phantom: PhantomData<(Item, Parsed)>,

    lo_scanner: LoScanner,
    hi_scanner: HiScanner,

    // Cached metadata
    restart_interval: u8,
    binary_index_step_size: u8,
    binary_index_offset: u32,
    binary_index_len: u32,
    hash_index_len: u32,
    hash_index_offset: u32,
    cached_entries_end: Option<usize>,
}

/// Pre-parsed block-trailer metadata, extracted once and reused across
/// many [`Decoder`] constructions over the same block.
///
/// All fields are small `Copy` integers read from the fixed trailer plus
/// the derived `cached_entries_end`. Parsing the trailer (six reads +
/// validation + `compute_entries_end`) is identical on every call, so a
/// long-lived index that decodes the same block on every point read
/// (e.g. [`crate::table::block_index::FullBlockIndex`]) can parse once at
/// construction and hand this struct to [`Decoder::from_meta`] thereafter,
/// turning per-lookup trailer parsing into a few field copies.
#[derive(Clone, Copy, Debug)]
pub struct DecoderMeta {
    restart_interval: u8,
    binary_index_step_size: u8,
    binary_index_offset: u32,
    binary_index_len: u32,
    hash_index_len: u32,
    hash_index_offset: u32,
    cached_entries_end: usize,
}

impl<'a, Item: Decodable<Parsed>, Parsed: ParsedItem<Item>> Decoder<'a, Item, Parsed> {
    #[must_use]
    pub fn restart_interval(&self) -> u8 {
        self.restart_interval
    }

    /// Byte length of the entry region (offset where the trailer begins).
    /// Test-only: lets headerless-decode tests slice the exact entry region.
    /// Only the zstd partial-decode / lazy-block tests use it, so it is gated
    /// to that feature to stay dead-code-clean in the default test build.
    #[cfg(all(test, feature = "zstd"))]
    #[must_use]
    pub(crate) fn entries_end_for_test(&self) -> Option<usize> {
        self.cached_entries_end
    }

    /// Entry-region bytes the decoder parses. Single seam for byte access so a
    /// lazily-decoded backing (decode only the inner blocks a read covers) can
    /// later replace the resident slice without touching every read site.
    /// Returns `'a`-lifetime bytes (tied to the block, not `&self`), preserving
    /// the disjoint borrow that lets reads coexist with scanner-state mutation.
    #[inline]
    fn entries(&self) -> &'a [u8] {
        &self.block.data
    }

    /// Extracts the reusable trailer metadata from a fully-constructed
    /// decoder so future decodes of the same block can skip the trailer
    /// parse via [`Self::from_meta`].
    ///
    /// # Panics
    ///
    /// Panics if `cached_entries_end` was never computed, which cannot
    /// happen for a decoder built by [`Self::try_new`] / [`Self::new`]
    /// (both populate it before returning).
    #[must_use]
    #[expect(
        clippy::expect_used,
        reason = "try_new/new always populate cached_entries_end before returning"
    )]
    pub fn meta(&self) -> DecoderMeta {
        DecoderMeta {
            restart_interval: self.restart_interval,
            binary_index_step_size: self.binary_index_step_size,
            binary_index_offset: self.binary_index_offset,
            binary_index_len: self.binary_index_len,
            hash_index_len: self.hash_index_len,
            hash_index_offset: self.hash_index_offset,
            cached_entries_end: self
                .cached_entries_end
                .expect("cached_entries_end populated by constructor"),
        }
    }

    /// Builds a decoder from a block plus already-parsed [`DecoderMeta`],
    /// skipping the trailer read + validation + `compute_entries_end` that
    /// [`Self::try_new`] performs. The caller guarantees `meta` was
    /// produced by [`Self::meta`] on a decoder of the SAME block layout
    /// (same trailer), so no validation is repeated.
    ///
    /// This is the parse-once fast path for hot read loops: the owning
    /// index parses the trailer once and reuses `meta` on every lookup.
    #[must_use]
    pub fn from_meta(block: &'a Block, meta: DecoderMeta) -> Self {
        Self {
            block,
            phantom: PhantomData,

            lo_scanner: LoScanner {
                offset: 0,
                remaining_in_interval: 0,
                base_key_offset: None,
                base_key_end: None,
            },

            hi_scanner: HiScanner {
                offset: 0,
                ptr_idx: usize::try_from(meta.binary_index_len).unwrap_or(usize::MAX),
                stack: Vec::new(),
                base_key_offset: None,
                base_key_end: None,
            },

            restart_interval: meta.restart_interval,
            binary_index_step_size: meta.binary_index_step_size,
            binary_index_offset: meta.binary_index_offset,
            binary_index_len: meta.binary_index_len,
            hash_index_len: meta.hash_index_len,
            hash_index_offset: meta.hash_index_offset,
            cached_entries_end: Some(meta.cached_entries_end),
        }
    }

    /// Creates a new block decoder, returning an error on malformed trailer
    /// fields instead of panicking.
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::InvalidTrailer`] when:
    /// - the block is too small to contain a trailer,
    /// - `restart_interval` is zero,
    /// - `binary_index_step_size` is not 2 or 4, or
    /// - binary/hash-index layout metadata is inconsistent.
    pub fn try_new(block: &'a Block) -> crate::Result<Self> {
        let trailer = Trailer::try_new(block)?;
        let mut reader = trailer.as_slice();

        let restart_interval = reader.read_u8().map_err(|_| crate::Error::InvalidTrailer)?;
        let binary_index_step_size = reader.read_u8().map_err(|_| crate::Error::InvalidTrailer)?;

        validate_trailer_fields(restart_interval, binary_index_step_size)?;

        let binary_index_len = reader
            .read_u32::<LittleEndian>()
            .map_err(|_| crate::Error::InvalidTrailer)?;
        let binary_index_offset = reader
            .read_u32::<LittleEndian>()
            .map_err(|_| crate::Error::InvalidTrailer)?;
        let hash_index_len = reader
            .read_u32::<LittleEndian>()
            .map_err(|_| crate::Error::InvalidTrailer)?;
        let hash_index_offset = reader
            .read_u32::<LittleEndian>()
            .map_err(|_| crate::Error::InvalidTrailer)?;

        let mut decoder = Self {
            block,
            phantom: PhantomData,

            lo_scanner: LoScanner {
                offset: 0,
                remaining_in_interval: 0,
                base_key_offset: None,
                base_key_end: None,
            },

            hi_scanner: HiScanner {
                offset: 0,
                ptr_idx: usize::try_from(binary_index_len).unwrap_or(usize::MAX),
                stack: Vec::new(),
                base_key_offset: None,
                base_key_end: None,
            },

            restart_interval,

            binary_index_step_size,
            binary_index_offset,
            binary_index_len,
            hash_index_len,
            hash_index_offset,
            cached_entries_end: None,
        };
        decoder.cached_entries_end = Some(
            decoder
                .compute_entries_end()
                .ok_or(crate::Error::InvalidTrailer)?,
        );
        Ok(decoder)
    }

    /// Creates a new block decoder.
    ///
    /// # Panics
    ///
    /// Panics on any block corruption detected by [`Self::try_new`]:
    /// undersized blocks, invalid trailer fields, or inconsistent
    /// binary/hash-index layout metadata. Prefer `try_new` in I/O paths
    /// where corrupt blocks should produce a structured error.
    #[must_use]
    #[expect(
        clippy::expect_used,
        reason = "infallible wrapper for test/non-I/O paths"
    )]
    pub fn new(block: &'a Block) -> Self {
        Self::try_new(block).expect("valid block trailer")
    }

    /// Builds a forward-only decoder over a block whose bytes are JUST the
    /// entry region, with NO trailer (restart array / binary index / hash
    /// index). Used for partial-decode reads, where `block.data` is the
    /// contiguous decompressed prefix of a subset of inner zstd blocks (the
    /// trailer lives in a later inner block that was skipped).
    ///
    /// Only forward iteration ([`Iterator::next`]) is valid: there is no
    /// index, so seeking / binary search / backward iteration are unsupported.
    /// `restart_interval` must match the value the block was encoded with
    /// (per-SST constant from `TableMeta`); restart heads are positional, so
    /// the forward scan reconstructs every key from byte 0 without the trailer.
    /// `entries_end` is the length of the entry region (== `block.data.len()`
    /// for a clean prefix); a partial prefix may end mid-entry, in which case
    /// the final truncated entry parses to `None` and iteration stops cleanly
    /// before it.
    ///
    /// # Panics
    ///
    /// Panics if `restart_interval == 0` (a zero interval has no restart heads,
    /// so positional restart tracking is undefined).
    #[cfg(feature = "zstd")]
    #[must_use]
    pub(crate) fn new_forward_headerless(
        block: &'a Block,
        restart_interval: u8,
        entries_end: usize,
    ) -> Self {
        assert!(
            restart_interval > 0,
            "restart_interval must be non-zero for headerless forward decode",
        );
        Self {
            block,
            phantom: PhantomData,
            lo_scanner: LoScanner {
                offset: 0,
                remaining_in_interval: 0,
                base_key_offset: None,
                base_key_end: None,
            },
            // No backward cursor: base_key_offset stays None so `next` never
            // treats the (zeroed) hi_scanner.offset as an upper bound.
            hi_scanner: HiScanner {
                offset: 0,
                ptr_idx: 0,
                stack: Vec::new(),
                base_key_offset: None,
                base_key_end: None,
            },
            restart_interval,
            // No trailer/index: keep these zeroed. Forward `next` reads neither.
            binary_index_step_size: 2,
            binary_index_offset: 0,
            binary_index_len: 0,
            hash_index_len: 0,
            hash_index_offset: 0,
            cached_entries_end: Some(entries_end),
        }
    }

    /// Forward-scan a headerless block, returning the byte offset of every
    /// restart head, the count of complete entries, and the byte offset just
    /// past the last complete entry. Used to synthesize a binary-index trailer
    /// over a decoded prefix (whose tail may be a truncated entry, which this
    /// scan stops cleanly before — see [`Self::new_forward_headerless`]).
    #[cfg(feature = "zstd")]
    pub(crate) fn scan_restart_offsets(mut self) -> (Vec<u32>, usize, usize) {
        let mut restart_offsets = Vec::new();
        let mut item_count = 0usize;
        // `next()` clobbers `lo_scanner.offset` to `data.len()` when it stops on
        // a truncated tail entry, so track the end of the last COMPLETE entry
        // ourselves rather than reading the post-loop scanner offset.
        let mut last_complete_end = 0usize;
        loop {
            let is_restart = self.lo_scanner.remaining_in_interval == 0;
            let head = self.lo_scanner.offset;
            if self.next().is_none() {
                break;
            }
            if is_restart {
                #[expect(
                    clippy::cast_possible_truncation,
                    reason = "block offsets are far below u32::MAX"
                )]
                restart_offsets.push(head as u32);
            }
            item_count += 1;
            last_complete_end = self.lo_scanner.offset;
        }
        (restart_offsets, item_count, last_complete_end)
    }

    fn binary_index_bounds(&self) -> Option<(usize, usize)> {
        let step_size = match self.binary_index_step_size {
            2 | 4 => usize::from(self.binary_index_step_size),
            _ => return None,
        };
        let binary_index_len = usize::try_from(self.binary_index_len).ok()?;
        if binary_index_len == 0 {
            return None;
        }
        let binary_index_bytes = binary_index_len.checked_mul(step_size)?;
        let binary_index_offset = usize::try_from(self.binary_index_offset).ok()?;
        let binary_index_end = binary_index_offset.checked_add(binary_index_bytes)?;
        let trailer_offset = Trailer::new(self.block).trailer_offset();
        let hash_index_len = usize::try_from(self.hash_index_len).ok()?;
        let hash_index_offset = usize::try_from(self.hash_index_offset).ok()?;

        if (hash_index_len == 0) != (hash_index_offset == 0) {
            return None;
        }

        if hash_index_offset > trailer_offset {
            return None;
        }

        if hash_index_offset > 0 {
            let hash_index_end = hash_index_offset.checked_add(hash_index_len)?;
            if hash_index_end != trailer_offset {
                return None;
            }
        }

        let binary_index_limit = if hash_index_offset > 0 {
            hash_index_offset
        } else {
            trailer_offset
        };
        if binary_index_offset == 0 || binary_index_end != binary_index_limit {
            return None;
        }

        Some((binary_index_offset, binary_index_end))
    }

    fn get_binary_index_reader(&self) -> Option<BinaryIndexReader<'_>> {
        let (binary_index_offset, _) = self.binary_index_bounds()?;
        if self.block.data.get(binary_index_offset - 1).copied()? != TRAILER_START_MARKER {
            return None;
        }

        Some(BinaryIndexReader::new(
            &self.block.data,
            self.binary_index_offset,
            self.binary_index_len,
            self.binary_index_step_size,
        ))
    }

    /// Binary index reader built from the trailer metadata cached at
    /// construction, with no re-parse of the block trailer. Returns `None`
    /// if the cached layout is inconsistent (mirrors the validation in
    /// [`Self::get_binary_index_reader`]).
    pub(crate) fn cached_binary_index_reader(&self) -> Option<BinaryIndexReader<'_>> {
        self.get_binary_index_reader()
    }

    /// Hash index reader built from the trailer metadata cached at
    /// construction, with no re-parse of the block trailer. Returns `None`
    /// when the block carries no hash index (`hash_index_len == 0`).
    pub(crate) fn cached_hash_index_reader(&self) -> Option<HashIndexReader<'_>> {
        if self.hash_index_len == 0 {
            return None;
        }
        Some(HashIndexReader::new(
            &self.block.data,
            self.hash_index_offset,
            self.hash_index_len,
        ))
    }

    fn reader_at(data: &[u8], offset: usize) -> Option<Cursor<&[u8]>> {
        if offset >= data.len() {
            return None;
        }
        Some(Cursor::new(data.get(offset..)?))
    }

    fn get_key_at(&self, pos: usize, entries_end: usize) -> Option<(&[u8], SeqNo)> {
        if pos >= entries_end {
            return None;
        }
        let bytes = self.entries();
        let mut cursor = Self::reader_at(bytes, pos)?;
        Item::parse_restart_key(&mut cursor, pos, bytes, entries_end)
    }

    fn partition_point<F>(&self, pred: F) -> Option<(/* offset */ usize, /* idx */ usize)>
    where
        F: Fn(&[u8], SeqNo) -> bool,
    {
        // The first pass over the binary index emulates `Iterator::partition_point` over the
        // restart heads that are in natural key order.  We keep track of both the byte offset and
        // the restart index because callers need the offset to seed the linear scanner, while the
        // index is sometimes reused (for example by `seek_upper`).
        //
        // In contrast to the usual `partition_point`, we intentionally return the *last* restart
        // entry when the predicate continues to hold for every head key.  Forward scans rely on
        // this behaviour to land on the final restart interval and resume the linear scan there
        // instead of erroneously reporting "not found".
        let binary_index = self.get_binary_index_reader()?;
        let entries_end = self.entries_end()?;

        debug_assert!(
            binary_index.len() >= 1,
            "binary index should never be empty",
        );

        let mut left: usize = 0;
        let mut right = binary_index.len();

        if right == 0 {
            return None;
        }

        while left < right {
            let mid = usize::midpoint(left, right);

            let offset = binary_index.get(mid);

            let (head_key, head_seqno) = self.get_key_at(offset, entries_end)?;

            if pred(head_key, head_seqno) {
                left = mid + 1;
            } else {
                right = mid;
            }
        }

        if left == 0 {
            return Some((0, 0));
        }

        if left == binary_index.len() {
            let idx = binary_index.len() - 1;
            let offset = binary_index.get(idx);
            return Some((offset, idx));
        }

        let offset = binary_index.get(left - 1);

        Some((offset, left - 1))
    }

    // TODO:
    fn partition_point_2<F>(&self, pred: F) -> Option<(/* offset */ usize, /* idx */ usize)>
    where
        F: Fn(&[u8], SeqNo) -> bool,
    {
        // `partition_point_2` mirrors `partition_point` but keeps the *next* restart entry instead
        // of the previous one. This variant is used exclusively by reverse scans (`seek_upper`)
        // that want the first restart whose head key exceeds the predicate. Returning the raw
        // offset preserves the ability to reuse linear scanning infrastructure without duplicating
        // decoder logic.
        let binary_index = self.get_binary_index_reader()?;
        let entries_end = self.entries_end()?;

        debug_assert!(
            binary_index.len() >= 1,
            "binary index should never be empty",
        );

        let mut left: usize = 0;
        let mut right = binary_index.len();

        if right == 0 {
            return None;
        }

        while left < right {
            let mid = usize::midpoint(left, right);

            let offset = binary_index.get(mid);

            let (head_key, head_seqno) = self.get_key_at(offset, entries_end)?;

            if pred(head_key, head_seqno) {
                left = mid + 1;
            } else {
                right = mid;
            }
        }

        if left == binary_index.len() {
            let idx = binary_index.len() - 1;
            let offset = binary_index.get(idx);
            return Some((offset, idx));
        }

        let offset = binary_index.get(left);

        Some((offset, left))
    }

    pub fn set_lo_offset(&mut self, offset: usize) {
        self.lo_scanner.offset = offset;
    }

    /// Resets reverse-scan state so the next `next_back()` starts from the
    /// right edge of the block rather than any previously seeded upper bound.
    pub fn reset_back_cursor(&mut self) {
        self.hi_scanner.ptr_idx = usize::try_from(self.binary_index_len).unwrap_or(usize::MAX);
        self.hi_scanner.stack.clear();
        self.hi_scanner.base_key_offset = None;
        self.hi_scanner.base_key_end = None;
    }

    fn poison_back_cursor(&mut self) {
        // Clamp rather than clear: setting base_key_offset = Some(0) keeps
        // the upper bound visible to next(), so forward iteration also stops.
        // Clearing to None would let next() continue past the corrupted
        // interval because it only enforces the ceiling when base_key_offset
        // is Some.
        self.clamp_upper_to_lo();
    }

    fn clamp_upper_to_lo(&mut self) {
        self.hi_scanner.offset = self.lo_scanner.offset;
        self.hi_scanner.ptr_idx = usize::MAX;
        self.hi_scanner.stack.clear();
        self.hi_scanner.base_key_offset = Some(0);
        self.hi_scanner.base_key_end = Some(0);
    }

    fn exhaust(&mut self) {
        let end = self.entries().len();

        self.lo_scanner.offset = end;
        self.lo_scanner.remaining_in_interval = 0;
        self.lo_scanner.base_key_offset = None;
        self.lo_scanner.base_key_end = None;

        self.hi_scanner.offset = end;
        self.hi_scanner.ptr_idx = usize::MAX;
        self.hi_scanner.stack.clear();
        self.hi_scanner.base_key_offset = Some(0);
        self.hi_scanner.base_key_end = Some(0);
    }

    /// Seeks using the given predicate.
    ///
    /// Returns `false` if the key does not possible exist.
    #[must_use]
    pub fn seek(&mut self, pred: impl Fn(&[u8], SeqNo) -> bool, second_partition: bool) -> bool {
        // Index blocks historically used `second_partition` because with restart_interval=1 each
        // restart head is also an item boundary, so the "next restart head" is the lower bound we
        // want. Once restart intervals can be larger than 1, jumping to the next restart head
        // would skip all items in the current interval. In that case we must keep using the
        // current interval and linearly scan within it.
        let use_next_restart = second_partition && self.restart_interval == 1;

        let result = if use_next_restart {
            self.partition_point_2(&pred)
        } else {
            self.partition_point(&pred)
        };

        // Binary index lookup
        let Some((offset, _)) = result else {
            self.exhaust();
            return false;
        };

        if use_next_restart
            && self
                .entries_end()
                .and_then(|entries_end| self.get_key_at(offset, entries_end))
                .is_some_and(|(key, seqno)| pred(key, seqno))
        {
            // `second_partition == true` means we ran the "look one restart ahead" search used by
            // index blocks. When the predicate is still true at the chosen restart head it means
            // the caller asked us to seek strictly beyond the last entry. In that case we skip any
            // costly parsing and flip both scanners into an "exhausted" state so the outer iterator
            // immediately reports EOF.
            self.exhaust();
            return false;
        }

        self.lo_scanner.offset = offset;
        self.lo_scanner.remaining_in_interval = 0;
        self.lo_scanner.base_key_offset = None;
        self.lo_scanner.base_key_end = None;

        true
    }

    /// Seeks the upper bound using the given predicate.
    ///
    /// Returns `false` if the key does not possible exist.
    #[must_use]
    pub fn seek_upper(
        &mut self,
        pred: impl Fn(&[u8], SeqNo) -> bool,
        second_partition: bool,
    ) -> bool {
        let use_next_restart = second_partition && self.restart_interval == 1;

        let result = if use_next_restart {
            self.partition_point_2(&pred)
        } else {
            self.partition_point(&pred)
        };

        // Binary index lookup
        let Some((offset, idx)) = result else {
            self.exhaust();
            return false;
        };

        self.hi_scanner.offset = offset;
        self.hi_scanner.ptr_idx = idx;
        self.hi_scanner.stack.clear();
        self.hi_scanner.base_key_offset = None;
        self.hi_scanner.base_key_end = None;

        self.fill_stack();
        if self.hi_scanner.stack.is_empty() {
            // `fill_stack` failed to decode the selected upper interval and poisoned the
            // reverse scanner. Fail closed by turning the upper cursor into a zero-width bound
            // at the current forward offset so bounded scans cannot continue past the limit.
            self.clamp_upper_to_lo();
            return false;
        }

        true
    }

    fn parse_current_item(
        reader: &mut Cursor<&[u8]>,
        offset: usize,
        base_key_offset: Option<usize>,
        base_key_end: Option<usize>,
        is_restart: bool,
        entries_end: usize,
    ) -> Option<Parsed> {
        if is_restart {
            Item::parse_full(reader, offset, entries_end)
        } else {
            #[expect(clippy::expect_used, reason = "we trust the is_restart flag")]
            Item::parse_truncated(
                reader,
                offset,
                base_key_offset.expect("should parse truncated item"),
                base_key_end.expect("should parse truncated item"),
                entries_end,
            )
        }
    }

    fn compute_entries_end(&self) -> Option<usize> {
        let (binary_index_offset, _) = self.binary_index_bounds()?;
        if self.block.data.get(binary_index_offset - 1).copied()? != TRAILER_START_MARKER {
            return None;
        }
        Some(binary_index_offset - 1)
    }

    fn entries_end(&self) -> Option<usize> {
        self.cached_entries_end
    }

    fn fill_stack(&mut self) {
        // Always rebuild from clean state: consume_stack_top may have partially
        // drained the stack (e.g. early return on offset < lo_scanner.offset),
        // leaving stale offsets from the previous interval.
        self.hi_scanner.stack.clear();
        self.hi_scanner.base_key_offset = None;
        self.hi_scanner.base_key_end = None;

        let Some(entries_end) = self.entries_end() else {
            self.poison_back_cursor();
            return;
        };
        let Some(binary_index) = self.get_binary_index_reader() else {
            self.poison_back_cursor();
            return;
        };
        if self.hi_scanner.ptr_idx >= binary_index.len() {
            // Stack/base_key already cleared at function entry.
            return;
        }

        {
            self.hi_scanner.offset = binary_index.get(self.hi_scanner.ptr_idx);

            let offset = self.hi_scanner.offset;
            let Some(mut reader) = Self::reader_at(self.entries(), offset) else {
                self.poison_back_cursor();
                return;
            };

            #[expect(
                clippy::cast_possible_truncation,
                reason = "blocks do not even come close to 4 GiB in size"
            )]
            let parsed_restart =
                Item::parse_full(&mut reader, offset, entries_end).inspect(|item| {
                    self.hi_scanner.offset += reader.position() as usize;
                    self.hi_scanner.base_key_offset = Some(item.key_offset());
                    self.hi_scanner.base_key_end = Some(item.key_end_offset());
                });

            if parsed_restart.is_some() {
                self.hi_scanner.stack.push(offset);
            } else {
                self.poison_back_cursor();
                return;
            }
        }

        for _ in 1..self.restart_interval {
            let offset = self.hi_scanner.offset;
            let Some(mut reader) = Self::reader_at(self.entries(), offset) else {
                self.poison_back_cursor();
                return;
            };

            #[expect(clippy::expect_used, reason = "base key offset is expected to exist")]
            #[expect(
                clippy::cast_possible_truncation,
                reason = "blocks do not even come close to 4 GiB in size"
            )]
            if Item::parse_truncated(
                &mut reader,
                offset,
                self.hi_scanner.base_key_offset.expect("should exist"),
                self.hi_scanner.base_key_end.expect("should exist"),
                entries_end,
            )
            .inspect(|_| {
                self.hi_scanner.offset += reader.position() as usize;
            })
            .is_some()
            {
                self.hi_scanner.stack.push(offset);
            } else {
                if offset < entries_end {
                    self.poison_back_cursor();
                    return;
                }
                break;
            }
        }
    }

    fn consume_stack_top(&mut self) -> Option<Parsed> {
        let offset = self.hi_scanner.stack.pop()?;
        let entries_end = self.entries_end()?;

        if self.lo_scanner.offset > 0 && offset < self.lo_scanner.offset {
            return None;
        }

        self.hi_scanner.offset = offset;

        let is_restart = self.hi_scanner.stack.is_empty();

        let mut reader = Self::reader_at(self.entries(), offset)?;

        Self::parse_current_item(
            &mut reader,
            offset,
            self.hi_scanner.base_key_offset,
            self.hi_scanner.base_key_end,
            is_restart,
            entries_end,
        )
    }

    pub fn advance_while(&mut self, pred: impl Fn(&Parsed, &[u8]) -> bool) {
        let Some(entries_end) = self.entries_end() else {
            return;
        };

        loop {
            let hi_offset = if self.hi_scanner.base_key_offset.is_some() {
                Some(self.hi_scanner.offset)
            } else {
                None
            };
            if hi_offset.is_some_and(|hi| self.lo_scanner.offset >= hi) {
                break;
            }

            let is_restart = self.lo_scanner.remaining_in_interval == 0;

            let Some(mut reader) = Self::reader_at(self.entries(), self.lo_scanner.offset) else {
                break;
            };

            let Some(item) = Self::parse_current_item(
                &mut reader,
                self.lo_scanner.offset,
                self.lo_scanner.base_key_offset,
                self.lo_scanner.base_key_end,
                is_restart,
                entries_end,
            ) else {
                break;
            };

            if !pred(&item, self.entries()) {
                break;
            }

            #[expect(
                clippy::cast_possible_truncation,
                reason = "blocks do not even come close to 4 GiB in size"
            )]
            {
                let Some(next_offset) = self
                    .lo_scanner
                    .offset
                    .checked_add(reader.position() as usize)
                else {
                    break;
                };
                if hi_offset.is_some_and(|hi| next_offset > hi) {
                    break;
                }
                self.lo_scanner.offset = next_offset;
            }

            if is_restart {
                self.lo_scanner.base_key_offset = Some(item.key_offset());
                self.lo_scanner.base_key_end = Some(item.key_end_offset());
                self.lo_scanner.remaining_in_interval = usize::from(self.restart_interval) - 1;
            } else {
                self.lo_scanner.remaining_in_interval -= 1;
            }
        }
    }

    pub fn trim_back_to_upper_bound(
        &mut self,
        cmp: impl Fn(&Parsed, &[u8]) -> core::cmp::Ordering,
    ) {
        let Some(entries_end) = self.entries_end() else {
            return;
        };

        let mut last_popped = None;

        loop {
            let Some(&offset) = self.hi_scanner.stack.last() else {
                break;
            };

            let is_restart = self.hi_scanner.stack.len() == 1;

            let Some(mut reader) = Self::reader_at(self.entries(), offset) else {
                break;
            };

            let Some(item) = Self::parse_current_item(
                &mut reader,
                offset,
                self.hi_scanner.base_key_offset,
                self.hi_scanner.base_key_end,
                is_restart,
                entries_end,
            ) else {
                break;
            };

            if cmp(&item, self.entries()) != core::cmp::Ordering::Greater {
                break;
            }

            last_popped = Some(offset);
            self.hi_scanner.stack.pop();
        }

        let Some(candidate_offset) = last_popped else {
            return;
        };

        let should_restore = if let Some(&offset) = self.hi_scanner.stack.last() {
            let is_restart = self.hi_scanner.stack.len() == 1;

            let Some(mut reader) = Self::reader_at(self.entries(), offset) else {
                return;
            };

            let Some(item) = Self::parse_current_item(
                &mut reader,
                offset,
                self.hi_scanner.base_key_offset,
                self.hi_scanner.base_key_end,
                is_restart,
                entries_end,
            ) else {
                return;
            };

            cmp(&item, self.entries()) == core::cmp::Ordering::Less
        } else {
            true
        };

        if should_restore {
            // `candidate_offset` is the first item with key > needle.
            //
            // For reverse upper seeks we intentionally keep that first-greater item:
            // it is the covering interval boundary that may still contain `needle`.
            // Dropping it would incorrectly move `next_back()` to the previous item
            // (< needle) and skip the covering block.
            self.hi_scanner.stack.push(candidate_offset);
        }

        let Some(&offset) = self.hi_scanner.stack.last() else {
            return;
        };
        let is_restart = self.hi_scanner.stack.len() == 1;

        let Some(mut reader) = Self::reader_at(self.entries(), offset) else {
            return;
        };

        #[expect(
            clippy::cast_possible_truncation,
            reason = "blocks do not even come close to 4 GiB in size"
        )]
        if Self::parse_current_item(
            &mut reader,
            offset,
            self.hi_scanner.base_key_offset,
            self.hi_scanner.base_key_end,
            is_restart,
            entries_end,
        )
        .is_some()
        {
            self.hi_scanner.offset = offset + reader.position() as usize;
        }
    }

    #[must_use]
    pub fn upper_stack_tail_cmp(
        &self,
        cmp: impl Fn(&Parsed, &[u8]) -> core::cmp::Ordering,
    ) -> Option<core::cmp::Ordering> {
        let &offset = self.hi_scanner.stack.last()?;
        let entries_end = self.entries_end()?;
        let is_restart = self.hi_scanner.stack.len() == 1;

        let mut reader = Self::reader_at(self.entries(), offset)?;

        let item = Self::parse_current_item(
            &mut reader,
            offset,
            self.hi_scanner.base_key_offset,
            self.hi_scanner.base_key_end,
            is_restart,
            entries_end,
        )?;

        Some(cmp(&item, self.entries()))
    }

    #[must_use]
    pub fn advance_upper_restart_interval(&mut self) -> bool {
        let Some(next_idx) = self.hi_scanner.ptr_idx.checked_add(1) else {
            return false;
        };
        let Ok(binary_index_len) = usize::try_from(self.binary_index_len) else {
            return false;
        };
        if next_idx >= binary_index_len {
            return false;
        }

        self.hi_scanner.ptr_idx = next_idx;
        self.hi_scanner.stack.clear();
        self.hi_scanner.base_key_offset = None;
        self.hi_scanner.base_key_end = None;
        self.fill_stack();

        if self.hi_scanner.stack.is_empty() {
            self.clamp_upper_to_lo();
            return false;
        }

        true
    }
}

impl<Item: Decodable<Parsed>, Parsed: ParsedItem<Item>> Iterator for Decoder<'_, Item, Parsed> {
    type Item = Parsed;

    fn next(&mut self) -> Option<Self::Item> {
        let entries_end = self.entries_end()?;
        if self.lo_scanner.offset >= self.entries().len() {
            return None;
        }

        if self.hi_scanner.base_key_offset.is_some()
            && self.lo_scanner.offset >= self.hi_scanner.offset
        {
            return None;
        }

        let is_restart: bool = self.lo_scanner.remaining_in_interval == 0;

        let mut reader = Cursor::new(self.entries().get(self.lo_scanner.offset..)?);

        #[expect(
            clippy::cast_possible_truncation,
            reason = "blocks do not even come close to 4 GiB in size"
        )]
        let item = Self::parse_current_item(
            &mut reader,
            self.lo_scanner.offset,
            self.lo_scanner.base_key_offset,
            self.lo_scanner.base_key_end,
            is_restart,
            entries_end,
        )
        .inspect(|item| {
            self.lo_scanner.offset += reader.position() as usize;

            if is_restart {
                self.lo_scanner.base_key_offset = Some(item.key_offset());
                self.lo_scanner.base_key_end = Some(item.key_end_offset());
            }
        });

        if item.is_some() {
            if is_restart {
                self.lo_scanner.remaining_in_interval = usize::from(self.restart_interval) - 1;
            } else {
                self.lo_scanner.remaining_in_interval -= 1;
            }
        } else {
            self.lo_scanner.offset = self.entries().len();
            self.lo_scanner.remaining_in_interval = 0;
            self.lo_scanner.base_key_offset = None;
            self.lo_scanner.base_key_end = None;
        }

        item
    }
}

impl<Item: Decodable<Parsed>, Parsed: ParsedItem<Item>> DoubleEndedIterator
    for Decoder<'_, Item, Parsed>
{
    fn next_back(&mut self) -> Option<Self::Item> {
        if let Some(top) = self.consume_stack_top() {
            return Some(top);
        }

        // NOTE: If we wrapped, we are at the end
        // This is safe to do, because there cannot be that many restart intervals
        if self.hi_scanner.ptr_idx == usize::MAX {
            return None;
        }

        self.hi_scanner.ptr_idx = self.hi_scanner.ptr_idx.wrapping_sub(1);

        // NOTE: If we wrapped, we are at the end
        // This is safe to do, because there cannot be that many restart intervals
        if self.hi_scanner.ptr_idx == usize::MAX {
            return None;
        }

        self.fill_stack();

        self.consume_stack_top()
    }
}

#[cfg(test)]
#[expect(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::cast_possible_truncation,
    reason = "corruption regression tests intentionally mutate encoded bytes and assert parser rejection paths"
)]
mod tests;