coordinode-lsm-tree 7.3.0

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

#[cfg(not(feature = "std"))]
use alloc::vec::Vec;

use super::{Block, BlockHandle, GlobalTableId};
use crate::path::Path;
use crate::{
    Cache, CompressionType, KeyRange, Table, encryption::EncryptionProvider,
    file_accessor::FileAccessor, table::block::BlockType, version::run::Ranged,
};

#[cfg(feature = "metrics")]
use crate::metrics::Metrics;

/// Returns the bounding key range of a table slice.
///
/// Takes `first().min()` and `last().max()` — no comparison needed because
/// callers pass tables that are already sorted in comparator order (via
/// `push_cmp` / `sort_by_cmp`). Works correctly for any comparator.
#[must_use]
pub fn aggregate_run_key_range(tables: &[Table]) -> KeyRange {
    #[expect(clippy::expect_used, reason = "runs are never empty by definition")]
    let lo = tables.first().expect("run should never be empty");
    #[expect(clippy::expect_used, reason = "runs are never empty by definition")]
    let hi = tables.last().expect("run should never be empty");
    KeyRange::new((lo.key_range().min().clone(), hi.key_range().max().clone()))
}

/// [start, end] slice indexes
#[derive(Debug)]
pub struct SliceIndexes(pub usize, pub usize);

/// Paces the block reads a walk issues, never for a block served from the
/// cache or refused before reading.
pub(crate) trait ReadPacer: Send + Sync {
    /// Whether reads are paced now. Decided once per read: a read decided
    /// unpaced is one read, uncharged.
    fn active(&self) -> bool;

    /// Told the file offset and length of each portion of a paced read just
    /// before that portion is read.
    fn pace(&self, offset: u64, len: u64);

    /// Told how long the read of a paced portion took, whether it returned
    /// bytes or an error. Only a `std` build times the reads.
    fn read_took(&self, _elapsed: core::time::Duration) {}

    /// Told the bytes every read of the walk returned from the file, paced
    /// or not.
    fn read_bytes(&self, _len: u64) {}
}

/// A walk's pacer, shared with the iterators that load its blocks.
pub(crate) type Pacer = alloc::sync::Arc<dyn ReadPacer>;

/// What a read of an [`FsFile`](crate::fs::FsFile) returns in this build.
#[cfg(feature = "std")]
type TimedRead = std::io::Result<usize>;
#[cfg(not(feature = "std"))]
type TimedRead = crate::io::Result<usize>;

/// A file read front to back through a buffer, telling a pacer, while it is
/// active, how long each read takes. An active pacer has the reader take a
/// portion at a time, as a paced verification does, so every sample is one
/// request of comparable size; inactive, a read goes as far as its buffer asks.
pub(crate) struct TimedFile {
    file: alloc::boxed::Box<dyn crate::fs::FsFile>,
    pace: Option<Pacer>,
}

impl TimedFile {
    pub(crate) fn new(file: alloc::boxed::Box<dyn crate::fs::FsFile>) -> Self {
        Self { file, pace: None }
    }

    /// Tells `pace` about every read from now on.
    pub(crate) fn set_pace(&mut self, pace: Pacer) {
        self.pace = Some(pace);
    }

    /// The file, for reads this wrapper does not time.
    pub(crate) fn into_inner(self) -> alloc::boxed::Box<dyn crate::fs::FsFile> {
        self.file
    }

    fn read_timed(&mut self, buf: &mut [u8]) -> TimedRead {
        #[cfg(not(feature = "std"))]
        use crate::io::Read;
        #[cfg(feature = "std")]
        use std::io::Read;

        // no-std: reads are timed only with std (see `ReadPacer::read_took`)
        #[cfg(feature = "std")]
        let read = if let Some(pace) = self.pace.as_ref().filter(|pace| pace.active()) {
            let len = buf.len().min(PACE_PORTION);
            let (portion, _) = buf.split_at_mut(len);
            let started = std::time::Instant::now();
            let read = self.file.read(portion);
            pace.read_took(started.elapsed());
            read
        } else {
            self.file.read(buf)
        };
        #[cfg(not(feature = "std"))]
        let read = self.file.read(buf);
        if let (Some(pace), Ok(n)) = (&self.pace, &read) {
            pace.read_bytes(*n as u64);
        }
        read
    }
}

#[cfg(feature = "std")]
impl std::io::Read for TimedFile {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.read_timed(buf)
    }
}

#[cfg(not(feature = "std"))]
impl crate::io::Read for TimedFile {
    fn read(&mut self, buf: &mut [u8]) -> crate::io::Result<usize> {
        self.read_timed(buf)
    }
}

#[cfg(feature = "std")]
impl std::io::Seek for TimedFile {
    fn seek(&mut self, to: std::io::SeekFrom) -> std::io::Result<u64> {
        self.file.seek(to)
    }
}

#[cfg(not(feature = "std"))]
impl crate::io::Seek for TimedFile {
    fn seek(&mut self, to: crate::io::SeekFrom) -> crate::io::Result<u64> {
        self.file.seek(to)
    }
}

/// The most a paced read takes from the file in one go: a larger read is
/// paced and made a portion at a time, so a shared limiter serves other
/// requests between the portions and the device never sees one read as large
/// as a block.
#[cfg(feature = "std")]
pub(crate) const PACE_PORTION: usize = 64 * 1024;
/// The no_std limiter never waits, so portions would only multiply backend
/// calls: a read there goes whole.
#[cfg(not(feature = "std"))]
pub(crate) const PACE_PORTION: usize = usize::MAX;

/// The table file one read takes its blocks from: taken from the descriptor
/// cache (or opened) at the first block the read has to fetch from the file,
/// and reused by every later one, so a lookup that reads its filter, index and
/// data blocks pays one descriptor-cache access instead of one per block.
///
/// Scoped to one read of one table under one [`ReadCharge`]: it holds the
/// descriptor only while the read runs. A read taken from a [`ScanFd`] starts
/// from the descriptor the scan's earlier reads resolved.
#[derive(Default)]
pub(crate) struct LookupFd {
    /// The descriptor this read resolved, with the descriptor cache's epoch
    /// read just before it was.
    fd: core::cell::OnceCell<(alloc::sync::Arc<dyn crate::fs::FsFile>, u64)>,
    held: Option<HeldFd>,
}

/// A descriptor an earlier read of the same scan resolved, held weakly so it
/// stays open only as long as the descriptor cache (or the pinned table)
/// keeps it, with the descriptor cache's epoch at the time it was resolved.
struct HeldFd {
    file: alloc::sync::Weak<dyn crate::fs::FsFile>,
    epoch: u64,
}

impl LookupFd {
    /// The table's file, resolved on the first call. An untraced read leaves
    /// the descriptor cache as it found it, as it leaves the block cache.
    pub(crate) fn resolve(
        &self,
        file_accessor: &FileAccessor,
        table_id: &GlobalTableId,
        path: &Path,
        charge: ReadCharge,
        #[cfg(feature = "metrics")] metrics: &Metrics,
    ) -> crate::io::Result<&dyn crate::fs::FsFile> {
        if let Some((fd, _)) = self.fd.get() {
            return Ok(fd.as_ref());
        }
        // Read before the descriptor is taken, so a removal that lands while
        // it is taken leaves the descriptor marked as resolved before it.
        let epoch = file_accessor.table_epoch();
        // The scan's descriptor is current while nothing has left the
        // descriptor cache since it was resolved: an in-place heal that
        // renames a copy over the path evicts the old inode's descriptor, and
        // a descriptor the cache evicted for room is closed once no read
        // holds it, which fails the upgrade.
        if let Some(held) = &self.held
            && held.epoch == epoch
            && let Some(fd) = held.file.upgrade()
        {
            return Ok(self.fd.get_or_init(|| (fd, epoch)).0.as_ref());
        }
        let fd = Self::open(
            file_accessor,
            table_id,
            path,
            charge,
            #[cfg(feature = "metrics")]
            metrics,
        )?;
        Ok(self.fd.get_or_init(|| (fd, epoch)).0.as_ref())
    }

    /// The table's file, owned: the one this read resolved, or else taken now,
    /// for a caller that keeps reading through it after the read.
    pub(crate) fn into_file(
        self,
        file_accessor: &FileAccessor,
        table_id: &GlobalTableId,
        path: &Path,
        charge: ReadCharge,
        #[cfg(feature = "metrics")] metrics: &Metrics,
    ) -> crate::io::Result<alloc::sync::Arc<dyn crate::fs::FsFile>> {
        match self.fd.into_inner() {
            Some((fd, _)) => Ok(fd),
            None => Self::open(
                file_accessor,
                table_id,
                path,
                charge,
                #[cfg(feature = "metrics")]
                metrics,
            ),
        }
    }

