coordinode-lsm-tree 5.3.0

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

//! Out-of-band inspection of a single SST file.
//!
//! Companion to [`crate::verify`]: while `verify_block_checksums` walks
//! every block and checks per-block XXH3, this module exposes a public
//! read-only view of an SST's stored metadata ([`TableProperties`](crate::inspect::TableProperties)) for
//! diagnostic tooling like `sst-dump properties` that needs the
//! metadata fields without spinning up a [`Tree`](crate::Tree) or
//! recovering the manifest.
//!
//! The reader follows the same TAIL-first / MID-fallback pattern as
//! [`Table::recover`](crate::Table) (see PR #295): if the canonical
//! `meta` section at the file tail fails to decode, the MID-mirror
//! `meta_mid` section is attempted, and only if both copies are
//! unreadable does the call return an error.

use crate::CompressionType;
// `Fs` brought in for trait method resolution: the `fs.open(path, ...)`
// call below dispatches through the `Fs` trait, so the trait must be
// in scope. Removing it breaks the build with `method `open` not found
// for this struct` (rustc E0599). rustc correctly classifies this as
// USED — no `#[allow]` / `#[expect]` is needed; static-analysis
// passes that flag it as unused are false positives.
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::meta::ParsedMeta;
use crate::table::regions::ParsedRegions;
use std::path::Path;

/// Read-only snapshot of an SST file's stored metadata.
///
/// Constructed by [`read_table_properties`]; not directly creatable by
/// external callers. Fields are a stable, documented subset of the
/// on-disk meta block (see
/// `src/table/writer/mod.rs::write_meta_section` for the full set of
/// emitted entries). The internal `ParsedMeta` parser carries
/// additional fields — notably the seqno range — that are not yet
/// exposed here; those are tracked as a separate API surface to keep
/// the public contract small while the meta layout still evolves.
/// `#[non_exhaustive]` so new fields can be added in a minor version
/// bump.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct TableProperties {
    /// Per-tree unique table id (the SST file's logical identifier).
    /// Plain `u64` rather than the crate-internal `TableId` alias so
    /// the public API does not couple to a `#[doc(hidden)]` type.
    pub id: u64,
    /// Logical size of the data-blocks region as recorded by the
    /// writer. This is `*self.meta.file_pos` at the moment the meta
    /// section was flushed — i.e. the byte offset just past the last
    /// data block — and does NOT include the index / filter /
    /// range-tombstone / linked-blob / meta / SFA-trailer sections
    /// that follow. To get the actual on-disk file size, `stat` the
    /// file directly. See `write_meta_section`'s `file_size` field
    /// for the writer-side definition.
    pub file_size: u64,
    /// Smallest user key present in the table. Owned `Vec<u8>` rather
    /// than the crate-internal `KeyRange` / `UserKey` types so the
    /// public API does not couple to `#[doc(hidden)]` re-exports.
    pub min_key: Vec<u8>,
    /// Largest user key present in the table. See [`Self::min_key`]
    /// for the rationale on the owned-bytes representation.
    pub max_key: Vec<u8>,
    /// Number of live (non-tombstone) KV entries.
    pub item_count: u64,
    /// Number of strong tombstone entries (delete markers that
    /// suppress all older versions of a key).
    pub tombstone_count: u64,
    /// Number of weak tombstone entries (single-version delete
    /// markers).
    pub weak_tombstone_count: u64,
    /// Number of weak tombstones eligible for reclamation during
    /// compaction (already paired with their matching value entry).
    pub weak_tombstone_reclaimable: u64,
    /// Number of data blocks emitted by the writer.
    pub data_block_count: u64,
    /// Number of index blocks. For full-index tables this is 1
    /// (the TLI itself acts as the index); for partitioned-index
    /// tables this is the count of leaf index blocks under the TLI.
    pub index_block_count: u64,
    /// Codec used to compress data blocks.
    pub data_block_compression: CompressionType,
    /// Codec used to compress index blocks. Often `None` since the
    /// TLI is small and compression overhead dominates the win.
    pub index_block_compression: CompressionType,
    /// Wall-clock nanoseconds since the Unix epoch when the writer
    /// finalised the table. Recovered identically from MID or TAIL
    /// meta — the writer snapshots `unix_timestamp()` once and emits
    /// the same value to both copies.
    pub created_at_nanos: u128,
    /// `true` when the table was written with a recognized, applicable Page
    /// ECC scheme (the read path sizes + recovers its parity trailers).
    pub page_ecc: bool,
    /// `true` when the table's ECC descriptor decodes to a scheme this build
    /// cannot apply (an unimplemented scheme, page granularity, an unknown
    /// kind, or a non-canonical descriptor). Block data still reads (verified
    /// by its checksum) but without ECC recovery; recompaction re-stamps the
    /// table with a supported scheme. Mutually exclusive with
    /// [`Self::page_ecc`].
    pub ecc_unrecognized: bool,
    /// Page-ECC scheme decoded from the table's `descriptor#page_ecc`
    /// value, in a public form independent of the crate-internal
    /// `EccParams` type. `Some(_)` whenever the SST carries an ECC
    /// descriptor (recognized OR not — [`Self::page_ecc`] /
    /// [`Self::ecc_unrecognized`] distinguish the two); `None` for a
    /// table written without Page ECC. Pair with
    /// [`EccSchemeInfo::parity_trailer_len`] and a block's `data_length`
    /// to report the per-block parity-trailer size a forensic dump would
    /// see following the payload.
    pub ecc_scheme: Option<EccSchemeInfo>,
}

/// Page-ECC scheme of an SST, in a public form decoupled from the
/// crate-internal (doc-hidden) `EccParams` enum.
///
/// Returned via [`TableProperties::ecc_scheme`] so diagnostic tooling can
/// report the parity layout — and, with a block's on-disk payload length,
/// the per-block parity-trailer size — without reaching into
/// `crate::table::block`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum EccSchemeInfo {
    /// Shard-based parity: the block payload is split into `data_shards`
    /// shards with `parity_shards` recovery shards appended (XOR / RAID-5
    /// when `parity_shards == 1`, Reed-Solomon when `>= 2`).
    Shard {
        /// Number of data shards the payload is split into (`>= 1`).
        data_shards: u8,
        /// Number of parity shards appended (`>= 1`).
        parity_shards: u8,
    },
    /// Per-word Hamming SEC-DED: one check byte per 8-byte data word.
    Secded,
}

impl EccSchemeInfo {
    /// Byte length of the parity trailer this scheme appends after a
    /// `payload_len`-byte block payload.
    ///
    /// The trailer length is not stored per block; the read path re-derives
    /// it from `data_length` plus the per-SST scheme descriptor, and this
    /// mirrors that derivation for out-of-band reporting.
    ///
    /// # Examples
    ///
    /// ```
    /// use lsm_tree::inspect::EccSchemeInfo;
    /// // RS(4,2): payload split into 4 shards, 2 parity shards appended.
    /// let rs = EccSchemeInfo::Shard { data_shards: 4, parity_shards: 2 };
    /// assert!(rs.parity_trailer_len(4096) > 0);
    /// // SEC-DED: one check byte per 8-byte word.
    /// assert_eq!(EccSchemeInfo::Secded.parity_trailer_len(64), 8);
    /// ```
    #[must_use]
    pub fn parity_trailer_len(self, payload_len: usize) -> usize {
        match self {
            Self::Shard {
                data_shards,
                parity_shards,
            } => {
                let ds = usize::from(data_shards);
                let ps = usize::from(parity_shards);
                if ds == 0 || ps == 0 {
                    return 0;
                }
                // Mirrors `crate::ecc::parity_len` (the writer-side source of
                // truth), inlined here because that module is gated behind the
                // `page_ecc` feature while this reporting API must work on any
                // build: per-shard bytes = ceil(payload / data_shards) rounded
                // up to a multiple of 2 (reed-solomon-simd's shard alignment;
                // XOR shares the layout), times the parity-shard count.
                let shard = payload_len.div_ceil(ds).div_ceil(2) * 2;
                shard * ps
            }
            // One check byte per 8-byte word, last word zero-padded.
            Self::Secded => payload_len.div_ceil(8),
        }
    }
}

