structured_zstd/encoding/

frame_compressor.rs

//! Utilities and interfaces for encoding an entire frame. Allows reusing resources

use alloc::vec::Vec;
#[cfg(feature = "hash")]
use twox_hash::XxHash64;

#[cfg(feature = "hash")]
use core::hash::Hasher;

use super::{
    CompressionLevel, LiteralCompressionMode, Matcher, block_header::BlockHeader,
    fastpath::EncodeTier, frame_header::FrameHeader, levels::*,
    match_generator::MatchGeneratorDriver, workspace::IngestBuffer,
};
use crate::common::MAX_BLOCK_SIZE;
use crate::fse::fse_encoder::{FSETable, default_ll_table, default_ml_table, default_of_table};

use crate::io::{Read, Write};

/// A dictionary prepared for the ENCODER side, analogous to zstd's `CDict`
/// (vs the decoder's [`Dictionary`](crate::decoding::Dictionary) / `DDict`).
///
/// It carries the entropy tables, content, and repeat-offset history the
/// compressor needs, but is a distinct type with **no decode path**: there is
/// no way to turn it into a [`DictionaryHandle`](crate::decoding::DictionaryHandle)
/// or feed it to a [`FrameDecoder`](crate::decoding::FrameDecoder). That keeps
/// the compress-only state (which may have been parsed without building the
/// decode lookup tables, see
/// [`set_dictionary_from_bytes`](FrameCompressor::set_dictionary_from_bytes))
/// from ever reaching the decode side — the encoder/decoder dictionary split
/// mirrors C zstd's `CDict` / `DDict`.
/// Cloning one is a handle, not a copy: it is attached to a compressor by
/// value, so a dictionary serving many frames would otherwise have its parsed
/// tables and content duplicated for each of them — on exactly the path where
/// one dictionary is prepared once precisely to be used again and again.
///
/// The encoder entropy tables a dictionary seeds into each frame's first block
/// (upstream zstd `cdict->cBlockState`) are built once, here, and shared by
/// every compressor the dictionary is attached to.
#[derive(Clone)]
pub struct EncoderDictionary {
    pub(crate) inner: SharedEncoderDictionary,
    /// Size of the serialized dictionary this was built from (header, entropy
    /// tables, repeat offsets and content); the CDict cParams tier key
    /// (upstream `ZSTD_createCDict(dictBuffer, dictSize, level)`). Falls back
    /// to the content length when the wrapped
    /// [`Dictionary`](crate::decoding::Dictionary) was handed over
    /// already parsed ([`Self::from_dictionary`]) — exact for raw-content
    /// dictionaries, a close lower bound otherwise.
    serialized_len: usize,
}

/// What an [`EncoderDictionary`] shares between its clones: the parsed
/// dictionary and the encoder entropy tables derived from it, in one
/// allocation so a clone is one reference count, not two. Reads through to the
/// [`Dictionary`](crate::decoding::Dictionary) for its content, offsets and id.
pub(crate) struct EncoderDictionaryParts {
    dictionary: crate::decoding::Dictionary,
    pub(crate) entropy: CachedDictionaryEntropy,
}

impl core::ops::Deref for EncoderDictionaryParts {
    type Target = crate::decoding::Dictionary;

    fn deref(&self) -> &Self::Target {
        &self.dictionary
    }
}

/// Shared owner of [`EncoderDictionaryParts`]: `Arc` on atomic-pointer
/// targets, `Rc` otherwise, as [`SharedFseTable`] is.
#[cfg(target_has_atomic = "ptr")]
pub(crate) type SharedEncoderDictionary = alloc::sync::Arc<EncoderDictionaryParts>;
#[cfg(not(target_has_atomic = "ptr"))]
pub(crate) type SharedEncoderDictionary = alloc::rc::Rc<EncoderDictionaryParts>;

impl EncoderDictionary {
    /// Share `dictionary` with the entropy tables it seeds, built once here.
    fn prepared(dictionary: crate::decoding::Dictionary, serialized_len: usize) -> Self {
        let entropy = CachedDictionaryEntropy::from_dictionary(&dictionary);
        Self {
            inner: SharedEncoderDictionary::new(EncoderDictionaryParts {
                dictionary,
                entropy,
            }),
            serialized_len,
        }
    }

    /// Wrap an already-parsed [`Dictionary`](crate::decoding::Dictionary) for
    /// encoder use. A fully-decoded dictionary is valid here; only the encoder
    /// entropy tables, content, and offset history are read. The CDict cParams
    /// tier is keyed by the content length here; prefer [`Self::from_bytes`]
    /// when the serialized blob is at hand — it keys the tier by the exact
    /// serialized size as upstream `ZSTD_createCDict` does.
    pub fn from_dictionary(dictionary: crate::decoding::Dictionary) -> Self {
        let serialized_len = dictionary.dict_content.len();
        Self::prepared(dictionary, serialized_len)
    }

    /// Parse a serialized dictionary blob for encoder use, skipping the decode
    /// lookup-table build the encoder never reads (see
    /// `Dictionary::decode_dict_for_encoding`). The encoder entropy tables — and
    /// thus the emitted frame — are identical to a full parse.
    pub fn from_bytes(
        raw_dictionary: &[u8],
    ) -> Result<Self, crate::decoding::errors::DictionaryDecodeError> {
        Ok(Self::prepared(
            crate::decoding::Dictionary::decode_dict_for_encoding(raw_dictionary)?,
            raw_dictionary.len(),
        ))
    }

    /// Load whichever kind of dictionary `raw_dictionary` holds, the way
    /// `zstd -D` does: a serialized blob is parsed, anything else is taken as
    /// raw content (see
    /// [`Dictionary::from_serialized_or_raw_content`](crate::decoding::Dictionary::from_serialized_or_raw_content)).
    ///
    /// Either way the blob's own length is what the compression-parameter tier
    /// is chosen by, which is why this exists rather than parsing and calling
    /// [`Self::from_dictionary`]: that keys the tier on the content length, and
    /// for a serialized dictionary the entropy tables in between can put the
    /// two on opposite sides of a boundary.
    pub fn from_serialized_or_raw_content(
        raw_dictionary: &[u8],
    ) -> Result<Self, crate::decoding::errors::DictionaryDecodeError> {
        // Parsed for the encoder, which reads the entropy probabilities, the
        // content and the offsets and never the decode lookup tables: routing a
        // serialized blob through the full parser builds those tables for
        // nothing. The emitted frame is identical either way — only the wasted
        // build is dropped (see `Dictionary::decode_dict_for_encoding`).
        if crate::decoding::Dictionary::is_serialized(raw_dictionary) {
            return Self::from_bytes(raw_dictionary);
        }
        Ok(Self::prepared(
            crate::decoding::Dictionary::from_raw_content(0, raw_dictionary.to_vec())?,
            raw_dictionary.len(),
        ))
    }

    /// Heap bytes the prepared dictionary holds: the shared allocation itself,
    /// the parsed dictionary's content and tables, and the entropy tables it
    /// seeds. Clones share all of it, so this counts once however many
    /// compressors the dictionary is attached to.
    ///
    /// # Examples
    /// ```
    /// use structured_zstd::encoding::EncoderDictionary;
    ///
    /// let dictionary =
    ///     EncoderDictionary::from_serialized_or_raw_content(b"raw content, no header").unwrap();
    /// assert!(dictionary.heap_size() >= b"raw content, no header".len());
    /// ```
    pub fn heap_size(&self) -> usize {
        Self::shared_allocation_size()
            + self.inner.dictionary.heap_bytes()
            + self.inner.entropy.heap_size()
    }

    /// Upper bound on [`Self::heap_size`] for a dictionary prepared from a
    /// `serialized_len`-byte blob, before preparing it: for budgeting memory
    /// ahead of time, as the C `ZSTD_estimateCDictSize` does.
    ///
    /// # Examples
    /// ```
    /// use structured_zstd::encoding::EncoderDictionary;
    ///
    /// let blob = b"raw content, no header";
    /// let dictionary = EncoderDictionary::from_serialized_or_raw_content(blob).unwrap();
    /// assert!(dictionary.heap_size() <= EncoderDictionary::estimated_heap_size(blob.len()));
    /// ```
    pub fn estimated_heap_size(serialized_len: usize) -> usize {
        // The content is at most the blob. The tables are bounded by their
        // largest form: every FSE table and the Huffman table at their maximum
        // accuracy, as a dictionary's entropy section can declare them.
        const PREPARED_TABLES_HEAP: usize = 64 * 1024;
        // Saturating on purpose: `usize::MAX` still bounds what no allocation
        // can exceed, where a wrapped sum would under-report.
        Self::shared_allocation_size()
            .saturating_add(serialized_len)
            .saturating_add(PREPARED_TABLES_HEAP)
    }

    /// Bytes of the shared allocation itself: `Arc` and `Rc` both place two
    /// word-sized counts ahead of the value, padded to its alignment.
    fn shared_allocation_size() -> usize {
        core::alloc::Layout::new::<[usize; 2]>()
            .extend(core::alloc::Layout::new::<EncoderDictionaryParts>())
            .expect("a fixed-size layout extends")
            .0
            .pad_to_align()
            .size()
    }

    /// Heap bytes that `handles` alone keep alive: each distinct dictionary
    /// among them once, and only when no handle outside them shares it.
    ///
    /// For an owner holding several handles, such as a context whose
    /// compressors each keep the dictionary they last ran with: what it reports
    /// should count one dictionary once, and none that someone else also holds.
    ///
    /// # Examples
    /// ```
    /// use structured_zstd::encoding::EncoderDictionary;
    ///
    /// let dictionary = EncoderDictionary::from_serialized_or_raw_content(b"some shared history").unwrap();
    /// let copy = dictionary.clone();
    /// // Both handles are in the set: counted once.
    /// assert_eq!(
    ///     EncoderDictionary::exclusive_heap_size([&dictionary, &copy]),
    ///     dictionary.heap_size()
    /// );
    /// // `copy` is held elsewhere: nothing is this set's alone.
    /// assert_eq!(EncoderDictionary::exclusive_heap_size([&dictionary]), 0);
    /// ```
    pub fn exclusive_heap_size<'a, I>(handles: I) -> usize
    where
        I: IntoIterator<Item = &'a EncoderDictionary>,
        I::IntoIter: Clone,
    {
        let handles = handles.into_iter();
        let mut total = 0;
        for (position, handle) in handles.clone().enumerate() {
            let same = |other: &&EncoderDictionary| {
                SharedEncoderDictionary::ptr_eq(&handle.inner, &other.inner)
            };
            // The first handle to an allocation speaks for all of them.
            if handles.clone().take(position).any(|other| same(&other)) {
                continue;
            }
            // Holders, not references: one handle listed twice is one count.
            let held_here = handles
                .clone()
                .enumerate()
                .filter(|(at, other)| {
                    same(other)
                        && !handles
                            .clone()
                            .take(*at)
                            .any(|earlier| core::ptr::eq(earlier, *other))
                })
                .count();
            if SharedEncoderDictionary::strong_count(&handle.inner) == held_here {
                total += handle.heap_size();
            }
        }
        total
    }

    /// The content and serialized sizes the encoder's matcher is hinted with.
    pub(crate) fn sizes(&self) -> crate::encoding::DictionarySizes {
        crate::encoding::DictionarySizes {
            content: self.inner.dict_content.len(),
            serialized: self.serialized_len,
        }
    }

    /// Whether frames compress against this dictionary: one prepared from a
    /// blob under 8 bytes takes no part, as zstd.h states for every dictionary
    /// entry point and upstream zstd applies when it inserts one
    /// (`zstd_compress.c:5206`). Attaching such a dictionary attaches none. A
    /// decoder still reads that blob as content, as upstream's does.
    ///
    /// ```
    /// use structured_zstd::encoding::EncoderDictionary;
    ///
    /// let tiny = EncoderDictionary::from_serialized_or_raw_content(b"tiny").unwrap();
    /// assert!(!tiny.is_used_for_compression());
    /// let history = EncoderDictionary::from_serialized_or_raw_content(b"some history").unwrap();
    /// assert!(history.is_used_for_compression());
    /// ```
    pub fn is_used_for_compression(&self) -> bool {
        self.serialized_len >= 8
    }

    /// Bytes of dictionary content, the part a frame can reference.
    ///
    /// This is the extent a caller weighs a source against when deciding
    /// whether a frame is still about the dictionary or about its own content.
    ///
    /// # Examples
    ///
    /// ```
    /// use structured_zstd::encoding::EncoderDictionary;
    ///
    /// let dictionary = EncoderDictionary::from_serialized_or_raw_content(&[7u8; 4096])
    ///     .expect("raw content is always usable");
    /// assert_eq!(dictionary.content_size(), 4096);
    /// ```
    pub fn content_size(&self) -> usize {
        self.inner.dict_content.len()
    }

    /// The dictionary content a frame's matches can reach back into.
    ///
    /// ```
    /// use structured_zstd::encoding::EncoderDictionary;
    ///
    /// let dictionary = EncoderDictionary::from_serialized_or_raw_content(b"history").unwrap();
    /// assert_eq!(dictionary.content(), b"history");
    /// ```
    pub fn content(&self) -> &[u8] {
        &self.inner.dict_content
    }

    /// The dictionary id.
    ///
    /// Zero is a raw-content dictionary, which has no header to carry an id.
    /// Such a dictionary attaches like any other; what changes is the frame,
    /// which omits the `Dictionary_ID` field rather than storing a zero, so a
    /// decoder has to be handed the same bytes explicitly.
    pub fn id(&self) -> u32 {
        self.inner.id
    }
}

/// How a [`FrameCompressor`] holds its dictionary: owned, as
/// [`FrameCompressor::set_encoder_dictionary`] attaches one, or borrowed for
/// the compressor's lifetime through
/// [`FrameCompressor::new_with_matcher_and_dictionary`], so a short-lived
/// compressor reads a prepared dictionary without taking a reference on it.
/// Implemented for [`EncoderDictionary`] and `&EncoderDictionary` only.
pub trait DictionaryHold: core::borrow::Borrow<EncoderDictionary> + hold::Sealed {
    /// Whether the compressor owns what it holds, so its footprint counts it.
    const OWNED: bool;
}

impl DictionaryHold for EncoderDictionary {
    const OWNED: bool = true;
}

impl DictionaryHold for &EncoderDictionary {
    const OWNED: bool = false;
}

mod hold {
    pub trait Sealed {}
    impl Sealed for super::EncoderDictionary {}
    impl Sealed for &super::EncoderDictionary {}
}

/// The dictionary `held` holds, however it holds it.
fn attached<D: DictionaryHold>(held: &Option<D>) -> Option<&EncoderDictionary> {
    held.as_ref().map(core::borrow::Borrow::borrow)
}

/// An interface for compressing arbitrary data with the ZStandard compression algorithm.
///
/// `FrameCompressor` will generally be used by:
/// 1. Initializing a compressor by providing a buffer of data using `FrameCompressor::new()`
/// 2. Starting compression and writing that compression into a vec using `FrameCompressor::begin`
///
/// # Examples
/// ```
/// use structured_zstd::encoding::{FrameCompressor, CompressionLevel};
/// let mock_data: &[_] = &[0x1, 0x2, 0x3, 0x4];
/// let mut output = std::vec::Vec::new();
/// // Initialize a compressor.
/// let mut compressor = FrameCompressor::new(CompressionLevel::Uncompressed);
/// compressor.set_source(mock_data);
/// compressor.set_drain(&mut output);
///
/// // `compress` writes the compressed output into the provided buffer.
/// compressor.compress();
/// ```
pub struct FrameCompressor<
    R: Read = &'static [u8],
    W: Write = Vec<u8>,
    M: Matcher = MatchGeneratorDriver,
    D: DictionaryHold = EncoderDictionary,
> {
    uncompressed_data: Option<R>,
    compressed_data: Option<W>,
    compression_level: CompressionLevel,
    dictionary: Option<D>,
    source_size_hint: Option<u64>,
    state: CompressState<M>,
    /// When true, emitted frames omit the 4-byte magic number prefix
    /// (`ZSTD_f_zstd1_magicless`). Default false. The caller is
    /// responsible for ensuring the decoder is configured for the
    /// matching format — wire-format only round-trips with a
    /// magicless-aware decoder.
    magicless: bool,
    /// Whether to emit a trailing XXH64 content checksum and set the frame
    /// header's `Content_Checksum_flag` (semantics of upstream
    /// `ZSTD_c_checksumFlag`). Default `false`, matching the upstream
    /// library default; combined with the `hash` feature at frame-build
    /// time, so without `hash` no checksum is emitted regardless. Set via
    /// [`Self::set_content_checksum`].
    content_checksum: bool,
    /// Diagnostic: skip the block pre-splitter and cut full blocks only
    /// (upstream's block structure under `ZSTD_generateSequences`, whose
    /// sequence-collecting mode never accrues the savings the splitter
    /// requires). Set via `set_pre_split_disabled` (a `bench-internals`
    /// diagnostic); default `false`.
    pre_split_disabled: bool,
    /// Whether to record `Frame_Content_Size` in the frame header when the
    /// total size is known (semantics of upstream `ZSTD_c_contentSizeFlag`).
    /// Default `true`, matching upstream. With the flag off the header
    /// carries a window descriptor instead (single-segment requires an FCS,
    /// so it is disabled too). Set via [`Self::set_content_size_flag`].
    content_size_flag: bool,
    /// Whether to record the dictionary ID in the frame header when a
    /// dictionary is attached (semantics of upstream `ZSTD_c_dictIDFlag`).
    /// Default `true`, matching upstream. Decoders can still decode the
    /// frame by being handed the right dictionary explicitly. Set via
    /// [`Self::set_dictionary_id_flag`].
    dict_id_flag: bool,
    /// Upper bound on emitted block sizes (semantics of upstream
    /// `ZSTD_c_targetCBlockSize`): capping the RAW block length at the
    /// target bounds every physical block's compressed payload at the
    /// target too (a compressed block never exceeds its raw input — the
    /// raw-block fallback fires otherwise), so blocks land at or under
    /// `target + 3` header bytes on the wire. `None` = no target (full
    /// 128 KiB blocks). Set via [`Self::set_target_block_size`].
    target_block_size: Option<u32>,
    /// Largest block the frame may carry (upstream `ZSTD_c_maxBlockSize`),
    /// below the format's 128 KiB. Set via [`Self::set_max_block_size`].
    max_block_size: Option<u32>,
    #[cfg(feature = "hash")]
    hasher: XxHash64,
    /// Block-layout introspection populated at the end of every
    /// successful `compress()`. `None` until the first call.
    /// Behind the `lsm` feature gate.
    #[cfg(feature = "lsm")]
    frame_emit_info: Option<crate::encoding::frame_emit_info::FrameEmitInfo>,
    /// When `true`, `compress()` XXH64-hashes each block's
    /// uncompressed bytes and appends the low-32-bit digest to
    /// `block_checksums`. Default `false` (zero cost). Gated on
    /// `all(lsm, hash)` because XXH64 lives behind the `hash`
    /// feature; an `lsm`-only build has no way to compute digests.
    #[cfg(all(feature = "lsm", feature = "hash"))]
    per_block_checksums_enabled: bool,
    /// Per-block XXH64 (low 32 bits) digests captured during
    /// `compress()` when `per_block_checksums_enabled` is set. Ordered
    /// by block-emit order. `None` until the first call after enabling.
    /// Gated on `all(lsm, hash)` (see `per_block_checksums_enabled`).
    #[cfg(all(feature = "lsm", feature = "hash"))]
    block_checksums: Option<alloc::vec::Vec<u32>>,
    /// Per-physical-block decompressed (regenerated) sizes captured
    /// during `compress()`, in block-emit order (1:1 with
    /// `frame_emit_info.blocks`). Always captured under `lsm` (no
    /// opt-in, unlike `block_checksums`) because `FrameEmitInfo` is
    /// always built under `lsm` and `decompressed_byte_range` needs
    /// the per-block sizes. Cleared and refilled per frame.
    #[cfg(feature = "lsm")]
    block_decompressed_sizes: alloc::vec::Vec<u32>,
    tuning: FrameTuning,
}