    /// Takes the table's file from the descriptor cache, or opens it, counting
    /// the access.
    fn open(
        file_accessor: &FileAccessor,
        table_id: &GlobalTableId,
        path: &Path,
        charge: ReadCharge,
        #[cfg(feature = "metrics")] metrics: &Metrics,
    ) -> crate::io::Result<alloc::sync::Arc<dyn crate::fs::FsFile>> {
        let (fd, cache_event) = if charge.touches_cache() {
            file_accessor.get_or_open_table(table_id, path)?
        } else {
            (file_accessor.peek_or_open_table(table_id, path)?, None)
        };

        // Only descriptor-table accesses are cache events; a pinned FD (None)
        // is not.
        #[cfg(feature = "metrics")]
        if let Some(hit) = cache_event
            && charge.is_counted()
        {
            use core::sync::atomic::Ordering::Relaxed;
            if hit {
                metrics.table_file_opened_cached.fetch_add(1, Relaxed);
            } else {
                metrics.table_file_opened_uncached.fetch_add(1, Relaxed);
            }
        }
        #[cfg(not(feature = "metrics"))]
        let _ = cache_event;

        Ok(fd)
    }
}

/// The table file one scan of one table reads its blocks through, carried
/// from each block's [`LookupFd`] to the next: the scan takes the descriptor
/// from the descriptor cache once, rather than once per block it reads from
/// the file, and takes it again only when it is no longer current.
#[derive(Default)]
pub(crate) struct ScanFd(core::cell::Cell<Option<HeldFd>>);

impl ScanFd {
    /// The read of the scan's next block (or index partition), starting from
    /// the descriptor its earlier reads resolved.
    pub(crate) fn lookup(&self) -> LookupFd {
        LookupFd {
            fd: core::cell::OnceCell::new(),
            held: self.0.take(),
        }
    }

    /// Keeps the descriptor `read` resolved for the scan's later reads, or the
    /// one it started from when it needed none (its block came from the
    /// cache).
    pub(crate) fn keep(&self, read: LookupFd) {
        let LookupFd { fd, held } = read;
        let kept = match (fd.into_inner(), held) {
            // The read went through the descriptor it started from: kept as
            // it is, without touching its reference counts again.
            (Some((file, epoch)), Some(held))
                if held.epoch == epoch
                    && core::ptr::addr_eq(held.file.as_ptr(), alloc::sync::Arc::as_ptr(&file)) =>
            {
                Some(held)
            }
            (Some((file, epoch)), _) => Some(HeldFd {
                file: alloc::sync::Arc::downgrade(&file),
                epoch,
            }),
            (None, held) => held,
        };
        self.0.set(kept);
    }
}

/// Loads a block from disk or block cache, if cached.
///
/// Also handles file descriptor opening and caching.
///
/// When this read recovers the payload from its Page-ECC parity (a checksum
/// mismatch the parity repaired) the returned bytes are correct, but the on-disk
/// copy still carries the fault. If `heal_hints` is `Some`, the loader then
/// re-reads the same block straight from disk to tell a persistent medium fault
/// (re-read again recovers) from a transient read-path glitch (re-read clean),
/// and records the SST in the sink on confirmed persistence so the compaction
/// picker can rewrite it clean. `None` disables that scheduling (the payload is
/// still healed and returned). A cache hit never triggers this path: the cached
/// bytes were already verified on their original read.
///
/// `charge` says whose read this is: only a foreground read reaches the read
/// and load counters, and a report neither fills the cache nor promotes what
/// it finds there. Page-ECC recoveries are counted for every reader.
#[expect(
    clippy::too_many_arguments,
    reason = "block loading requires table id, path, file accessor, cache, handle, block type, compression, and heal context"
)]
pub fn load_block(
    table_id: GlobalTableId,
    path: &Path,
    file_accessor: &FileAccessor,
    cache: &Cache,
    handle: &BlockHandle,
    block_type: BlockType,
    compression: CompressionType,
    encryption: Option<&dyn EncryptionProvider>,
    ecc: Option<crate::table::block::EccParams>,
    #[cfg(zstd_any)] zstd_dict: Option<&crate::compression::ZstdDictionary>,
    heal_hints: Option<&crate::heal_hints::HealHints>,
    #[cfg(feature = "metrics")] metrics: &Metrics,
    charge: ReadCharge,
) -> crate::Result<Block> {
    load_block_paced(
        table_id,
        path,
        file_accessor,
        cache,
        handle,
        block_type,
        compression,
        encryption,
        ecc,
        #[cfg(zstd_any)]
        zstd_dict,
        heal_hints,
        #[cfg(feature = "metrics")]
        metrics,
        charge,
        None,
        &LookupFd::default(),
    )
}

/// [`load_block`], with `pace`, when given, told the offset and length of
/// each portion of the read just before that portion is read from the file,
/// and the file taken through `fd`, which the read's other blocks share.
#[expect(
    clippy::too_many_arguments,
    reason = "block loading requires table id, path, file accessor, cache, handle, block type, compression, and heal context"
)]
pub(crate) fn load_block_paced(
    table_id: GlobalTableId,
    path: &Path,
    file_accessor: &FileAccessor,
    cache: &Cache,
    handle: &BlockHandle,
    block_type: BlockType,
    compression: CompressionType,
    encryption: Option<&dyn EncryptionProvider>,
    ecc: Option<crate::table::block::EccParams>,
    #[cfg(zstd_any)] zstd_dict: Option<&crate::compression::ZstdDictionary>,
    heal_hints: Option<&crate::heal_hints::HealHints>,
    #[cfg(feature = "metrics")] metrics: &Metrics,
    charge: ReadCharge,
    pace: Option<&dyn ReadPacer>,
    fd: &LookupFd,
) -> crate::Result<Block> {
    log::trace!("load {block_type:?} block {handle:?}");

    // Invariant: manifest Blocks have their own reader path and
    // never reach the SST block cache. Surface a typed error
    // (rather than panic) so a caller that wires up an SST loader
    // with a manifest BlockType gets a routable failure instead
    // of a process abort. The check stays outside the metrics
    // cfg so the contract holds on every build.
    if matches!(block_type, BlockType::Manifest | BlockType::ManifestFooter) {
        return Err(crate::Error::InvalidTag(("BlockType", block_type.into())));
    }

    let cached = if charge.touches_cache() {
        cache.get_block(table_id, handle.offset())
    } else {
        cache.peek_block(table_id, handle.offset())
    };
    if let Some(block) = cached {
        // Per-KV checking is a header flag, not a block type, so a data
        // block is always BlockType::Data on disk — an exact role match is
        // the right swap-defence check here.
        if block.header.block_type != block_type {
            return Err(crate::Error::InvalidTag((
                "BlockType",
                block.header.block_type.into(),
            )));
        }

        #[cfg(feature = "metrics")]
        if charge.is_counted() {
            record_block_load_cached(metrics, block_type);
        }

        return Ok(block);
    }

    // A read that leaves the cache as it found it loads alone: it must not
    // take a block a foreground read decoded, and its own read verifies
    // the disk afresh. So does a paced read: its pacer may hold it on a rate
    // budget in the middle of the read, which a read waiting on it would
    // wait out too, and which is no measure of how long a load takes.
    let coordinated = charge.touches_cache() && pace.is_none();
    let lead = if coordinated {
        match cache.claim_block(
            table_id,
            handle.offset(),
            crate::cache::LoadIdentity::Block {
                block_type,
                size: handle.size(),
                compression,
                encrypted: encryption.is_some(),
            },
        ) {
            crate::cache::BlockClaim::Shared(block) => {
                #[cfg(feature = "metrics")]
                if charge.is_counted() {
                    metrics
                        .block_load_shared
                        .fetch_add(1, core::sync::atomic::Ordering::Relaxed);
                }
                // The read that led may not have cached the block (a
                // multi-get past the room it keeps for its blocks): this read
                // caches it, as it would have reading the block itself.
                if !cache.has_block(table_id, handle.offset()) {
                    cache.insert_block(table_id, handle.offset(), block.clone());
                }
                return Ok(block);
            }
            crate::cache::BlockClaim::Lead(lead) => {
                // A load that finished between the miss above and this claim
                // left its block in the cache.
                if let Some(block) = cache.get_block(table_id, handle.offset())
                    && block.header.block_type == block_type
                {
                    #[cfg(feature = "metrics")]
                    if charge.is_counted() {
                        record_block_load_cached(metrics, block_type);
                    }
                    lead.publish(&block);
                    return Ok(block);
                }
                Some(lead)
            }
            crate::cache::BlockClaim::Alone => None,
        }
    } else {
        None
    };

    // Timed from here, where it no longer waits on anything but the read and
    // decode, for the estimate that decides whether misses share loads.
    let timer = crate::cache::Cache::time_load();
    let fd = fd.resolve(
        file_accessor,
        &table_id,
        path,
        charge,
        #[cfg(feature = "metrics")]
        metrics,
    )?;

    let transform = build_block_transform(
        compression,
        encryption,
        ecc,
        #[cfg(zstd_any)]
        zstd_dict,
    )?;
    // Charged as the read is issued, before it is validated: a block that then
    // fails its checksum, decryption or decompression was still asked of the
    // filesystem, while a handle refused before reading asked nothing.
    let mut produced = 0;
    let read = Block::from_file_issuing_paced(
        fd,
        *handle,
        crate::table::block::BlockIdentity {
            table_id: table_id.table_id(),
            block_type,
            dict_id: compression.dict_id(),
            window_log: 0,
        },
        &transform,
        || {
            #[cfg(feature = "metrics")]
            if charge.is_counted() {
                record_block_read(metrics, block_type, handle.size().into());
            }
        },
        pace,
        &mut produced,
    );
    // What the transform produced, counted once per block that actually ran
    // one. Paired with the per-role `*_io_requested` above: those record what
    // was asked of the filesystem, this records what came out the other side
    // of decompression, decryption and ECC. The ratio is the compression the
    // read actually paid for, and its absolute value is what separates a
    // physical projection from a cosmetic one. Charged before the result is
    // judged: a transform whose output the length or role check then refuses
    // still ran.
    #[cfg(feature = "metrics")]
    record_block_decoded(metrics, charge, produced);
    let (block, ecc_status, recovery) = read?;
    let block = admit_read_block(
        table_id,
        path,
        file_accessor,
        handle,
        block_type,
        compression,
        encryption,
        ecc,
        #[cfg(zstd_any)]
        zstd_dict,
        heal_hints,
        #[cfg(feature = "metrics")]
        metrics,
        charge,
        block,
        ecc_status,
        recovery,
    )?;
    // A read that must leave no trace feeds no estimate either: its loads
    // would otherwise switch on the waiting of the reads after it.
    if coordinated {
        cache.record_block_load(table_id, handle.offset(), timer);
    }
    if charge.touches_cache() {
        cache.insert_block(table_id, handle.offset(), block.clone());
    }
    if let Some(lead) = lead {
        lead.publish(&block);
    }
    Ok(block)
}