/// Reads `path` and returns its on-disk metadata as
/// [`TableProperties`].
///
/// Always opens through [`StdFs`]: this is a path-based out-of-band
/// helper, not tied to a live `Tree`'s configured `Fs` backend.
/// `IoUringFs` also operates on real on-disk paths and would work
/// here mechanically; the reason for not threading the backend
/// through is that the caller (typically a diagnostic CLI) starts
/// from a filesystem path with no `Tree` in scope, so `StdFs` is
/// always the right default. Parses the SFA trailer to locate the
/// `meta` and optional `meta_mid` sections, then decodes the meta
/// block via the same `ParsedMeta` machinery `Table::recover` uses
/// on a live tree open.
///
/// Recovery semantics mirror [`Table::recover`](crate::Table):
/// the canonical tail `meta` section is tried first; on
/// decode/checksum/decrypt failure the MID mirror `meta_mid` is tried.
/// Both copies failing returns the original tail error. Tables
/// written before the meta-mirror change (#295) have no `meta_mid`
/// section and the reader falls straight through to the tail copy.
///
/// Encryption: this function does not accept an encryption provider
/// because it is intended for out-of-band diagnostic use. Encrypted
/// SSTs will fail to decode their meta block here. Encrypted-aware
/// property inspection is tracked under #251 / #256 once the
/// AAD-bound wire format lands.
///
/// # Errors
///
/// Returns an error if:
/// - the file cannot be opened or read (`std::io::Error`),
/// - the SFA trailer is missing / malformed and cannot be parsed to
///   locate the `meta` / `meta_mid` sections,
/// - the canonical tail `meta` block fails to decode AND either the
///   `meta_mid` section is absent or the mid copy also fails to
///   decode (block header / XXH3 / structural mismatch). In the
///   both-copies-fail case the original tail error is returned and
///   the mid failure is dropped on the floor (no diagnostic logging
///   here — callers wanting per-copy attribution should walk the
///   meta sections themselves),
/// - the table is encrypted: this function does not take an
///   encryption provider, so the AEAD-protected meta block fails to
///   decode and the failure surfaces as a regular decrypt error.
#[cfg(feature = "std")]
pub fn read_table_properties(path: &Path) -> crate::Result<TableProperties> {
    let fs = StdFs;
    let mut file = fs.open(path, &FsOpenOptions::new().read(true))?;

    let sfa_reader = crate::sfa::Reader::from_reader(&mut file)?;
    let toc = sfa_reader.toc();
    let regions = ParsedRegions::parse_from_toc(toc)?;

    // TAIL first (authoritative copy by convention). On decode /
    // decrypt / checksum failure, fall back to MID. Mirrors
    // `Table::recover` so a corrupted-tail SST that the live open
    // path can still recover also produces inspectable properties
    // here.
    let meta = match ParsedMeta::load_with_handle(&*file, &regions.metadata, None, None) {
        Ok(m) => m,
        Err(tail_err) => {
            if let Some(mid_handle) = regions.metadata_mid {
                match ParsedMeta::load_with_handle(&*file, &mid_handle, None, None) {
                    Ok(mid) => mid,
                    Err(_) => return Err(tail_err),
                }
            } else {
                return Err(tail_err);
            }
        }
    };

    Ok(TableProperties {
        id: meta.id,
        file_size: meta.file_size,
        min_key: meta.key_range.min().to_vec(),
        max_key: meta.key_range.max().to_vec(),
        item_count: meta.item_count,
        tombstone_count: meta.tombstone_count,
        weak_tombstone_count: meta.weak_tombstone_count,
        weak_tombstone_reclaimable: meta.weak_tombstone_reclaimable,
        data_block_count: meta.data_block_count,
        index_block_count: meta.index_block_count,
        data_block_compression: meta.data_block_compression,
        index_block_compression: meta.index_block_compression,
        created_at_nanos: *meta.created_at,
        page_ecc: meta.page_ecc,
        ecc_unrecognized: meta.ecc_unrecognized,
        ecc_scheme: meta.ecc_params.map(ecc_scheme_info),
    })
}

/// Projects the crate-internal (doc-hidden) `EccParams` onto the public
/// [`EccSchemeInfo`] reported by [`TableProperties::ecc_scheme`].
fn ecc_scheme_info(params: crate::table::block::EccParams) -> EccSchemeInfo {
    match params {
        crate::table::block::EccParams::Shard {
            data_shards,
            parity_shards,
        } => EccSchemeInfo::Shard {
            data_shards,
            parity_shards,
        },
        crate::table::block::EccParams::Secded => EccSchemeInfo::Secded,
    }
}

/// One entry parsed from an SST's top-level index (TLI) block.
///
/// What this points at depends on the SST's index layout:
///
/// - **Full-index tables**: the SST has no separate `index` SFA
///   section. The TLI directly carries data-block handles, so each
///   `IndexEntry` corresponds to one data block in the SST.
/// - **Partitioned-index tables**: the SST has an `index` SFA section
///   containing sub-index leaf blocks. The TLI carries handles
///   pointing at those leaves, so each `IndexEntry` corresponds to
///   one leaf, NOT a data block; walking the leaves to enumerate
///   individual data blocks is a separate operation.
///
/// The distinction is not derivable from `TableProperties` fields
/// alone — in particular `TableProperties.index_block_count == 1`
/// can mean either a full-index table OR a partitioned-index table
/// with a single leaf partition. The authoritative signal is the
/// presence of an `index` SFA section in the SFA TOC; this facade
/// does not currently expose that signal. Callers needing to
/// classify the layout should consult the SST file's TOC directly.
///
/// `#[non_exhaustive]` so new fields can be added in a minor version
/// bump.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct IndexEntry {
    /// Last user-key covered by the pointed-at block. For binary
    /// search the TLI is searched on `end_key`, so this is the
    /// authoritative sort key for the entry.
    pub end_key: Vec<u8>,
    /// Highest seqno of any item in the pointed-at block. Used by
    /// the read path to decide whether a snapshot can skip the
    /// block entirely.
    pub seqno: u64,
    /// On-disk byte offset of the pointed-at block (data-block start
    /// for full-index tables, sub-index-block start for partitioned).
    pub offset: u64,
    /// On-disk size of the pointed-at block in bytes, including the
    /// block `Header` prefix.
    pub size: u32,
}