/// What a compressor keeps of the fine-grained parameters (#27) beside the
/// level: the parts the literal-compression gates and the block splitter read
/// at frame start, where the matcher holds the rest. Shared by
/// [`FrameCompressor`] and [`CompressionContext`](crate::encoding::CompressionContext).
#[derive(Clone, Copy, Default)]
pub(crate) struct FrameTuning {
    /// A public-parameter [`Strategy`](crate::encoding::Strategy) override:
    /// its tag and lazy depth (the collapsed `Lazy` tag needs the depth for
    /// the pre-split tier). `Some` wins over the level-derived
    /// `state.strategy_tag` so the gates and the dict-attach cutoff see the
    /// strategy the matcher runs; `None` keeps the level's.
    pub(crate) strategy: Option<(crate::encoding::strategy::StrategyTag, u8)>,
    /// Public `target_length` override, kept so the raw-literals gate can be
    /// recomputed per frame: a dictionary attached or cleared after
    /// `set_parameters` can move the strategy the gate reads.
    pub(crate) target_length: Option<u32>,
    /// Public literal-compression mode (upstream
    /// `ZSTD_c_literalCompressionMode`), kept for the same per-frame
    /// recomputation of the raw-literals gate.
    pub(crate) literal_compression: LiteralCompressionMode,
}

impl FrameTuning {
    /// The tuning `overrides` asks for.
    pub(crate) fn from_overrides(overrides: &crate::encoding::parameters::ParamOverrides) -> Self {
        Self {
            strategy: overrides.strategy.map(|s| (s.tag(), s.lazy_depth())),
            target_length: overrides.target_length,
            literal_compression: overrides.literal_compression,
        }
    }
}

#[derive(Clone, Default)]
pub(crate) struct CachedDictionaryEntropy {
    pub(crate) huff: Option<crate::huff0::huff0_encoder::HuffmanTable>,
    pub(crate) ll_previous: Option<PreviousFseTable>,
    pub(crate) ml_previous: Option<PreviousFseTable>,
    pub(crate) of_previous: Option<PreviousFseTable>,
}

impl CachedDictionaryEntropy {
    /// Heap bytes the cached dictionary entropy holds: the literals Huffman
    /// table plus any `Custom` LL/ML/OF FSE tables (the `Arc`-boxed `FSETable`
    /// payload and its flat state array). `Default` / `Rle` variants own no heap.
    pub(crate) fn heap_size(&self) -> usize {
        let mut total = self.huff.as_ref().map_or(0, |h| h.heap_size());
        for prev in [&self.ll_previous, &self.ml_previous, &self.of_previous] {
            if let Some(PreviousFseTable::Custom(table)) = prev {
                total +=
                    core::mem::size_of::<crate::fse::fse_encoder::FSETable>() + table.heap_size();
            }
        }
        total
    }

    /// Derive the encoder-side entropy tables a dictionary seeds for the first
    /// block of each frame (the upstream zstd `cdict->cBlockState`): the literals
    /// Huffman table plus the literal-length / match-length / offset FSE
    /// "previous" tables. Shared by [`FrameCompressor`] and
    /// [`crate::encoding::StreamingEncoder`] so both seed identically.
    pub(crate) fn from_dictionary(dictionary: &crate::decoding::Dictionary) -> Self {
        Self {
            huff: dictionary.huf.table.to_encoder_table(),
            ll_previous: dictionary
                .fse
                .literal_lengths
                .to_encoder_table()
                .map(|table| PreviousFseTable::Custom(SharedFseTable::new(table))),
            ml_previous: dictionary
                .fse
                .match_lengths
                .to_encoder_table()
                .map(|table| PreviousFseTable::Custom(SharedFseTable::new(table))),
            of_previous: dictionary
                .fse
                .offsets
                .to_encoder_table()
                .map(|table| PreviousFseTable::Custom(SharedFseTable::new(table))),
        }
    }
}

/// Shared owner for a custom "previous" FSE encoder table. `Arc` on
/// atomic-pointer targets, `Rc` otherwise (keeps `no_std` no-atomics
/// builds compiling, single-thread there anyway), mirroring
/// `decoding::dictionary::SharedDictionary`. Cloning the cached
/// dictionary entropy into the per-frame state is then a refcount bump,
/// not a full `FSETable` copy — the upstream zstd references `cdict->cBlockState`
/// instead of rebuilding it per frame.
#[cfg(target_has_atomic = "ptr")]
pub(crate) type SharedFseTable = alloc::sync::Arc<FSETable>;
#[cfg(not(target_has_atomic = "ptr"))]
pub(crate) type SharedFseTable = alloc::rc::Rc<FSETable>;

/// Bytes a [`SharedFseTable`] allocation carries in FRONT of the table: the two
/// reference counts, then whatever padding the table's alignment adds. One
/// allocation holds both, so a caller sizing a context is told about the whole
/// of it rather than the payload alone.
pub(crate) const fn shared_table_overhead() -> usize {
    let counts = 2 * core::mem::size_of::<usize>();
    let align = core::mem::align_of::<FSETable>();
    counts.div_ceil(align) * align
}

#[derive(Clone)]
pub(crate) enum PreviousFseTable {
    // Default tables are immutable and already stored alongside the state, so
    // repeating them only needs a lightweight marker instead of cloning FSETable.
    Default,
    // Shared handle: cloning (per-frame dictionary entropy seed) is a refcount
    // bump. The table is only ever read or REPLACED wholesale (a block that
    // builds a new table swaps in a fresh `SharedFseTable`), never mutated in
    // place, so sharing is sound.
    Custom(SharedFseTable),
    Rle(u8),
}

impl PreviousFseTable {
    pub(crate) fn as_table<'a>(&'a self, default: &'a FSETable) -> Option<&'a FSETable> {
        match self {
            Self::Default => Some(default),
            Self::Custom(table) => Some(table),
            Self::Rle(_) => None,
        }
    }
}

pub(crate) struct FseTables {
    /// The three predefined LL/ML/OF tables are functions of
    /// compile-time-constant distributions. The
    /// [`FseDefaultTable`](crate::fse::fse_encoder::FseDefaultTable) type alias resolves to
    /// `&'static FSETable` when a process-wide cache is available
    /// (atomic-pointer targets, or no-atomic targets with the
    /// `critical-section` feature) and to `Box<FSETable>` on the
    /// cache-less no-atomic path (one per-frame allocation, dropped
    /// with the compressor — no `Box::leak`, no unbounded growth).
    /// Both arms `Deref` to `FSETable`, so consumers in
    /// `encoding/blocks/compressed.rs` borrow through `&` uniformly
    /// without seeing the per-target divergence.
    pub(crate) ll_default: crate::fse::fse_encoder::FseDefaultTable,
    pub(crate) ll_previous: Option<PreviousFseTable>,
    pub(crate) ml_default: crate::fse::fse_encoder::FseDefaultTable,
    pub(crate) ml_previous: Option<PreviousFseTable>,
    pub(crate) of_default: crate::fse::fse_encoder::FseDefaultTable,
    pub(crate) of_previous: Option<PreviousFseTable>,
    /// Where a block builds the table it is about to emit, before that table
    /// becomes the axis's `*_previous`.
    ///
    /// Upstream's `ZSTD_blockState_t` keeps `prevCBlock` and `nextCBlock` for
    /// exactly this: the entropy build reads the previous tables and writes the
    /// next ones, so the two never alias and committing a block is a pointer
    /// swap (`ZSTD_blockState_confirmRepcodesAndEntropyTables`). A block that
    /// ends up raw simply does not swap. Holding the slot here is what lets a
    /// table be built in place instead of on the stack.
    /// `None` until an axis first builds a table. Lazy because a state that
    /// never emits a custom table must not pay for a slot: the block splitter
    /// makes one of these per probe, and eagerly giving each three tables
    /// faulted in pages for buffers most probes never wrote to.
    pub(crate) ll_next: Option<SharedFseTable>,
    pub(crate) ml_next: Option<SharedFseTable>,
    pub(crate) of_next: Option<SharedFseTable>,
}

impl FseTables {
    /// Keep the frame's tables as buffers before the next frame overwrites
    /// them.
    ///
    /// A frame start replaces every previous slot — with the dictionary's
    /// seed, or with nothing — and dropping the table that was there costs a
    /// reused compressor an eleven-kilobyte allocation per axis on the next
    /// frame that builds one. The slot the axis builds into is exactly where
    /// it belongs. Only an unshared table is worth keeping: one still held by
    /// the dictionary entropy cache cannot be built into anyway.
    pub(crate) fn park_previous_before_frame(&mut self) {
        for (previous, next) in [
            (&mut self.ll_previous, &mut self.ll_next),
            (&mut self.ml_previous, &mut self.ml_next),
            (&mut self.of_previous, &mut self.of_next),
        ] {
            // Decided before taking: a handle the cache still holds stays
            // where it is rather than being pulled out and dropped.
            let worth_keeping = matches!(
                previous.as_ref(),
                Some(PreviousFseTable::Custom(handle))
                    if SharedFseTable::strong_count(handle) == 1
            );
            // An occupied slot is not the same as a usable one. A frame that
            // settled on a predefined table parks the dictionary cache's handle
            // here, and a shared handle cannot be built into — so leaving it
            // there while the uniquely owned table is dropped costs the
            // allocation this exists to avoid, on the first custom build of
            // every frame.
            let next_is_a_buffer = next
                .as_ref()
                .is_some_and(|handle| SharedFseTable::strong_count(handle) == 1);
            if !next_is_a_buffer
                && worth_keeping
                && let Some(PreviousFseTable::Custom(handle)) = previous.take()
            {
                *next = Some(handle);
            }
        }
    }

    /// Heap bytes the retained encoder tables hold.
    ///
    /// Both slots of an axis count: the previous table is what the next block
    /// reads, the next slot is the buffer it builds into, and each lives as
    /// long as the compressor. A handle shared with the dictionary entropy
    /// cache is left out, because that cache reports its own tables and the
    /// caller would otherwise be told about the same allocation twice.
    pub(crate) fn heap_size(&self) -> usize {
        // The table AND the reference counts in front of it: a shared handle is
        // one allocation holding both, and reporting only the payload
        // understates every retained table by the control block and whatever
        // padding its alignment adds.
        let per_table = core::mem::size_of::<FSETable>() + shared_table_overhead();
        let mut total = 0;
        for previous in [&self.ll_previous, &self.ml_previous, &self.of_previous] {
            if let Some(PreviousFseTable::Custom(handle)) = previous
                && SharedFseTable::strong_count(handle) == 1
            {
                total += per_table;
            }
        }
        for next in [&self.ll_next, &self.ml_next, &self.of_next] {
            // Same ownership test as the previous slots, and for the same
            // reason: a dictionary-seeded axis that settles on a predefined
            // table parks the CACHE's handle here, and that cache reports the
            // table itself.
            if let Some(handle) = next
                && SharedFseTable::strong_count(handle) == 1
            {
                total += per_table;
            }
        }
        // Where there is neither a pointer atomic nor `critical-section` to
        // guard a process-wide cache, each default table is an owned box built
        // per compressor, so it is this struct's allocation to report. With the
        // cache it is a `&'static` shared by every compressor and counts as
        // nothing.
        #[cfg(not(any(target_has_atomic = "ptr", feature = "critical-section")))]
        {
            total += 3 * core::mem::size_of::<FSETable>();
        }
        total
    }

    /// The most [`Self::heap_size`] reports: both slots of all three axes
    /// owned, and the defaults where they are boxed per compressor.
    pub(crate) const fn max_heap_size() -> usize {
        let owned = 6 * (core::mem::size_of::<FSETable>() + shared_table_overhead());
        if cfg!(any(target_has_atomic = "ptr", feature = "critical-section")) {
            owned
        } else {
            owned + 3 * core::mem::size_of::<FSETable>()
        }
    }

    pub fn new() -> Self {
        Self {
            ll_next: None,
            ml_next: None,
            of_next: None,
            ll_default: default_ll_table(),
            ll_previous: None,
            ml_default: default_ml_table(),
            ml_previous: None,
            of_default: default_of_table(),
            of_previous: None,
        }
    }

    /// Borrow the LL default table as `&FSETable`.
    ///
    /// Test-only now: the encoder and the estimator both destructure
    /// `FseTables` so they can hold a `*_next` slot mutably, and a method
    /// borrowing the whole struct cannot coexist with that. Tests keep it
    /// because they touch one table at a time.
    ///
    /// Abstracts the cfg
    /// split in [`crate::fse::fse_encoder::FseDefaultTable`] —
    /// `&'static FSETable` (atomic / `critical-section`) auto-derefs
    /// directly; `Box<FSETable>` (cache-less no-atomic) derefs
    /// through `Box`. Both arms yield `&FSETable` uniformly so
    /// downstream consumers can stay cfg-agnostic.
    #[inline]
    #[allow(clippy::borrow_deref_ref)]
    #[cfg(test)]
    pub(crate) fn ll_default_ref(&self) -> &FSETable {
        &*self.ll_default
    }

    /// Borrow the ML default table as `&FSETable`. See [`Self::ll_default_ref`].
    #[inline]
    #[allow(clippy::borrow_deref_ref)]
    #[cfg(test)]
    pub(crate) fn ml_default_ref(&self) -> &FSETable {
        &*self.ml_default
    }

    /// Borrow the OF default table as `&FSETable`. See [`Self::ll_default_ref`].
    #[inline]
    #[allow(clippy::borrow_deref_ref)]
    #[cfg(test)]
    pub(crate) fn of_default_ref(&self) -> &FSETable {
        &*self.of_default
    }
}

const PRESPLIT_BLOCK_MIN: usize = 3500;
const PRESPLIT_THRESHOLD_PENALTY_RATE: u64 = 16;
const PRESPLIT_THRESHOLD_BASE: u64 = PRESPLIT_THRESHOLD_PENALTY_RATE - 2;
const PRESPLIT_THRESHOLD_PENALTY: i32 = 3;
const PRESPLIT_CHUNK_SIZE: usize = 8 << 10;
const PRESPLIT_HASH_LOG_MAX: usize = 10;
const PRESPLIT_HASH_TABLE_SIZE: usize = 1 << PRESPLIT_HASH_LOG_MAX;
const PRESPLIT_KNUTH: u32 = 0x9E37_79B9;
/// Upstream zstd `SEGMENT_SIZE` in `ZSTD_splitBlock_fromBorders` (`zstd_preSplit.c:201`).
/// Two `SEGMENT_SIZE`-byte fingerprints — one from the start, one from the end —
/// drive the cheap border heuristic; a third one from the middle disambiguates
/// where in the block the transition sits.
const PRESPLIT_BORDERS_SEGMENT: usize = 512;

#[derive(Clone)]
struct PreSplitFingerprint {
    events: [u32; PRESPLIT_HASH_TABLE_SIZE],
    nb_events: usize,
}

impl Default for PreSplitFingerprint {
    fn default() -> Self {
        Self {
            events: [0; PRESPLIT_HASH_TABLE_SIZE],
            nb_events: 0,
        }
    }
}

/// Grow `out` ahead of the next block so block emission never lands on an
/// amortized-doubling reallocation mid-frame (whose transient old+new copy
/// spikes peak memory to ~3x the output), sizing the reservation from the
/// compression ratio observed so far instead of the whole-input worst case.
///
/// `blocks_start` is where this frame's blocks begin in `out`, `consumed`
/// the input bytes already emitted as blocks, `remaining` the input
/// bytes still to compress (an estimate is fine: a low one only means one
/// more re-estimate later), and `block_capacity` the active block-size cap
/// (`FrameCompressor::block_capacity`) so a small `targetCBlockSize` does
/// not keep a 128 KiB floor in the buffer or undercount header density.
/// Incompressible input re-estimates to ~the full `compress_bound` after
/// the first block — the old up-front policy's worst case — while
/// compressible input stays at output scale.
fn reserve_for_next_block(
    out: &mut Vec<u8>,
    blocks_start: usize,
    consumed: u64,
    remaining: usize,
    block_capacity: usize,
) {
    // Worst-case single-block output: 3-byte header + raw payload, plus
    // slack for the 4-byte frame checksum trailer and a few extra sub-block
    // headers from the post-split emitters, so neither can reallocate.
    let block_bound = remaining.min(block_capacity) + 3 + 16;
    if out.capacity() - out.len() >= block_bound {
        return;
    }
    let produced = (out.len() - blocks_start) as u64;
    let estimate = if consumed == 0 {
        // No ratio signal yet (capacity exhausted before the first block —
        // only reachable with a caller-shrunk `out`): one block's bound.
        block_bound
    } else {
        // remaining * observed ratio + per-block headers + 1/16 slack so a
        // slightly-worsening tail doesn't force a reallocation per block.
        // u128 keeps the product exact for multi-GiB frames.
        let scaled = ((remaining as u128 * produced as u128) / consumed as u128) as u64;
        let headers = (remaining as u64 / block_capacity.max(1) as u64 + 1) * 3;
        usize::try_from(scaled + scaled / 16 + headers + 64).unwrap_or(usize::MAX)
    };
    // `reserve_exact`: the estimate already carries its own slack, and the
    // whole-buffer doubling policy is exactly what this function exists to
    // avoid. The `produced`-sized floor keeps growth geometric when the
    // ratio estimate lands BELOW one block's bound (highly compressible
    // input): without it every block would trigger a block-sized
    // reallocation — O(blocks) buffer copies — while with it the buffer at
    // least doubles its produced span per reallocation (O(log) copies) and
    // the peak stays at output scale.
    out.reserve_exact(estimate.max(block_bound + produced as usize));
}

/// The rate and the width stay ARGUMENTS here, though upstream generates one
/// function per tier and says the speed of the pass relies on compile-time
/// constant propagation (`zstd_preSplit.c:32`, `ZSTD_GEN_RECORD_FINGERPRINT` at
/// `:87`). Monomorphising the walk on both was tried and measured on 8 MiB of
/// repeated log lines at level 6, three interleaved readings a side: cycles
/// overlapping (966-984 M against 973-997 M) and instructions UP, 1,252 M to
/// 1,268 M. Rust inlines this into a walk the caller already picked a tier for,
/// so the propagation is there without the four copies.
fn presplit_hash2(bytes: &[u8], hash_log: usize) -> usize {
    debug_assert!(hash_log >= 8);
    if hash_log == 8 {
        return bytes[0] as usize;
    }
    debug_assert!(hash_log <= PRESPLIT_HASH_LOG_MAX);
    let value = u16::from_le_bytes([bytes[0], bytes[1]]) as u32;
    (value.wrapping_mul(PRESPLIT_KNUTH) >> (32 - hash_log)) as usize
}

fn presplit_record_fingerprint(
    fp: &mut PreSplitFingerprint,
    src: &[u8],
    sampling_rate: usize,
    hash_log: usize,
) {
    // Only the slots this hash can reach. The table is sized for the widest
    // hash the splitter uses; a tier on a narrower one was clearing four times
    // the memory it then touched, sixteen times a block.
    fp.events[..1usize << hash_log].fill(0);
    fp.nb_events = 0;
    if src.len() < 2 {
        return;
    }
    let limit = src.len() - 1;
    let mut n = 0usize;
    while n < limit {
        fp.events[presplit_hash2(&src[n..], hash_log)] += 1;
        n += sampling_rate;
    }
    // Upstream zstd parity: zstd_preSplit.c records the integer division, not the
    // rounded-up number of sampled events from the loop above.
    fp.nb_events += limit / sampling_rate;
}

/// Single-byte histogram pass — matches upstream zstd `HIST_add` over a small
/// segment with `hashLog == 8` (the `hash2` shortcut at
/// `zstd_preSplit.c:36` returns the raw byte). The byChunks path uses
/// 2-byte hashing for `hashLog >= 9`; this helper exists so the borders
/// heuristic doesn't pay for that wider hash on its 512-byte windows.
fn presplit_record_byte_histogram(fp: &mut PreSplitFingerprint, src: &[u8]) {
    // Only the byte slots: a `hash_log == 8` fingerprint is recorded, compared
    // and read in those alone.
    fp.events[..256].fill(0);
    for &b in src {
        fp.events[b as usize] += 1;
    }
    // Upstream zstd `HIST_add` returns the maximum symbol; the caller then sets
    // `nbEvents = SEGMENT_SIZE` explicitly (see `zstd_preSplit.c:213`).
    fp.nb_events = src.len();
}