/// Takes a block just read from disk into the engine: counts it, checks its
/// role, and schedules a heal when ECC had to correct it. Caching it is the
/// caller's, which knows the form later reads want it in: a block as read, or
/// a row group's directory already decoded.
///
/// Everything [`load_block`] does after the read and its decode charge, split
/// out so a caller that read several blocks in one request — a columnar row
/// group's directory and pages — accounts for each of them exactly as a block
/// read on its own is accounted for. Two paths that disagreed here would make
/// the counters, the cache and auto-heal depend on how a block happened to be
/// fetched.
#[expect(
    clippy::too_many_arguments,
    reason = "the whole block read context, plus the block it admits"
)]
pub(crate) fn admit_read_block(
    table_id: GlobalTableId,
    path: &Path,
    file_accessor: &FileAccessor,
    handle: &BlockHandle,
    block_type: BlockType,
    compression: CompressionType,
    encryption: Option<&dyn EncryptionProvider>,
    ecc: Option<crate::table::block::EccParams>,
    #[cfg(zstd_any)] zstd_dict: Option<&crate::compression::ZstdDictionary>,
    heal_hints: Option<&crate::heal_hints::HealHints>,
    #[cfg(feature = "metrics")] metrics: &Metrics,
    charge: ReadCharge,
    block: Block,
    ecc_status: crate::table::block::EccStatus,
    recovery: Option<crate::table::block::EccRecoveryKind>,
) -> crate::Result<Block> {
    // Count the on-read ECC recovery (by mechanism) at this primary read site,
    // whoever the reader is: it is a health signal about the medium. The
    // persistence-confirming re-read below goes through a path that does NOT
    // count, so a single fault is counted exactly once.
    #[cfg(feature = "metrics")]
    if let Some(kind) = recovery {
        metrics.record_ecc_recovery(kind);
    }
    #[cfg(not(feature = "metrics"))]
    let _ = recovery;
    let corrected = matches!(ecc_status, crate::table::block::EccStatus::Corrected);

    if block.header.block_type != block_type {
        return Err(crate::Error::InvalidTag((
            "BlockType",
            block.header.block_type.into(),
        )));
    }

    #[cfg(feature = "metrics")]
    if charge.is_counted() {
        record_block_loaded(metrics, block_type);
    }

    // ECC recovered this block's payload from parity. The bytes returned below
    // are correct, but the on-disk copy is still faulty: when auto-heal is on,
    // confirm the fault is persistent and queue the SST for a healing rewrite.
    if corrected {
        maybe_record_persistent_heal(
            table_id,
            path,
            file_accessor,
            handle,
            block_type,
            compression,
            encryption,
            ecc,
            #[cfg(zstd_any)]
            zstd_dict,
            heal_hints,
            #[cfg(feature = "metrics")]
            metrics,
            charge,
        );
    }

    Ok(block)
}

/// Counts one block served from the block cache under its role.
///
/// A columnar page directory counts as index: it is what turns a row group
/// into the addresses of its pages, which is the index's job one level down.
/// Counting it as data would make the data-block hit rate report an address
/// lookup as a payload read.
#[cfg(feature = "metrics")]
pub(crate) fn record_block_load_cached(metrics: &Metrics, block_type: BlockType) {
    use core::sync::atomic::Ordering::Relaxed;
    let cached = match block_type {
        BlockType::Filter => &metrics.filter_block_load_cached,
        BlockType::Index | BlockType::ColumnPageDirectory | BlockType::ColumnZones => {
            &metrics.index_block_load_cached
        }
        BlockType::RangeTombstone => &metrics.range_tombstone_block_load_cached,
        BlockType::Data | BlockType::Meta | BlockType::ColumnPage => {
            &metrics.data_block_load_cached
        }
        // Manifest variants are rejected by `load_block`'s guard before any
        // cache lookup; the remaining sections are loaded once on open, or on
        // a table's drop, via `Block::from_file`, never through this cached
        // path.
        BlockType::Manifest
        | BlockType::ManifestFooter
        | BlockType::BlockLayout
        | BlockType::Locator
        | BlockType::SeqnoBounds
        | BlockType::ZoneMap
        | BlockType::DeleteBitmap
        | BlockType::OwnedBlobObjects => return,
    };
    cached.fetch_add(1, Relaxed);
}

/// Charges one uncached block read to its role's requested-bytes counter:
/// `on_disk` bytes asked of the filesystem. Every path that reads a block
/// without going through the block cache calls this at the point the read is
/// issued, so a read that then fails validation is still counted. Whether the
/// block then loaded is a separate count, [`record_block_loaded`].
#[cfg(feature = "metrics")]
pub(crate) fn record_block_read(metrics: &Metrics, block_type: BlockType, on_disk: u64) {
    use core::sync::atomic::Ordering::Relaxed;
    let requested = match block_type {
        BlockType::Filter => &metrics.filter_block_io_requested,
        BlockType::Index | BlockType::ColumnPageDirectory | BlockType::ColumnZones => {
            &metrics.index_block_io_requested
        }
        BlockType::RangeTombstone => &metrics.range_tombstone_block_io_requested,
        BlockType::Data | BlockType::Meta | BlockType::ColumnPage => {
            &metrics.data_block_io_requested
        }
        // Manifest variants never reach a table read path; the remaining
        // sections are loaded once on open, or on a table's drop, via
        // `Block::from_file`, outside these per-read counters.
        BlockType::Manifest
        | BlockType::ManifestFooter
        | BlockType::BlockLayout
        | BlockType::Locator
        | BlockType::SeqnoBounds
        | BlockType::ZoneMap
        | BlockType::DeleteBitmap
        | BlockType::OwnedBlobObjects => return,
    };
    requested.fetch_add(on_disk, Relaxed);
}