/// Reads `path` and returns the parsed entries of its top-level index
/// (TLI) block.
///
/// Same out-of-band path-based open semantics as
/// [`read_table_properties`] (uses [`StdFs`], no encryption provider).
/// Recovery semantics mirror [`Table::read_tli`](crate::Table)'s
/// TAIL-first / HEAD-fallback path from #296: the tail `tli_tail`
/// mirror section is attempted first when present; on
/// decode / decrypt / checksum failure the canonical head `tli`
/// section is tried; both copies failing returns the original tail
/// error. Tables written before the TLI-mirror change have no
/// `tli_tail` section and fall straight through to the head copy.
///
/// The function returns owned `IndexEntry` records — the inner
/// `IndexBlock` and its underlying `Slice` are dropped on return, so
/// the caller does not need to keep the file mapping alive. Memory
/// cost is `O(entries × end_key.len())`.
///
/// # Errors
///
/// Returns an error if:
/// - the file cannot be opened or read (`std::io::Error`),
/// - the SFA trailer is missing / malformed and cannot be parsed,
/// - the `meta` block fails to decode (needed to determine
///   `index_block_compression`),
/// - both the head `tli` and the tail `tli_tail` (if present) fail
///   to decode (block header / XXH3 / structural mismatch). In the
///   both-copies-fail case the original tail error is returned,
/// - the TLI block trailer is malformed (e.g. `restart_interval == 0`),
/// - the table is encrypted: this function does not take an
///   encryption provider, so the AEAD-protected blocks fail to
///   decode and the failure surfaces as a regular decrypt error.
#[cfg(feature = "std")]
pub fn read_top_level_index_entries(path: &Path) -> crate::Result<Vec<IndexEntry>> {
    use crate::table::block_index::iter::OwnedIndexBlockIter;
    use crate::table::{IndexBlock, KeyedBlockHandle};

    let fs = StdFs;
    let mut file = fs.open(path, &FsOpenOptions::new().read(true))?;

    let sfa_reader = crate::sfa::Reader::from_reader(&mut file)?;
    let toc = sfa_reader.toc();
    let regions = ParsedRegions::parse_from_toc(toc)?;

    // Load meta to know the index block's compression codec. Same
    // TAIL-first / MID-fallback as `read_table_properties` above —
    // factored together so this function gives identical recovery
    // behaviour on a meta-corrupted SST.
    let meta = match ParsedMeta::load_with_handle(&*file, &regions.metadata, None, None) {
        Ok(m) => m,
        Err(tail_err) => {
            if let Some(mid_handle) = regions.metadata_mid {
                match ParsedMeta::load_with_handle(&*file, &mid_handle, None, None) {
                    Ok(mid) => mid,
                    Err(_) => return Err(tail_err),
                }
            } else {
                return Err(tail_err);
            }
        }
    };
    let index_compression = meta.index_block_compression;
    let ecc = meta.ecc_params;
    let table_id = meta.id;

    // TLI tail mirror tried first when present (most-recently fsynced
    // copy); on failure fall back to the head copy. Mirrors
    // `Table::read_tli` so a partially-corrupted TLI behaves the
    // same here as in a live open.
    let tli_block = if let Some(tail_handle) = regions.tli_tail {
        match load_index_block(&*file, tail_handle, table_id, index_compression, ecc) {
            Ok(b) => b,
            Err(tail_err) => {
                match load_index_block(&*file, regions.tli, table_id, index_compression, ecc) {
                    Ok(b) => b,
                    Err(_) => return Err(tail_err),
                }
            }
        }
    } else {
        load_index_block(&*file, regions.tli, table_id, index_compression, ecc)?
    };

    let block = IndexBlock::new(tli_block);
    let iter = OwnedIndexBlockIter::from_block(block, crate::comparator::default_comparator())?;

    let entries = iter
        .map(|h: KeyedBlockHandle| IndexEntry {
            end_key: h.end_key().to_vec(),
            seqno: h.seqno(),
            offset: *h.offset(),
            size: h.size(),
        })
        .collect();

    Ok(entries)
}

/// Helper: load and validate a single index block from disk. Shared
/// between the tail and head TLI load sites so both paths produce the
/// same error shape on a malformed block.
#[cfg(feature = "std")]
fn load_index_block(
    file: &dyn crate::fs::FsFile,
    handle: crate::table::BlockHandle,
    table_id: crate::table::TableId,
    compression: CompressionType,
    ecc: Option<crate::table::block::EccParams>,
) -> crate::Result<crate::table::Block> {
    use crate::table::block::{Block, BlockIdentity, BlockType};

    let block = Block::from_file(
        file,
        handle,
        BlockIdentity {
            table_id,
            // A block's byte offset is intentionally NOT part of the AAD
            // identity (offset-independent AAD keeps encryption parallel),
            // so nothing per-block needs threading here.
            block_type: BlockType::Index,
            dict_id: 0,
            window_log: 0,
        },
        // Inspect-side index loader doesn't accept an encryption
        // provider (out-of-band facade) and never threads a zstd
        // dict. The transform therefore resolves to either:
        //   - `Plain` when `compression == None`, or
        //   - `Compressed` for `Lz4` / `Zstd(level)`.
        // `CompressionType::ZstdDict { .. }` is rejected here with
        // `Error::ZstdDictMismatch` because the dict can't be
        // recovered without a live `Tree`; inspect such SSTs through
        // the owning tree instead.
        //
        // Index blocks omit the block_flags byte, so ECC presence is a
        // per-SST descriptor property: upgrade to the `*Ecc` transform when
        // the SST was written with Page ECC, sizing the parity trailer the
        // reader must skip from the descriptor scheme. Identity without the
        // feature.
        &{
            let t = crate::table::block::BlockTransform::from_parts(
                compression,
                None,
                #[cfg(zstd_any)]
                None,
            )?;
            if let Some(ecc) = ecc {
                t.with_ecc(ecc)
            } else {
                t
            }
        },
    )?;

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

    Ok(block)
}

/// One KV entry materialised from an SST data block.
///
/// `#[non_exhaustive]` so new fields can be added in a minor version
/// bump.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct DataEntry {
    /// The user key bytes (no internal-key suffix, no seqno).
    pub key: Vec<u8>,
    /// The value bytes.
    ///
    /// Three reasons this can be empty:
    /// 1. The entry is a tombstone (see [`Self::is_tombstone`]).
    /// 2. The iterator was put into keys-only mode via
    ///    [`DataBlockEntryIter::keys_only`]; in that mode `Value`
    ///    entries also yield `value: Vec::new()` because the
    ///    underlying `Slice::to_vec()` allocation is deliberately
    ///    skipped. The caller asked not to pay it.
    /// 3. The on-disk value is a zero-length payload (legal for
    ///    `Value` entries: an empty byte-string is distinct from
    ///    a tombstone).
    ///
    /// Use [`Self::value_type`] to disambiguate (1) from the
    /// other two, and the iterator construction site to know
    /// whether (2) applies; this field cannot tell them apart on
    /// its own.
    ///
    /// For [`crate::ValueType::Indirection`] entries the bytes
    /// are the encoded blob handle, NOT the resolved blob payload.
    pub value: Vec<u8>,
    /// Per-entry sequence number as stored in the internal key on
    /// disk. This is the **table-local** (or "local") seqno: the
    /// `Table::get` / `Iter` read path adds the table's `global_seqno`
    /// offset on top of this for ingested SSTs (see
    /// `src/table/mod.rs` for the translation), so the value the
    /// running tree sees can be higher than the byte here. For an
    /// SST written by the normal writer path `global_seqno == 0`
    /// and the local seqno IS the visible seqno; for an ingested
    /// SST the visible seqno is `global_seqno + local_seqno` and
    /// this field gives the local component only.
    ///
    /// Diagnostic surface: this facade deliberately exposes the
    /// on-disk byte, not the translated value, because operators
    /// using `sst-dump` are usually trying to correlate against the
    /// raw on-disk state. If you need the live-tree-visible seqno,
    /// pair this with the `global_seqno` stored in the table's meta
    /// block (currently not exposed on [`TableProperties`]; tracked
    /// for a future API addition).
    pub seqno: u64,
    /// Underlying internal-key value type. Data blocks can contain
    /// `Value`, `Tombstone`, `WeakTombstone`, `MergeOperand`, and
    /// `Indirection` (blob pointer) entries; the convenience predicate
    /// [`Self::is_tombstone`] folds the two tombstone variants into a
    /// single boolean, but callers that need to distinguish merge
    /// operands or blob pointers from regular values must read this
    /// field directly. Going through the underlying enum keeps the
    /// facade aligned with the crate's own tombstone predicate
    /// ([`crate::ValueType::is_tombstone`]) as a single source of
    /// truth so a future variant addition does not silently start
    /// flagging unrelated entries as tombstones.
    pub value_type: crate::ValueType,
}