fn presplit_distance(lhs: &PreSplitFingerprint, rhs: &PreSplitFingerprint, hash_log: usize) -> u64 {
    let slots = 1usize << hash_log;
    // 64-bit, as upstream's `fpDistance` is (`zstd_preSplit.c`). Both factors
    // are per-block sample counts bounded by the block size, so each product is
    // under 2^34 and the sum over at most 2^10 slots is under 2^44 — the 128-bit
    // arithmetic this used to do could not overflow either, it just cost several
    // instructions a slot on a loop the splitter runs sixteen times a block.
    let rn = rhs.nb_events as u64;
    let ln = lhs.nb_events as u64;
    // Four accumulators over slices the optimiser can see the length of, which
    // is what lets this go wide: the splitter runs this loop once per chunk and
    // sixteen times a block, so it counts more slots than the sampling reads
    // bytes.
    let (l, r) = (&lhs.events[..slots], &rhs.events[..slots]);
    let mut acc = [0u64; 4];
    let mut idx = 0;
    while idx + 4 <= slots {
        for lane in 0..4 {
            let left = u64::from(l[idx + lane]) * rn;
            let right = u64::from(r[idx + lane]) * ln;
            acc[lane] += left.abs_diff(right);
        }
        idx += 4;
    }
    let mut distance = acc[0] + acc[1] + acc[2] + acc[3];
    while idx < slots {
        distance += (u64::from(l[idx]) * rn).abs_diff(u64::from(r[idx]) * ln);
        idx += 1;
    }
    distance
}

fn presplit_fingerprints_differ(
    reference: &PreSplitFingerprint,
    new_fp: &PreSplitFingerprint,
    penalty: i32,
    hash_log: usize,
) -> bool {
    debug_assert!(reference.nb_events > 0);
    debug_assert!(new_fp.nb_events > 0);
    let p50 = reference.nb_events as u64 * new_fp.nb_events as u64;
    let deviation = presplit_distance(reference, new_fp, hash_log);
    // Plain `*`: p50 <= (block-sample-count)^2 and the (base+penalty) factor is
    // a small constant, so the product stays well under u64::MAX.
    let threshold =
        p50 * (PRESPLIT_THRESHOLD_BASE + penalty as u64) / PRESPLIT_THRESHOLD_PENALTY_RATE;
    deviation >= threshold
}

fn presplit_merge_events(
    acc: &mut PreSplitFingerprint,
    new_fp: &PreSplitFingerprint,
    hash_log: usize,
) {
    // Plain `+`: `acc` accumulates only the chunks of a single block (caller
    // loops within one block, <= MAX_BLOCK_SIZE), so the merged sample counts
    // stay far under u32 / usize bounds — no overflow.
    //
    // Only the slots this tier's hash reaches: the rest are never recorded into
    // and never read, so merging them is a quarter of the table for nothing.
    // Paired slices rather than indices, so the bound is one check instead of
    // one per slot and the adds can go wide.
    let slots = 1usize << hash_log;
    for (into, from) in acc.events[..slots].iter_mut().zip(&new_fp.events[..slots]) {
        *into += *from;
    }
    acc.nb_events += new_fp.nb_events;
}

fn split_block_by_chunks(block: &[u8], level: usize) -> usize {
    debug_assert_eq!(block.len(), MAX_BLOCK_SIZE as usize);
    debug_assert!((1..=4).contains(&level));
    let (sampling_rate, hash_log) = match level - 1 {
        0 => (43, 8),
        1 => (11, 9),
        2 => (5, 10),
        _ => (1, 10),
    };

    let mut past = PreSplitFingerprint::default();
    let mut new_events = PreSplitFingerprint::default();
    let mut penalty = PRESPLIT_THRESHOLD_PENALTY;
    presplit_record_fingerprint(
        &mut past,
        &block[..PRESPLIT_CHUNK_SIZE],
        sampling_rate,
        hash_log,
    );
    // No pre-check on the ends before the walk. It reads as free — two
    // fingerprints against the sixty-three the walk takes — but a sample of the
    // ends cannot stand in for the walk: an A-B-A block has matching ends and a
    // boundary in the middle, and four megabytes repeated at nearly the window
    // distance lost 13% to exactly that. Upstream has no such check either; the
    // cheap gate that keeps the walk off hopeless input is `savings`, which the
    // caller already applies.
    let mut pos = PRESPLIT_CHUNK_SIZE;
    while pos <= block.len() - PRESPLIT_CHUNK_SIZE {
        presplit_record_fingerprint(
            &mut new_events,
            &block[pos..pos + PRESPLIT_CHUNK_SIZE],
            sampling_rate,
            hash_log,
        );
        if presplit_fingerprints_differ(&past, &new_events, penalty, hash_log) {
            return pos;
        }
        presplit_merge_events(&mut past, &new_events, hash_log);
        if penalty > 0 {
            penalty -= 1;
        }
        pos += PRESPLIT_CHUNK_SIZE;
    }
    block.len()
}

/// Upstream zstd port of `ZSTD_splitBlock_fromBorders` (`zstd_preSplit.c:198`).
/// Records two 512-byte byte-histograms — one from each end of a 128 KB
/// block — and a third from the middle as a tie-breaker; returns either
/// a quantised split point (32 KB / 64 KB / 96 KB) or the full block
/// size when the two ends look indistinguishable. Cheaper than the
/// chunk-based path because it touches at most 1.5 KB of input
/// regardless of block size.
fn split_block_from_borders(block: &[u8]) -> usize {
    debug_assert_eq!(block.len(), MAX_BLOCK_SIZE as usize);
    let block_size = block.len();
    let mut past = PreSplitFingerprint::default();
    let mut new_fp = PreSplitFingerprint::default();
    presplit_record_byte_histogram(&mut past, &block[..PRESPLIT_BORDERS_SEGMENT]);
    presplit_record_byte_histogram(&mut new_fp, &block[block_size - PRESPLIT_BORDERS_SEGMENT..]);
    // Upstream zstd uses `penalty = 0, hash_log = 8` — i.e. raw byte histogram
    // distance with no threshold padding (`zstd_preSplit.c:214`).
    if !presplit_fingerprints_differ(&past, &new_fp, 0, 8) {
        return block_size;
    }

    let mut middle = PreSplitFingerprint::default();
    let mid_start = block_size / 2 - PRESPLIT_BORDERS_SEGMENT / 2;
    presplit_record_byte_histogram(
        &mut middle,
        &block[mid_start..mid_start + PRESPLIT_BORDERS_SEGMENT],
    );

    let dist_from_begin = presplit_distance(&past, &middle, 8);
    let dist_from_end = presplit_distance(&new_fp, &middle, 8);
    // Upstream zstd `SEGMENT_SIZE * SEGMENT_SIZE / 3` (`zstd_preSplit.c:221`):
    // if the middle is roughly equidistant from both ends, the change
    // sits near the centre — split at the midpoint.
    let min_distance = (PRESPLIT_BORDERS_SEGMENT as u64) * (PRESPLIT_BORDERS_SEGMENT as u64) / 3;
    if dist_from_begin.abs_diff(dist_from_end) < min_distance {
        return 64 * 1024;
    }
    // Larger `dist_from_begin` (i.e. `middle` farther from the head
    // fingerprint, equivalently closer to the tail) means the new
    // statistics already dominate the centre — the transition
    // happened EARLY → emit a small 32 KB head and let the 96 KB
    // tail absorb the rest. Inverse case: `dist_from_end` larger
    // (middle still resembles the head) means the transition is
    // LATE → emit a 96 KB head so the trailing 32 KB carries the
    // new statistics alone.
    if dist_from_begin > dist_from_end {
        32 * 1024
    } else {
        96 * 1024
    }
}

/// XXH64 (low 32 bits, seed 0) over `data`. Shared helper for the
/// per-physical-block checksum sidecar so encoder and decoder hash
/// the exact same byte ranges with the exact same parameters. Gated
/// at `all(lsm, hash)` because the only consumer is the lsm-side
/// `block_checksums` sidecar; non-lsm builds carry no reference to
/// this helper at all.
#[cfg(all(feature = "lsm", feature = "hash"))]
#[inline]
pub(crate) fn xxh64_block_low32(data: &[u8]) -> u32 {
    let mut h = XxHash64::with_seed(0);
    h.write(data);
    h.finish() as u32
}

/// Bench-only entry point for the upstream zstd-parity comparator test in
/// `tests/block_splitter_parity.rs`. Dispatches to the same
/// `_from_borders` (split_level == 0) / `_by_chunks` (split_level ∈
/// 1..=4) ports that `optimal_block_size` itself routes
/// through. Caller is responsible for passing exactly
/// `MAX_BLOCK_SIZE` bytes (per upstream zstd `ZSTD_splitBlock` contract —
/// "@blockSize must be == 128 KB" in `zstd_preSplit.h`).
#[cfg(feature = "bench-internals")]
pub(crate) fn block_splitter_decision_for_bench(block: &[u8], split_level: usize) -> usize {
    assert_eq!(
        block.len(),
        MAX_BLOCK_SIZE as usize,
        "block_splitter_decision_for_bench expects exactly MAX_BLOCK_SIZE bytes"
    );
    assert!(
        split_level <= 4,
        "block_splitter_decision_for_bench: split_level must be in 0..=4, got {split_level}"
    );
    if split_level == 0 {
        split_block_from_borders(block)
    } else {
        split_block_by_chunks(block, split_level)
    }
}

/// Pull a pre-split window into cache with one bandwidth-bound sequential
/// pass before the strided fingerprint histogram + match scan read it.
///
/// The borrowed (no-copy) over-window path matches in place on the caller's
/// input, so the pre-split fingerprint is the FIRST touch of that 128 KiB
/// region — a cache-cold read. `presplit_record_fingerprint` reads it with a
/// `sampling_rate` stride and interleaved random writes into the 1 KiB events
/// table, a latency-bound pattern that pays full DRAM miss latency per line
/// (measured ~3x the cost of an ERMS streaming read of the same bytes). The
/// owned path never hits this because its history-mirror copy already warmed
/// the bytes; this restores that warmth without the copy's write half. One
/// dependent load per 64-byte line (the i9 line size) streams under the
/// hardware prefetcher, so the cold read is paid once at memory bandwidth and
/// every subsequent strided sample lands in L1/L2. `black_box` keeps the loop
/// from being optimized away as a dead read.
#[inline]
fn warm_presplit_window(window: &[u8]) {
    let mut acc = 0u8;
    let mut i = 0usize;
    while i < window.len() {
        acc ^= window[i];
        i += 64;
    }
    core::hint::black_box(acc);
}

/// [`optimal_block_size_with`] at a level's default pre-split tier; the
/// frame loop resolves the tier itself, so only tests and the
/// `bench-internals` block-boundary probe read this form.
#[cfg(any(test, feature = "bench-internals"))]
pub(crate) fn optimal_block_size(
    level: CompressionLevel,
    block: &[u8],
    remaining_src_size: usize,
    block_size_max: usize,
    savings: i64,
) -> usize {
    optimal_block_size_with(
        crate::encoding::levels::config::level_pre_split(level),
        block,
        remaining_src_size,
        block_size_max,
        savings,
    )
}

/// Size of the next block to cut from `block`, given the pre-split level
/// already resolved for the frame (`None` = never split, only full blocks).
///
/// Out of line on purpose: inlined into the per-block frame loop it grew the
/// loop body and shifted the code layout around `run_fast_kernel_block`,
/// costing the Fast levels 17-24 % on 1 MiB+ inputs on x86 (measured on the
/// i9; the kernel's own instructions were byte-identical). One call per block
/// is noise; the compact caller is not.
#[inline(never)]
pub(crate) fn optimal_block_size_with(
    pre_split: Option<usize>,
    block: &[u8],
    remaining_src_size: usize,
    block_size_max: usize,
    savings: i64,
) -> usize {
    let Some(split_level) = pre_split else {
        return remaining_src_size.min(block_size_max);
    };
    if remaining_src_size < MAX_BLOCK_SIZE as usize || block_size_max < MAX_BLOCK_SIZE as usize {
        return remaining_src_size.min(block_size_max);
    }
    if savings < 3 {
        return MAX_BLOCK_SIZE as usize;
    }
    if block.len() < MAX_BLOCK_SIZE as usize {
        return remaining_src_size.min(block_size_max);
    }
    // Upstream zstd `ZSTD_splitBlock` dispatch (`zstd_preSplit.c:234`):
    // `split_level == 0` → cheap borders heuristic;
    // `split_level == 1..=4` → byChunks with internal sampling level
    // `split_level - 1`.
    // NOTE: gating the sampling tier behind the cheap borders tier (run it
    // only when the block's ends already disagree) was measured and REJECTED:
    // it costs 2.8-3.4% of the decode corpus across levels 3-15, because the
    // boundaries the sampling finds are mostly inside blocks whose ends look
    // alike. The cost of the sampling on uniform input has to come off some
    // other way.
    let raw_split = if split_level == 0 {
        split_block_from_borders(&block[..MAX_BLOCK_SIZE as usize])
    } else {
        split_block_by_chunks(&block[..MAX_BLOCK_SIZE as usize], split_level)
    };
    raw_split
        .max(PRESPLIT_BLOCK_MIN)
        .min(MAX_BLOCK_SIZE as usize)
}

/// Load `dict` into the matcher for the frame about to start, `size_hint`
/// bytes long when known, once `state.strategy_tag` holds the strategy the
/// frame runs. Shared by the frame compressor and the streaming context so a
/// reused one of either loads its dictionary the same way.
///
/// A dictionary a reused matcher kept resident across the reset (its bytes
/// and index still in place) only gets its offset history back: priming it
/// again would commit it into the window a second time. Otherwise it is
/// primed, or restored from the matcher's copy-mode snapshot when the source
/// is past the strategy's attach cutoff.
pub(crate) fn load_frame_dictionary<M: Matcher, K: EncodeTier>(
    state: &mut CompressState<M>,
    level: CompressionLevel,
    dict: &EncoderDictionary,
    size_hint: Option<u64>,
) {
    // Upstream zstd `ZSTD_shouldAttachDict` (`zstd_compress.c`): a
    // precomputed-dictionary table is COPIED into the working context
    // only when the source is larger than a per-strategy cutoff; at or
    // below it (and for unknown size) the upstream zstd ATTACHES the dictionary
    // tables by reference (no per-frame table touch at all). We don't
    // have an attach-by-reference path yet, so:
    //   - large source (> cutoff): reuse the captured prime snapshot
    //     (a table copy) instead of re-hashing the dictionary — the
    //     upstream zstd COPY regime, where the copy is cheaper than re-priming;
    //   - small / unknown source: re-prime (the snapshot copy of the
    //     whole table would cost MORE than the sparse re-prime here,
    //     which is exactly why the upstream zstd attaches by reference instead).
    // `attachDictSizeCutoffs` per strategy: fast 8K, dfast 16K,
    // greedy/lazy/btopt 32K, btultra/btultra2 8K. Expressed as the
    // ceil-log bucket (8K = 2^13, 16K = 2^14, 32K = 2^15) so the
    // decision uses the SAME bucketed representation as the driver's
    // attach/copy gate (`reset_size_log`) — comparing
    // `source_size_ceil_log(hint)` on the full u64 avoids the `as usize`
    // truncation that could diverge from the driver on 32-bit targets.
    // For a power-of-two cutoff `2^k`, `ceil_log2(hint) > k` is exactly
    // `hint > 2^k`, so this is identical to the raw `hint > cutoff` on
    // 64-bit.
    let cutoff_log = match state.strategy_tag {
        // Keep the copy-snapshot gate in sync with the matcher's own
        // attach cutoff, so Fast never captures or restores a snapshot
        // for a mode it did not resolve.
        crate::encoding::strategy::StrategyTag::Fast => {
            crate::encoding::levels::config::FAST_ATTACH_DICT_CUTOFF_LOG
        }
        crate::encoding::strategy::StrategyTag::BtUltra
        | crate::encoding::strategy::StrategyTag::BtUltra2 => 13,
        crate::encoding::strategy::StrategyTag::Dfast => 14,
        crate::encoding::strategy::StrategyTag::Greedy
        | crate::encoding::strategy::StrategyTag::Lazy
        | crate::encoding::strategy::StrategyTag::Btlazy2
        | crate::encoding::strategy::StrategyTag::BtOpt => 15,
    };
    if state.matcher.dictionary_is_resident() {
        // Re-borrow fast path: the previous frame's reset kept this
        // dict's bytes + cached index resident, so skip the re-commit /
        // re-index and only reapply the offset history.
        state
            .matcher
            .reapply_resident_dictionary(dict.inner.offset_hist);
        return;
    }
    let prefer_copy_snapshot = crate::encoding::levels::config::restores_primed_snapshot(
        size_hint.map(crate::encoding::levels::config::source_size_ceil_log),
        cutoff_log,
    );
    let restored = prefer_copy_snapshot && state.matcher.restore_primed_dictionary(level);
    if !restored {
        let content = dict.inner.dict_content.as_slice();
        // The built-in matcher indexes the dictionary under the frame's tier.
        match state.matcher.builtin_driver() {
            Some(driver) => driver.prime_with_dictionary_with::<K>(content, dict.inner.offset_hist),
            None => state
                .matcher
                .prime_with_dictionary(content, dict.inner.offset_hist),
        }
        if prefer_copy_snapshot {
            state.matcher.capture_primed_dictionary(level);
        }
    }
}

/// Record in `state` the strategy the matcher runs for the next frame (the
/// honoured public override, else the size- and dictionary-adaptive
/// resolution in `params`) and its pre-split tier; the literal gates and the
/// block splitter read these, and upstream indexes `splitLevels` by the
/// effective strategy too. Shared by the frame compressor and the streaming
/// encoder so both entry points cut and gate blocks identically.
pub(crate) fn sync_effective_strategy<M: Matcher>(
    state: &mut CompressState<M>,
    level: CompressionLevel,
    params: &crate::encoding::levels::config::LevelParams,
    strategy_override: Option<(crate::encoding::strategy::StrategyTag, u8)>,
) {
    match strategy_override {
        Some((tag, lazy_depth)) => {
            state.strategy_tag = tag;
            state.pre_split = Some(crate::encoding::levels::config::pre_split_for(
                tag, lazy_depth,
            ));
        }
        None => {
            state.strategy_tag = params.strategy_tag;
            state.pre_split = if matches!(level, CompressionLevel::Uncompressed) {
                None
            } else {
                params.pre_split()
            };
        }
    }
}

/// Upstream `ZSTD_literalsCompressionIsDisabled`: an explicit
/// [`LiteralCompressionMode`] decides outright; under `Auto`, raw literals iff
/// the EFFECTIVE cParams are the fast strategy with `targetLength > 0`.
/// The effective strategy tag gates this (a strategy override can move a
/// negative level off fast). For the fast strategy the level table sets
/// `targetLength > 0` exactly on the negative (acceleration) rows (a CDict's
/// too), so where [`gate_target_length`] has no value `level < 0` is that
/// test.
pub(crate) fn literal_compression_disabled(
    strategy_tag: crate::encoding::strategy::StrategyTag,
    level: CompressionLevel,
    target_length_override: Option<u32>,
    mode: LiteralCompressionMode,
) -> bool {
    match mode {
        LiteralCompressionMode::Disable => true,
        LiteralCompressionMode::Enable => false,
        LiteralCompressionMode::Auto => {
            strategy_tag == crate::encoding::strategy::StrategyTag::Fast
                && target_length_override.map_or_else(
                    || matches!(level, CompressionLevel::Level(n) if n < 0),
                    |tl| tl > 0,
                )
        }
    }
}

/// The `target_length` the raw-literals gate reads for a frame: the caller's,
/// else, for a dictionary frame a strategy knob moves onto the fast strategy,
/// the CDict row's, which the fast matcher runs as its step (upstream
/// `ZSTD_overrideCParams` replaces only the strategy, and
/// `ZSTD_literalsCompressionIsDisabled` reads the resulting cParams). Every
/// other frame's fast targetLength is what the level says, which
/// [`literal_compression_disabled`] derives itself.
pub(crate) fn gate_target_length(
    level: CompressionLevel,
    tuning: &FrameTuning,
    dictionary: Option<&EncoderDictionary>,
) -> Option<u32> {
    match (tuning.target_length, tuning.strategy, dictionary) {
        (Some(target_length), _, _) => Some(target_length),
        (None, Some((crate::encoding::strategy::StrategyTag::Fast, _)), Some(dict))
            if !dict.inner.dict_content.is_empty() =>
        {
            Some(
                crate::encoding::cparams::get_cdict_cparams(
                    crate::encoding::levels::config::numeric_level(level),
                    dict.sizes().serialized,
                    &crate::encoding::parameters::ParamOverrides::default(),
                )
                .target_length,
            )
        }
        _ => None,
    }
}