/// Counts one block that came back from disk usable, after its checksum,
/// transform and type checks passed. A read that failed any of them asked the
/// filesystem for bytes ([`record_block_read`]) but loaded nothing, so it must
/// not lower the cache hit rates the load counters feed.
#[cfg(feature = "metrics")]
pub(crate) fn record_block_loaded(metrics: &Metrics, block_type: BlockType) {
    use core::sync::atomic::Ordering::Relaxed;
    let loads = match block_type {
        BlockType::Filter => &metrics.filter_block_load_io,
        BlockType::Index | BlockType::ColumnPageDirectory | BlockType::ColumnZones => {
            &metrics.index_block_load_io
        }
        BlockType::RangeTombstone => &metrics.range_tombstone_block_load_io,
        BlockType::Data | BlockType::Meta | BlockType::ColumnPage => &metrics.data_block_load_io,
        BlockType::Manifest
        | BlockType::ManifestFooter
        | BlockType::BlockLayout
        | BlockType::Locator
        | BlockType::SeqnoBounds
        | BlockType::ZoneMap
        | BlockType::DeleteBitmap
        | BlockType::OwnedBlobObjects => return,
    };
    loads.fetch_add(1, Relaxed);
}

/// Charges what one block's transform produced to `bytes_decoded`, for a
/// counted read. Called as soon as the transform ran, before its output is
/// judged: a transform whose output the length or role check then refuses
/// still ran.
#[cfg(feature = "metrics")]
pub(crate) fn record_block_decoded(metrics: &Metrics, charge: ReadCharge, produced: usize) {
    if charge.is_counted() {
        metrics
            .block_bytes_decoded
            .fetch_add(produced as u64, core::sync::atomic::Ordering::Relaxed);
    }
}

/// Schedules a healing recompaction for `table_id` when a just-read block was
/// ECC-corrected and the fault is confirmed persistent.
///
/// Call this on any read path that recovers a block from parity (full
/// [`load_block`], the zstd partial-decode path, and [`scrub_block`] all feed
/// it). No-op unless `heal_hints` is `Some` and enabled (`auto_heal`). On a
/// corrected read it re-reads the block straight from disk (cache-bypassing) to
/// tell a persistent medium fault from a transient read-path glitch, recording
/// the SST only on confirmed persistence. A re-read error is treated as
/// non-persistent: a genuinely faulty medium resurfaces on the next read, and
/// this must never mask a live-path I/O failure.
///
/// Returns `true` when this call newly queued `table_id` for healing (confirmed
/// persistent fault, scheduling enabled, not already queued). Read paths ignore
/// the return; the patrol scrub uses it to count distinct SSTs it scheduled.
///
/// `charge` says whose read this confirms: a foreground read's confirming
/// re-read goes into the read-byte counters like the read itself, and is
/// charged only once the re-read is actually issued; an untraced one, such as
/// a patrol scrub's, is neither counted nor allowed to touch the caches.
#[expect(
    clippy::too_many_arguments,
    reason = "mirrors the block read context needed for the confirming re-read"
)]
pub(crate) fn maybe_record_persistent_heal(
    table_id: GlobalTableId,
    path: &Path,
    file_accessor: &FileAccessor,
    handle: &BlockHandle,
    block_type: BlockType,
    compression: CompressionType,
    encryption: Option<&dyn EncryptionProvider>,
    ecc: Option<crate::table::block::EccParams>,
    #[cfg(zstd_any)] zstd_dict: Option<&crate::compression::ZstdDictionary>,
    heal_hints: Option<&crate::heal_hints::HealHints>,
    #[cfg(feature = "metrics")] metrics: &Metrics,
    charge: ReadCharge,
) -> bool {
    let Some(hints) = heal_hints else {
        return false;
    };
    if !hints.is_enabled() {
        return false;
    }
    // The confirming re-read of a foreground read is a second request for the
    // whole block, charged as issued like the first, and so is what its
    // transform produced, whether or not the block then checks out.
    let mut decoded = 0;
    let reread = reread_block_is_corrected(
        table_id,
        path,
        file_accessor,
        handle,
        block_type,
        compression,
        encryption,
        ecc,
        #[cfg(zstd_any)]
        zstd_dict,
        #[cfg(feature = "metrics")]
        metrics,
        charge,
        &mut decoded,
    );
    #[cfg(feature = "metrics")]
    if charge.is_counted() {
        metrics
            .block_bytes_decoded
            .fetch_add(decoded as u64, core::sync::atomic::Ordering::Relaxed);
    }
    match reread {
        Ok(corrected) => {
            if !corrected {
                log::debug!(
                    "Transient ECC correction on table {table_id:?} block {handle:?}; \
                     re-read clean, not scheduling"
                );
                return false;
            }
            if hints.record(table_id) {
                #[cfg(feature = "metrics")]
                metrics
                    .ecc_auto_heal_scheduled
                    .fetch_add(1, core::sync::atomic::Ordering::Relaxed);
                log::warn!(
                    "Persistent ECC correction on table {table_id:?} block {handle:?}; \
                     queued for healing recompaction"
                );
                return true;
            }
            false
        }
        Err(e) => {
            log::debug!(
                "ECC re-read confirmation for table {table_id:?} block {handle:?} failed: {e:?}"
            );
            false
        }
    }
}

/// Whose read a block read belongs to, which decides whether it is charged to
/// the read counters and whether it may fill the block cache.
///
/// Page-ECC recoveries are counted whoever read the block: they are a health
/// signal about the medium, not a cost of the read.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum ReadCharge {
    /// A read made for a caller: counted, and cached like any read.
    Foreground,
    /// Maintenance, such as compaction or a patrol scrub: not counted.
    Maintenance,
    /// A read that must leave no trace: not counted, and it neither fills the
    /// block or descriptor cache nor touches the recency of what they hold,
    /// and skips the partial-decode tier, which lives in the block cache. A
    /// monitoring report
    /// (polling it leaves nothing a later read can see), and verification that
    /// must not displace the workload's blocks.
    Untraced,
}

impl ReadCharge {
    /// Whether this read is charged to the read counters.
    #[cfg(feature = "metrics")]
    #[must_use]
    pub(crate) const fn is_counted(self) -> bool {
        matches!(self, Self::Foreground)
    }

    /// Whether this read may leave a trace in the block cache.
    #[must_use]
    pub(crate) const fn touches_cache(self) -> bool {
        !matches!(self, Self::Untraced)
    }
}

/// Outcome of patrol-scrubbing a single block via [`scrub_block`].
//
// Std-gated: the sole consumer (`Table::scrub_data_blocks` → `crate::scrub`)
// is std-only, so gating keeps the no_std build free of dead code.
#[cfg(feature = "std")]
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub(crate) enum BlockScrubOutcome {
    /// The block read clean: its checksum verified with no parity correction
    /// (covers [`EccStatus::Ok`](crate::table::block::EccStatus::Ok) and
    /// [`EccStatus::Unrecognized`](crate::table::block::EccStatus::Unrecognized)
    /// — in both cases the payload verified against its stored checksum).
    Clean,
    /// The block was recovered from its Page-ECC parity (a latent on-disk fault
    /// was corrected in-flight). `scheduled` is `true` when this read confirmed
    /// the fault is persistent and newly queued the owning SST for a healing
    /// rewrite (`auto_heal` on and the SST not already queued).
    Corrected {
        /// `true` iff this block newly scheduled the SST for healing.
        scheduled: bool,
    },
}