impl DataEntry {
    /// `true` iff [`Self::value_type`] is `Tombstone` or `WeakTombstone`.
    ///
    /// Convenience wrapper that delegates to
    /// [`crate::ValueType::is_tombstone`]; `MergeOperand` and
    /// `Indirection` entries return `false` here even though their
    /// `value` bytes do not represent a regular plaintext value.
    #[must_use]
    pub fn is_tombstone(&self) -> bool {
        self.value_type.is_tombstone()
    }
}

/// Streaming iterator over every KV entry in an SST's data section.
///
/// Returned by [`iter_data_block_entries`]. Owns the underlying file
/// handle and the list of pending data-block handles; loads exactly
/// one data block at a time and drops it before moving to the next.
///
/// Memory cost is **`O(N_data_blocks) * 16 B + largest_data_block +
/// key.len + value.len`**, where the first term is the upfront
/// `Vec<BlockHandle>` collected from the TLI at construction time
/// (one 16-byte handle per data block). For typical SSTs this is
/// negligible: a 64 MiB SST with 4 KiB data blocks has ~16 K
/// handles ≈ 256 KiB — small compared to a single decompressed
/// data block. A 1 GiB SST gives ~4 MiB of handles. The trade-off
/// buys constant-time `next()` per block lookup; truly streaming
/// the TLI through the iterator would require nested self-cells
/// (the `OwnedIndexBlockIter` already self-references its
/// `IndexBlock`) which is structurally awkward for the rare case
/// where the handle vec actually matters in absolute terms.
///
/// Encrypted SSTs and partitioned-index SSTs both return an error
/// at construction time — see [`iter_data_block_entries`] for the
/// full contract.
#[cfg(feature = "std")]
pub struct DataBlockEntryIter {
    file: Box<dyn crate::fs::FsFile>,
    table_id: crate::table::TableId,
    data_block_compression: CompressionType,
    /// Per-SST Page-ECC flag from the parsed meta: data blocks omit the
    /// `block_flags` byte, so the read transform needs the parity scheme
    /// to size + recover the trailer. `None` = no parity.
    ecc: Option<crate::table::block::EccParams>,
    /// Per-SST per-KV-footer flag (`kv_checksum_algo.is_some()`) from the
    /// parsed meta: tells `from_loaded` whether to strip a footer.
    has_kv_footer: bool,
    /// Remaining data-block handles (FIFO). Drained as blocks are
    /// loaded. Held as a `Vec` rather than a `VecDeque` because the
    /// TLI walks the handles in sorted order already, and `Vec::pop`
    /// from the tail with reversed input is cheaper than the
    /// double-ended queue overhead.
    ///
    /// Stores plain `BlockHandle` (offset + size) rather than
    /// `KeyedBlockHandle`. `KeyedBlockHandle` carries `end_key:
    /// Slice` which is a view into the TLI block buffer; collecting
    /// the keyed variant would keep the entire TLI block alive for
    /// the iterator's lifetime. Stripping to the bare handle lets
    /// the TLI block drop after enumeration completes, leaving only
    /// the 16-byte-per-block `Vec` here. See the struct-level
    /// docstring for the memory-cost analysis.
    remaining_handles: Vec<crate::table::BlockHandle>,
    /// Iterator over the currently-loaded block, or `None` when we
    /// haven't loaded a block yet or just finished one.
    current: Option<crate::table::iter::OwnedDataBlockIter>,
    /// When `true`, skip materialising `DataEntry.value` — leave it
    /// as an empty `Vec`. Toggled via [`Self::keys_only`]. Saves the
    /// `Slice::to_vec()` allocation per entry for callers (e.g.
    /// `sst-dump dump --keys-only`) that don't need the value bytes.
    keys_only: bool,
}

#[cfg(feature = "std")]
impl DataBlockEntryIter {
    /// Suppress value materialisation: yielded `DataEntry` values
    /// have `value: Vec::new()` instead of the per-entry
    /// `to_vec()`-copy of the on-disk bytes. Use for keys-only
    /// walks (the entire SST data section can be processed without
    /// allocating any value bytes; only the key copy plus the
    /// constant-size struct fields stay).
    ///
    /// Builder-style — chain off the constructor:
    ///
    /// ```ignore
    /// let iter = iter_data_block_entries(path)?.keys_only();
    /// ```
    #[must_use]
    pub const fn keys_only(mut self) -> Self {
        self.keys_only = true;
        self
    }
}

#[cfg(feature = "std")]
impl Iterator for DataBlockEntryIter {
    type Item = crate::Result<DataEntry>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            // Drain the current block before considering the next one.
            if let Some(iter) = self.current.as_mut() {
                if let Some(internal) = iter.next() {
                    let value = if self.keys_only {
                        // Skip the `Slice::to_vec()` allocation; the
                        // caller has opted out of value bytes for
                        // this walk. An empty `Vec` is cheap (no
                        // heap allocation in current `Vec::new()`
                        // implementations on stable Rust).
                        Vec::new()
                    } else {
                        internal.value.to_vec()
                    };
                    let entry = DataEntry {
                        key: internal.key.user_key.to_vec(),
                        value,
                        seqno: internal.key.seqno,
                        value_type: internal.key.value_type,
                    };
                    return Some(Ok(entry));
                }
                self.current = None;
            }

            // Advance to the next data block.
            let handle = self.remaining_handles.pop()?;
            match load_data_block_iter(
                &*self.file,
                &handle,
                self.table_id,
                self.data_block_compression,
                self.ecc,
                self.has_kv_footer,
            ) {
                Ok(iter) => {
                    self.current = Some(iter);
                }
                Err(e) => {
                    // Surface the block-load failure but stop the
                    // walk: subsequent blocks would likely fail the
                    // same way and the operator only needs the first
                    // diagnostic.
                    self.remaining_handles.clear();
                    return Some(Err(e));
                }
            }
        }
    }
}