/// The level params the matcher's reset resolves for a frame: through the
/// dictionary's CDict tier when a dictionary is in play, else by source size.
/// Resolved without the caller's knobs: the frame gates read only the
/// strategy from these, and a strategy override replaces it
/// ([`sync_effective_strategy`]) on a dictionary frame as on any other.
pub(crate) fn resolve_frame_params(
    level: CompressionLevel,
    hint: Option<u64>,
    dictionary: Option<&EncoderDictionary>,
) -> crate::encoding::levels::config::LevelParams {
    match dictionary {
        Some(dict) if !dict.inner.dict_content.is_empty() => {
            crate::encoding::levels::config::resolve_level_params_with_dict(
                level,
                hint,
                dict.sizes(),
                &crate::encoding::parameters::ParamOverrides::default(),
            )
            .0
        }
        _ => crate::encoding::levels::config::resolve_level_params(level, hint),
    }
}

pub(crate) struct CompressState<M: Matcher> {
    pub(crate) matcher: M,
    /// The parse every block took, recorded while this is `Some`, each block
    /// closed by a delimiter carrying its trailing literals.
    pub(crate) sequence_capture: Option<Vec<crate::encoding::sequences::ExternalSequence>>,
    /// Grid fingerprints of the frame's blocks, which is what tells the
    /// raw-skip that a block duplicating an earlier one is not the noise it
    /// looks like. See [`SeenContentGrid`](crate::encoding::incompressible::SeenContentGrid).
    pub(crate) seen_content: crate::encoding::incompressible::SeenContentGrid,
    /// CPU tier the literal gather runs under, resolved once when the
    /// compressor is built and dispatched on once per block.
    pub(crate) copy_kernel: crate::encoding::fastpath::FastpathKernel,
    pub(crate) last_huff_table: Option<crate::huff0::huff0_encoder::HuffmanTable>,
    /// Recycled `HuffmanTable` buffers: when a block clears or replaces
    /// `last_huff_table`, the old table parks here instead of dropping, so
    /// the next frame's dictionary entropy seed `clone_from`s into existing
    /// allocations. Without this, every dict-seeded frame whose last block
    /// ended raw/RLE paid a fresh two-Vec table clone per frame.
    pub(crate) huff_table_spare: Option<crate::huff0::huff0_encoder::HuffmanTable>,
    /// The Huffman weight builder's three buffers, kept across blocks and
    /// frames. The cheap build path takes a tree and two weight buffers per
    /// call, and it runs once per block plus once per split candidate wherever
    /// the block splitter probes, so taking them fresh each time was the single
    /// largest source of per-frame allocations: 1,880 of a frame's 4,000 at
    /// level 3. Lives here rather than in the per-block scratch, which the
    /// block emitter takes out of this state while a block is in flight.
    pub(crate) huff_weights: crate::huff0::huff0_encoder::WeightScratch,
    pub(crate) fse_tables: FseTables,
    pub(crate) block_scratch: crate::encoding::blocks::CompressedBlockScratch,
    /// The one allocation the match finder's tables and the per-block buffers
    /// are carved from, laid out at each frame start by
    /// [`Self::reset_for_frame`] and [`Self::finish_layout`].
    pub(crate) workspace: crate::encoding::workspace::Workspace,
    /// Offset history for repeat offset encoding: [rep0, rep1, rep2].
    /// Initialized to [1, 4, 8] per RFC 8878 §3.1.2.5.
    pub(crate) offset_hist: [u32; 3],
    /// Strategy tag resolved from the current `CompressionLevel` at every
    /// `matcher.reset()` call. Used by the literal-compression gates
    /// (`min_literals_to_compress`, `min_gain`) in
    /// `encoding::blocks::compressed` to mirror upstream zstd's strategy-aware
    /// thresholds (`zstd_compress_literals.c:114-127, 187-188`).
    ///
    /// **Invariant (required of every construction site):** must be
    /// initialized from the active `CompressionLevel` via
    /// `StrategyTag::for_compression_level`, and re-synced from the
    /// active level alongside every `matcher.reset()` call so the
    /// level-aware gates stay correct after a level change. The two
    /// reset sites that own this sync are `FrameCompressor::compress`
    /// and `StreamingEncoder::ensure_frame_started`. There is no
    /// `Default` impl — production constructors
    /// (`FrameCompressor::new`, `new_with_matcher`, the streaming
    /// encoder constructor) plumb this explicitly. Tests that build
    /// `CompressState` by hand must also supply a value.
    pub(crate) strategy_tag: crate::encoding::strategy::StrategyTag,
    /// Pre-split tier of the effective strategy (upstream `splitLevels`),
    /// synced with `strategy_tag`; `None` never pre-splits (raw frames).
    pub(crate) pre_split: Option<u8>,
    /// Whether the HUF literal table build runs the #167 table-log search
    /// (`true`) or the cheap single-build (`false`). The search is a clean
    /// ratio win over upstream zstd but costs ~1.5 us per literal section —
    /// negligible on large inputs, ~20% on small ones. The Fast and DoubleFast
    /// matchers are byte-faithful to upstream zstd, so the cheap path ties them;
    /// the search is therefore gated ON only for large (> 128 KiB) Fast and
    /// DoubleFast frames. Higher strategies always keep it (their matchers
    /// diverge, making the search load-bearing for ratio). Set per frame
    /// alongside `strategy_tag` via [`huf_search_enabled`].
    pub(crate) huf_optimal_search: bool,
    /// Mirror of upstream zstd's `ZSTD_literalsCompressionIsDisabled`
    /// (zstd_compress_internal.h): in the default (`auto`) literal-compression
    /// mode the literals section is emitted RAW (no Huffman) when
    /// `strategy == ZSTD_fast && targetLength > 0`. For the levels we resolve,
    /// that is exactly the negative levels (Fast strategy with `targetLength =
    /// -level > 0`; L1/L2 are Fast with `targetLength == 0`). C trades the
    /// literal-Huffman pass for speed there, so matching it keeps both the frame
    /// size and the encode cost in parity on the negative band. Set per frame
    /// alongside `strategy_tag`.
    pub(crate) literal_compression_disabled: bool,
    /// Whether a matched block may be cut into several blocks after matching,
    /// where its strategy and window call for it. `true` for every frame but a
    /// dictionary's analysis, which has to see one written block per matched
    /// block, as upstream's single-block analysis does (`ZDICT_countEStats`
    /// compresses with `ZSTD_compressBlock`, which never splits).
    pub(crate) post_split_allowed: bool,
}

/// Whether the HUF literal build should run the #167 table-log search for a
/// frame of `source_size` bytes (see [`CompressState::huf_optimal_search`]).
/// Upstream gates the optimal-depth tableLog probe to
/// `HUF_OPTIMAL_DEPTH_THRESHOLD = ZSTD_btultra` (huf.h:117): only btultra /
/// btultra2 search the tableLog, every lower strategy (fast .. btopt) takes the
/// single-shot fast path (`HUF_optimalTableLog`, huf_compress.c:1284-1287).
/// Mirror that so our literal tableLog choice tracks upstream's instead of
/// spending the search to beat it on ratio at a speed cost.
pub(crate) fn huf_search_enabled(
    strategy: crate::encoding::strategy::StrategyTag,
    _source_size: Option<u64>,
) -> bool {
    use crate::encoding::strategy::StrategyTag;
    matches!(strategy, StrategyTag::BtUltra | StrategyTag::BtUltra2)
}

impl<M: Matcher> CompressState<M> {
    /// Resets the matcher for the next frame at `level` and lets it lay its
    /// tables and history out in the workspace, reserving room behind them for
    /// the block buffers of blocks up to `block_target` bytes (capped by the
    /// frame's window). `ingest` is how the frame's input reaches the matcher,
    /// which sizes its history. [`Self::finish_layout`] completes the layout.
    pub(crate) fn reset_for_frame(
        &mut self,
        level: CompressionLevel,
        block_target: usize,
        ingest: crate::encoding::workspace::IngestPlan,
    ) {
        // A raw frame builds no compressed block, so it reserves no buffers
        // for one.
        let trailing_for = if ingest == crate::encoding::workspace::IngestPlan::Raw {
            crate::encoding::workspace::no_trailing
        } else {
            crate::encoding::blocks::CompressedBlockScratch::workspace_bytes
        };
        self.workspace
            .begin_layout(block_target, trailing_for, ingest);
        self.matcher.reset_in_workspace(level, &mut self.workspace);
    }

    /// Carves the block buffers for blocks of up to `block_capacity` bytes, or
    /// of the frame's length when that is exact and shorter, opening the layout
    /// first when the matcher took no tables from it.
    ///
    /// # Panics
    ///
    /// Panics when `block_capacity` exceeds what the layout reserved for, which
    /// would mean the frame and the matcher disagree on the window.
    pub(crate) fn finish_layout(&mut self, block_capacity: usize) {
        if !self.workspace.is_open() {
            self.workspace.open(0, block_capacity);
        }
        let reserved = self.workspace.block_capacity();
        // A frame of exact length never carries a block longer than itself, and
        // the layout reserved for no more.
        let largest = block_capacity.min(self.workspace.ingest().exact_len().unwrap_or(usize::MAX));
        assert!(
            largest <= reserved,
            "a {largest}-byte block exceeds the {reserved} bytes the workspace reserved"
        );
        if self.workspace.ingest() == crate::encoding::workspace::IngestPlan::Raw {
            self.block_scratch.unbind();
        } else {
            self.block_scratch.bind(&mut self.workspace, reserved);
        }
        // Everything is laid out in the current allocation by now.
        self.workspace.release_retired();
    }

    /// Heap bytes the compressor keeps between blocks and frames beyond the
    /// match finder: the FSE tables both slots of each axis hold, the block
    /// scratch with everything it holds (the splitter's workspace and the
    /// nested estimator scratch), and the context workspace the literal and
    /// sequence buffers are carved from.
    ///
    /// All of it survives a frame, so a caller sizing a context has to see it.
    pub(crate) fn retained_scratch_heap_size(&self) -> usize {
        self.fse_tables.heap_size()
            + self.block_scratch.retained_heap_size()
            + self.workspace.heap_bytes()
    }

    /// Heap bytes of the parse recorded while capture is on, held until taken.
    pub(crate) fn captured_parse_heap_size(&self) -> usize {
        self.sequence_capture.as_ref().map_or(0, |capture| {
            capture.capacity()
                * core::mem::size_of::<crate::encoding::sequences::ExternalSequence>()
        })
    }

    /// Clears `last_huff_table`, parking the table's buffers in
    /// `huff_table_spare` for reuse instead of dropping them.
    #[inline]
    pub(crate) fn clear_huff_table(&mut self) {
        if let Some(table) = self.last_huff_table.take() {
            self.park_huff_table(table);
        }
    }

    /// Replaces `last_huff_table` with `table`, parking any displaced table
    /// in `huff_table_spare` for reuse.
    #[inline]
    pub(crate) fn replace_huff_table(&mut self, table: crate::huff0::huff0_encoder::HuffmanTable) {
        if let Some(old) = self.last_huff_table.replace(table) {
            self.park_huff_table(old);
        }
    }

    /// Keeps a table's buffers rather than dropping them. The dictionary seed
    /// wants one spare to `clone_from` into, once per frame; every further
    /// table a block displaces goes to the weight builder instead, which takes
    /// one per block and per split candidate. Overwriting the single spare
    /// dropped the previous table on every block, so the builds that followed
    /// allocated their buffers again.
    #[inline]
    pub(crate) fn park_huff_table(&mut self, table: crate::huff0::huff0_encoder::HuffmanTable) {
        if self.huff_table_spare.is_none() {
            self.huff_table_spare = Some(table);
        } else {
            self.huff_weights.recycle(table);
        }
    }
}

/// Per-frame setup resolved once by [`FrameCompressor::prepare_frame`] and
/// consumed by the block loop + [`FrameCompressor::finish_frame`]. Lets the
/// owned `compress()` and the borrowed one-shot path share identical
/// reset / dict-prime / entropy-seed setup and frame-tail emission.
struct FramePrep {
    window_size: u64,
    use_dictionary_state: bool,
    initial_size_hint: Option<u64>,
}

/// Initial capacity for the `all_blocks` accumulator, by source-size hint.
/// The frame header is written only after all input is read (so
/// Frame_Content_Size is known), so compressed blocks accumulate in memory
/// first. Seed-size tiers (mirrors upstream zstd `ZSTD_CStreamOutSize` naming):
/// - tiny (`<= 4 KiB` hint): payload-bound seed, `>=` anything a tiny input's
///   compressed output could need.
/// - small (`<= 64 KiB` hint): absorbs one or two `Vec::extend` doublings
///   without over-allocating.
/// - default (one upstream zstd block, `130 KiB`): the value the rest of the encoder
///   is sized around; larger inputs amortise the first doublings cheaply and
///   the residue is dominated by internal `compress_block_encoded` buffers.
///
/// Shared by the owned (`run_owned_block_loop`) and borrowed
/// (`run_borrowed_block_loop`) paths so the tier table can't drift between them.
///
/// `block_capacity` (the active `targetCBlockSize` cap, or the 128 KiB
/// format ceiling) bounds every tier: with a small target the first
/// allocation tracks one capped block + header/checksum slack instead of
/// keeping the upstream zstd-sized floor that only later growth respects.
fn initial_all_blocks_cap(initial_size_hint: Option<u64>, block_capacity: usize) -> usize {
    const TINY_THRESHOLD: u64 = 4 * 1024;
    const SMALL_THRESHOLD: u64 = 64 * 1024;
    const TINY_CAP: usize = 4 * 1024;
    const SMALL_CAP: usize = 16 * 1024;
    const DEFAULT_CAP: usize = 130 * 1024;
    let first_block_cap = block_capacity + 3 + 16;
    match initial_size_hint {
        Some(h) if h <= TINY_THRESHOLD => TINY_CAP.min(first_block_cap),
        Some(h) if h <= SMALL_THRESHOLD => SMALL_CAP.min(first_block_cap),
        _ => DEFAULT_CAP.min(first_block_cap),
    }
}

/// The block ceiling a target block size and a maximum block size leave
/// together: the smaller one that is set, or the format's 128 KiB.
pub(crate) fn block_ceiling(target: Option<u32>, max: Option<u32>) -> usize {
    let ceiling = match (target, max) {
        (Some(target), Some(max)) => target.min(max),
        (Some(size), None) | (None, Some(size)) => size,
        (None, None) => crate::common::MAX_BLOCK_SIZE,
    };
    ceiling as usize
}

/// Per-block feeder for `run_owned_block_loop`.
///
/// `fill_block` appends source bytes to `buf` (which already holds any
/// carried pre-split suffix) until `buf.len() == block_capacity` or the
/// source is exhausted, returning `(bytes_appended, reached_eof)`.
/// `reached_eof` is true when no more input follows this block: either the
/// block could not be filled to `block_capacity`, or it filled exactly and the
/// source is confirmed exhausted (the slice knows its length; the reader probes
/// one byte ahead). An input that is an exact multiple of the block size
/// therefore marks its final full block `last_block` rather than emitting a
/// spurious trailing empty block.
///
/// The slice impl exists so the slice entry points
/// (`compress_independent_frame_into`, `compress_oneshot_*` fallbacks)
/// append with one `extend_from_slice` — the generic reader impl must
/// `resize` an initialized target region before `Read::read` can fill it,
/// which costs a zero-fill memset of the whole block on every frame.
///
/// `buf` is the matcher's history on the in-place path and a staging `Vec`
/// otherwise.
pub(crate) trait OwnedBlockSource {
    fn fill_block<B: IngestBuffer>(
        &mut self,
        buf: &mut B,
        block_capacity: usize,
        size_hint_remaining: Option<u64>,
    ) -> (usize, bool);

    /// Bytes left, when the source knows them exactly. The matcher reserves
    /// room for a read before it is made, so a read held to this asks for no
    /// room past the end of the input.
    fn exact_remaining(&self) -> Option<usize> {
        None
    }
}

impl OwnedBlockSource for &[u8] {
    fn exact_remaining(&self) -> Option<usize> {
        Some(self.len())
    }

    fn fill_block<B: IngestBuffer>(
        &mut self,
        buf: &mut B,
        block_capacity: usize,
        _size_hint_remaining: Option<u64>,
    ) -> (usize, bool) {
        let want = block_capacity - buf.len();
        let take = want.min(self.len());
        buf.extend_from_slice(&self[..take]);
        *self = &self[take..];
        // EOF when this fill could not top the block to `block_capacity`
        // (`take < want`) OR it exactly consumed the last input bytes
        // (`self` now empty). The slice knows its own length, so a block that
        // exactly fills capacity at end-of-input is reported as the final
        // block here — the loop marks it `last_block` instead of emitting a
        // spurious trailing empty block on the next iteration. Mirrors the C
        // encoder, which marks the last real block last on `ZSTD_e_end`.
        (take, take < want || self.is_empty())
    }
}

/// Adapter routing a generic [`Read`] source through [`OwnedBlockSource`]:
/// preserves the historical sizing behaviour — an initialized target region
/// bounded by the source-size hint, grown (doubling, capped) only when the
/// hint under-counted.
/// `peeked` holds a single look-ahead byte: when a block fills exactly to
/// `block_capacity`, `fill_block` reads one more byte to learn whether the
/// stream ended on that boundary. A `None` from that probe sets EOF (so the
/// just-filled block is marked last, mirroring the C encoder on `ZSTD_e_end`);
/// a byte is stashed here and prepended to the next block instead of leaking a
/// spurious trailing empty block when the input is an exact multiple of the
/// block size.
pub(crate) struct ReaderBlockSource<Rd> {
    pub(crate) reader: Rd,
    peeked: Option<u8>,
}

impl<Rd> ReaderBlockSource<Rd> {
    pub(crate) fn new(reader: Rd) -> Self {
        Self {
            reader,
            peeked: None,
        }
    }
}

impl<Rd: Read> OwnedBlockSource for ReaderBlockSource<Rd> {
    fn fill_block<B: IngestBuffer>(
        &mut self,
        buf: &mut B,
        block_capacity: usize,
        size_hint_remaining: Option<u64>,
    ) -> (usize, bool) {
        let start = buf.len();
        let mut filled = start;
        let mut reached_eof = false;
        // Prepend the look-ahead byte read past the previous full block. In
        // stream order it follows any carried pre-split suffix already in
        // `buf`, so it is appended after that suffix and counted as part of
        // this block's appended bytes.
        if let Some(b) = self.peeked.take() {
            buf.push(b);
            filled += 1;
        }
        // Size the read buffer to the bytes this block actually expects
        // rather than always zero-filling a full MAX_BLOCK_SIZE: a small
        // frame otherwise pays a 128 KiB `resize(_, 0)` memset per block
        // just to read a few KiB (the zero-fill past `filled` is then
        // truncated away).
        //
        // Overflow-free by construction (no `saturating_*` masking):
        // `filled <= block_capacity` always (the read only ever targets
        // `[filled..len]` with `len <= block_capacity`, and a carried-over
        // pre-split suffix is a `split_off` below `block_capacity`), so
        // `block_capacity - filled` never underflows; pinning `remaining`
        // to `block_capacity` before the `usize` cast keeps the cast and
        // the final add within `usize` on every target.
        let initial_target = match size_hint_remaining {
            Some(remaining) => {
                let remaining = remaining.min(block_capacity as u64) as usize;
                filled + remaining.min(block_capacity - filled)
            }
            // Unknown hint, or an inexact hint already met by prior blocks:
            // read against the full block window.
            None => block_capacity,
        };
        if buf.len() < initial_target {
            buf.resize(initial_target, 0);
        }
        loop {
            if reached_eof || filled == block_capacity {
                break;
            }
            if filled == buf.len() {
                // Hint under-counted the block; grow toward block_capacity
                // (doubling, capped) so reading continues without paying a
                // full-buffer zero up front. `len <= block_capacity` so the
                // double stays well within `usize`; `filled < block_capacity`
                // here (the `== block_capacity` break fired otherwise), so
                // `filled + 1 <= block_capacity`.
                let grow_to = (buf.len() * 2).clamp(filled + 1, block_capacity);
                buf.resize(grow_to, 0);
            }
            let read_end = buf.len();
            let new_bytes = self.reader.read(&mut buf[filled..read_end]).unwrap();
            if new_bytes == 0 {
                reached_eof = true;
                break;
            }
            filled += new_bytes;
        }
        // Look ahead one byte when the block filled exactly to capacity: a
        // 0-byte read means the stream ended on the block boundary, so this
        // block is the last one (the loop marks it `last_block`); otherwise
        // stash the byte for the next block. Without this, an input that is an
        // exact multiple of the block size would emit a spurious trailing
        // empty block (the next iteration reads 0 and serializes an empty
        // last Raw block). A blocking reader's probe read is consistent with
        // the existing pull model — the next `fill_block` would block on the
        // same byte anyway.
        if !reached_eof && filled == block_capacity {
            let mut probe = [0u8; 1];
            if self.reader.read(&mut probe).unwrap() == 0 {
                reached_eof = true;
            } else {
                self.peeked = Some(probe[0]);
            }
        }
        buf.truncate(filled);
        (filled - start, reached_eof)
    }
}