/// Reads one block straight from disk (bypassing the block cache) and runs the
/// full Page-ECC verify+correct path, recording a heal hint on a confirmed
/// persistent correction.
///
/// This is the per-block primitive of the patrol scrub: it proactively reads a
/// (typically cold) block so latent single-block bit-rot is corrected on read
/// and the SST scheduled for a clean rewrite before a second fault in the same
/// block exceeds the parity budget. Unlike [`load_block`] it deliberately
/// **bypasses the block cache** in both directions: it neither serves a cached
/// (already-clean) copy that would hide the on-disk fault, nor inserts the cold
/// block and evicts the live working set.
///
/// Returns the [`BlockScrubOutcome`], or `Err` when the block is uncorrectable
/// (checksum failed and parity could not recover it) or unreadable; the caller
/// records that as an uncorrectable finding rather than silently skipping it.
///
/// Scope: FRAME integrity only — checksum, decompress, decrypt, ECC. The
/// block body (restart trailer, entry framing, value-type tags) is
/// deliberately NOT decoded here: a scrub is a lightweight bit-rot patrol,
/// and any body-level fault in a checksum-clean block is (a) not producible
/// by media corruption (the checksum covers the payload) and (b) the domain
/// of callers that fully decode and validate block bodies (e.g. the salvage
/// walk's row materialization). [`crate::verify::verify_sst_file`] is likewise
/// a frame-level verifier. Both the plain patrol scrub and the in-place heal
/// share exactly this depth.
#[cfg(feature = "std")]
#[expect(
    clippy::too_many_arguments,
    reason = "mirrors load_block's read context minus the cache"
)]
pub(crate) fn scrub_block(
    table_id: GlobalTableId,
    path: &Path,
    file_accessor: &FileAccessor,
    handle: &BlockHandle,
    block_type: BlockType,
    compression: CompressionType,
    encryption: Option<&dyn EncryptionProvider>,
    ecc: Option<crate::table::block::EccParams>,
    #[cfg(zstd_any)] zstd_dict: Option<&crate::compression::ZstdDictionary>,
    heal_hints: Option<&crate::heal_hints::HealHints>,
    #[cfg(feature = "metrics")] metrics: &Metrics,
) -> crate::Result<BlockScrubOutcome> {
    // The patrol leaves the descriptor cache as it leaves the block cache:
    // walking every table must not evict the workload's descriptors.
    let fd = file_accessor.peek_or_open_table(&table_id, path)?;
    let transform = build_block_transform(
        compression,
        encryption,
        ecc,
        #[cfg(zstd_any)]
        zstd_dict,
    )?;
    let (block, ecc_status, recovery) = Block::from_file_with_recovery(
        fd.as_ref(),
        *handle,
        crate::table::block::BlockIdentity {
            table_id: table_id.table_id(),
            block_type,
            dict_id: compression.dict_id(),
            window_log: 0,
        },
        &transform,
    )?;
    // Role check, mirroring `load_block`'s swap-defence: an index entry
    // misdirected at another (checksum-valid) block of a different role must
    // surface as an uncorrectable finding, not scrub clean.
    if block.header.block_type != block_type {
        return Err(crate::Error::InvalidTag((
            "BlockType",
            block.header.block_type.into(),
        )));
    }

    Ok(match ecc_status {
        crate::table::block::EccStatus::Corrected => {
            // Primary read for the scrub: count the recovery by mechanism here
            // (the confirming re-read inside maybe_record_persistent_heal does
            // not count).
            #[cfg(feature = "metrics")]
            if let Some(kind) = recovery {
                metrics.record_ecc_recovery(kind);
            }
            #[cfg(not(feature = "metrics"))]
            let _ = recovery;
            let scheduled = maybe_record_persistent_heal(
                table_id,
                path,
                file_accessor,
                handle,
                block_type,
                compression,
                encryption,
                ecc,
                #[cfg(zstd_any)]
                zstd_dict,
                heal_hints,
                #[cfg(feature = "metrics")]
                metrics,
                ReadCharge::Untraced,
            );
            BlockScrubOutcome::Corrected { scheduled }
        }
        crate::table::block::EccStatus::Ok | crate::table::block::EccStatus::Unrecognized => {
            BlockScrubOutcome::Clean
        }
    })
}

/// Builds the [`BlockTransform`](crate::table::block::BlockTransform) for a
/// block read from its per-SST codec context.
///
/// ECC presence is a per-SST descriptor property (`ecc`): the transform is
/// upgraded to its `*Ecc` variant when this SST was written with a recognized
/// Page ECC scheme. On a build WITHOUT the `page_ecc` feature `with_ecc` is the
/// identity function — the parity trailer then reads as an unrecognized opaque
/// trailer (the read frames the payload by `data_length`, verifies its
/// checksum, and reports
/// [`EccStatus::Unrecognized`](crate::table::block::EccStatus::Unrecognized)),
/// so the data still loads without ECC recovery rather than failing closed.
// The lifetime is only elidable when the dictionary parameter is compiled
// out: with it two input references share `'a`, and the transform must borrow
// from both for exactly as long.
#[cfg_attr(
    not(zstd_any),
    expect(
        clippy::elidable_lifetime_names,
        reason = "named to stay valid in the zstd build, where a second reference parameter shares it"
    )
)]
pub(crate) fn build_block_transform<'a>(
    compression: CompressionType,
    encryption: Option<&'a dyn EncryptionProvider>,
    ecc: Option<crate::table::block::EccParams>,
    #[cfg(zstd_any)] zstd_dict: Option<&'a crate::compression::ZstdDictionary>,
) -> crate::Result<crate::table::block::BlockTransform<'a>> {
    let t = crate::table::block::BlockTransform::from_parts(
        compression,
        encryption,
        #[cfg(zstd_any)]
        zstd_dict,
    )?;
    Ok(if let Some(ecc) = ecc {
        t.with_ecc(ecc)
    } else {
        t
    })
}

/// Re-reads a block straight from disk (bypassing the block cache) and reports
/// whether its on-disk bytes are *still* ECC-corrected.
///
/// Used to confirm that a correction observed on a cache-miss read reflects a
/// **persistent** on-disk fault rather than a transient read-path glitch (bad
/// RAM / DMA / cable during the first read): a second independent read of the
/// same offset that again recovers from parity proves the bytes on the medium
/// are faulty. Returns whether the re-read was itself ECC-corrected (`false`
/// when it read clean, a transient fault, or carried no recognized parity),
/// and the length the re-read decoded to, for the caller's byte counters.
///
/// Runs only on the cold corrected-read path, so the extra disk read costs
/// nothing on clean reads.
#[expect(
    clippy::too_many_arguments,
    reason = "mirrors load_block's read context minus the cache"
)]
fn reread_block_is_corrected(
    table_id: GlobalTableId,
    path: &Path,
    file_accessor: &FileAccessor,
    handle: &BlockHandle,
    block_type: BlockType,
    compression: CompressionType,
    encryption: Option<&dyn EncryptionProvider>,
    ecc: Option<crate::table::block::EccParams>,
    #[cfg(zstd_any)] zstd_dict: Option<&crate::compression::ZstdDictionary>,
    #[cfg(feature = "metrics")] metrics: &Metrics,
    charge: ReadCharge,
    decoded: &mut usize,
) -> crate::Result<bool> {
    // The confirmation belongs to the read it confirms: an untraced one leaves
    // the descriptor cache as it found it.
    let fd = if charge.touches_cache() {
        file_accessor.get_or_open_table(&table_id, path)?.0
    } else {
        file_accessor.peek_or_open_table(&table_id, path)?
    };
    let transform = build_block_transform(
        compression,
        encryption,
        ecc,
        #[cfg(zstd_any)]
        zstd_dict,
    )?;
    let (_block, ecc_status, _recovery) = Block::from_file_issuing(
        fd.as_ref(),
        *handle,
        crate::table::block::BlockIdentity {
            table_id: table_id.table_id(),
            block_type,
            dict_id: compression.dict_id(),
            window_log: 0,
        },
        &transform,
        // Charged when the read is issued, not before the descriptor opens.
        || {
            #[cfg(feature = "metrics")]
            if charge.is_counted() {
                record_block_read(metrics, block_type, handle.size().into());
            }
        },
        decoded,
    )?;
    Ok(matches!(
        ecc_status,
        crate::table::block::EccStatus::Corrected
    ))
}

// Cached, `no_std`-friendly CPU-feature tokens for the SIMD dispatch below.
// `cpufeatures::new!` generates a module whose `get()` runs CPUID once (atomic
// load thereafter) AND verifies OS AVX-state enablement via XCR0, so the AVX
// paths cannot SIGILL on a CPU/OS that does not actually support them. These
// work identically under `std` and `no_std` — which is why the dispatch is no
// longer `std`-gated and `no_std` x86 builds keep the wide SIMD lanes instead
// of dropping to the scalar tail.
#[cfg(any(target_arch = "x86_64", target_arch = "x86"))]
cpufeatures::new!(cpu_avx512bw, "avx512bw");
#[cfg(any(target_arch = "x86_64", target_arch = "x86"))]
cpufeatures::new!(cpu_avx2, "avx2");
// SSE2 is the mandatory `x86_64` ISA baseline, so it is only runtime-detected on
// 32-bit x86 (pre-Pentium-4 may lack it); `x86_64` takes the SSE2 arm directly.
#[cfg(target_arch = "x86")]
cpufeatures::new!(cpu_sse2, "sse2");