/// Reads `path` and returns a streaming iterator over every KV entry
/// in the SST's data blocks.
///
/// Same out-of-band path-based open semantics as
/// [`read_table_properties`] etc.: uses [`StdFs`], no encryption
/// provider. Walks the top-level index (TLI) to enumerate data-block
/// handles, then yields entries one block at a time so memory cost
/// stays bounded.
///
/// # Scope
///
/// Only **full-index** SSTs are supported by this facade. SSTs with
/// a separate `index` SFA section (partitioned index) point their
/// TLI entries at sub-index leaves rather than data blocks; walking
/// the leaves to enumerate individual data blocks is a separate
/// operation not yet wired here. Partitioned-index SSTs return an
/// `Io(ErrorKind::Unsupported)` error.
///
/// **Custom user comparator: forward iteration works, but bounds /
/// seek operations do not.** Forward iteration over the data section
/// is positional — the index walks blocks in their on-disk file
/// order and the block iterator yields entries in their on-disk
/// encoded order, neither of which depends on the comparator. So a
/// bounds-free walk of a custom-comparator SST through this facade
/// streams entries in the SST's own sort order correctly. What this
/// facade can't do is run `crate::comparator::default_comparator`
/// for seek / point-read / range-bounded operations layered on top
/// (e.g. `sst-dump dump --from/--to` applies bytewise bounds and
/// breaks early on the upper bound, which only makes sense for the
/// default lexicographic comparator). Callers that need
/// comparator-correct range queries on a custom-comparator SST
/// should use the owning `Tree`'s regular read APIs.
///
/// **Zstd-dictionary blocks: not supported.** This facade calls
/// [`crate::table::Block::from_file`] with no
/// [`crate::compression::ZstdDictionary`] attached (it has no way to
/// fetch one without a live `Tree`), so blocks compressed with
/// [`crate::CompressionType::ZstdDict`] fail with
/// [`crate::Error::ZstdDictMismatch`] even though they are otherwise
/// valid. Inspect such SSTs through the owning `Tree` instead.
///
/// # Errors
///
/// Returns an error if:
/// - the file cannot be opened or read,
/// - the SFA trailer is missing / malformed,
/// - the `meta` block fails to decode (needed for
///   `data_block_compression`),
/// - the SST has a partitioned-index layout (`index` SFA section
///   present),
/// - the TLI block cannot be loaded or its trailer is malformed,
/// - the table is encrypted: this function does not take an
///   encryption provider, so AEAD-protected blocks fail to decode.
///
/// Per-entry errors are surfaced by the returned iterator's
/// `Item = Result<DataEntry>` shape: a single block failing to load
/// yields one `Err` and ends the walk.
#[cfg(feature = "std")]
pub fn iter_data_block_entries(path: &Path) -> crate::Result<DataBlockEntryIter> {
    use crate::table::IndexBlock;
    use crate::table::block_index::iter::OwnedIndexBlockIter;

    let fs = StdFs;
    let mut file = fs.open(path, &FsOpenOptions::new().read(true))?;

    let sfa_reader = crate::sfa::Reader::from_reader(&mut file)?;
    let toc = sfa_reader.toc();
    let regions = ParsedRegions::parse_from_toc(toc)?;

    if regions.index.is_some() {
        return Err(crate::Error::from(std::io::Error::new(
            std::io::ErrorKind::Unsupported,
            "partitioned-index SST (separate `index` section present) is not yet supported \
             by iter_data_block_entries; walking sub-index leaves to enumerate data blocks \
             is a follow-up surface",
        )));
    }

    let meta = match ParsedMeta::load_with_handle(&*file, &regions.metadata, None, None) {
        Ok(m) => m,
        Err(tail_err) => {
            if let Some(mid_handle) = regions.metadata_mid {
                match ParsedMeta::load_with_handle(&*file, &mid_handle, None, None) {
                    Ok(mid) => mid,
                    Err(_) => return Err(tail_err),
                }
            } else {
                return Err(tail_err);
            }
        }
    };
    let table_id = meta.id;
    let data_block_compression = meta.data_block_compression;
    let index_compression = meta.index_block_compression;
    let ecc = meta.ecc_params;

    // Load TLI (with tail-mirror fallback). For full-index tables
    // each TLI entry IS a data-block handle, so this list is what
    // we stream over.
    let tli_block = if let Some(tail_handle) = regions.tli_tail {
        match load_index_block(&*file, tail_handle, table_id, index_compression, ecc) {
            Ok(b) => b,
            Err(tail_err) => {
                match load_index_block(&*file, regions.tli, table_id, index_compression, ecc) {
                    Ok(b) => b,
                    Err(_) => return Err(tail_err),
                }
            }
        }
    } else {
        load_index_block(&*file, regions.tli, table_id, index_compression, ecc)?
    };

    let block = IndexBlock::new(tli_block);
    let iter = OwnedIndexBlockIter::from_block(block, crate::comparator::default_comparator())?;
    // Materialise each handle to the bare `BlockHandle` (offset +
    // size) right here, so the surrounding `IndexBlock` and its
    // backing `Slice` can drop as soon as this collect finishes.
    // `KeyedBlockHandle.end_key` is a view into the TLI block bytes;
    // holding `KeyedBlockHandle` in the iterator struct would keep
    // the entire TLI block alive for the iterator's lifetime, which
    // contradicts the "memory bounded to one data block" guarantee.
    // `BlockHandle` is `Copy`, so `into_inner()` is cheap and
    // releases the `Slice` view at the same time.
    let mut handles: Vec<crate::table::BlockHandle> = iter
        .map(crate::table::KeyedBlockHandle::into_inner)
        .collect();
    // Reverse so `pop` from the tail yields the smallest end_key
    // first — matches the natural left-to-right read order
    // operators expect from `dump`.
    handles.reverse();

    Ok(DataBlockEntryIter {
        file,
        table_id,
        data_block_compression,
        ecc: meta.ecc_params,
        has_kv_footer: meta.kv_checksum_algo.is_some(),
        remaining_handles: handles,
        current: None,
        keys_only: false,
    })
}