impl<R: Read, W: Write> FrameCompressor<R, W, MatchGeneratorDriver> {
    /// Create a new `FrameCompressor`
    pub fn new(compression_level: CompressionLevel) -> Self {
        Self {
            uncompressed_data: None,
            compressed_data: None,
            compression_level,
            dictionary: None,
            source_size_hint: None,
            state: CompressState {
                matcher: MatchGeneratorDriver::new(1024 * 128, 1),
                sequence_capture: None,
                copy_kernel: crate::encoding::fastpath::select_kernel(),
                last_huff_table: None,
                huff_table_spare: None,
                huff_weights: Default::default(),
                seen_content: Default::default(),
                fse_tables: FseTables::new(),
                block_scratch: crate::encoding::blocks::CompressedBlockScratch::new(),
                workspace: crate::encoding::workspace::Workspace::new(),
                offset_hist: [1, 4, 8],
                strategy_tag: crate::encoding::strategy::StrategyTag::for_compression_level(
                    compression_level,
                ),
                pre_split: crate::encoding::levels::config::level_pre_split(compression_level)
                    .map(|tier| tier as u8),
                huf_optimal_search: true,
                literal_compression_disabled: matches!(
                    compression_level,
                    crate::encoding::CompressionLevel::Level(n) if n < 0
                ),
                post_split_allowed: true,
            },
            magicless: false,
            content_checksum: false,
            pre_split_disabled: false,
            content_size_flag: true,
            dict_id_flag: true,
            target_block_size: None,
            max_block_size: None,
            #[cfg(feature = "hash")]
            hasher: XxHash64::with_seed(0),
            #[cfg(feature = "lsm")]
            frame_emit_info: None,
            #[cfg(all(feature = "lsm", feature = "hash"))]
            per_block_checksums_enabled: false,
            #[cfg(all(feature = "lsm", feature = "hash"))]
            block_checksums: None,
            #[cfg(feature = "lsm")]
            block_decompressed_sizes: alloc::vec::Vec::new(),
            tuning: FrameTuning::default(),
        }
    }

    /// Ask `producer` for each block's parse in place of the built-in
    /// matchers; `None` removes it. With `fallback` a block the producer fails
    /// on is matched as usual; without, it goes out uncompressed and
    /// [`Self::take_sequence_producer_failure`] reports it. With long-distance
    /// matching on, the long matches are taken outright and the producer
    /// parses the literals between them.
    pub fn set_sequence_producer(
        &mut self,
        producer: Option<alloc::boxed::Box<dyn crate::encoding::sequences::SequenceProducer>>,
        fallback: bool,
    ) {
        self.state.matcher.set_sequence_producer(producer, fallback);
    }

    /// Whether a block went out uncompressed because the producer failed on it
    /// with no fallback, since this was last asked.
    pub fn take_sequence_producer_failure(&mut self) -> bool {
        self.state.matcher.take_producer_failure()
    }

    /// Configure fine-grained compression parameters.
    ///
    /// Resets the base [`CompressionLevel`] to the parameters' level and installs the per-knob overrides
    /// (window/hash/chain/search logs, strategy, LDM) applied at the next
    /// frame. Pass `None`-equivalent (a builder that overrides nothing)
    /// to fall back to plain level-based compression.
    ///
    /// ```rust
    /// use structured_zstd::encoding::{
    ///     CompressionLevel, CompressionParameters, FrameCompressor, Strategy,
    /// };
    /// let params = CompressionParameters::builder(CompressionLevel::Level(19))
    ///     .strategy(Strategy::Btultra2)
    ///     .enable_long_distance_matching(true)
    ///     .build()
    ///     .unwrap();
    /// let mut compressor: FrameCompressor = FrameCompressor::new(CompressionLevel::Default);
    /// compressor.set_parameters(&params);
    /// let compressed = compressor.compress_independent_frame(b"some data to compress");
    /// assert!(!compressed.is_empty());
    /// ```
    pub fn set_parameters(&mut self, params: &crate::encoding::CompressionParameters) {
        self.compression_level = params.level();
        let overrides = params.overrides();
        self.tuning = FrameTuning::from_overrides(&overrides);
        // Keep `state.strategy_tag` consistent immediately so the borrowed
        // one-shot eligibility gate (`borrowed_eligible`) and literal gates
        // are correct even before the next `compress()` re-sync. Resolve it
        // size-adaptively (same `resolve_level_params` path `prepare_frame`
        // uses) so a hint already set here yields the same strategy the matcher
        // will run, not the bare level-only mapping.
        // The dictionary counts only when the frame will prime it (same gate
        // as `prepare_frame`'s `use_dictionary_state`): uncompressed mode
        // ignores an attached dictionary and has no CDict tier to resolve.
        let with_dictionary = !matches!(self.compression_level, CompressionLevel::Uncompressed)
            && self.state.matcher.supports_dictionary_priming();
        let params = self.resolve_frame_params(self.source_size_hint, with_dictionary);
        self.sync_effective_strategy(&params);
        self.state.huf_optimal_search =
            huf_search_enabled(self.state.strategy_tag, self.source_size_hint);
        self.state.literal_compression_disabled = literal_compression_disabled(
            self.state.strategy_tag,
            self.compression_level,
            gate_target_length(
                self.compression_level,
                &self.tuning,
                self.dictionary.as_ref().filter(|_| with_dictionary),
            ),
            self.tuning.literal_compression,
        );
        self.state.matcher.set_param_overrides(Some(overrides));
    }

    /// Whether the borrowed (no per-block history copy) one-shot loop is
    /// valid for an `input_len`-byte slice under the resolved `prep`.
    ///
    /// `Uncompressed` resolves to `StrategyTag::Fast` but must emit stored
    /// Raw blocks, which the borrowed loop's
    /// `compress_block_encoded_borrowed` (RLE/raw-fast/compressed) does NOT
    /// do, so exclude it; it then takes the owned path's dedicated
    /// Uncompressed arm.
    ///
    /// No window-size gate: over-window inputs are handled too. The owned
    /// path bounds matches to the last `advertised_window` bytes via
    /// `window_low` and evicts its history, sliding the table's stored
    /// positions down with it; the borrowed path
    /// computes the identical `window_low = block_end - advertised_window`
    /// and the kernel rejects any hash candidate below it, while the
    /// per-position `put` during the scan keeps in-window slots current,
    /// so it produces byte-identical output to the owned (evicting) path
    /// without ever copying the input into `history`, even when the input
    /// far exceeds the window.
    ///
    /// BUT gate on `input_len <= u32::MAX`: the Fast kernel stores ABSOLUTE
    /// positions in a `u32` hash table, and the borrowed scan walks
    /// absolute input offsets up to `block_end == input.len()`. Past 4 GiB
    /// those offsets truncate / overflow the `u32` position math
    /// (`base_off + ip0 as u32`, `window_low`), panicking or corrupting.
    /// The owned/evicting path keeps the scanned window bounded (positions
    /// stay small), so >4 GiB inputs fall back to it.
    fn borrowed_eligible(&self, input_len: usize, prep: &FramePrep) -> bool {
        if matches!(self.compression_level, CompressionLevel::Uncompressed)
            || input_len > u32::MAX as usize
        {
            return false;
        }
        if prep.use_dictionary_state {
            // The borrowed dict scan runs in VIRTUAL `[dict][input]` coordinates,
            // so the position space is `dict_content.len() + input_len`, not just
            // `input_len`. A large attached dictionary plus an otherwise-allowed
            // input can exceed the `u32` floor the kernel asserts — fall back to
            // the owned (copy) path in that case.
            let fits_u32 = self
                .dictionary
                .as_ref()
                .and_then(|dict| dict.inner.dict_content.len().checked_add(input_len))
                .is_some_and(|virtual_len| virtual_len <= u32::MAX as usize);
            if !fits_u32 {
                return false;
            }
            // Dictionary frames: only the Simple (Fast) backend in attach mode
            // has a borrowed (no input copy) dict scan. Copy-mode dict frames
            // and the other backends still take the owned path. The reset laid
            // the history out on a prediction of this, which must hold, or the
            // history is either unused or outgrown.
            let attached = self.state.matcher.borrowed_dict_supported();
            debug_assert_eq!(
                attached,
                self.state.matcher.frame_scans_in_place(),
                "the reset's in-place prediction disagrees with the primed dictionary mode",
            );
            return attached;
        }
        // The borrowed (no-copy, in-place over-window) scan exists for the
        // Simple (Fast), Dfast, and Row backends; the optimal parsers stay on
        // the owned path. Every borrowed scan applies the per-position
        // `window_low = abs_ip - advertised_window` offset cap so over-window
        // inputs are matched in place (no input->history copy), matching C's
        // continuous-index + windowLow one-shot behaviour. The reset decided
        // it, and laid the history out to match.
        self.state.matcher.frame_scans_in_place()
    }

    /// Compress `input` as one frame's worth of blocks into `out` (appended
    /// from its current end): the borrowed in-place loop when
    /// [`Self::borrowed_eligible`], else the owned (history-copying) loop fed
    /// an in-place `&[u8]` cursor. Returns `total_uncompressed`; the caller
    /// emits the frame header (before this call, when the content size is
    /// known) or the drain tail.
    fn run_one_frame<K: EncodeTier>(
        &mut self,
        input: &[u8],
        prep: &FramePrep,
        out: &mut Vec<u8>,
    ) -> u64 {
        if self.borrowed_eligible(input.len(), prep) {
            // A reused Dfast matcher's tables hold earlier frames, which its
            // borrowed kernel (numbering this input from zero) would read as
            // its own: emptied, the frame is the one a fresh matcher writes
            // (upstream: first and later uses of a context compress the
            // same). Measured cheaper than the owned path, which retires them
            // by moving the floor but copies the input and runs slower: level 3
            // on a reused ZSTD_compress2 context, the owned path costs 3.5% at
            // 1 KiB frames, 4.1% at 4 KiB and 10.8% at 1 MiB, and gains at most
            // 1% below 256 bytes, within noise. The emptied tables are sized
            // from the frame's own source, so a tiny frame clears a tiny table.
            if !self.state.matcher.borrowed_frame_is_independent() {
                self.state.matcher.forget_earlier_frames();
            }
            self.run_borrowed_block_loop::<K>(input, out)
        } else {
            let mut cursor: &[u8] = input;
            self.run_owned_block_loop::<_, K>(&mut cursor, prep.initial_size_hint, true, out)
        }
    }

    /// Compress one contiguous `&[u8]` as a single independent Zstd frame,
    /// writing the frame bytes into `out` (its previous contents are
    /// replaced and its allocation reused), reusing this compressor's heavy
    /// state across calls.
    ///
    /// This is the reusable-compression-context (CCtx-equivalent) entry
    /// point, mirroring C `ZSTD_compress2` over a reused `ZSTD_CCtx`:
    /// construct ONE `FrameCompressor` and call this in a loop to emit N
    /// independent, self-describing frames (each carrying its own header,
    /// blocks, and checksum, decodable in isolation, with no cross-frame
    /// match history). Every call resets the per-frame state: only the
    /// allocations are kept, so the
    /// dominant per-frame setup cost (table allocation + dictionary prime)
    /// is paid once instead of N times. Passing the same `out` buffer each
    /// call additionally reuses the output allocation, matching C's
    /// caller-owned `dst` buffer (no per-frame output allocation).
    ///
    /// Reusing the context + `out` across many small frames (the typical
    /// per-block-frame workload) is far cheaper than a fresh
    /// [`compress_slice_to_vec`](crate::encoding::compress_slice_to_vec)
    /// per block, which allocates and primes from scratch each time.
    ///
    /// The input is read in place: no [`Self::set_source`] /
    /// [`Self::set_drain`] setup is required, and the input lifetime is not
    /// baked into the compressor type, so successive calls may pass slices
    /// with unrelated lifetimes. When the Fast (Simple) backend is active
    /// and no dictionary is set, the matcher references the input directly
    /// (no per-block history copy); other backends / dictionary use copy
    /// each block into history exactly as the streaming
    /// [`compress`](Self::compress) path does. The source-size hint is
    /// derived from the input length on every call, so per-frame table
    /// sizing tracks each frame's actual size regardless of any earlier
    /// hint.
    ///
    /// A sticky dictionary set via
    /// [`set_dictionary`](Self::set_dictionary) (or its variants) is primed
    /// into every frame, mirroring `ZSTD_CCtx_loadDictionary` /
    /// `ZSTD_CCtx_refCDict`.
    ///
    /// # Panics
    ///
    /// Panics on encoder error, matching [`Self::compress`] and
    /// [`compress_slice_to_vec`](crate::encoding::compress_slice_to_vec).
    pub fn compress_independent_frame_into(&mut self, input: &[u8], out: &mut Vec<u8>) {
        // The frame's one choice of kernel tier: its setup, every block's cut,
        // search and entropy stage run compiled for it.
        crate::encoding::fastpath::with_encode_tier!(
            self.state.copy_kernel,
            K => self.compress_independent_frame_with::<K>(input, out)
        )
    }

    /// [`Self::compress_independent_frame_into`] compiled for the kernel tier
    /// `K`.
    fn compress_independent_frame_with<K: EncodeTier>(&mut self, input: &[u8], out: &mut Vec<u8>) {
        // Size the next frame from the actual payload, not a stale hint a
        // previous call may have left behind (a wrong hint would change the
        // resolved window/header and could flip borrowed eligibility).
        self.source_size_hint = Some(input.len() as u64);
        let prep =
            self.prepare_frame::<K>(crate::encoding::workspace::IngestPlan::Slice(input.len()));
        // Content size is known up front (one-shot), so write the frame
        // header FIRST and emit blocks STRAIGHT into `out` — no separate
        // `all_blocks` accumulator and no header+blocks copy (which was the
        // dominant per-frame memmove + the only un-amortized per-frame alloc
        // even when the compressor is reused).
        let total_uncompressed = input.len() as u64;
        let emit_checksum = cfg!(feature = "hash") && self.content_checksum;
        let checksum_len = if emit_checksum { 4 } else { 0 };
        out.clear();
        // Reserve the header plus ONE block's worst case up front; the block
        // loops then grow `out` from the compression ratio observed so far
        // (`reserve_for_next_block`). Reserving `compress_bound(input_len)`
        // here held a whole-input-sized allocation for the entire frame —
        // ~100 MiB peak on a 100 MiB stream whose compressed output is a few
        // MiB, where the reference implementation's context peaks at
        // window-sized state. Small frames (<= one block) still get their
        // full bound in one shot, so the reused-`out` steady state is
        // unchanged. 18 = max frame header (magic 4 + descriptor 1 + window
        // 1 + dict id 4 + FCS 8).
        let first_block_bound = input.len().min(self.block_capacity()) + 3;
        out.reserve(18 + first_block_bound + checksum_len);
        self.append_frame_header(total_uncompressed, &prep, out);
        let header_len = out.len();
        let _ = self.run_one_frame::<K>(input, &prep, out);
        #[cfg(feature = "hash")]
        if self.content_checksum {
            out.extend_from_slice(&(self.hasher.finish() as u32).to_le_bytes());
        }
        #[cfg(feature = "lsm")]
        {
            let blocks_end = out.len() - checksum_len;
            self.populate_frame_emit_info(header_len, &out[header_len..blocks_end], emit_checksum);
        }
        #[cfg(not(feature = "lsm"))]
        let _ = header_len;
    }

    /// Convenience wrapper over [`Self::compress_independent_frame_into`]
    /// that allocates and returns a fresh `Vec` per call. Prefer the
    /// `_into` form in tight per-block-frame loops to reuse one output
    /// buffer across frames (the CCtx-equivalent zero-per-call-alloc
    /// output, matching C's caller-owned `dst`).
    ///
    /// ```rust
    /// use structured_zstd::encoding::{FrameCompressor, CompressionLevel};
    /// let mut cctx: FrameCompressor = FrameCompressor::new(CompressionLevel::Default);
    /// let frame_a = cctx.compress_independent_frame(b"first block payload");
    /// let frame_b = cctx.compress_independent_frame(b"second block payload");
    /// assert!(!frame_a.is_empty() && !frame_b.is_empty());
    /// ```
    pub fn compress_independent_frame(&mut self, input: &[u8]) -> Vec<u8> {
        let mut out = Vec::new();
        self.compress_independent_frame_into(input, &mut out);
        out
    }