/// Returns the length of the longest shared byte prefix of `s1` and `s2`.
///
/// This is on the hot path of block encoding during flush and compaction —
/// every truncated entry pays one call against the restart base key.
///
/// Dispatch:
/// - **`x86_64` / `x86` with AVX-512BW** (runtime-detected): 64-byte vectorized lanes via
///   `_mm512_cmpeq_epi8_mask`. Checked first so AVX-512 hosts use the widest lane.
/// - **`x86_64` / `x86` with AVX2** (runtime-detected): 32-byte vectorized lanes via `_mm256_cmpeq_epi8`.
/// - **`x86_64` with SSE2**: 16-byte lanes via `_mm_cmpeq_epi8` — SSE2 is the mandatory `x86_64`
///   ISA baseline, so this path needs no runtime check, only the AVX2 negative result.
/// - **`x86` (32-bit) with SSE2** (runtime-detected): same 16-byte kernel, but SSE2 is *not*
///   guaranteed on 32-bit x86 (pre-Pentium-4 lacks it), so it is runtime-detected; pre-SSE2
///   hosts fall through to the scalar kernel below.
/// - **`aarch64` little-endian**: 16-byte vectorized lanes via NEON (`ARMv8` baseline — no runtime check).
/// - **Everything else** (incl. big-endian aarch64, pre-SSE2 32-bit x86, riscv, powerpc): 8-byte word
///   stride via XOR. First-mismatch position uses `trailing_zeros() / 8` on little-endian
///   targets and `leading_zeros() / 8` on big-endian, so the byte ordering of the word matches
///   the byte ordering of the source slice on either endianness.
///
/// CPU-feature detection is cached (one CPUID on first call via `cpufeatures`,
/// an atomic load thereafter) and works under `no_std`, so the per-call dispatch
/// cost is one to three cached atomic loads on x86 and the wide lanes are taken
/// on capable hosts regardless of whether the `std` feature is enabled.
///
/// This convenience wrapper resolves the kernel on every call. In a hot loop,
/// resolve once with [`resolve_lsp_kernel`] at loop entry and call the returned
/// pointer per item so feature detection does not run per comparison.
#[must_use]
pub fn longest_shared_prefix_length(s1: &[u8], s2: &[u8]) -> usize {
    // SAFETY: `resolve_lsp_kernel` only returns a kernel whose required CPU
    // feature it has just verified present on this host (or the always-safe
    // scalar kernel), so invoking it on this same host cannot execute an
    // unsupported instruction.
    unsafe { resolve_lsp_kernel()(s1, s2) }
}

/// Function-pointer type for the longest-shared-prefix kernel.
///
/// `unsafe` because the SIMD kernels require their target CPU feature to be
/// present; [`resolve_lsp_kernel`] only ever hands back a pointer whose feature
/// it verified (or the always-safe scalar kernel), so a caller invoking the
/// returned pointer on the same host upholds that contract.
pub type LspKernel = unsafe fn(&[u8], &[u8]) -> usize;

/// Resolves the best longest-shared-prefix kernel for the current CPU **once**
/// and returns it as a function pointer.
///
/// Call this at the entry to a hot loop (e.g. when a block encoder is built) and
/// invoke the returned pointer per item, so CPU-feature detection runs a single
/// time instead of on every comparison. Selection order matches
/// [`longest_shared_prefix_length`]: AVX-512BW, then AVX2, then SSE2 (the
/// `x86_64` baseline; runtime-detected on 32-bit `x86`), then NEON on LE
/// `aarch64`, then the portable scalar kernel.
#[must_use]
pub fn resolve_lsp_kernel() -> LspKernel {
    // Each `cfg` block below is a tail expression and exactly one compiles per
    // target, so there is no `return`/unreachable-tail bookkeeping. The x86
    // path detects features at runtime and defers the actual choice to the pure
    // `select_lsp_kernel`, which is exhaustively unit-tested on any host (the
    // detection itself can only exercise the current CPU's lane).
    #[cfg(any(target_arch = "x86_64", target_arch = "x86"))]
    {
        select_lsp_kernel(LspCpuFeatures {
            avx512bw: cpu_avx512bw::get(),
            avx2: cpu_avx2::get(),
            // SSE2 is the mandatory x86_64 ISA baseline (always present); on
            // 32-bit x86 it is runtime-detected.
            #[cfg(target_arch = "x86_64")]
            sse2: true,
            #[cfg(target_arch = "x86")]
            sse2: cpu_sse2::get(),
        })
    }
    #[cfg(all(target_arch = "aarch64", target_endian = "little"))]
    {
        // NEON is mandatory in the ARMv8 baseline `target_arch = "aarch64"` implies.
        lsp_neon
    }
    #[cfg(not(any(
        target_arch = "x86_64",
        target_arch = "x86",
        all(target_arch = "aarch64", target_endian = "little")
    )))]
    {
        // Portable fallback (big-endian aarch64, riscv, powerpc, …).
        lsp_scalar
    }
}

/// Detected x86 SIMD features the longest-shared-prefix dispatch selects on.
#[cfg(any(target_arch = "x86_64", target_arch = "x86"))]
#[derive(Clone, Copy)]
struct LspCpuFeatures {
    avx512bw: bool,
    avx2: bool,
    sse2: bool,
}

/// Pure kernel selection from detected x86 features: widest lane first
/// (AVX-512BW, then AVX2, then SSE2), else the portable scalar kernel.
///
/// Split out from [`resolve_lsp_kernel`]'s runtime detection so every branch is
/// exhaustively unit-testable on any host — the detection can only ever
/// exercise the lane the test CPU happens to expose.
#[cfg(any(target_arch = "x86_64", target_arch = "x86"))]
fn select_lsp_kernel(f: LspCpuFeatures) -> LspKernel {
    if f.avx512bw {
        lsp_avx512
    } else if f.avx2 {
        lsp_avx2
    } else if f.sse2 {
        lsp_sse2
    } else {
        lsp_scalar
    }
}

/// 8-byte word-stride scalar implementation — works on every platform, no intrinsics.
///
/// Compares 8 bytes at a time via `u64` XOR and locates the first mismatching byte
/// using an endian-aware bit-count:
/// - **Little-endian** (`target_endian = "little"`): `trailing_zeros() / 8` — the
///   lowest-numbered bit in the XOR word corresponds to the first source byte.
/// - **Big-endian** (`target_endian = "big"`): `leading_zeros() / 8` — the highest-numbered
///   bit in the XOR word corresponds to the first source byte.
///
/// Tail shorter than 8 bytes falls back to a byte-by-byte loop.
// On little-endian aarch64 the NEON kernel is selected unconditionally (NEON is
// in the ARMv8 baseline), so the portable scalar fallback has no caller on that
// target alone — it is still the live kernel on x86 without SSE2, big-endian
// aarch64, riscv, powerpc, etc. `expect` (not `allow`) keeps this honest: it
// errors if the fallback ever becomes reachable here, or if it stops being dead.
#[cfg_attr(
    all(target_arch = "aarch64", target_endian = "little", not(test)),
    expect(
        dead_code,
        reason = "portable LSP fallback; aarch64-LE always selects the NEON kernel \
                  (unit tests still call it directly, hence not(test))"
    )
)]
#[must_use]
pub(crate) fn lsp_scalar(s1: &[u8], s2: &[u8]) -> usize {
    let min_len = s1.len().min(s2.len());
    let mut i = 0;

    while i + 8 <= min_len {
        // SAFETY: i + 8 <= min_len <= s{1,2}.len() — both 8-byte reads are in-bounds.
        // `read_unaligned` documents that the pointer needs no alignment, so the
        // `*const u8 -> *const u64` cast is sound. Clippy's `cast_ptr_alignment`
        // does NOT fire here (verified across all CI targets including BE powerpc64)
        // because the cast feeds directly into `read_unaligned`, which clippy
        // recognises as an unaligned-load idiom.
        #[expect(unsafe_code, reason = "bounds checked by loop guard above")]
        let (a, b) = unsafe {
            (
                s1.as_ptr().add(i).cast::<u64>().read_unaligned(),
                s2.as_ptr().add(i).cast::<u64>().read_unaligned(),
            )
        };
        let diff = a ^ b;
        if diff != 0 {
            // Endian-independent: position of first byte-level difference.
            // On LE the lowest mismatching byte is at trailing_zeros / 8;
            // on BE it is at leading_zeros / 8. Use the matching primitive.
            #[cfg(target_endian = "little")]
            let byte_off = (diff.trailing_zeros() / 8) as usize;
            #[cfg(target_endian = "big")]
            let byte_off = (diff.leading_zeros() / 8) as usize;
            return i + byte_off;
        }
        i += 8;
    }

    while i < min_len {
        // SAFETY: i < min_len <= s{1,2}.len()
        #[expect(unsafe_code, reason = "i < min_len bounds-checked above")]
        let (a, b) = unsafe { (*s1.get_unchecked(i), *s2.get_unchecked(i)) };
        if a != b {
            return i;
        }
        i += 1;
    }

    min_len
}