#[cfg(feature = "std")]
fn load_data_block_iter(
    file: &dyn crate::fs::FsFile,
    handle: &crate::table::BlockHandle,
    table_id: crate::table::TableId,
    compression: CompressionType,
    ecc: Option<crate::table::block::EccParams>,
    has_kv_footer: bool,
) -> crate::Result<crate::table::iter::OwnedDataBlockIter> {
    use crate::table::DataBlock;
    use crate::table::block::{Block, BlockIdentity, BlockType};
    use crate::table::iter::OwnedDataBlockIter;

    // Inspect-side data-block loader: no encryption provider (out-
    // of-band facade) and never a zstd dict. Same transform-
    // resolution rule as load_index_block above:
    //   - compression == None  →  BlockTransform::Plain
    //   - compression == Lz4 / Zstd(level)  →  BlockTransform::Compressed
    //   - compression == ZstdDict { .. }  →  rejected via
    //     Error::ZstdDictMismatch (no dict can be threaded without a
    //     live Tree).
    let block = Block::from_file(
        file,
        *handle,
        BlockIdentity {
            table_id,
            // A block's byte offset is intentionally NOT part of the AAD
            // identity, so this facade needs no per-block offset for any of
            // its meta / TLI / filter / data loads.
            block_type: BlockType::Data,
            dict_id: 0,
            window_log: 0,
        },
        &{
            // Data blocks omit the block_flags byte, so ECC presence is a
            // per-SST property: upgrade to the `*Ecc` transform when the SST
            // was written with Page ECC. Identity without the feature.
            let t = crate::table::block::BlockTransform::from_parts(
                compression,
                None,
                #[cfg(zstd_any)]
                None,
            )?;
            if let Some(ecc) = ecc {
                t.with_ecc(ecc)
            } else {
                t
            }
        },
    )?;

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

    // `from_loaded` strips the per-KV checksum footer when this SST carries
    // one (per-SST `has_kv_footer`; data blocks omit the byte) so the
    // inspection iterator decodes the inner payload normally.
    let data_block = DataBlock::from_loaded(block, has_kv_footer)?;
    OwnedDataBlockIter::try_new(data_block, |b| {
        b.try_iter(crate::comparator::default_comparator())
    })
}
/// Read-only stats for a single-block (full) `BuRR` / Ribbon filter
/// section.
///
/// Returned by [`read_filter_stats`]. For partitioned-filter tables
/// (those with a `filter_tli` SFA section) this struct is not
/// populated — see `read_filter_stats` for the contract. Stats are
/// derived from the public
/// [`BurrFilterReader`](crate::table::filter::ribbon::burr::BurrFilterReader)
/// surface plus the SFA section size; no internal filter bits are
/// exposed here.
///
/// `#[non_exhaustive]` so new fields can be added in a minor version
/// bump.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct FilterStats {
    /// On-disk size of the `filter` SFA section in bytes, including
    /// the block `Header` prefix.
    pub filter_section_bytes: u64,
    /// Number of `BuRR` / Ribbon "layers" the writer emitted. Each
    /// layer is a Bumped-Ribbon-Retrieval pass; more layers means a
    /// larger filter at a given key count but tighter FPR. Fixed-width
    /// `u64` so the public layout doesn't vary between 32-bit and
    /// 64-bit targets, matching the rest of the inspect facade.
    pub layer_count: u64,
    /// Number of keys the meta block reports for the table. Used as
    /// the denominator for `bits_per_key`; sourced from
    /// `TableProperties.item_count` and copied here so callers can
    /// compute the rate without a second `read_table_properties`
    /// call.
    pub item_count: u64,
    /// Approximate average bits-per-key the filter consumes:
    /// `filter_section_bytes * 8 / max(item_count, 1)`. This is a
    /// SIZE metric, not the true theoretical `BuRR` overhead — it
    /// includes the block `Header`, the `BuRR` wire-format header,
    /// per-layer payload framing, and any zero-padding bits at the
    /// end of each layer's storage word array. Treat it as an
    /// upper bound on the actual ribbon parameter `bits_per_key`.
    pub bits_per_key: f64,
}
/// Reads `path` and returns `BuRR` filter sizing stats for the SST's
/// `filter` section, or `Ok(None)` if the SST has no filter section
/// at all (filter-less table).
///
/// **Scope:** only the single-block (full) filter layout is
/// supported by this facade. SSTs with a `filter_tli` SFA section
/// (partitioned filter) return an error — per-partition stats need
/// a different surface that walks the TLI and reports a
/// `Vec<FilterStats>` or aggregate metrics, and that is a separate
/// public-API decision not yet made.
///
/// Same out-of-band path-based open semantics as
/// [`read_table_properties`]: uses [`StdFs`], no encryption provider.
/// Recovery for the meta block (needed to source `item_count`)
/// mirrors the TAIL-first / MID-fallback pattern from #295. The
/// filter block itself is written uncompressed (see
/// `FullFilterWriter::finish`), so no compression-codec dependency on
/// the read path here.
///
/// # Errors
///
/// Returns an error if:
/// - the file cannot be opened or read,
/// - the SFA trailer is missing / malformed,
/// - the `meta` block fails to decode (needed for `item_count`),
/// - the table has a `filter_tli` SFA section (partitioned filter,
///   not supported by this facade): returned as
///   `Error::FeatureUnsupported("filter_tli")` (see
///   [`crate::Error::FeatureUnsupported`]) so callers can match the
///   typed variant instead of parsing message strings,
/// - the `filter` block header / payload is malformed,
/// - the `BuRR` wire format cannot be parsed (magic mismatch,
///   unsupported version, structurally invalid header).
///
/// Returns `Ok(None)` for both on-disk shapes of "no filter
/// installed":
///
/// 1. the SST has no `filter` SFA section at all, or
/// 2. the section is present but carries a zero-byte payload (the
///    [`crate::table::filter::block::FilterBlock`] sentinel for "no
///    filter"; the writer emits it when the filter policy resolves
///    to no usable filter at flush time, which is structurally
///    equivalent to the absent-section case).
///
/// A tree configured with
/// [`FilterPolicy::disabled`](crate::config::FilterPolicy::disabled)
/// (or any policy whose per-level entry is
/// [`FilterPolicyEntry::None`](crate::config::FilterPolicyEntry::None))
/// produces filter-less SSTs that take one of these two shapes;
/// either way callers see the same `Ok(None)` result.
#[cfg(feature = "std")]
pub fn read_filter_stats(path: &Path) -> crate::Result<Option<FilterStats>> {
    use crate::table::block::{Block, BlockIdentity, BlockType};
    use crate::table::filter::ribbon::burr::BurrFilterReader;

    let fs = StdFs;
    let mut file = fs.open(path, &FsOpenOptions::new().read(true))?;

    let sfa_reader = crate::sfa::Reader::from_reader(&mut file)?;
    let toc = sfa_reader.toc();
    let regions = ParsedRegions::parse_from_toc(toc)?;

    if regions.filter_tli.is_some() {
        // Partitioned filter: a `filter_tli` SFA section is present
        // alongside `filter`, and the contents of `filter` are a
        // concatenation of per-partition `BuRR` payloads, not a
        // single parseable wire buffer. Surface this as the typed
        // `Error::FeatureUnsupported("filter_tli")` so callers can
        // match on the marker without parsing message strings; the
        // payload literal is the SFA section name an operator can
        // confirm via the TOC.
        return Err(crate::Error::FeatureUnsupported("filter_tli"));
    }

    let Some(filter_handle) = regions.filter else {
        return Ok(None);
    };
    let filter_section_bytes = u64::from(filter_handle.size());

    // Meta block carries `item_count`. Same TAIL-first / MID-fallback
    // as `read_table_properties` so a partially-corrupted meta still
    // yields stats from the surviving copy.
    let meta = match ParsedMeta::load_with_handle(&*file, &regions.metadata, None, None) {
        Ok(m) => m,
        Err(tail_err) => {
            if let Some(mid_handle) = regions.metadata_mid {
                match ParsedMeta::load_with_handle(&*file, &mid_handle, None, None) {
                    Ok(mid) => mid,
                    Err(_) => return Err(tail_err),
                }
            } else {
                return Err(tail_err);
            }
        }
    };
    let item_count = meta.item_count;
    let table_id = meta.id;

    // Inspect-side filter loader. Two design choices here:
    //
    // 1. **Transform is `Plain` — chosen here, not dictated by the
    //    writer.** `FullFilterWriter::finish` does NOT always emit
    //    filter blocks as `Plain`: when the tree has an encryption
    //    provider configured, it emits `Encrypted` instead (no
    //    compression in either case — filter blocks aren't worth
    //    compressing). This facade chooses `Plain` because it has
    //    no encryption provider plumbed through (out-of-band path,
    //    no live `Tree`), so for SSTs whose filter block was
    //    written encrypted, `Block::from_file` here verifies the
    //    XXH3 over the ciphertext bytes and then fails downstream —
    //    typically `InvalidHeader` from `uncompressed_length`
    //    disagreeing with the ciphertext length, not a clean
    //    `Error::Decrypt`. Encrypted-aware filter inspection is
    //    tracked alongside #256.
    //
    // 2. **A block's byte offset is not part of the AAD identity** —
    //    offset-independent AAD keeps encryption parallel, so neither the
    //    writer nor this reader threads `filter_handle.offset()` into the
    //    BlockIdentity; there is nothing per-block to agree on here.
    let block = Block::from_file(
        &*file,
        filter_handle,
        BlockIdentity {
            table_id,
            block_type: BlockType::Filter,
            dict_id: 0,
            window_log: 0,
        },
        // Filter blocks omit the block_flags byte, so ECC presence is a
        // per-SST descriptor property: upgrade the `Plain` transform to its
        // `*Ecc` variant when the SST was written with Page ECC, so the
        // parity trailer is sized + skipped with the descriptor scheme.
        // Identity without the feature.
        &{
            let t = crate::table::block::BlockTransform::PLAIN;
            if let Some(ecc) = meta.ecc_params {
                t.with_ecc(ecc)
            } else {
                t
            }
        },
    )?;
    if block.header.block_type != BlockType::Filter {
        return Err(crate::Error::InvalidTag((
            "BlockType",
            block.header.block_type.into(),
        )));
    }

    // Empty data slice is the "no filter installed" sentinel (see
    // `FilterBlock::maybe_contains_hash`). The on-disk shape of "no
    // filter present" can be either the section absent entirely
    // OR the section present with a zero-byte payload; both mean
    // the same thing semantically. Collapse the empty-payload case
    // to the same `Ok(None)` result the section-absent branch
    // above returns, so the CLI prints the documented
    // "no filter section installed" line for either shape and the
    // public API stays consistent with the FilterBlock sentinel
    // contract.
    if block.data.is_empty() {
        return Ok(None);
    }

    // BurrFilterReader::layer_count returns usize; widen to u64 at
    // the public-API boundary. usize -> u64 is lossless on every
    // target Rust supports (u64 is at least as wide as usize on
    // 32-bit and identical on 64-bit).
    let layer_count: u64 = BurrFilterReader::new(&block.data)?.layer_count() as u64;

    // `item_count` is `u64`; cast to `f64` is lossy for values above
    // 2^53 (~9 quadrillion), which is well past anything a real SST
    // holds. The lossy cast is the standard pattern for size
    // statistics here — clippy's `cast_precision_loss` lint is
    // already allowed crate-wide for this kind of arithmetic.
    #[expect(
        clippy::cast_precision_loss,
        reason = "filter stats are diagnostic; precision loss above 2^53 keys is irrelevant"
    )]
    let denom = item_count.max(1) as f64;
    #[expect(
        clippy::cast_precision_loss,
        reason = "filter stats are diagnostic; precision loss above 2^53 bytes is irrelevant"
    )]
    let bits = (filter_section_bytes * 8) as f64;
    let bits_per_key = bits / denom;

    Ok(Some(FilterStats {
        filter_section_bytes,
        layer_count,
        item_count,
        bits_per_key,
    }))
}