    /// Borrowed one-shot block loop: walks `input` in `MAX_BLOCK_SIZE`
    /// strides (the Fast backend never pre-splits, so boundaries match the
    /// owned loop), scanning each block range in place against the
    /// borrowed window via `compress_block_encoded_borrowed`, with no
    /// per-block copy into the history. Returns `(all_blocks,
    /// total_uncompressed)`. Caller guarantees Fast backend + no
    /// dictionary; over-window inputs are fine (matches are bounded by
    /// `window_low` exactly as the owned evicting path).
    fn run_borrowed_block_loop<K: EncodeTier>(&mut self, input: &[u8], out: &mut Vec<u8>) -> u64 {
        // Blocks are appended to `out` starting here. `out` may already hold
        // the frame header (the one-shot compress-into-Vec path writes it
        // first, since the content size is known up front, and the loop
        // emits blocks straight after it — no separate `all_blocks` Vec and
        // no header+blocks copy). Output-size reads below are taken RELATIVE
        // to `blocks_start` so a header prefix never skews the upstream zstd split
        // `savings` gate (which would change block boundaries / wire output).
        let blocks_start = out.len();
        let total_uncompressed = input.len() as u64;
        // Empty input: emit a single empty last Raw block (mirrors the
        // owned loop's empty-file special case).
        if input.is_empty() {
            let header = BlockHeader {
                last_block: true,
                block_type: crate::blocks::block::BlockType::Raw,
                block_size: 0,
            };
            header.serialize(out);
            #[cfg(feature = "lsm")]
            self.block_decompressed_sizes.push(0);
            #[cfg(all(feature = "lsm", feature = "hash"))]
            if let Some(checksums) = self.block_checksums.as_mut() {
                checksums.push(xxh64_block_low32(&[]));
            }
            return total_uncompressed;
        }
        // SAFETY: `input` outlives this call (held by the caller across
        // the call) and is not mutated. Only the Simple backend is active
        // (gated by `compress_oneshot_borrowed`).
        unsafe {
            self.state.matcher.set_borrowed_window(input);
        }
        // Fixed for the whole frame, so resolved before the loop takes the
        // state.
        let block_capacity = self.block_capacity();
        let pre_split = self.pre_split_level();
        let dict_active =
            self.dictionary.is_some() && self.state.matcher.supports_dictionary_priming();
        // Panic-safety: clear the borrowed `(ptr, len)` on EVERY exit,
        // including an unwind from an `assert!` inside the block loop, so
        // a caught-and-reused compressor never retains a dangling window.
        // (The next frame's `reset()` also clears it before any read, but
        // this guard makes the invariant local and unwind-proof.) The guard
        // holds the state's `&mut` and the loop reaches the state through it:
        // a raw pointer kept beside the loop's own `&mut self.state` would be
        // invalidated by it (Stacked Borrows), making the drop's access
        // undefined.
        struct ClearBorrowedOnDrop<'a>(&'a mut CompressState<MatchGeneratorDriver>);
        impl Drop for ClearBorrowedOnDrop<'_> {
            fn drop(&mut self) {
                self.0.matcher.clear_borrowed_window();
            }
        }
        let guard = ClearBorrowedOnDrop(&mut self.state);
        let mut start = 0usize;
        while start < input.len() {
            reserve_for_next_block(
                out,
                blocks_start,
                start as u64,
                input.len() - start,
                block_capacity,
            );
            // Upstream zstd `ZSTD_compress_frameChunk`: size each block via the cheap
            // fingerprint pre-splitter so a full 128 KiB block is cut at a
            // statistical boundary when it pays. `savings = consumed -
            // produced` mirrors the upstream zstd gate (the first block and
            // incompressible input keep the full 128 KiB). The borrowed window
            // already spans the whole input, so a smaller block is just a
            // narrower `(block_start, block_end)` range into it.
            let savings = start as i64 - (out.len() - blocks_start) as i64;
            // Borrowed path only: warm the pre-split window before the
            // cache-cold strided fingerprint read. Gated to exactly the
            // conditions under which `optimal_block_size` reads the whole
            // window (a sampling pre-split level, a full 128 KiB block
            // remaining, the block-size cap admits a full block, and
            // `savings >= 3` so the splitter actually runs) — so other
            // levels, the first block, and the trailing partial block pay
            // nothing. Level 0 reads three 512-byte segments
            // (`ZSTD_splitBlock_fromBorders`, `zstd_preSplit.c:198-226`), not
            // the window, so warming 128 KiB for it was 2048 loads a block
            // for 1.5 KiB read. See `warm_presplit_window`.
            if savings >= 3
                && input.len() - start >= MAX_BLOCK_SIZE as usize
                && block_capacity >= MAX_BLOCK_SIZE as usize
                && pre_split.is_some_and(|split_level| split_level != 0)
            {
                warm_presplit_window(&input[start..start + MAX_BLOCK_SIZE as usize]);
            }
            let block_len = optimal_block_size_with(
                pre_split,
                &input[start..],
                input.len() - start,
                block_capacity,
                savings,
            );
            // A cut the splitter proposes inside a long match is moved out of
            // it, as the owned loop does for the full block it has read.
            let available = (input.len() - start).min(block_capacity);
            let block_len = if block_len < available && available == block_capacity {
                guard
                    .0
                    .matcher
                    .settle_block_cut_with::<K>(available, block_len)
            } else {
                block_len
            };
            let end = (start + block_len).min(input.len());
            let block = &input[start..end];
            let last_block = end == input.len();
            #[cfg(feature = "hash")]
            if self.content_checksum {
                self.hasher.write(block);
            }
            crate::encoding::levels::compress_block_encoded_borrowed::<K>(
                &mut *guard.0,
                self.compression_level,
                last_block,
                block,
                start,
                end,
                out,
                dict_active,
                #[cfg(feature = "lsm")]
                Some(&mut self.block_decompressed_sizes),
                #[cfg(all(feature = "lsm", feature = "hash"))]
                self.block_checksums.as_mut(),
            );
            start = end;
        }
        // `guard` drops here, clearing the borrowed window.
        drop(guard);
        total_uncompressed
    }
}

impl<R: Read, W: Write, M: Matcher> FrameCompressor<R, W, M> {
    /// Create a new `FrameCompressor` with a custom matching algorithm implementation
    pub fn new_with_matcher(matcher: M, compression_level: CompressionLevel) -> Self {
        Self::with_matcher(matcher, compression_level)
    }

    /// Attach a pre-parsed dictionary to be used for subsequent compressions.
    ///
    /// In compressed modes, the dictionary id is written only when the active
    /// matcher supports dictionary priming.
    /// Uncompressed mode and non-priming matchers ignore the attached dictionary
    /// at encode time.
    pub fn set_dictionary(
        &mut self,
        dictionary: crate::decoding::Dictionary,
    ) -> Result<Option<EncoderDictionary>, crate::decoding::errors::DictionaryDecodeError> {
        self.attach_dictionary(EncoderDictionary::from_dictionary(dictionary))
    }

    /// Parse and attach a dictionary blob, in either of the two forms upstream
    /// `ZSTD_CCtx_loadDictionary` takes (`ZSTD_dct_auto`): a
    /// [serialized](crate::decoding::Dictionary::is_serialized) dictionary,
    /// and anything else is raw content, which is why any file can
    /// be handed to `zstd -D`. Raw content has no id, so the frame records none
    /// and the decoder must be given the same bytes explicitly.
    ///
    /// A serialized blob parses through the encoder-only path (skips the
    /// FSE/HUF decode lookup-table build the encoder never reads); the entropy
    /// ENCODER tables — and thus the emitted frame — are identical to a full
    /// parse. To reject anything but a serialized dictionary, parse with
    /// [`EncoderDictionary::from_bytes`] (upstream `ZSTD_dct_fullDict`) and
    /// attach the result.
    ///
    /// An empty buffer is how the same upstream entry point is told there is
    /// no dictionary: it clears whatever was attached and succeeds, returning
    /// it, rather than reporting a dictionary too small to use.
    pub fn set_dictionary_from_bytes(
        &mut self,
        raw_dictionary: &[u8],
    ) -> Result<Option<EncoderDictionary>, crate::decoding::errors::DictionaryDecodeError> {
        if raw_dictionary.is_empty() {
            return Ok(self.clear_dictionary());
        }
        self.attach_dictionary(EncoderDictionary::from_serialized_or_raw_content(
            raw_dictionary,
        )?)
    }
}

impl<R: Read, W: Write, M: Matcher, D: DictionaryHold> FrameCompressor<R, W, M, D> {
    /// A compressor with a custom matcher that compresses every frame with
    /// `dictionary`, or with none, held as given: a `&EncoderDictionary` is
    /// borrowed for the compressor's lifetime, so a compressor built for one
    /// frame reads a dictionary prepared once without taking a reference on it.
    ///
    /// # Errors
    /// As [`Self::set_encoder_dictionary`].
    ///
    /// # Examples
    /// ```
    /// use structured_zstd::encoding::sequences::{ExternalSequence, SequenceReplay};
    /// use structured_zstd::encoding::{CompressionLevel, EncoderDictionary, FrameCompressor};
    ///
    /// let dictionary = EncoderDictionary::from_serialized_or_raw_content(b"shared history").unwrap();
    /// let input = b"shared history, once more";
    /// // One block of literals; a parse may also reach into the dictionary.
    /// let parse = [ExternalSequence::delimiter(input.len() as u32)];
    /// let mut frame = Vec::new();
    /// let mut compressor: FrameCompressor<&[u8], &mut Vec<u8>, _, _> =
    ///     FrameCompressor::new_with_matcher_and_dictionary(
    ///         SequenceReplay::new(&parse, 1 << 17).with_dictionary(),
    ///         CompressionLevel::Default,
    ///         Some(&dictionary),
    ///     )
    ///     .unwrap();
    /// compressor.set_source(&input[..]);
    /// compressor.set_drain(&mut frame);
    /// compressor.compress();
    /// assert!(compressor.dictionary().is_some());
    /// drop(compressor);
    /// assert!(!frame.is_empty());
    /// ```
    pub fn new_with_matcher_and_dictionary(
        matcher: M,
        compression_level: CompressionLevel,
        dictionary: Option<D>,
    ) -> Result<Self, crate::decoding::errors::DictionaryDecodeError> {
        let mut compressor = Self::with_matcher(matcher, compression_level);
        if let Some(dictionary) = dictionary {
            compressor.attach_dictionary(dictionary)?;
        }
        Ok(compressor)
    }

    fn with_matcher(matcher: M, compression_level: CompressionLevel) -> Self {
        Self {
            uncompressed_data: None,
            compressed_data: None,
            dictionary: None,
            source_size_hint: None,
            state: CompressState {
                matcher,
                sequence_capture: None,
                copy_kernel: crate::encoding::fastpath::select_kernel(),
                last_huff_table: None,
                huff_table_spare: None,
                huff_weights: Default::default(),
                seen_content: Default::default(),
                fse_tables: FseTables::new(),
                block_scratch: crate::encoding::blocks::CompressedBlockScratch::new(),
                workspace: crate::encoding::workspace::Workspace::new(),
                offset_hist: [1, 4, 8],
                strategy_tag: crate::encoding::strategy::StrategyTag::for_compression_level(
                    compression_level,
                ),
                pre_split: crate::encoding::levels::config::level_pre_split(compression_level)
                    .map(|tier| tier as u8),
                huf_optimal_search: true,
                literal_compression_disabled: matches!(
                    compression_level,
                    crate::encoding::CompressionLevel::Level(n) if n < 0
                ),
                post_split_allowed: true,
            },
            compression_level,
            magicless: false,
            content_checksum: false,
            pre_split_disabled: false,
            content_size_flag: true,
            dict_id_flag: true,
            target_block_size: None,
            max_block_size: None,
            #[cfg(feature = "hash")]
            hasher: XxHash64::with_seed(0),
            #[cfg(feature = "lsm")]
            frame_emit_info: None,
            #[cfg(all(feature = "lsm", feature = "hash"))]
            per_block_checksums_enabled: false,
            #[cfg(all(feature = "lsm", feature = "hash"))]
            block_checksums: None,
            #[cfg(feature = "lsm")]
            block_decompressed_sizes: alloc::vec::Vec::new(),
            tuning: FrameTuning::default(),
        }
    }

    /// Enable or disable magicless frame format (`ZSTD_f_zstd1_magicless`).
    ///
    /// When set to `true`, emitted frames omit the 4-byte magic number
    /// prefix. The matching decoder must be configured to expect a
    /// magicless stream — wire-format only round-trips with a
    /// magicless-aware decoder.
    pub fn set_magicless(&mut self, magicless: bool) {
        self.magicless = magicless;
    }

    /// Enable or disable the trailing XXH64 content checksum
    /// (semantics of upstream `ZSTD_c_checksumFlag`). Default `false`,
    /// matching the upstream library default (`ZSTD_c_checksumFlag = 0`)
    /// so out-of-the-box frames carry the same layout and pay the same
    /// costs as the reference implementation.
    ///
    /// When `false`, emitted frames set `Content_Checksum_flag = 0` and carry
    /// no trailing digest; such frames are valid (RFC 8878) and decode
    /// correctly in any [`ContentChecksum`](crate::decoding::ContentChecksum)
    /// mode. Without the `hash` feature no checksum is emitted regardless of
    /// this setting.
    pub fn set_content_checksum(&mut self, emit: bool) {
        self.content_checksum = emit;
    }

    /// Enable or disable recording `Frame_Content_Size` in the frame header
    /// when the total size is known (semantics of upstream
    /// `ZSTD_c_contentSizeFlag`). Default `true`, matching upstream. With
    /// the flag off the header carries a window descriptor instead (and the
    /// single-segment layout, which requires an FCS, is disabled).
    pub fn set_content_size_flag(&mut self, emit: bool) {
        self.content_size_flag = emit;
    }

    /// Enable or disable recording the dictionary ID in the frame header
    /// when a dictionary is attached (semantics of upstream
    /// `ZSTD_c_dictIDFlag`). Default `true`, matching upstream. Frames
    /// emitted with the flag off still decode when the decoder is handed
    /// the dictionary explicitly.
    pub fn set_dictionary_id_flag(&mut self, emit: bool) {
        self.dict_id_flag = emit;
    }

    /// Set an upper bound on emitted block sizes (semantics of upstream
    /// `ZSTD_c_targetCBlockSize`): every physical block's payload is capped
    /// at `target` bytes (+3-byte block header on the wire), trading some
    /// ratio for bounded per-block latency. The value is clamped to
    /// `[MIN_TARGET_BLOCK_SIZE, MAX_BLOCK_SIZE]` (the upstream bounds).
    /// `None` removes the target.
    pub fn set_target_block_size(&mut self, target: Option<u32>) {
        self.target_block_size = target.map(|t| {
            t.clamp(
                crate::common::MIN_TARGET_BLOCK_SIZE,
                crate::common::MAX_BLOCK_SIZE,
            )
        });
    }

    /// Cap every block of the frame at `size` bytes (semantics of upstream
    /// `ZSTD_c_maxBlockSize`), for a decoder that holds only that much of a
    /// block at a time. Clamped to `[MIN_MAX_BLOCK_SIZE, MAX_BLOCK_SIZE]`;
    /// `None` restores the format's 128 KiB. Combines with
    /// [`Self::set_target_block_size`]: the smaller of the two applies.
    ///
    /// ```
    /// use structured_zstd::encoding::{CompressionLevel, FrameCompressor};
    ///
    /// let mut compressor = FrameCompressor::new(CompressionLevel::Default);
    /// compressor.set_max_block_size(Some(1024));
    /// let mut frame = Vec::new();
    /// compressor.set_source(&[7u8; 4096][..]);
    /// compressor.set_drain(&mut frame);
    /// compressor.compress();
    /// ```
    pub fn set_max_block_size(&mut self, size: Option<u32>) {
        self.max_block_size = size.map(|s| {
            s.clamp(
                crate::common::MIN_MAX_BLOCK_SIZE,
                crate::common::MAX_BLOCK_SIZE,
            )
        });
    }

    /// Record the parse every block takes, from the next one on, until turned
    /// off: its sequences with plain offsets, each block closed by a delimiter
    /// carrying its trailing literals, and a block written out unsearched as a
    /// delimiter alone (C `ZSTD_generateSequences`). An empty input records
    /// nothing, as upstream's does. The frames themselves are unchanged.
    ///
    /// ```
    /// use structured_zstd::encoding::{CompressionLevel, FrameCompressor};
    ///
    /// let input = b"capture capture capture capture".repeat(10);
    /// let mut compressor: FrameCompressor = FrameCompressor::new(CompressionLevel::Default);
    /// compressor.set_sequence_capture(true);
    /// compressor.compress_independent_frame(&input);
    /// let parse = compressor.take_captured_sequences();
    /// let covered: usize = parse.iter().map(|s| (s.lit_len + s.match_len) as usize).sum();
    /// assert_eq!(covered, input.len());
    /// ```
    pub fn set_sequence_capture(&mut self, capture: bool) {
        self.state.sequence_capture = capture.then(Vec::new);
    }

    /// The parse recorded since [`Self::set_sequence_capture`] turned it on,
    /// or since it was last taken; empty when nothing records it.
    pub fn take_captured_sequences(&mut self) -> Vec<crate::encoding::sequences::ExternalSequence> {
        self.state
            .sequence_capture
            .as_mut()
            .map(core::mem::take)
            .unwrap_or_default()
    }

    /// The configured block ceiling before the window is considered: the
    /// smaller of the target and the maximum, or the format's 128 KiB.
    fn block_ceiling(&self) -> usize {
        block_ceiling(self.target_block_size, self.max_block_size)
    }

    /// The active block-size cap: the configured ceiling, held to the window.
    fn block_capacity(&self) -> usize {
        let requested = self.block_ceiling();
        // Upstream zstd sizes a block as `MIN(maxBlockSize, windowSize)`
        // (`ZSTD_compress.c`). A block wider than the window can never be
        // held by the matcher, which asserts on it, so a small `window_log`
        // must shrink the block rather than overrun the window.
        let window = self.state.matcher.window_size() as usize;
        if window == 0 {
            requested
        } else {
            requested.min(window)
        }
    }

    /// Before calling [FrameCompressor::compress] you need to set the source.
    ///
    /// This is the data that is compressed and written into the drain.
    pub fn set_source(&mut self, uncompressed_data: R) -> Option<R> {
        self.uncompressed_data.replace(uncompressed_data)
    }

    /// Before calling [FrameCompressor::compress] you need to set the drain.
    ///
    /// As the compressor compresses data, the drain serves as a place for the output to be writte.
    pub fn set_drain(&mut self, compressed_data: W) -> Option<W> {
        self.compressed_data.replace(compressed_data)
    }

    /// Diagnostic switch: cut full blocks only, never pre-split. Used by
    /// the sequence capture so its block structure matches upstream's
    /// `ZSTD_generateSequences` (see `pre_split_disabled`).
    #[cfg(feature = "bench-internals")]
    pub fn set_pre_split_disabled(&mut self, disabled: bool) {
        self.pre_split_disabled = disabled;
    }

    /// The pre-split level the block loops apply: the effective strategy's
    /// tier (`state.pre_split`) unless the diagnostic switch is on.
    fn pre_split_level(&self) -> Option<usize> {
        if self.pre_split_disabled {
            None
        } else {
            self.state.pre_split.map(usize::from)
        }
    }

    /// The level params the matcher's reset resolves for the next frame
    /// (dictionary-aware when a dictionary will be used).
    fn resolve_frame_params(
        &self,
        hint: Option<u64>,
        with_dictionary: bool,
    ) -> crate::encoding::levels::config::LevelParams {
        resolve_frame_params(
            self.compression_level,
            hint,
            attached(&self.dictionary).filter(|_| with_dictionary),
        )
    }

    /// Record the strategy the matcher actually runs for the next frame (a
    /// public-parameter override, else the size- and dictionary-adaptive
    /// resolution in `params`) and its pre-split tier: the literal gates and
    /// the block splitter read these, and upstream indexes `splitLevels` by
    /// the effective strategy too.
    fn sync_effective_strategy(&mut self, params: &crate::encoding::levels::config::LevelParams) {
        sync_effective_strategy(
            &mut self.state,
            self.compression_level,
            params,
            self.tuning.strategy,
        );
    }

    /// Provide a hint about the total uncompressed size for the next frame.
    ///
    /// When set, the encoder selects smaller hash tables and windows for
    /// small inputs, matching the C zstd source-size-class behavior.
    ///
    /// This hint applies only to frame payload bytes (`size`). Dictionary
    /// history is primed separately and does not inflate the hinted size or
    /// advertised frame window.
    /// Must be called before [`compress`](Self::compress).
    pub fn set_source_size_hint(&mut self, size: u64) {
        self.source_size_hint = Some(size);
    }

    /// Total heap bytes this compressor's allocations hold, excluding the
    /// inline struct: the match-finder tables / history / recycled buffers and
    /// the primed-dictionary snapshot (via the matcher), the retained
    /// Huffman tables (active + recycled spare), the retained dictionary
    /// content, the cached dictionary entropy tables (literals Huffman +
    /// LL/ML/OF FSE), and the per-block sidecar buffers. Lets a context
    /// report its true footprint through `ZSTD_sizeof_CCtx`.
    pub fn heap_size(&self) -> usize {
        let mut total = self.state.matcher.heap_size();
        total += self
            .state
            .last_huff_table
            .as_ref()
            .map_or(0, |table| table.heap_size());
        total += self
            .state
            .huff_table_spare
            .as_ref()
            .map_or(0, |table| table.heap_size());
        // The weight builder's buffers are kept between blocks and frames, so
        // a reused compressor holds them for as long as it lives.
        total += self.state.huff_weights.heap_size();
        total += self.state.retained_scratch_heap_size();
        total += self.state.seen_content.heap_size();
        total += self.state.captured_parse_heap_size();
        // A borrowed dictionary belongs to its owner, as upstream's
        // ZSTD_sizeof_CCtx counts a local dictionary and not a referenced CDict.
        if D::OWNED {
            total += attached(&self.dictionary).map_or(0, EncoderDictionary::heap_size);
        }
        #[cfg(all(feature = "lsm", feature = "hash"))]
        {
            total += self
                .block_checksums
                .as_ref()
                .map_or(0, |v| v.capacity() * core::mem::size_of::<u32>());
        }
        #[cfg(feature = "lsm")]
        {
            total += self.block_decompressed_sizes.capacity() * core::mem::size_of::<u32>();
        }
        total
    }

    /// Compress the uncompressed data from the provided source as one Zstd frame and write it to the provided drain
    ///
    /// This will repeatedly call [Read::read] on the source to fill up blocks until the source returns 0 on the read call.
    /// All compressed blocks are buffered in memory so that the frame header can include the
    /// `Frame_Content_Size` field (which requires knowing the total uncompressed size). The
    /// entire frame — header, blocks, and optional checksum — is then written to the drain
    /// at the end. This means peak memory usage is O(compressed_size).
    ///
    /// To avoid endlessly encoding from a potentially endless source (like a network socket) you can use the
    /// [Read::take] function.
    pub fn compress(&mut self) {
        // The frame's one choice of kernel tier, as in
        // `compress_independent_frame_into`.
        crate::encoding::fastpath::with_encode_tier!(
            self.state.copy_kernel,
            K => self.compress_with::<K>()
        )
    }