/// AVX2 implementation — 32 bytes per iteration via `_mm256_cmpeq_epi8`.
///
/// # Safety
///
/// Caller must ensure the host CPU supports AVX2 (`cpu_avx2::get()`).
// Compiled on every x86 / x86_64 build (std and no_std): the runtime dispatch
// above selects it via cached CPU-feature detection, and the #[cfg(test)]
// kernel tests below exercise it directly.
#[cfg(any(target_arch = "x86_64", target_arch = "x86"))]
#[target_feature(enable = "avx2")]
#[expect(unsafe_code, reason = "intrinsics require unsafe")]
#[must_use]
unsafe fn lsp_avx2(s1: &[u8], s2: &[u8]) -> usize {
    #[cfg(target_arch = "x86")]
    use core::arch::x86::{__m256i, _mm256_cmpeq_epi8, _mm256_loadu_si256, _mm256_movemask_epi8};
    #[cfg(target_arch = "x86_64")]
    use core::arch::x86_64::{
        __m256i, _mm256_cmpeq_epi8, _mm256_loadu_si256, _mm256_movemask_epi8,
    };

    let min_len = s1.len().min(s2.len());
    let mut i = 0;

    while i + 32 <= min_len {
        // SAFETY: i + 32 <= min_len ≤ s{1,2}.len() — both 32-byte loads are in-bounds.
        // `_mm256_loadu_si256` is the *unaligned* load, so the u8→__m256i pointer cast
        // does not require 32-byte alignment (the pointer is only used by `loadu`).
        #[expect(
            clippy::cast_ptr_alignment,
            reason = "_mm256_loadu_si256 explicitly performs an unaligned 32-byte load"
        )]
        let (va, vb) = unsafe {
            (
                _mm256_loadu_si256(s1.as_ptr().add(i).cast::<__m256i>()),
                _mm256_loadu_si256(s2.as_ptr().add(i).cast::<__m256i>()),
            )
        };
        // Register-only AVX2 intrinsics under #[target_feature(enable = "avx2")] —
        // no `unsafe` block needed; the function-level `unsafe` covers their availability.
        let cmp = _mm256_cmpeq_epi8(va, vb);
        // `_mm256_movemask_epi8` returns the byte-mask as a signed `i32`. We treat the
        // bit pattern as `u32` for trailing-zeros math — `cast_unsigned()` makes the
        // sign-preserving reinterpretation explicit.
        let mask = _mm256_movemask_epi8(cmp).cast_unsigned();
        if mask != u32::MAX {
            return i + (!mask).trailing_zeros() as usize;
        }
        i += 32;
    }

    // Tail: byte-stride (≤31 bytes left, not worth dispatching a narrower kernel).
    while i < min_len {
        // SAFETY: i < min_len ≤ s{1,2}.len()
        let (a, b) = unsafe { (*s1.get_unchecked(i), *s2.get_unchecked(i)) };
        if a != b {
            return i;
        }
        i += 1;
    }

    min_len
}

/// AVX-512BW implementation — 64 bytes per iteration via `_mm512_cmpeq_epi8_mask`.
///
/// The widest `x86_64` lane: one iteration consumes a full 64-byte cache line, so
/// keys that share a long prefix (time-series, tenant-prefixed, sorted UUIDs)
/// settle in half the iterations of the AVX2 kernel. `_mm512_cmpeq_epi8_mask`
/// folds the 64-lane byte comparison directly into a `__mmask64`, avoiding the
/// separate `movemask` step the AVX2/SSE2 kernels need.
///
/// # Safety
///
/// Caller must ensure the host CPU supports AVX-512BW (`cpu_avx512bw::get()`).
/// BW implies the F subset, so the 512-bit load and the byte-granular
/// compare-mask are both available.
// See `lsp_avx2`: compiled on every x86 build (std and no_std).
#[cfg(any(target_arch = "x86_64", target_arch = "x86"))]
// List both ISA features the body relies on: `_mm512_loadu_si512` is AVX-512F,
// `_mm512_cmpeq_epi8_mask` is AVX-512BW. BW implies F (so `avx512bw` alone would
// compile), but naming both keeps the gate matching the actual requirements and
// guards against a future edit dropping the BW-only compare without noticing the
// F load is still gated. Runtime detection on `avx512bw` is sufficient because
// any CPU exposing BW necessarily implements F.
#[target_feature(enable = "avx512bw,avx512f")]
#[expect(unsafe_code, reason = "intrinsics require unsafe")]
#[must_use]
unsafe fn lsp_avx512(s1: &[u8], s2: &[u8]) -> usize {
    #[cfg(target_arch = "x86")]
    use core::arch::x86::{__m512i, _mm512_cmpeq_epi8_mask, _mm512_loadu_si512};
    #[cfg(target_arch = "x86_64")]
    use core::arch::x86_64::{__m512i, _mm512_cmpeq_epi8_mask, _mm512_loadu_si512};

    let min_len = s1.len().min(s2.len());
    let mut i = 0;

    while i + 64 <= min_len {
        // SAFETY: i + 64 <= min_len ≤ s{1,2}.len() — both 64-byte loads are in-bounds.
        // `_mm512_loadu_si512` is the *unaligned* load, so the u8→__m512i pointer cast
        // does not require 64-byte alignment (the pointer is only used by `loadu`).
        #[expect(
            clippy::cast_ptr_alignment,
            reason = "_mm512_loadu_si512 explicitly performs an unaligned 64-byte load"
        )]
        let (va, vb) = unsafe {
            (
                _mm512_loadu_si512(s1.as_ptr().add(i).cast::<__m512i>()),
                _mm512_loadu_si512(s2.as_ptr().add(i).cast::<__m512i>()),
            )
        };
        // `_mm512_cmpeq_epi8_mask` yields a 64-bit mask: bit j is set iff byte j is
        // equal. A full-match lane is `u64::MAX`; the first mismatch is the lowest
        // zero bit of the mask, i.e. the lowest set bit of its complement.
        let mask = _mm512_cmpeq_epi8_mask(va, vb);
        if mask != u64::MAX {
            return i + (!mask).trailing_zeros() as usize;
        }
        i += 64;
    }

    // Tail: byte-stride (≤63 bytes left, not worth dispatching a narrower kernel).
    while i < min_len {
        // SAFETY: i < min_len ≤ s{1,2}.len()
        let (a, b) = unsafe { (*s1.get_unchecked(i), *s2.get_unchecked(i)) };
        if a != b {
            return i;
        }
        i += 1;
    }

    min_len
}