/// Little-endian zstd frame magic (`0xFD2FB528`), per RFC 8878 §3.1.1.
const ZSTD_FRAME_MAGIC: [u8; 4] = [0x28, 0xB5, 0x2F, 0xFD];

/// Type of a single inner zstd block, decoded from the 2 type bits of its
/// 3-byte header (RFC 8878 §3.1.1.2.1).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ZstdBlockType {
    /// Uncompressed bytes stored verbatim (`Block_Size` content bytes follow).
    Raw,
    /// Run-length: a single byte follows, regenerated `Block_Size` times.
    Rle,
    /// Zstd-compressed content (`Block_Size` compressed bytes follow).
    Compressed,
}

/// One inner zstd block within a compressed block payload, enumerated
/// key-free by walking the frozen block-header chain WITHOUT decompressing.
///
/// Returned in frame order by [`census_zstd_frame`]. The walk reads only the
/// 3-byte block headers (RFC 8878 §3.1.1.2), so it is cheap and immune to
/// decompression bombs, but cannot recover a `Compressed` block's regenerated
/// size (that lives in the compressed stream, not the header).
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct InnerZstdBlock {
    /// Zero-based position of this block within the frame.
    pub index: u32,
    /// Block kind from the header's 2 type bits.
    pub block_type: ZstdBlockType,
    /// `true` for the frame's final block (the header's `Last_Block` bit).
    pub last: bool,
    /// Byte offset of this block's 3-byte header within the frame payload.
    pub header_offset: u64,
    /// On-disk content bytes following the 3-byte header: the `Block_Size`
    /// field for `Raw` / `Compressed`, and `1` for `Rle` (a single repeated
    /// byte is stored regardless of the regenerated size).
    pub content_len: u32,
    /// Regenerated (decompressed) size when derivable from the header alone:
    /// the `Block_Size` field for `Raw` / `Rle`. `None` for `Compressed`,
    /// whose output size is not in the header.
    pub decompressed_len: Option<u32>,
}

/// Key-free structural census of a single zstd frame, produced by
/// [`census_zstd_frame`].
///
/// `#[non_exhaustive]` so new frame-header fields can be surfaced in a minor
/// version bump.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct ZstdFrameCensus {
    /// Length in bytes of the frame header (magic + descriptor + optional
    /// window / dictionary-id / frame-content-size fields).
    pub frame_header_len: u8,
    /// Decoded window size in bytes (from the window descriptor, or the frame
    /// content size for a single-segment frame). `None` if neither was
    /// present in the header.
    pub window_size: Option<u64>,
    /// Dictionary id declared in the frame header, if any (`0` is treated as
    /// "no dictionary" per the spec and reported as `None`).
    pub dictionary_id: Option<u32>,
    /// Declared uncompressed content size of the whole frame, if the header
    /// carried a `Frame_Content_Size` field.
    pub frame_content_size: Option<u64>,
    /// `true` if the frame header's `Content_Checksum_flag` is set (a 4-byte
    /// XXH64 content checksum trails the last block).
    pub content_checksum: bool,
    /// Inner blocks in frame order.
    pub blocks: Vec<InnerZstdBlock>,
}

/// Returns `true` if `payload` begins with the zstd frame magic.
///
/// A cheap pre-check before [`census_zstd_frame`] so a forensic caller can
/// distinguish a zstd-compressed block payload from a raw / lz4 / otherwise
/// non-zstd one without attempting a full header parse.
#[must_use]
pub fn is_zstd_frame(payload: &[u8]) -> bool {
    payload.starts_with(&ZSTD_FRAME_MAGIC)
}

/// Walks a zstd frame's block structure key-free, WITHOUT decompressing.
///
/// Parses the frame header (RFC 8878 §3.1.1) to locate the first block, then
/// walks the 3-byte block-header chain to the final block, recording each
/// inner block's type, on-disk length, and (for `Raw` / `Rle`) regenerated
/// size. No block content is decompressed, so a multi-block cold-tier frame
/// can be inspected — and a truncated / malformed block localised — without
/// the cost or decode-bomb exposure of a full decode, and without the
/// encryption key (the caller passes the already-plaintext compressed
/// payload; encrypted blocks have an opaque ciphertext body and cannot be
/// walked this way).
///
/// # Errors
///
/// Returns [`crate::Error::InvalidHeader`] if `payload` does not start with a
/// zstd frame, the frame header is truncated, a block declares a reserved
/// type, or a block's content runs past the end of `payload` (a truncated or
/// forged frame).
///
/// # Examples
///
/// ```
/// use lsm_tree::inspect::{census_zstd_frame, is_zstd_frame, ZstdBlockType};
/// // A minimal hand-built zstd frame: magic + single-segment header
/// // (declared content size 5) + one RLE last-block regenerating 5 bytes
/// // of 0x41. Hand-built so the example needs no compression feature.
/// let frame = [0x28, 0xB5, 0x2F, 0xFD, 0x20, 0x05, 0x2B, 0x00, 0x00, 0x41];
/// assert!(is_zstd_frame(&frame));
/// let census = census_zstd_frame(&frame).unwrap();
/// assert_eq!(census.blocks.len(), 1);
/// assert_eq!(census.blocks[0].block_type, ZstdBlockType::Rle);
/// assert!(census.blocks[0].last);
/// assert_eq!(census.blocks[0].decompressed_len, Some(5));
/// ```
pub fn census_zstd_frame(payload: &[u8]) -> crate::Result<ZstdFrameCensus> {
    use crate::Error::InvalidHeader;

    // Magic (4) + frame descriptor (1) is the minimum to read.
    if !payload.starts_with(&ZSTD_FRAME_MAGIC) {
        return Err(InvalidHeader("zstd frame magic"));
    }
    let desc = *payload
        .get(4)
        .ok_or(InvalidHeader("zstd frame descriptor"))?;
    let fcs_flag = desc >> 6;
    let single_segment = (desc >> 5) & 1 == 1;
    let content_checksum = (desc >> 2) & 1 == 1;
    // Field-size tables straight from RFC 8878 §3.1.1.1.
    let dict_id_size = match desc & 0x3 {
        0 => 0usize,
        1 => 1,
        2 => 2,
        // dict-id flag is 2 bits, so the only remaining value is 3 (= 4 bytes).
        _ => 4,
    };
    let fcs_size = match fcs_flag {
        0 => usize::from(single_segment),
        1 => 2,
        2 => 4,
        // fcs_flag is 2 bits, so the only remaining value is 3.
        _ => 8,
    };

    let mut pos = 5usize;

    // Window descriptor is present iff the frame is not single-segment.
    let window_from_descriptor = if single_segment {
        None
    } else {
        let wd = *payload
            .get(pos)
            .ok_or(InvalidHeader("zstd window descriptor"))?;
        pos += 1;
        let exp = u64::from(wd >> 3);
        let mantissa = u64::from(wd & 0x7);
        let window_base = 1u64 << (10 + exp);
        Some(window_base + (window_base / 8) * mantissa)
    };

    // Dictionary id (little-endian, `dict_id_size` bytes; 0 means "none").
    let mut dictionary_id = None;
    if dict_id_size > 0 {
        let bytes = payload
            .get(pos..pos + dict_id_size)
            .ok_or(InvalidHeader("zstd dictionary id"))?;
        let mut id = 0u32;
        for (i, &b) in bytes.iter().enumerate() {
            id |= u32::from(b) << (8 * i);
        }
        if id != 0 {
            dictionary_id = Some(id);
        }
        pos += dict_id_size;
    }

    // Frame content size (little-endian; the 2-byte form has a +256 bias).
    let mut frame_content_size = None;
    if fcs_size > 0 {
        let bytes = payload
            .get(pos..pos + fcs_size)
            .ok_or(InvalidHeader("zstd frame content size"))?;
        let mut fcs = 0u64;
        for (i, &b) in bytes.iter().enumerate() {
            fcs |= u64::from(b) << (8 * i);
        }
        if fcs_size == 2 {
            fcs += 256;
        }
        frame_content_size = Some(fcs);
        pos += fcs_size;
    }

    let frame_header_len =
        u8::try_from(pos).map_err(|_| InvalidHeader("zstd frame header too long"))?;
    // Single-segment frames omit the window descriptor: the window equals the
    // declared content size.
    let window_size = window_from_descriptor.or(frame_content_size);

    // Walk the 3-byte block-header chain to the last block. `pos` advances by
    // at least 3 each iteration and is bounded by `payload.len()`, so the loop
    // terminates without an explicit block-count cap.
    let mut blocks = Vec::new();
    let mut index = 0u32;
    loop {
        // Exactly three bytes; the slice pattern avoids per-byte indexing
        // (the `else` is unreachable since `get(..3)` yields a 3-byte slice).
        let &[b0, b1, b2] = payload
            .get(pos..pos + 3)
            .ok_or(InvalidHeader("zstd block header"))?
        else {
            return Err(InvalidHeader("zstd block header"));
        };
        let raw = u32::from(b0) | (u32::from(b1) << 8) | (u32::from(b2) << 16);
        let last = raw & 1 == 1;
        let block_size = raw >> 3; // 21-bit Block_Size
        let header_offset = pos as u64;
        pos += 3;

        let (block_type, content_len, decompressed_len) = match (raw >> 1) & 0x3 {
            0 => (ZstdBlockType::Raw, block_size, Some(block_size)),
            1 => (ZstdBlockType::Rle, 1u32, Some(block_size)),
            2 => (ZstdBlockType::Compressed, block_size, None),
            // 3 = Reserved: per spec this is corruption.
            _ => return Err(InvalidHeader("zstd reserved block type")),
        };

        let content_end = pos
            .checked_add(content_len as usize)
            .ok_or(InvalidHeader("zstd block content overflow"))?;
        if content_end > payload.len() {
            return Err(InvalidHeader("zstd block content truncated"));
        }
        pos = content_end;

        blocks.push(InnerZstdBlock {
            index,
            block_type,
            last,
            header_offset,
            content_len,
            decompressed_len,
        });
        index += 1;
        if last {
            break;
        }
    }

    Ok(ZstdFrameCensus {
        frame_header_len,
        window_size,
        dictionary_id,
        frame_content_size,
        content_checksum,
        blocks,
    })
}