    /// [`Self::compress`] compiled for the kernel tier `K`.
    fn compress_with<K: EncodeTier>(&mut self) {
        // `prepare_frame` / `finish_frame` are shared with the borrowed
        // one-shot path, so both run the same reset / dict-prime /
        // entropy-seed setup and frame tail.
        let prep = self.prepare_frame::<K>(crate::encoding::workspace::IngestPlan::Stream);
        // Take the reader out so `run_owned_block_loop` can borrow it
        // mutably alongside `&mut self` (the rest of the loop touches
        // `self.state` / `self.hasher`, disjoint from the reader). Restored
        // before the frame tail so a reused compressor keeps its source.
        //
        // Deliberately NOT restored on unwind: if the block loop panics the
        // source has been partially consumed, so handing it back would let a
        // `catch_unwind` caller "successfully" compress the remaining tail
        // from an arbitrary midpoint — silent data corruption. Leaving the
        // slot empty makes any post-panic reuse fail loudly at the `expect`
        // below (matcher/entropy state is equally unre-usable after an
        // unwind; the reference implementation likewise requires a context
        // reset after an error).
        let mut source = self
            .uncompressed_data
            .take()
            .expect("source must be set via set_source before compress()");
        // Streaming drain: the content size is only known at EOF, so the
        // frame header can't precede the blocks — accumulate them in a local
        // buffer and let `finish_frame` write header + blocks to the drain.
        let mut all_blocks: Vec<u8> = Vec::with_capacity(initial_all_blocks_cap(
            prep.initial_size_hint,
            self.block_capacity(),
        ));
        let mut block_source = ReaderBlockSource::new(&mut source);
        let total_uncompressed = self.run_owned_block_loop::<_, K>(
            &mut block_source,
            prep.initial_size_hint,
            false,
            &mut all_blocks,
        );
        self.uncompressed_data = Some(source);
        self.finish_frame(all_blocks, total_uncompressed, &prep);
    }

    /// Write every matched block as one block, never cut after matching: what
    /// a dictionary's analysis needs to tell each block's kind from the frame.
    #[cfg(feature = "dict-builder")]
    pub(crate) fn forbid_post_split(&mut self) {
        self.state.post_split_allowed = false;
    }

    /// [`Self::compress`] for a source of `total` bytes, the frame written
    /// over `out` rather than to the drain: with the content size known up
    /// front the header goes first and the blocks straight after it, with no
    /// block accumulator and no copy into the drain. The frame is the one
    /// `compress` writes for the same source, hint and settings.
    #[cfg(feature = "dict-builder")]
    pub(crate) fn compress_known_into(&mut self, total: u64, out: &mut Vec<u8>) {
        crate::encoding::fastpath::with_encode_tier!(
            self.state.copy_kernel,
            K => self.compress_known_into_with::<K>(total, out)
        )
    }

    /// [`Self::compress_known_into`] compiled for the kernel tier `K`.
    #[cfg(feature = "dict-builder")]
    fn compress_known_into_with<K: EncodeTier>(&mut self, total: u64, out: &mut Vec<u8>) {
        let prep = self.prepare_frame::<K>(crate::encoding::workspace::IngestPlan::Stream);
        let mut source = self
            .uncompressed_data
            .take()
            .expect("source must be set via set_source before compress_known_into()");
        out.clear();
        self.append_frame_header(total, &prep, out);
        let header_len = out.len();
        let mut block_source = ReaderBlockSource::new(&mut source);
        let read = self.run_owned_block_loop::<_, K>(
            &mut block_source,
            prep.initial_size_hint,
            false,
            out,
        );
        self.uncompressed_data = Some(source);
        assert_eq!(
            read, total,
            "the source held another length than its frame header"
        );
        #[cfg(feature = "hash")]
        if self.content_checksum {
            out.extend_from_slice(&(self.hasher.finish() as u32).to_le_bytes());
        }
        #[cfg(feature = "lsm")]
        {
            let emit_checksum = cfg!(feature = "hash") && self.content_checksum;
            let blocks_end = out.len() - if emit_checksum { 4 } else { 0 };
            self.populate_frame_emit_info(header_len, &out[header_len..blocks_end], emit_checksum);
        }
        #[cfg(not(feature = "lsm"))]
        let _ = header_len;
    }

    /// Resets the compressor for the next frame, whose input reaches the
    /// match finder as `ingest` says.
    fn prepare_frame<K: EncodeTier>(
        &mut self,
        ingest: crate::encoding::workspace::IngestPlan,
    ) -> FramePrep {
        // The raw-skip's memory of what this frame has already emitted. Frames
        // are independent, so carrying it over would let one frame's content
        // hold the skip off for the next; the allocation is kept.
        self.state.seen_content.reset_for_frame();
        // Reset per-frame introspection state so a re-used compressor
        // doesn't carry over the previous frame's layout/checksums.
        #[cfg(feature = "lsm")]
        {
            self.frame_emit_info = None;
            // Always captured under lsm (drives `decompressed_byte_range`);
            // clear, keep the allocation for a reused compressor.
            self.block_decompressed_sizes.clear();
        }
        #[cfg(all(feature = "lsm", feature = "hash"))]
        {
            if self.per_block_checksums_enabled {
                self.block_checksums = Some(alloc::vec::Vec::new());
            } else {
                self.block_checksums = None;
            }
        }
        let initial_size_hint = self.source_size_hint;
        let use_dictionary_state =
            !matches!(self.compression_level, CompressionLevel::Uncompressed)
                && self.state.matcher.supports_dictionary_priming()
                && self.dictionary.is_some();
        if let Some(size_hint) = self.source_size_hint.take() {
            // Keep source-size hint scoped to payload bytes; dictionary priming
            // is applied separately and should not force larger matcher sizing.
            self.state.matcher.set_source_size_hint(size_hint);
        }
        // Hand the matcher the dictionary's sizes so the frame runs the CDict's
        // cParams tier and sizes its dictionary tables from the content. Set
        // before `reset` (which consumes it) and only when a dictionary will
        // actually be primed.
        if use_dictionary_state && let Some(dict) = attached(&self.dictionary) {
            self.state.matcher.set_dictionary_size_hint(dict.sizes());
        }
        // Clearing buffers to allow re-using of the compressor
        let block_target = self.block_ceiling();
        // Raw frames emit straight from the staged buffer and never consult
        // the match finder, so they keep no history.
        let ingest = if matches!(self.compression_level, CompressionLevel::Uncompressed) {
            crate::encoding::workspace::IngestPlan::Raw
        } else {
            ingest
        };
        self.state
            .reset_for_frame(self.compression_level, block_target, ingest);
        let block_capacity = self.block_capacity();
        self.state.finish_layout(block_capacity);
        self.state.offset_hist = [1, 4, 8];
        // Sync `state.strategy_tag` to the level resolved at this reset so
        // the literal-compression gates (`min_literals_to_compress` /
        // `min_gain` in `encoding::blocks::compressed`) see the correct
        // strategy for the next frame. Frame-by-frame level changes go
        // through this same `compress()` entry point, so re-syncing here
        // covers level switches without touching the matcher dispatch.
        // A public-parameter strategy override (#27) wins over the level's
        // derived tag so the literal-compression gates and dict-attach cutoff
        // below see the strategy the matcher actually runs. Otherwise resolve
        // the strategy SIZE-ADAPTIVELY through the same path the matcher's reset
        // used (`resolve_level_params` -> `get_cparams`, the port of upstream
        // `ZSTD_getCParams`): a small frame promotes a level to a higher
        // strategy (e.g. L13 over a <=16 KiB frame becomes btultra). Re-deriving
        // from the bare level would make the literal-compression / HUF-search
        // gates disagree with the matcher's actual parse on small frames (the
        // gate would think btlazy2 and skip the HUF table-log search the btultra
        // frame runs, costing a few bytes on small literal sections).
        // A dictionary frame runs the CDict's strategy (upstream
        // `ZSTD_resetCCtx_usingCDict`), so resolve through the same
        // dictionary-aware path the matcher's reset took.
        let params = self.resolve_frame_params(initial_size_hint, use_dictionary_state);
        self.sync_effective_strategy(&params);
        // `initial_size_hint` (captured before the `.take()` above) — by here
        // `self.source_size_hint` is None.
        self.state.huf_optimal_search =
            huf_search_enabled(self.state.strategy_tag, initial_size_hint);
        // Recomputed per frame from the persisted override: attaching or
        // clearing a dictionary after `set_parameters` can move the strategy
        // this gate reads.
        self.state.literal_compression_disabled = literal_compression_disabled(
            self.state.strategy_tag,
            self.compression_level,
            gate_target_length(
                self.compression_level,
                &self.tuning,
                attached(&self.dictionary).filter(|_| use_dictionary_state),
            ),
            self.tuning.literal_compression,
        );
        let cached_entropy = if use_dictionary_state {
            attached(&self.dictionary).map(|dict| &dict.inner.entropy)
        } else {
            None
        };
        if use_dictionary_state && let Some(dict) = attached(&self.dictionary) {
            // This state drives sequence encoding, while matcher priming below updates
            // the match generator's internal repeat-offset history for match finding.
            self.state.offset_hist = dict.inner.offset_hist;
            load_frame_dictionary::<M, K>(
                &mut self.state,
                self.compression_level,
                dict,
                initial_size_hint,
            );
        }
        if let Some(cache) = cached_entropy {
            // Refill an empty slot from the recycled spare before
            // `clone_from`: `Option::clone_from(None ← Some)` falls back to
            // a fresh clone (two Vec allocations), while `Some ← Some`
            // delegates to the table's buffer-reusing `clone_from`. Frames
            // whose last block cleared the table would otherwise re-clone
            // the dict seed every frame.
            match &cache.huff {
                Some(src) => {
                    if self.state.last_huff_table.is_none() {
                        self.state.last_huff_table = self.state.huff_table_spare.take();
                    }
                    match &mut self.state.last_huff_table {
                        Some(dst) => dst.clone_from(src),
                        slot => *slot = Some(src.clone()),
                    }
                }
                None => self.state.clear_huff_table(),
            }
        } else {
            self.state.clear_huff_table();
        }
        // Whatever the last frame ended on is about to be replaced. Keep it as
        // this frame's build buffer rather than dropping it, or a reused
        // compressor that emits one custom-table block per frame allocates a
        // table per axis every frame.
        self.state.fse_tables.park_previous_before_frame();
        // `clone_from` keeps frame-to-frame seeding cheap for reused compressors by
        // reusing existing allocations where possible instead of reallocating every frame.
        if let Some(cache) = cached_entropy {
            self.state
                .fse_tables
                .ll_previous
                .clone_from(&cache.ll_previous);
            self.state
                .fse_tables
                .ml_previous
                .clone_from(&cache.ml_previous);
            self.state
                .fse_tables
                .of_previous
                .clone_from(&cache.of_previous);
        } else {
            self.state.fse_tables.ll_previous = None;
            self.state.fse_tables.ml_previous = None;
            self.state.fse_tables.of_previous = None;
        }
        let ll_entropy = cached_entropy.and_then(|cache| match cache.ll_previous.as_ref() {
            Some(PreviousFseTable::Custom(table)) => Some(table.as_ref()),
            _ => None,
        });
        let ml_entropy = cached_entropy.and_then(|cache| match cache.ml_previous.as_ref() {
            Some(PreviousFseTable::Custom(table)) => Some(table.as_ref()),
            _ => None,
        });
        let of_entropy = cached_entropy.and_then(|cache| match cache.of_previous.as_ref() {
            Some(PreviousFseTable::Custom(table)) => Some(table.as_ref()),
            _ => None,
        });
        self.state.matcher.seed_dictionary_entropy(
            self.state.last_huff_table.as_ref(),
            ll_entropy,
            ml_entropy,
            of_entropy,
        );
        #[cfg(feature = "hash")]
        {
            self.hasher = XxHash64::with_seed(0);
        }
        let window_size = self.state.matcher.window_size();
        assert!(
            window_size != 0,
            "matcher reported window_size == 0, which is invalid"
        );
        FramePrep {
            window_size,
            use_dictionary_state,
            initial_size_hint,
        }
    }

    /// Owned streaming block loop: reads blocks from the caller-provided
    /// `source` reader, optionally pre-splits, hashes for the content
    /// checksum, and emits each block via `compress_block_encoded`,
    /// accumulating the block bytes. Returns `(all_blocks,
    /// total_uncompressed)`. The source is passed in (rather than read
    /// from `self.uncompressed_data`) so the streaming `compress` path can
    /// feed the configured reader while the slice paths
    /// (`compress_oneshot_borrowed`, `compress_independent_frame`) feed an
    /// in-place `&[u8]` cursor without baking its lifetime into the
    /// compressor type.
    fn run_owned_block_loop<S: OwnedBlockSource, K: EncodeTier>(
        &mut self,
        source: &mut S,
        initial_size_hint: Option<u64>,
        // Whether `initial_size_hint` is the input's exact length (the
        // one-shot slice paths) or a caller-provided estimate (the streaming
        // `Read` path, where `set_source_size_hint` is advisory). An exact
        // hint drives the one-shot ratio reservation; an estimate is only
        // trusted up to a small lookahead past the bytes actually read.
        hint_is_exact: bool,
        out: &mut Vec<u8>,
    ) -> u64 {
        if matches!(self.compression_level, CompressionLevel::Uncompressed) {
            return self.run_raw_block_loop(source, initial_size_hint, out);
        }
        // Compressed blocks are appended to `out` from its current end. The
        // streaming drain path passes a fresh buffer (the frame header is
        // written to the drain afterward, since Frame_Content_Size is only
        // known once the reader hits EOF); the one-shot compress-into-Vec
        // path passes `out` already holding the header. The upstream zstd split
        // `savings` gate below accumulates block-relative (`before_len`)
        // output deltas, so a header prefix never skews it.
        let blocks_start = out.len();
        let mut total_uncompressed: u64 = 0;
        let mut reached_eof = false;
        let mut savings = 0i64;
        // The matcher's history was laid out for the whole frame when the
        // context was (`frame_history_bytes`), so the blocks read into it below
        // never reallocate it.
        let block_capacity = self.block_capacity();
        loop {
            // Top the matcher's uncommitted input back up to one block. Reading
            // straight into its history means neither the block nor a pre-split
            // remainder is ever copied: the remainder of the previous iteration
            // is already there, uncommitted, and heads this block, as upstream's
            // ZSTD_compress_frameChunk() runs over one contiguous input buffer.
            if !reached_eof {
                // Remaining-bytes expectation for the reader source's sizing
                // (`None` = unknown, or an inexact hint already met by prior
                // blocks). The slice source appends directly and ignores it.
                let size_hint_remaining = match initial_size_hint {
                    Some(hint) if hint > total_uncompressed => Some(hint - total_uncompressed),
                    _ => None,
                };
                // A carried remainder is what a pre-split left of a full block,
                // so it is shorter than one. Held to what an exact source has
                // left: its history was laid out for its bytes and no more.
                let want = (block_capacity - self.state.matcher.uncommitted_input().len())
                    .min(source.exact_remaining().unwrap_or(usize::MAX));
                let (appended, eof) = self.state.matcher.fill_in_place(want, &mut |buf| {
                    source.fill_block(buf, buf.len() + want, size_hint_remaining)
                });
                total_uncompressed += appended as u64;
                reached_eof = eof;
            }
            let available = self.state.matcher.uncommitted_input().len();
            let mut last_block = reached_eof;
            let remaining_for_split = if reached_eof {
                available
            } else {
                block_capacity
            };
            // Length this block will actually claim. The pre-split pass may
            // shorten it; the remainder stays uncommitted and heads the next
            // block.
            let mut block_len = available;
            let fixed_len = self.state.matcher.fixed_block_len();
            if let Some(len) = fixed_len {
                // A block of no bytes would read below as the empty frame's
                // terminal block and drop the input; such a cut is ignored.
                if len > 0 && len < available {
                    block_len = len;
                    last_block = false;
                }
            } else if available == block_capacity {
                let split_at = optimal_block_size_with(
                    self.pre_split_level(),
                    self.state.matcher.uncommitted_input(),
                    remaining_for_split,
                    block_capacity,
                    savings,
                );
                // The splitter sees byte statistics only; a cut it proposes
                // inside a match the matcher already knows of is moved out.
                let split_at = if split_at < available {
                    match self.state.matcher.builtin_driver() {
                        Some(driver) => driver.settle_block_cut_with::<K>(available, split_at),
                        None => self.state.matcher.settle_block_cut(available, split_at),
                    }
                } else {
                    split_at
                };
                if split_at < available {
                    block_len = split_at;
                    last_block = false;
                }
            }
            // As we read, hash that data too (skipped when the content
            // checksum is disabled).
            #[cfg(feature = "hash")]
            if self.content_checksum {
                self.hasher
                    .write(&self.state.matcher.uncommitted_input()[..block_len]);
            }
            // Per-physical-block XXH64 (low 32 bits) for the optional
            // per-block checksum sidecar. Hashing happens INSIDE the
            // block emitters (RLE / Raw fast-path / Compressed /
            // post-split partitions), so the digests vector has
            // exactly one entry per physical Block_Header written to
            // `all_blocks` — 1:1 with `FrameEmitInfo.blocks`. See
            // `enable_per_block_checksums` rustdoc.
            // Size the output ahead of this block's emission from the ratio
            // observed so far (see `reserve_for_next_block`); with no usable
            // size hint, ensure one block's worst case and let the doubling
            // growth policy amortize across blocks.
            // Bytes already emitted as blocks: everything read so far minus what
            // is still uncommitted, this block included.
            let emitted = total_uncompressed - available as u64;
            match initial_size_hint {
                Some(hint) if hint >= total_uncompressed => {
                    // An advisory hint (streaming path) is only trusted up to
                    // a small lookahead past the bytes actually read: a hint
                    // far above the real input would otherwise reserve the
                    // whole phantom remainder up front.
                    let hint_remaining = hint - emitted;
                    let remaining = if hint_is_exact {
                        hint_remaining
                    } else {
                        let buffered = total_uncompressed - emitted;
                        const HINT_LOOKAHEAD: u64 = 64 * 1024;
                        hint_remaining.min(buffered + HINT_LOOKAHEAD)
                    };
                    reserve_for_next_block(
                        out,
                        blocks_start,
                        emitted,
                        remaining as usize,
                        self.block_capacity(),
                    );
                }
                _ => {
                    out.reserve(block_len + 3 + 16);
                }
            }
            // Special handling is needed for compression of a totally empty file
            if block_len == 0 {
                let header = BlockHeader {
                    last_block: true,
                    block_type: crate::blocks::block::BlockType::Raw,
                    block_size: 0,
                };
                header.serialize(out);
                #[cfg(feature = "lsm")]
                self.block_decompressed_sizes.push(0);
                #[cfg(all(feature = "lsm", feature = "hash"))]
                if let Some(checksums) = self.block_checksums.as_mut() {
                    checksums.push(xxh64_block_low32(&[]));
                }
                break;
            }

            let before_len = out.len();
            // A primed dictionary makes "incompressible-looking" blocks
            // matchable against the dict, so the raw-fast-path inside must be
            // bypassed (it skips matching). Mirror prepare_frame's
            // `use_dictionary_state`: a dict is only PRIMED (and thus matchable)
            // when the matcher supports priming; a non-priming matcher ignores
            // an attached dictionary, so the raw-fast-path must stay enabled
            // for it.
            let dict_active =
                self.dictionary.is_some() && self.state.matcher.supports_dictionary_priming();
            // A block the matcher laid out stays one physical block.
            let post_split = self.state.post_split_allowed;
            self.state.post_split_allowed = post_split && fixed_len.is_none();
            compress_block_encoded::<M, K>(
                &mut self.state,
                self.compression_level,
                last_block,
                block_len,
                out,
                dict_active,
                #[cfg(feature = "lsm")]
                Some(&mut self.block_decompressed_sizes),
                #[cfg(all(feature = "lsm", feature = "hash"))]
                self.block_checksums.as_mut(),
            );
            self.state.post_split_allowed = post_split;
            savings += block_len as i64 - (out.len() - before_len) as i64;
            // A block is last only when it took everything read, so nothing
            // uncommitted is left behind it.
            if last_block {
                break;
            }
        }
        total_uncompressed
    }