/// SSE2 implementation — 16 bytes per iteration via `_mm_cmpeq_epi8`.
///
/// Used on `x86_64` hosts that lack AVX2 (older Intel Atoms, some sandboxed
/// VMs / containers, AMD pre-Excavator, low-power embedded `x86_64`) and on
/// 32-bit `x86` hosts with SSE2 but without AVX2.
///
/// # Safety
///
/// Caller must ensure the host supports SSE2. On `x86_64` this is the mandatory
/// ISA baseline (always true); on 32-bit `x86` it must be runtime-detected via
/// `cpu_sse2::get()` because pre-Pentium-4 CPUs lack it.
// See `lsp_avx2`: compiled on every x86 build (std and no_std).
#[cfg(any(target_arch = "x86_64", target_arch = "x86"))]
#[target_feature(enable = "sse2")]
#[expect(unsafe_code, reason = "intrinsics require unsafe")]
#[must_use]
unsafe fn lsp_sse2(s1: &[u8], s2: &[u8]) -> usize {
    #[cfg(target_arch = "x86")]
    use core::arch::x86::{__m128i, _mm_cmpeq_epi8, _mm_loadu_si128, _mm_movemask_epi8};
    #[cfg(target_arch = "x86_64")]
    use core::arch::x86_64::{__m128i, _mm_cmpeq_epi8, _mm_loadu_si128, _mm_movemask_epi8};

    let min_len = s1.len().min(s2.len());
    let mut i = 0;

    while i + 16 <= min_len {
        // SAFETY: i + 16 <= min_len ≤ s{1,2}.len() — both 16-byte loads are in-bounds.
        // `_mm_loadu_si128` is the *unaligned* load, so the u8→__m128i pointer cast
        // does not require 16-byte alignment (the pointer is only used by `loadu`).
        #[expect(
            clippy::cast_ptr_alignment,
            reason = "_mm_loadu_si128 explicitly performs an unaligned 16-byte load"
        )]
        let (va, vb) = unsafe {
            (
                _mm_loadu_si128(s1.as_ptr().add(i).cast::<__m128i>()),
                _mm_loadu_si128(s2.as_ptr().add(i).cast::<__m128i>()),
            )
        };
        // Register-only SSE2 intrinsics under #[target_feature(enable = "sse2")] —
        // safe in stable Rust without an inner `unsafe` block.
        let cmp = _mm_cmpeq_epi8(va, vb);
        // `_mm_movemask_epi8` returns the 16-bit byte-mask as a signed `i32`
        // (low 16 bits used, high 16 zero). Reinterpret as `u32` for trailing-zeros math.
        let mask = _mm_movemask_epi8(cmp).cast_unsigned();
        // SSE2 mask is 16 bits, so a full-match lane is `0xFFFF`, not `u32::MAX`.
        if mask != 0xFFFF {
            return i + (!mask).trailing_zeros() as usize;
        }
        i += 16;
    }

    // Tail: byte-stride (≤15 bytes left).
    while i < min_len {
        // SAFETY: i < min_len ≤ s{1,2}.len()
        let (a, b) = unsafe { (*s1.get_unchecked(i), *s2.get_unchecked(i)) };
        if a != b {
            return i;
        }
        i += 1;
    }

    min_len
}

/// NEON implementation — 16 bytes per iteration via `vceqq_u8` + byte-wise mask reduction.
///
/// Restricted to **little-endian** aarch64 because the lane-to-memory mapping of
/// `vgetq_lane_u64` and the `trailing_zeros() / 8` mismatch-position math both
/// assume LE byte order. Big-endian aarch64 falls back to the scalar kernel.
///
/// # Safety
///
/// NEON is part of the `ARMv8` baseline and is always available on `target_arch = "aarch64"`,
/// so no runtime detection is needed. The `unsafe` is required only because the intrinsics
/// themselves are `unsafe fn`.
#[cfg(all(target_arch = "aarch64", target_endian = "little"))]
#[target_feature(enable = "neon")]
#[expect(unsafe_code, reason = "intrinsics require unsafe")]
#[must_use]
unsafe fn lsp_neon(s1: &[u8], s2: &[u8]) -> usize {
    use core::arch::aarch64::{
        vandq_u8, vceqq_u8, vdupq_n_u8, vgetq_lane_u64, vld1q_u8, vreinterpretq_u64_u8,
    };

    let min_len = s1.len().min(s2.len());
    let mut i = 0;

    // 16-byte equality mask: lanes are 0xFF when bytes match, 0x00 when they differ.
    // Reduce to a 128-bit value and inspect its halves as u64 for first-mismatch position.
    while i + 16 <= min_len {
        // SAFETY: i + 16 <= min_len ≤ s{1,2}.len() — both 16-byte loads are in-bounds.
        let (va, vb) = unsafe { (vld1q_u8(s1.as_ptr().add(i)), vld1q_u8(s2.as_ptr().add(i))) };
        // Register-only NEON intrinsics — safe in stable Rust under the `neon` target feature.
        let cmp = vceqq_u8(va, vb);
        // Trim to bit-per-byte mask via AND with 0xFF (no-op for the equality result,
        // but keeps the intent explicit); reinterpret as two u64 halves.
        let masked = vandq_u8(cmp, vdupq_n_u8(0xFF));
        let as_u64 = vreinterpretq_u64_u8(masked);
        let lo = vgetq_lane_u64(as_u64, 0);
        let hi = vgetq_lane_u64(as_u64, 1);

        if lo != u64::MAX {
            // First mismatching byte is in the low half.
            return i + (!lo).trailing_zeros() as usize / 8;
        }
        if hi != u64::MAX {
            // First mismatching byte is in the high half.
            return i + 8 + (!hi).trailing_zeros() as usize / 8;
        }
        i += 16;
    }

    // Tail: byte-stride for the ≤15 remaining bytes.
    while i < min_len {
        // SAFETY: i < min_len ≤ s{1,2}.len()
        let (a, b) = unsafe { (*s1.get_unchecked(i), *s2.get_unchecked(i)) };
        if a != b {
            return i;
        }
        i += 1;
    }

    min_len
}

/// Compares the conceptual concatenation `prefix + suffix` against `needle`
/// using the given comparator.
///
/// For the default lexicographic comparator this performs a zero-allocation
/// bytewise comparison. Custom comparators fall back to concatenating prefix
/// and suffix into a temporary `Vec` so that `UserComparator::compare` always
/// receives a complete key.
#[must_use]
pub fn compare_prefixed_slice(
    prefix: &[u8],
    suffix: &[u8],
    needle: &[u8],
    cmp: &dyn crate::comparator::UserComparator,
) -> core::cmp::Ordering {
    // Fast path: zero-allocation bytewise comparison for the default
    // (lexicographic) comparator. This is the hot path for block index
    // and data block binary searches.
    if cmp.is_lexicographic() {
        return compare_prefixed_slice_lexicographic(prefix, suffix, needle);
    }

    // Slow path: materialize prefix+suffix into a contiguous buffer for
    // custom comparators. Uses a stack buffer for typical key sizes to
    // avoid heap allocation on the hot binary-search path.
    let total_len = prefix.len() + suffix.len();

    if total_len <= 256 {
        let mut buf = [0_u8; 256];

        // SAFETY (indexing): total_len <= 256 == buf.len(), and
        // prefix.len() + suffix.len() == total_len, so all slices are in bounds.
        #[expect(clippy::indexing_slicing, reason = "total_len <= 256 checked above")]
        {
            buf[..prefix.len()].copy_from_slice(prefix);
            buf[prefix.len()..total_len].copy_from_slice(suffix);
        }

        #[expect(clippy::indexing_slicing, reason = "total_len <= 256 checked above")]
        return cmp.compare(&buf[..total_len], needle);
    }

    // Fallback for unusually large keys: allocate a temporary Vec.
    let mut full_key = Vec::with_capacity(total_len);
    full_key.extend_from_slice(prefix);
    full_key.extend_from_slice(suffix);
    cmp.compare(&full_key, needle)
}

/// Zero-allocation lexicographic comparison of `prefix + suffix` against `needle`.
#[must_use]
fn compare_prefixed_slice_lexicographic(
    prefix: &[u8],
    suffix: &[u8],
    needle: &[u8],
) -> core::cmp::Ordering {
    use core::cmp::Ordering::{Equal, Greater};

    if needle.is_empty() {
        let combined_len = prefix.len() + suffix.len();
        return if combined_len > 0 { Greater } else { Equal };
    }

    let max_pfx_len = prefix.len().min(needle.len());

    {
        // SAFETY: max_pfx_len = min(prefix.len(), needle.len()), so both
        // slices [0..max_pfx_len] are within bounds by construction.
        #[expect(
            unsafe_code,
            reason = "max_pfx_len <= prefix.len() && max_pfx_len <= needle.len()"
        )]
        let pfx = unsafe { prefix.get_unchecked(0..max_pfx_len) };

        #[expect(
            unsafe_code,
            reason = "max_pfx_len <= prefix.len() && max_pfx_len <= needle.len()"
        )]
        let ndl = unsafe { needle.get_unchecked(0..max_pfx_len) };

        match pfx.cmp(ndl) {
            Equal => {}
            ordering => return ordering,
        }
    }

    // Clamp-to-zero: when `needle` is longer than `prefix` there is no remainder
    // (the prefix is exhausted), which the `> 0` check below treats correctly.
    let rest_len = prefix.len().saturating_sub(needle.len());
    if rest_len > 0 {
        return Greater;
    }

    // SAFETY: rest_len == 0 means prefix.len() <= needle.len(), so
    // max_pfx_len == prefix.len() <= needle.len() and needle[max_pfx_len..] is in-bounds.
    #[expect(
        unsafe_code,
        reason = "max_pfx_len <= needle.len() guaranteed by rest_len == 0 guard above"
    )]
    let remaining_needle = unsafe { needle.get_unchecked(max_pfx_len..) };
    suffix.cmp(remaining_needle)
}

#[cfg(test)]
mod tests;