#[cfg(test)]
#[allow(
    clippy::unwrap_used,
    clippy::indexing_slicing,
    clippy::expect_used,
    reason = "test code"
)]
mod census_tests {
    use super::*;

    /// Hand-built minimal zstd frame: magic + single-segment header (declared
    /// content size 5) + one RLE last-block regenerating 5 bytes of 0x41.
    /// Built without the compression feature so the structural-walk tests run
    /// in the default build profile.
    const RLE_FRAME: [u8; 10] = [0x28, 0xB5, 0x2F, 0xFD, 0x20, 0x05, 0x2B, 0x00, 0x00, 0x41];

    #[test]
    fn is_zstd_frame_matches_only_on_magic() {
        assert!(is_zstd_frame(&RLE_FRAME));
        assert!(!is_zstd_frame(b"not a frame"));
        assert!(!is_zstd_frame(&[0x28, 0xB5, 0x2F])); // too short
    }

    #[test]
    fn census_zstd_frame_decodes_single_rle_block() {
        let census = census_zstd_frame(&RLE_FRAME).unwrap();
        assert_eq!(census.frame_header_len, 6);
        // Single-segment frame: window size falls back to the declared
        // content size (5).
        assert_eq!(census.window_size, Some(5));
        assert_eq!(census.frame_content_size, Some(5));
        assert!(!census.content_checksum);
        assert_eq!(census.blocks.len(), 1);
        let b = &census.blocks[0];
        assert_eq!(b.index, 0);
        assert_eq!(b.block_type, ZstdBlockType::Rle);
        assert!(b.last);
        assert_eq!(b.header_offset, 6);
        // RLE stores a single repeated byte on disk; regenerated size is 5.
        assert_eq!(b.content_len, 1);
        assert_eq!(b.decompressed_len, Some(5));
    }

    #[test]
    fn census_zstd_frame_rejects_non_zstd_payload() {
        let err = census_zstd_frame(b"definitely not zstd").unwrap_err();
        assert!(matches!(err, crate::Error::InvalidHeader(_)));
    }

    #[test]
    fn census_zstd_frame_rejects_truncated_frame() {
        // Drop the RLE block's single content byte: the block content now runs
        // past EOF.
        let err = census_zstd_frame(&RLE_FRAME[..RLE_FRAME.len() - 1]).unwrap_err();
        assert!(matches!(err, crate::Error::InvalidHeader(_)));
    }

    #[test]
    fn census_zstd_frame_rejects_reserved_block_type() {
        // Same header shape as RLE_FRAME but block type bits = 3 (Reserved):
        // block-header raw = (5 << 3) | (3 << 1) | 1 = 0x2F.
        let mut frame = RLE_FRAME;
        frame[6] = 0x2F;
        let err = census_zstd_frame(&frame).unwrap_err();
        assert!(matches!(err, crate::Error::InvalidHeader(_)));
    }

    #[test]
    fn ecc_scheme_parity_trailer_len() {
        // RS(4,2): four data shards + two parity shards → non-zero trailer.
        let rs = EccSchemeInfo::Shard {
            data_shards: 4,
            parity_shards: 2,
        };
        assert!(rs.parity_trailer_len(4096) > 0);
        // SEC-DED: one check byte per 8-byte word.
        assert_eq!(EccSchemeInfo::Secded.parity_trailer_len(64), 8);
        assert_eq!(EccSchemeInfo::Secded.parity_trailer_len(65), 9);
    }

    // Real-compression coverage: walk a frame produced by the actual zstd
    // backend (multi-byte, multi-block-capable). Gated on a zstd build since
    // the backend is only present then.
    #[cfg(zstd_any)]
    mod with_zstd {
        use super::*;
        use crate::compression::CompressionProvider;

        #[test]
        fn census_real_frame_walks_to_last_block() {
            let frame = crate::compression::ZstdBackend::compress(&vec![0x5Au8; 8192], 3).unwrap();
            assert!(is_zstd_frame(&frame));
            let census = census_zstd_frame(&frame).unwrap();
            assert!(!census.blocks.is_empty());
            assert!(census.frame_header_len >= 5);
            // Exactly one last block, and it is the final entry.
            assert_eq!(census.blocks.iter().filter(|b| b.last).count(), 1);
            assert!(census.blocks.last().unwrap().last);
            // Indices contiguous from zero; headers in-bounds and non-overlapping.
            let mut prev = u64::from(census.frame_header_len);
            for (i, b) in census.blocks.iter().enumerate() {
                assert_eq!(b.index as usize, i);
                assert!(b.header_offset >= prev);
                let end = b.header_offset + 3 + u64::from(b.content_len);
                assert!(end <= frame.len() as u64);
                prev = end;
            }
        }

        #[test]
        fn census_real_frame_rejects_truncation() {
            let frame = crate::compression::ZstdBackend::compress(&vec![0x7Eu8; 4096], 3).unwrap();
            let truncated = &frame[..frame.len() - 8];
            let err = census_zstd_frame(truncated).unwrap_err();
            assert!(matches!(err, crate::Error::InvalidHeader(_)));
        }
    }
}