    /// Block loop of an uncompressed frame: each block is read straight into
    /// `out` behind its header, which is filled in once the block's length is
    /// known, so the payload is never staged and the match finder never sees
    /// it.
    fn run_raw_block_loop<S: OwnedBlockSource>(
        &mut self,
        source: &mut S,
        initial_size_hint: Option<u64>,
        out: &mut Vec<u8>,
    ) -> u64 {
        let block_capacity = self.block_capacity();
        let mut total_uncompressed: u64 = 0;
        loop {
            let size_hint_remaining = match initial_size_hint {
                Some(hint) if hint > total_uncompressed => Some(hint - total_uncompressed),
                _ => None,
            };
            let header_at = out.len();
            out.extend_from_slice(&[0; 3]);
            let payload_at = out.len();
            let (block_len, last_block) =
                source.fill_block(out, payload_at + block_capacity, size_hint_remaining);
            total_uncompressed += block_len as u64;
            let header = BlockHeader {
                last_block,
                block_type: crate::blocks::block::BlockType::Raw,
                // At most one block, which a block header's 21-bit size holds.
                block_size: block_len as u32,
            };
            out[header_at..payload_at].copy_from_slice(&header.to_le_bytes());
            // A stored block is a delimiter alone in a captured parse. The
            // empty block of an empty input records none, as upstream's
            // block loop never runs for one (zstd_compress.c:4610).
            if block_len > 0
                && let Some(capture) = self.state.sequence_capture.as_mut()
            {
                capture.push(crate::encoding::sequences::ExternalSequence::delimiter(
                    block_len as u32,
                ));
            }
            #[cfg(feature = "hash")]
            if self.content_checksum {
                self.hasher.write(&out[payload_at..]);
            }
            #[cfg(feature = "lsm")]
            self.block_decompressed_sizes.push(block_len as u32);
            #[cfg(all(feature = "lsm", feature = "hash"))]
            if let Some(checksums) = self.block_checksums.as_mut() {
                checksums.push(xxh64_block_low32(&out[payload_at..]));
            }
            if last_block {
                break;
            }
        }
        total_uncompressed
    }

    /// Append the frame header bytes onto `out` once the total payload size
    /// is known (so `Frame_Content_Size` / `single_segment` can be set).
    /// Appends rather than returns so the one-shot path serializes straight
    /// into the reused output buffer with no per-frame header `Vec`.
    fn append_frame_header(&self, total_uncompressed: u64, prep: &FramePrep, out: &mut Vec<u8>) {
        // The header goes out after the content, so its size is known whether
        // or not it was hinted. A windowed frame cannot encode a content size
        // below 256 in fewer than 4 FCS bytes (the 1-byte class is
        // single-segment-only, see `find_fcs_field_size`), so small payloads
        // gain the most.
        let frame_content_size = self.content_size_flag.then_some(total_uncompressed);
        let single_segment =
            crate::encoding::frame_header::single_segment(frame_content_size, prep.window_size);
        let header = FrameHeader {
            frame_content_size,
            single_segment,
            content_checksum: cfg!(feature = "hash") && self.content_checksum,
            dictionary_id: if prep.use_dictionary_state && self.dict_id_flag {
                // Id 0 is a raw-content dictionary: RFC 8878 spells "no
                // dictionary ID" as an absent field, not as a stored zero.
                attached(&self.dictionary)
                    .map(|dict| dict.inner.id)
                    .filter(|id| *id != 0)
                    .map(u64::from)
            } else {
                None
            },
            window_size: if single_segment {
                None
            } else {
                Some(prep.window_size)
            },
            magicless: self.magicless,
        };
        header.serialize(out);
    }

    /// Write the frame header, accumulated block bytes, and optional
    /// trailing content checksum to the configured drain; populate
    /// `frame_emit_info` (lsm). Header and blocks are written separately to
    /// avoid shifting `all_blocks` to prepend the header. Used by
    /// `compress` and `compress_oneshot_borrowed`.
    fn finish_frame(&mut self, all_blocks: Vec<u8>, total_uncompressed: u64, prep: &FramePrep) {
        let mut header_buf: Vec<u8> = Vec::with_capacity(18);
        self.append_frame_header(total_uncompressed, prep, &mut header_buf);
        // Snapshot the checksum before borrowing the drain field so the
        // `self.hasher` read and the `self.compressed_data` write don't
        // both need `&mut self` simultaneously.
        #[cfg(feature = "hash")]
        let checksum_bytes = self
            .content_checksum
            .then(|| (self.hasher.finish() as u32).to_le_bytes());
        let drain = self.compressed_data.as_mut().unwrap();
        drain.write_all(&header_buf).unwrap();
        drain.write_all(&all_blocks).unwrap();
        // With the `hash` feature AND the content checksum enabled, the header
        // set `Content_Checksum_flag` and the 32-bit digest is written at the
        // end of the frame. Disabled => no trailing bytes, flag stays 0.
        #[cfg(feature = "hash")]
        if let Some(checksum_bytes) = checksum_bytes {
            drain.write_all(&checksum_bytes).unwrap();
        }
        #[cfg(feature = "lsm")]
        {
            let emit_checksum = cfg!(feature = "hash") && self.content_checksum;
            self.populate_frame_emit_info(header_buf.len(), &all_blocks, emit_checksum);
        }
    }

    /// Assemble the frame (header + blocks + optional checksum) into the
    /// caller-provided `out` buffer, replacing its contents, and populate
    /// `frame_emit_info` (lsm). `out` is cleared first (its allocation is
    /// reused, the CCtx-equivalent zero-per-call-alloc output path) then
    /// grown once to the exact frame size. Used by
    /// `compress_independent_frame_into`. The single `all_blocks` copy into
    /// `out` is the same one copy `finish_frame` performs writing
    /// `all_blocks` into a `Vec` drain, no extra buffering vs the drain
    /// path.
    /// Walk `all_blocks` to recover per-block layout and store it in
    /// `frame_emit_info`. Each Block_Header is 3 bytes LE packing
    /// `(block_size << 3) | (block_type << 1) | last_block`. Physical body
    /// size differs by type: RLE bodies are always 1 byte (the repeated
    /// byte), Raw/Compressed bodies span `block_size`. `header_len` is the
    /// serialized frame-header length (frame offset of the first block).
    #[cfg(feature = "lsm")]
    fn populate_frame_emit_info(
        &mut self,
        header_len: usize,
        all_blocks: &[u8],
        emit_checksum: bool,
    ) {
        use crate::blocks::block::BlockType as BT;
        use crate::encoding::frame_emit_info::{FrameBlock, FrameEmitInfo};
        // All frame-offset arithmetic below is bounded by u32 on the wire
        // (Block_Size is a 21-bit field, frames bounded by MAX_BLOCK_SIZE *
        // #blocks). A pathologically large frame whose total emitted size
        // exceeds u32::MAX would overflow the cast; bail out by leaving
        // `frame_emit_info` at `None` rather than handing the caller a
        // silently-truncated layout. The overflow path is statically
        // unreachable on every realistic frame so the predictor amortises
        // the branch to zero cost.
        let frame_header_len: u32 = match u32::try_from(header_len) {
            Ok(v) => v,
            Err(_) => return,
        };
        let all_blocks_len_u32: u32 = match u32::try_from(all_blocks.len()) {
            Ok(v) => v,
            Err(_) => return,
        };
        let mut blocks: Vec<FrameBlock> = Vec::new();
        let mut cursor: usize = 0;
        while cursor + 3 <= all_blocks.len() {
            let mut header_u32 = [0u8; 4];
            header_u32[..3].copy_from_slice(&all_blocks[cursor..cursor + 3]);
            let raw = u32::from_le_bytes(header_u32);
            let last_block = (raw & 1) != 0;
            let block_type = match (raw >> 1) & 0b11 {
                0 => BT::Raw,
                1 => BT::RLE,
                2 => BT::Compressed,
                _ => BT::Reserved,
            };
            let block_size_field = raw >> 3;
            // RLE bodies are always 1 byte physical on the wire (the single
            // repeated byte); the spec's Block_Size field carries the
            // logical repeat count. Raw and Compressed bodies physically
            // span block_size_field bytes. Store the physical length in
            // body_size so the 'offset + header + body_size' arithmetic
            // always lands on the next block boundary, and surface the raw
            // spec field separately as block_size_field.
            let physical_body: u32 = match block_type {
                BT::RLE => 1,
                _ => block_size_field,
            };
            let cursor_u32: u32 = match u32::try_from(cursor) {
                Ok(v) => v,
                Err(_) => return,
            };
            let offset_in_frame = match frame_header_len.checked_add(cursor_u32) {
                Some(v) => v,
                None => return,
            };
            // Decompressed (regenerated) size, captured per physical block
            // during emit (1:1 with the wire blocks scanned here). Raw/RLE are
            // wire-derivable (`block_size_field`), so a short sidecar still
            // yields the correct value for them. A Compressed block's size is
            // NOT on the wire: if the sidecar is missing its entry, fabricating
            // 0 would publish a silently-wrong `decompressed_byte_range`. Since
            // this metadata is the authoritative mapping for a successful
            // encode, bail out (leave `frame_emit_info` at `None`) rather than
            // hand back a corrupt layout; the 1:1 push invariant makes this
            // unreachable in practice (debug_assert catches a regression).
            let decompressed_size = match self.block_decompressed_sizes.get(blocks.len()).copied() {
                Some(size) => size,
                None if matches!(block_type, BT::Raw | BT::RLE) => block_size_field,
                None => {
                    debug_assert!(
                        false,
                        "missing decompressed-size sidecar entry for compressed block {}",
                        blocks.len()
                    );
                    return;
                }
            };
            blocks.push(FrameBlock {
                offset_in_frame,
                header_size: 3,
                body_size: physical_body,
                block_size_field,
                block_type,
                last_block,
                decompressed_size,
            });
            cursor += 3 + physical_body as usize;
            if last_block {
                break;
            }
        }
        // Fail closed on a structurally incomplete scan: the loop must have
        // consumed the whole block section AND ended on a parsed last block.
        // A premature `last_block` (bytes left over) or a run-off without any
        // last block would otherwise publish an invalid public `FrameEmitInfo`.
        // Unreachable for a well-formed self-produced frame (debug_assert
        // catches a regression); on release we bail, leaving `frame_emit_info`
        // at `None` rather than handing back a corrupt layout.
        if cursor != all_blocks.len() || !blocks.last().is_some_and(|b| b.last_block) {
            debug_assert!(
                false,
                "incomplete block scan in populate_frame_emit_info: cursor={} len={} last_block={:?}",
                cursor,
                all_blocks.len(),
                blocks.last().map(|b| b.last_block)
            );
            return;
        }
        let checksum_range = if emit_checksum {
            let cs_start = match frame_header_len.checked_add(all_blocks_len_u32) {
                Some(v) => v,
                None => return,
            };
            let cs_end = match cs_start.checked_add(4) {
                Some(v) => v,
                None => return,
            };
            Some(cs_start..cs_end)
        } else {
            None
        };
        let body_total = match frame_header_len.checked_add(all_blocks_len_u32) {
            Some(v) => v,
            None => return,
        };
        let total_size = if checksum_range.is_some() {
            match body_total.checked_add(4) {
                Some(v) => v,
                None => return,
            }
        } else {
            body_total
        };
        self.frame_emit_info = Some(FrameEmitInfo {
            frame_header_range: 0..frame_header_len,
            blocks,
            checksum_range,
            total_size,
        });
    }

    /// Layout of the most recently emitted frame.
    ///
    /// Returns `None` if [`compress`](Self::compress) has not been
    /// called yet on this compressor. After a successful `compress()`
    /// the returned `FrameEmitInfo` describes the frame header range,
    /// every emitted block's offset / size / type, and the optional
    /// trailing content-checksum range — all in frame-absolute byte
    /// offsets matching the bytes written to the drain.
    ///
    /// Behind the `lsm` Cargo feature.
    #[cfg(feature = "lsm")]
    pub fn last_frame_emit_info(&self) -> Option<&crate::encoding::frame_emit_info::FrameEmitInfo> {
        self.frame_emit_info.as_ref()
    }

    /// Opt in to per-block XXH64 checksum computation during
    /// [`compress`](Self::compress). Default off; zero cost when
    /// disabled. The captured digests are accessible via
    /// [`last_frame_block_checksums`](Self::last_frame_block_checksums).
    ///
    /// One checksum is emitted per physical FrameBlock written to
    /// the drain: 1:1 cardinality with
    /// [`last_frame_emit_info`](Self::last_frame_emit_info)'s
    /// `blocks` vector. On the post-split optimization path
    /// (Level 16-22 with large window) the per-partition decompressed
    /// range is hashed inside the partition loop so the digest count
    /// still matches the emitted block count. The decoder collects
    /// per-physical-block digests on the same granularity, so
    /// element-wise equality holds round-trip.
    ///
    /// Behind `all(feature = "lsm", feature = "hash")` — the XXH64
    /// primitive lives behind the `hash` feature, so this method only
    /// compiles when both are enabled.
    #[cfg(all(feature = "lsm", feature = "hash"))]
    pub fn enable_per_block_checksums(&mut self) {
        self.per_block_checksums_enabled = true;
    }

    /// Per-block XXH64 (low 32 bits) digests captured during the most
    /// recent `compress()` call. `None` unless
    /// [`enable_per_block_checksums`](Self::enable_per_block_checksums)
    /// was called before `compress()`.
    ///
    /// Behind `all(feature = "lsm", feature = "hash")`.
    #[cfg(all(feature = "lsm", feature = "hash"))]
    pub fn last_frame_block_checksums(&self) -> Option<&[u32]> {
        self.block_checksums.as_deref()
    }

    /// Get a mutable reference to the source
    pub fn source_mut(&mut self) -> Option<&mut R> {
        self.uncompressed_data.as_mut()
    }

    /// Get a mutable reference to the drain
    pub fn drain_mut(&mut self) -> Option<&mut W> {
        self.compressed_data.as_mut()
    }

    /// Get a reference to the source
    pub fn source(&self) -> Option<&R> {
        self.uncompressed_data.as_ref()
    }

    /// Get a reference to the drain
    pub fn drain(&self) -> Option<&W> {
        self.compressed_data.as_ref()
    }

    /// Retrieve the source
    pub fn take_source(&mut self) -> Option<R> {
        self.uncompressed_data.take()
    }

    /// Retrieve the drain
    pub fn take_drain(&mut self) -> Option<W> {
        self.compressed_data.take()
    }

    /// The matcher, for a caller that reads what it recorded between frames.
    #[cfg(feature = "dict-builder")]
    pub(crate) fn matcher_mut(&mut self) -> &mut M {
        &mut self.state.matcher
    }

    /// Whether the last frame's blocks coded offsets with the fast matcher's
    /// repeat policy, for a caller counting offset codes as they did.
    #[cfg(feature = "dict-builder")]
    pub(crate) fn uses_fast_offset_codes(&self) -> bool {
        crate::encoding::blocks::uses_fast_offset_codes(self.state.strategy_tag)
    }

    /// Before calling [FrameCompressor::compress] you can replace the matcher
    pub fn replace_matcher(&mut self, mut match_generator: M) -> M {
        core::mem::swap(&mut match_generator, &mut self.state.matcher);
        // The outgoing matcher may hold tables and history in this
        // compressor's workspace, which it must not outlive.
        match_generator.leave_workspace();
        match_generator
    }

    /// Before calling [FrameCompressor::compress] you can replace the compression level.
    ///
    /// This also clears any fine-grained parameter overrides installed via
    /// [`set_parameters`](Self::set_parameters): reverting to a bare level
    /// means plain level-based tuning, not the previous frame's customized
    /// strategy / LDM / log / literal-mode overrides. To keep overriding, call
    /// [`set_parameters`](Self::set_parameters) again with the new base level.
    pub fn set_compression_level(
        &mut self,
        compression_level: CompressionLevel,
    ) -> CompressionLevel {
        let old = self.compression_level;
        self.compression_level = compression_level;
        // Resync the raw-literals gate: negative levels disable literal (Huffman)
        // compression (C `ZSTD_literalsCompressionIsDisabled`). `prepare_frame`
        // never recomputes this, so it must be refreshed on the level switch the
        // same way the constructors and `set_parameters` do.
        self.state.literal_compression_disabled = matches!(
            compression_level,
            CompressionLevel::Level(n) if n < 0
        );
        // Drop sticky overrides so the level switch yields plain geometry,
        // the literal mode included: the gate above is the bare level's rule.
        self.tuning = FrameTuning::default();
        self.state.matcher.clear_param_overrides();
        old
    }

    /// Get the current compression level
    pub fn compression_level(&self) -> CompressionLevel {
        self.compression_level
    }

    /// Attach an already-parsed [`EncoderDictionary`] without reparsing a raw
    /// blob, held the way this compressor holds one (see [`DictionaryHold`]).
    ///
    /// Accepts an `EncoderDictionary` produced once via
    /// [`EncoderDictionary::from_bytes`] / [`EncoderDictionary::from_dictionary`]
    /// or handed back by [`Self::clear_dictionary`] / the `set_dictionary*`
    /// return value, so callers can reattach or reuse a prepared dictionary
    /// across compressions without re-running the dictionary parse each time.
    /// Returns the previously-attached dictionary, if any.
    ///
    /// # Errors
    /// [`ZeroRepeatOffsetInDictionary`](crate::decoding::errors::DictionaryDecodeError::ZeroRepeatOffsetInDictionary)
    /// when a repeat offset of the dictionary is zero.
    pub fn set_encoder_dictionary(
        &mut self,
        dictionary: D,
    ) -> Result<Option<D>, crate::decoding::errors::DictionaryDecodeError> {
        self.attach_dictionary(dictionary)
    }

    /// The dictionary frames are compressed with, if one is attached.
    ///
    /// # Examples
    /// ```
    /// use structured_zstd::encoding::{CompressionLevel, EncoderDictionary, FrameCompressor};
    ///
    /// let dictionary = EncoderDictionary::from_serialized_or_raw_content(b"some shared history").unwrap();
    /// let mut compressor: FrameCompressor = FrameCompressor::new(CompressionLevel::Default);
    /// assert!(compressor.dictionary().is_none());
    /// compressor.set_encoder_dictionary(dictionary).unwrap();
    /// assert_eq!(compressor.dictionary().map(EncoderDictionary::id), Some(0));
    /// ```
    pub fn dictionary(&self) -> Option<&EncoderDictionary> {
        attached(&self.dictionary)
    }

    /// Remove the attached dictionary, returning it as it was held.
    pub fn clear_dictionary(&mut self) -> Option<D> {
        // Drop the CDict prime snapshot — it is keyed to the dictionary
        // being removed and must not be restored against a different (or no)
        // dictionary on the next frame.
        self.state.matcher.invalidate_primed_dictionary();
        self.dictionary.take()
    }

    /// Validate `enc`, store it, and return the previously-attached
    /// dictionary; its entropy tables come with it, built when it was
    /// prepared. Shared by every public attach entry point: `set_dictionary`,
    /// `set_dictionary_from_bytes`, and `set_encoder_dictionary`.
    fn attach_dictionary(
        &mut self,
        enc: D,
    ) -> Result<Option<D>, crate::decoding::errors::DictionaryDecodeError> {
        // A zero id is not an error here: it marks a raw-content dictionary,
        // which has no header to carry one. The frame then records no
        // dictionary ID, so the decoder has to be handed the same bytes.
        let prepared: &EncoderDictionary = core::borrow::Borrow::borrow(&enc);
        let dictionary = &prepared.inner;
        if let Some(index) = dictionary.offset_hist.iter().position(|&rep| rep == 0) {
            return Err(
                crate::decoding::errors::DictionaryDecodeError::ZeroRepeatOffsetInDictionary {
                    index: index as u8,
                },
            );
        }
        // A previously-captured CDict prime snapshot belongs to the OLD
        // dictionary; drop it so the first frame with the new dictionary
        // re-primes (and re-captures) instead of restoring stale tables.
        self.state.matcher.invalidate_primed_dictionary();
        if !prepared.is_used_for_compression() {
            return Ok(self.dictionary.take());
        }
        Ok(self.dictionary.replace(enc))
    }
}

#[cfg(test)]
mod tests;