structured_zstd/decoding/

frame_decoder.rs

//! Framedecoder is the main low-level struct users interact with to decode zstd frames
//!
//! Zstandard compressed data is made of one or more frames. Each frame is independent and can be
//! decompressed independently of other frames. This module contains structures
//! and utilities that can be used to decode a frame.

use super::frame;
use crate::decoding;
use crate::decoding::block_decoder::BlockDecoder;
use crate::decoding::buffer_backend::BufferBackend;
use crate::decoding::decode_buffer::DecodeBuffer;
use crate::decoding::dictionary::{Dictionary, DictionaryHandle};
use crate::decoding::errors::{DecodeBlockContentError, FrameDecoderError};
use crate::decoding::flat_buf::FlatBuf;
use crate::decoding::ringbuffer::RingBuffer;
use crate::decoding::scratch::DecoderScratch;
use crate::io::{Error, Read, Write};
use alloc::collections::BTreeMap;
use alloc::vec::Vec;
use core::convert::TryInto;

use crate::common::MAXIMUM_ALLOWED_WINDOW_SIZE;

/// Low level Zstandard decoder that can be used to decompress frames with fine control over when and how many bytes are decoded.
///
/// This decoder is able to decode frames only partially and gives control
/// over how many bytes/blocks will be decoded at a time (so you don't have to decode a 10GB file into memory all at once).
/// It reads bytes as needed from a provided source and can be read from to collect partial results.
///
/// If you want to just read the whole frame with an `io::Read` without having to deal with manually calling [FrameDecoder::decode_blocks]
/// you can use the provided [crate::decoding::StreamingDecoder] wich wraps this FrameDecoder.
///
/// Workflow is as follows:
/// ```
/// use structured_zstd::decoding::BlockDecodingStrategy;
///
/// # #[cfg(feature = "std")]
/// use std::io::{Read, Write};
///
/// // no_std environments can use the crate's own Read traits
/// # #[cfg(not(feature = "std"))]
/// use structured_zstd::io::{Read, Write};
///
/// fn decode_this(mut file: impl Read) {
///     //Create a new decoder
///     let mut frame_dec = structured_zstd::decoding::FrameDecoder::new();
///     let mut result = Vec::new();
///
///     // Use reset or init to make the decoder ready to decode the frame from the io::Read
///     frame_dec.reset(&mut file).unwrap();
///
///     // Loop until the frame has been decoded completely
///     while !frame_dec.is_finished() {
///         // decode (roughly) batch_size many bytes
///         frame_dec.decode_blocks(&mut file, BlockDecodingStrategy::UptoBytes(1024)).unwrap();
///
///         // read from the decoder to collect bytes from the internal buffer
///         let bytes_read = frame_dec.read(result.as_mut_slice()).unwrap();
///
///         // then do something with it
///         do_something(&result[0..bytes_read]);
///     }
///
///     // handle the last chunk of data
///     while frame_dec.can_collect() > 0 {
///         let x = frame_dec.read(result.as_mut_slice()).unwrap();
///
///         do_something(&result[0..x]);
///     }
/// }
///
/// fn do_something(data: &[u8]) {
/// # #[cfg(feature = "std")]
///     std::io::stdout().write_all(data).unwrap();
/// }
/// ```
pub struct FrameDecoder {
    state: Option<FrameDecoderState>,
    owned_dicts: BTreeMap<u32, Dictionary>,
    #[cfg(target_has_atomic = "ptr")]
    shared_dicts: BTreeMap<u32, DictionaryHandle>,
    #[cfg(not(target_has_atomic = "ptr"))]
    shared_dicts: (),
    /// `ZSTD_f_zstd1_magicless` — when true, [`init`] / [`reset`]
    /// expect frames without the 4-byte magic number prefix.
    /// Default false (standard zstd format).
    magicless: bool,
    /// Pinned `Dictionary_ID` expectation set via
    /// [`Self::expect_dict_id`]. `None` (default) disables the
    /// check; `Some(0)` matches frames whose header omits the
    /// optional dict_id (treated as "no dictionary"). Validated in
    /// [`Self::reset`] AFTER the frame header parses successfully
    /// and BEFORE any block decode work.
    #[cfg(feature = "lsm")]
    expect_dict_id: Option<u32>,
    /// Pinned `Window_Descriptor` byte expectation set via
    /// [`Self::expect_window_descriptor`]. `None` (default)
    /// disables the check. Validated in [`Self::reset`] AFTER the
    /// frame header parses successfully and BEFORE any block
    /// decode work. Single-segment frames (which omit the
    /// `Window_Descriptor` byte from the wire) surface as
    /// [`crate::decoding::errors::FrameDecoderError::UnexpectedWindowDescriptor`]
    /// with `found: None`.
    #[cfg(feature = "lsm")]
    expect_window_descriptor: Option<u8>,
    /// When `true`, the per-block decode loop XXH64-hashes each
    /// block's decompressed bytes and stores the low-32-bit digest in
    /// [`Self::computed_block_checksums`]. Default `false` (zero
    /// cost). Set via [`Self::enable_per_block_checksums`]. Gated on
    /// `all(lsm, hash)` because XXH64 lives behind the `hash`
    /// feature.
    #[cfg(all(feature = "lsm", feature = "hash"))]
    per_block_checksums_enabled: bool,
    /// Per-block XXH64 (low 32 bits) digests captured during the
    /// current frame's decode when `per_block_checksums_enabled` is
    /// set. Reset at the start of every new frame. Gated on
    /// `all(lsm, hash)` (see `per_block_checksums_enabled`).
    #[cfg(all(feature = "lsm", feature = "hash"))]
    computed_block_checksums: alloc::vec::Vec<u32>,
}

/// Backend-tagged decode scratch — chosen at frame-reset time based
/// on the parsed `FrameHeader.descriptor.single_segment_flag()` and
/// kept stable through the lifetime of the frame. The match in each
/// helper below dispatches **once per call** (e.g. once per block in
/// `decode_block_content`, once per drain in `drain_to_writer`) —
/// never inside the hot push/repeat loop, which is fully
/// monomorphised through the `DecoderScratch<B>` generic.
enum DecoderScratchKind {
    Ring(DecoderScratch<RingBuffer>),
    Flat(DecoderScratch<FlatBuf>),
}

impl DecoderScratchKind {
    fn new_ring(window_size: usize) -> Self {
        // Lazy ring-buffer allocation: do NOT `reserve(window_size)` here.
        // The direct-decode path (`run_direct_decode`) writes through
        // `UserSliceBackend` and never touches the ring; allocating it
        // eagerly wastes one full window of peak memory on the common
        // direct-eligible frame. On the non-direct path the window is
        // pre-reserved once at frame entry (`decode_all_impl` and
        // `decode_blocks` both call `DecoderScratchKind::reserve_buffer`
        // before any block writes), so multi-block frames pay one
        // amortised grow instead of repeated `reserve_amortized` steps
        // per block. Issue #279 round 2.
        let s = DecoderScratch::<RingBuffer>::new(window_size);
        Self::Ring(s)
    }

    /// Construct a flat-backed scratch sized for a single-segment
    /// frame. `frame_content_size` is the upcoming output size in
    /// bytes (== `window_size` when the flag is set).
    fn new_flat(frame_content_size: usize) -> Self {
        let flat = FlatBuf::with_capacity(frame_content_size);
        // DecoderScratch's default ctor would discard the pre-sized
        // FlatBuf — go through from_backend so the buffer carries the
        // capacity the constructor wants.
        let mut s = DecoderScratch::<FlatBuf>::new(frame_content_size);
        s.buffer = DecodeBuffer::from_backend(flat, frame_content_size);
        Self::Flat(s)
    }

    /// Reset (or transition between) backends for a new frame.
    /// Reuses the existing `DecoderScratch` allocations (FSE / HUF
    /// tables, sequence vec, etc.) when the backend kind is unchanged
    /// — only the underlying buffer is re-sized for the new frame.
    /// Building a fresh `DecoderScratch` on every frame would
    /// re-allocate everything and was measured at +255 % vs ring on
    /// small frames; reusing it keeps the small-frame cost flat.
    fn reset(&mut self, frame: &frame::FrameHeader, window_size: usize) {
        if frame.descriptor.single_segment_flag() {
            match self {
                Self::Flat(s) => {
                    s.reset(window_size);
                    // DecodeBuffer::reset clears + reserves
                    // window_size; FlatBuf's reserve grows the
                    // backing Vec if the new FCS is larger than
                    // what's already allocated. No alloc when the
                    // previous flat frame had >= this capacity.
                }
                Self::Ring(_) => *self = Self::new_flat(window_size),
            }
        } else {
            match self {
                Self::Ring(s) => s.reset(window_size),
                Self::Flat(_) => *self = Self::new_ring(window_size),
            }
        }
    }

    fn init_from_dict(&mut self, dict: &Dictionary) {
        match self {
            Self::Ring(s) => s.init_from_dict(dict),
            Self::Flat(s) => s.init_from_dict(dict),
        }
    }

    #[inline]
    fn buffer_len(&self) -> usize {
        match self {
            Self::Ring(s) => s.buffer.len(),
            Self::Flat(s) => s.buffer.len(),
        }
    }

    /// Pre-reserve the backing buffer to `window_size` in a single
    /// allocation. Called once on the non-direct (`decode_blocks`) path
    /// after direct-eligibility is ruled out, so multi-segment fallback
    /// decodes don't pay repeated `reserve_amortized` grow steps
    /// (128 KiB → 256 KiB → ... → window) as blocks accumulate.
    ///
    /// Direct-eligible ring-backed frames never call this and pay zero
    /// ring allocation for the window. Flat-backed (single-segment)
    /// frames eagerly size their `FlatBuf` to `frame_content_size` at
    /// `new_flat` construction time, so a direct-eligible flat frame
    /// already paid for that capacity before this helper is reachable;
    /// the "zero buffer allocation" guarantee here applies to the
    /// ring path only.
    #[inline]
    fn reserve_buffer(&mut self, window_size: usize) {
        match self {
            Self::Ring(s) => s.buffer.reserve(window_size),
            Self::Flat(s) => s.buffer.reserve(window_size),
        }
    }

    /// Last `n` bytes of the visible buffer as `(s1, s2)` (wrap-aware).
    /// Routes through whichever backend the current scratch holds.
    #[cfg(all(feature = "lsm", feature = "hash"))]
    fn last_n_as_slices(&self, n: usize) -> (&[u8], &[u8]) {
        match self {
            Self::Ring(s) => s.buffer.last_n_as_slices(n),
            Self::Flat(s) => s.buffer.last_n_as_slices(n),
        }
    }

    fn buffer_drain(&mut self) -> Vec<u8> {
        match self {
            Self::Ring(s) => s.buffer.drain(),
            Self::Flat(s) => s.buffer.drain(),
        }
    }

    fn buffer_drain_to_window_size(&mut self) -> Option<Vec<u8>> {
        match self {
            Self::Ring(s) => s.buffer.drain_to_window_size(),
            Self::Flat(s) => s.buffer.drain_to_window_size(),
        }
    }

    fn buffer_drain_to_writer(&mut self, sink: impl Write) -> Result<usize, Error> {
        match self {
            Self::Ring(s) => s.buffer.drain_to_writer(sink),
            Self::Flat(s) => s.buffer.drain_to_writer(sink),
        }
    }

    fn buffer_drain_to_window_size_writer(&mut self, sink: impl Write) -> Result<usize, Error> {
        match self {
            Self::Ring(s) => s.buffer.drain_to_window_size_writer(sink),
            Self::Flat(s) => s.buffer.drain_to_window_size_writer(sink),
        }
    }

    fn buffer_can_drain(&self) -> usize {
        match self {
            Self::Ring(s) => s.buffer.can_drain(),
            Self::Flat(s) => s.buffer.can_drain(),
        }
    }

    fn buffer_can_drain_to_window_size(&self) -> Option<usize> {
        match self {
            Self::Ring(s) => s.buffer.can_drain_to_window_size(),
            Self::Flat(s) => s.buffer.can_drain_to_window_size(),
        }
    }

    fn buffer_read(&mut self, target: &mut [u8]) -> Result<usize, Error> {
        match self {
            Self::Ring(s) => s.buffer.read(target),
            Self::Flat(s) => s.buffer.read(target),
        }
    }

    fn buffer_read_all(&mut self, target: &mut [u8]) -> Result<usize, Error> {
        match self {
            Self::Ring(s) => s.buffer.read_all(target),
            Self::Flat(s) => s.buffer.read_all(target),
        }
    }

    fn decode_block_content<R: Read>(
        &mut self,
        decoder: &mut BlockDecoder,
        header: &crate::blocks::block::BlockHeader,
        source: R,
    ) -> Result<u64, DecodeBlockContentError> {
        match self {
            Self::Ring(s) => decoder.decode_block_content(header, s, source),
            Self::Flat(s) => decoder.decode_block_content(header, s, source),
        }
    }

    #[cfg(feature = "hash")]
    fn hash_finish(&self) -> u64 {
        use core::hash::Hasher;
        match self {
            Self::Ring(s) => s.buffer.hash.finish(),
            Self::Flat(s) => s.buffer.hash.finish(),
        }
    }
}

struct FrameDecoderState {
    pub frame_header: frame::FrameHeader,
    decoder_scratch: DecoderScratchKind,
    frame_finished: bool,
    block_counter: usize,
    bytes_read_counter: u64,
    check_sum: Option<u32>,
    using_dict: Option<u32>,
}

pub enum BlockDecodingStrategy {
    All,
    UptoBlocks(usize),
    UptoBytes(usize),
}

impl FrameDecoderState {
    /// Construct a new frame decoder state, reading the frame header
    /// from `source`. When `magicless` is `true`, the 4-byte magic
    /// number prefix is NOT consumed (donor `ZSTD_f_zstd1_magicless`).
    /// Crate-internal — reached only via `FrameDecoder::init` /
    /// `FrameDecoder::init_with_dict_handle`. For multi-segment
    /// (ring-backed) frames the decode buffer is allocated lazily —
    /// direct-eligible frames pay zero buffer allocation, and the
    /// non-direct fallback reserves `window_size` once in
    /// `decode_all_impl` via `reserve_buffer`. Single-segment
    /// (flat-backed) frames eagerly size the backing `FlatBuf` to
    /// `frame_content_size` because the flat path writes the entire
    /// output into the same buffer and cannot defer that allocation.
    pub(crate) fn new_with_format(
        source: impl Read,
        magicless: bool,
    ) -> Result<FrameDecoderState, FrameDecoderError> {
        let (frame, header_size) = frame::read_frame_header_with_format(source, magicless)?;
        let window_size = frame.window_size()?;

        if window_size > MAXIMUM_ALLOWED_WINDOW_SIZE {
            return Err(FrameDecoderError::WindowSizeTooBig {
                requested: window_size,
            });
        }

        let decoder_scratch = if frame.descriptor.single_segment_flag() {
            DecoderScratchKind::new_flat(window_size as usize)
        } else {
            DecoderScratchKind::new_ring(window_size as usize)
        };
        Ok(FrameDecoderState {
            frame_header: frame,
            frame_finished: false,
            block_counter: 0,
            decoder_scratch,
            bytes_read_counter: u64::from(header_size),
            check_sum: None,
            using_dict: None,
        })
    }

    /// Reset this state for a new frame read from `source`, reusing
    /// existing allocations. When `magicless` is `true`, the frame
    /// header is read WITHOUT expecting a magic-number prefix
    /// (donor `ZSTD_f_zstd1_magicless`). Crate-internal — reached
    /// only via `FrameDecoder::reset`.
    ///
    /// `DecodeBuffer::reset` no longer reserves window_size for either
    /// backend — capacity decisions live one layer up. For ring-backed
    /// scratch, direct-eligible frames pay zero allocation here and
    /// the non-direct path is pre-reserved by `decode_all_impl` /
    /// `decode_blocks` via `DecoderScratchKind::reserve_buffer(window_size)`
    /// before any block writes. Flat-backed scratch is sized to
    /// `frame_content_size` by `DecoderScratchKind::new_flat` at first
    /// construction (and on Ring → Flat transition); a reused flat
    /// scratch whose new frame fits within the prior FCS reuses the
    /// existing capacity, and one whose new FCS is larger is also
    /// pre-reserved by the same `reserve_buffer(window_size)` call at
    /// frame entry (single-segment frames have window_size == FCS).
    pub(crate) fn reset_with_format(
        &mut self,
        source: impl Read,
        magicless: bool,
    ) -> Result<(), FrameDecoderError> {
        let (frame_header, header_size) = frame::read_frame_header_with_format(source, magicless)?;
        let window_size = frame_header.window_size()?;

        if window_size > MAXIMUM_ALLOWED_WINDOW_SIZE {
            return Err(FrameDecoderError::WindowSizeTooBig {
                requested: window_size,
            });
        }

        self.decoder_scratch
            .reset(&frame_header, window_size as usize);
        self.frame_header = frame_header;
        self.frame_finished = false;
        self.block_counter = 0;
        self.bytes_read_counter = u64::from(header_size);
        self.check_sum = None;
        self.using_dict = None;
        Ok(())
    }
}

impl Default for FrameDecoder {
    fn default() -> Self {
        Self::new()
    }
}

impl FrameDecoder {
    /// This will create a new decoder without allocating anything yet.
    /// init()/reset() will allocate all needed buffers if it is the first time this decoder is used
    /// else they just reset these buffers with not further allocations
    pub fn new() -> FrameDecoder {
        FrameDecoder {
            state: None,
            owned_dicts: BTreeMap::new(),
            #[cfg(target_has_atomic = "ptr")]
            shared_dicts: BTreeMap::new(),
            #[cfg(not(target_has_atomic = "ptr"))]
            shared_dicts: (),
            magicless: false,
            #[cfg(feature = "lsm")]
            expect_dict_id: None,
            #[cfg(feature = "lsm")]
            expect_window_descriptor: None,
            #[cfg(all(feature = "lsm", feature = "hash"))]
            per_block_checksums_enabled: false,
            #[cfg(all(feature = "lsm", feature = "hash"))]
            computed_block_checksums: alloc::vec::Vec::new(),
        }
    }

    /// Opt in to per-block XXH64 verification during decode.
    /// Default off; zero cost when disabled. Each block's decompressed
    /// bytes are XXH64-hashed (low 32 bits) and appended to
    /// [`Self::computed_block_checksums`] as the decode progresses.
    /// Callers compare the captured digests against externally-stored
    /// expected values (e.g. from a per-block sidecar in the
    /// containing application protocol).
    ///
    /// 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
    /// current frame's decode. Empty unless
    /// [`Self::enable_per_block_checksums`] was called before
    /// [`Self::decode_all`] / [`Self::reset`].
    ///
    /// Reset at the start of every new frame.
    ///
    /// Behind `all(feature = "lsm", feature = "hash")`.
    #[cfg(all(feature = "lsm", feature = "hash"))]
    pub fn computed_block_checksums(&self) -> &[u32] {
        &self.computed_block_checksums
    }

    /// Pin the expected `Dictionary_ID` for the next frame.
    ///
    /// When `expected` is set, [`Self::init`] / [`Self::reset`]
    /// validate it against the parsed frame header BEFORE any
    /// block decode work runs. A mismatch returns
    /// [`crate::decoding::errors::FrameDecoderError::UnexpectedDictId`]
    /// before any block decode and before any output is produced.
    /// Scratch buffer allocation / reservation for the decode
    /// pipeline happens during frame-header parsing, which is
    /// already complete when this validation fires — the cost of
    /// scratch sizing is paid even on a mismatched header. The
    /// guarantee is "no block decode, no XXH64 init, no partial
    /// output", not "zero allocation".
    ///
    /// `Some(0)` is treated as "no dictionary expected": a frame
    /// whose header omits the optional `Dictionary_ID` field
    /// (flag value 0) passes the check; a frame that carries an
    /// explicit non-zero id fails.
    ///
    /// `None` (default) disables the check.
    ///
    /// Primary use case: post-AEAD-decrypt sanity check in
    /// wire-format consumers (e.g. lsm-tree's encrypted block
    /// format pins the `dict_id` baked into the AAD against the
    /// inner zstd frame's `dict_id` to defeat dict-substitution
    /// attacks).
    ///
    /// NOT a replacement for AEAD authentication. NOT the same
    /// semantic as donor `ZSTD_d_windowLogMax` (which is a
    /// ceiling-style limit, separate concern).
    #[cfg(feature = "lsm")]
    #[cfg_attr(docsrs, doc(cfg(feature = "lsm")))]
    pub fn expect_dict_id(&mut self, expected: Option<u32>) {
        self.expect_dict_id = expected;
    }

    /// Pin the expected raw `Window_Descriptor` byte (RFC 8878
    /// §3.1.1.1.2 layout: `(exp << 3) | mantissa`) for the next
    /// frame.
    ///
    /// When `expected` is set, [`Self::init`] / [`Self::reset`]
    /// validate it against the parsed frame header BEFORE any
    /// block decode work runs. A mismatch returns
    /// [`crate::decoding::errors::FrameDecoderError::UnexpectedWindowDescriptor`].
    ///
    /// Single-segment frames omit the `Window_Descriptor` byte
    /// from the wire entirely. Setting an expectation while
    /// receiving a single-segment frame fails the check with
    /// `found: None` — there is no on-wire byte to match against,
    /// which is reported explicitly rather than silently passing.
    ///
    /// `None` (default) disables the check.
    ///
    /// Byte-exact equality, NOT a ceiling. Donor
    /// `ZSTD_d_windowLogMax` is a separate ceiling-style limit
    /// available through the C FFI surface; this method is for
    /// strict equality validation against a pinned expectation
    /// (e.g. lsm-tree's wire format pins the window descriptor
    /// from the AAD to defeat decompression-bomb-swap attacks).
    #[cfg(feature = "lsm")]
    #[cfg_attr(docsrs, doc(cfg(feature = "lsm")))]
    pub fn expect_window_descriptor(&mut self, expected: Option<u8>) {
        self.expect_window_descriptor = expected;
    }

    /// Validate the just-parsed frame header against any pinned
    /// expectations set via [`Self::expect_dict_id`] /
    /// [`Self::expect_window_descriptor`].
    ///
    /// Returns the typed error variant on mismatch and leaves
    /// `self.state` in a re-resettable shape — a subsequent
    /// `reset()` will overwrite `frame_header` from the new source
    /// without needing intermediate cleanup.
    #[cfg(feature = "lsm")]
    fn validate_expectations(
        &self,
        frame_header: &frame::FrameHeader,
    ) -> Result<(), FrameDecoderError> {
        if let Some(expected) = self.expect_dict_id {
            let found = frame_header.dictionary_id();
            // `Some(0)` is the "no dictionary expected" sentinel —
            // matches a frame whose header omits the optional
            // dict_id field (which is reported as `None` by the
            // parser). All other values must match exactly.
            let matches = match (expected, found) {
                (0, None) => true,
                (e, Some(f)) => e == f,
                _ => false,
            };
            if !matches {
                return Err(FrameDecoderError::UnexpectedDictId {
                    expected: Some(expected),
                    found,
                });
            }
        }
        if let Some(expected) = self.expect_window_descriptor {
            let found = frame_header.window_descriptor();
            if found != Some(expected) {
                return Err(FrameDecoderError::UnexpectedWindowDescriptor { expected, found });
            }
        }
        Ok(())
    }

    /// Enable or disable magicless frame format
    /// (`ZSTD_f_zstd1_magicless`). When set to `true`, subsequent
    /// [`init`] / [`reset`] calls expect the frame header to begin
    /// directly with the frame-header descriptor — no 4-byte magic
    /// number prefix. Default false. Must match the encoder's
    /// magicless setting; the format is unambiguous only when the
    /// caller knows it out-of-band.
    ///
    /// Note: magicless mode also disables skippable-frame detection.
    /// The `0x184D2A50..=0x184D2A5F` skippable-frame magic range is
    /// only recognised when the 4-byte magic prefix is consumed, so
    /// `decode_all` / `init` / `reset` will treat a skippable frame
    /// at the head of a magicless stream as a malformed frame header
    /// (bad descriptor / window-size error) instead of skipping it.
    /// Mixed-format streams that interleave skippable frames must be
    /// pre-split by the caller; `set_magicless(true)` is only safe
    /// when the entire stream is known to be magicless zstd frames.
    pub fn set_magicless(&mut self, magicless: bool) {
        self.magicless = magicless;
    }

    #[cfg(target_has_atomic = "ptr")]
    fn shared_dict_exists(&self, dict_id: u32) -> bool {
        self.shared_dicts.contains_key(&dict_id)
    }

    #[cfg(not(target_has_atomic = "ptr"))]
    fn shared_dict_exists(&self, _dict_id: u32) -> bool {
        false
    }

    fn validate_registered_dictionary(dict: &Dictionary) -> Result<(), FrameDecoderError> {
        use crate::decoding::errors::DictionaryDecodeError as dict_err;

        if dict.id == 0 {
            return Err(FrameDecoderError::from(dict_err::ZeroDictionaryId));
        }
        if let Some(index) = dict.offset_hist.iter().position(|&rep| rep == 0) {
            return Err(FrameDecoderError::from(
                dict_err::ZeroRepeatOffsetInDictionary { index: index as u8 },
            ));
        }
        Ok(())
    }

    /// init() will allocate all needed buffers if it is the first time this decoder is used
    /// else they just reset these buffers with not further allocations
    ///
    /// Note that all bytes currently in the decodebuffer from any previous frame will be lost. Collect them with collect()/collect_to_writer()
    ///
    /// equivalent to reset()
    pub fn init(&mut self, source: impl Read) -> Result<(), FrameDecoderError> {
        self.reset(source)
    }

    /// Initialize the decoder for a new frame using a pre-parsed dictionary handle.
    ///
    /// If the frame header has a dictionary ID, this validates it against
    /// `dict.id()` and returns [`FrameDecoderError::DictIdMismatch`] on mismatch.
    ///
    /// If the header omits the optional dictionary ID, this still applies the
    /// provided dictionary handle.
    ///
    /// # Warning
    ///
    /// This method always applies `dict` unless the frame header contains a
    /// non-matching dictionary ID. Callers must only use this API when they
    /// already know the frame was encoded with the provided dictionary, even if
    /// the frame header omits the dictionary ID or encodes an explicit
    /// dictionary ID of `0`.
    ///
    /// Passing a dictionary for a frame that was not encoded with it can
    /// silently corrupt the decoded output.
    pub fn init_with_dict_handle(
        &mut self,
        source: impl Read,
        dict: &DictionaryHandle,
    ) -> Result<(), FrameDecoderError> {
        self.reset_with_dict_handle(source, dict)
    }

    /// reset() will allocate all needed buffers if it is the first time this decoder is used
    /// else they just reset these buffers with not further allocations
    ///
    /// Note that all bytes currently in the decodebuffer from any previous frame will be lost. Collect them with collect()/collect_to_writer()
    ///
    /// equivalent to init()
    pub fn reset(&mut self, source: impl Read) -> Result<(), FrameDecoderError> {
        use FrameDecoderError as err;
        // Fresh frame → start with an empty per-block checksum vec so
        // the values for the next frame don't carry over from the
        // previous one.
        #[cfg(all(feature = "lsm", feature = "hash"))]
        self.computed_block_checksums.clear();
        let magicless = self.magicless;
        let dict_id = match &mut self.state {
            Some(s) => {
                s.reset_with_format(source, magicless)?;
                s.frame_header.dictionary_id()
            }
            None => {
                self.state = Some(FrameDecoderState::new_with_format(source, magicless)?);
                self.state
                    .as_ref()
                    .and_then(|state| state.frame_header.dictionary_id())
            }
        };
        // Validate any pinned expectations BEFORE block decode work
        // runs. Catches dict_id substitution / window-descriptor
        // tampering on inputs already authenticated by an outer
        // layer (e.g. AEAD). Returning here leaves `self.state` in
        // a re-resettable shape — next `reset()` re-parses the
        // frame header without intermediate cleanup.
        #[cfg(feature = "lsm")]
        if let Some(state) = self.state.as_ref() {
            self.validate_expectations(&state.frame_header)?;
        }
        if let Some(dict_id) = dict_id {
            let state = self.state.as_mut().expect("state initialized");
            let owned_dicts = &self.owned_dicts;
            #[cfg(target_has_atomic = "ptr")]
            let shared_dicts = &self.shared_dicts;
            let dict = owned_dicts
                .get(&dict_id)
                .or_else(|| {
                    #[cfg(target_has_atomic = "ptr")]
                    {
                        shared_dicts.get(&dict_id).map(DictionaryHandle::as_dict)
                    }
                    #[cfg(not(target_has_atomic = "ptr"))]
                    {
                        None
                    }
                })
                .ok_or(err::DictNotProvided { dict_id })?;
            state.decoder_scratch.init_from_dict(dict);
            state.using_dict = Some(dict_id);
        }
        Ok(())
    }

    /// Reset this decoder for a new frame using a pre-parsed dictionary handle.
    ///
    /// If the frame header has a dictionary ID, this validates it against
    /// `dict.id()` and returns [`FrameDecoderError::DictIdMismatch`] on mismatch.
    ///
    /// If the header omits the optional dictionary ID, this still applies the
    /// provided dictionary handle.
    ///
    /// # Warning
    ///
    /// This method always applies `dict` unless the frame header contains a
    /// non-matching dictionary ID. Callers must only use this API when they
    /// already know the frame was encoded with the provided dictionary, even if
    /// the frame header omits the dictionary ID or encodes an explicit
    /// dictionary ID of `0`.
    ///
    /// Passing a dictionary for a frame that was not encoded with it can
    /// silently corrupt the decoded output.
    pub fn reset_with_dict_handle(
        &mut self,
        source: impl Read,
        dict: &DictionaryHandle,
    ) -> Result<(), FrameDecoderError> {
        use FrameDecoderError as err;
        // Fresh frame → drop the previous frame's per-block checksum
        // digests so the next decode starts with an empty vec.
        // Mirrors the same clear in `reset()`; reset_with_dict_handle
        // is a parallel entry point so it needs its own call.
        #[cfg(all(feature = "lsm", feature = "hash"))]
        self.computed_block_checksums.clear();
        Self::validate_registered_dictionary(dict.as_dict())?;
        let magicless = self.magicless;
        // Scope the &mut borrow of `self.state` to the header parse
        // alone, so the subsequent `validate_expectations(&self, ...)`
        // call below can take a fresh shared borrow of self without
        // tripping the borrow checker.
        match &mut self.state {
            Some(s) => s.reset_with_format(source, magicless)?,
            None => {
                self.state = Some(FrameDecoderState::new_with_format(source, magicless)?);
            }
        }
        // Single source of truth: route through the same
        // `validate_expectations` used by `reset()`. Routing through
        // the helper keeps the two code paths from drifting (e.g.,
        // if expect-semantics or error wiring changes later).
        #[cfg(feature = "lsm")]
        {
            let header = &self
                .state
                .as_ref()
                .expect("state populated by reset_with_format/new_with_format")
                .frame_header;
            self.validate_expectations(header)?;
        }
        let state = self
            .state
            .as_mut()
            .expect("state populated by reset_with_format/new_with_format");
        if let Some(dict_id) = state.frame_header.dictionary_id()
            && dict_id != dict.id()
        {
            return Err(err::DictIdMismatch {
                expected: dict_id,
                provided: dict.id(),
            });
        }
        state.decoder_scratch.init_from_dict(dict.as_dict());
        state.using_dict = Some(dict.id());
        Ok(())
    }

    /// Add a dictionary that can be selected dynamically by frame dictionary ID.
    ///
    /// Returns [`FrameDecoderError::DictAlreadyRegistered`] if the ID is already
    /// registered (either as owned or shared).
    pub fn add_dict(&mut self, dict: Dictionary) -> Result<(), FrameDecoderError> {
        Self::validate_registered_dictionary(&dict)?;
        let dict_id = dict.id;
        if self.owned_dicts.contains_key(&dict_id) || self.shared_dict_exists(dict_id) {
            return Err(FrameDecoderError::DictAlreadyRegistered { dict_id });
        }
        self.owned_dicts.insert(dict_id, dict);
        Ok(())
    }

    /// Parse and add a serialized dictionary blob.
    pub fn add_dict_from_bytes(&mut self, raw_dictionary: &[u8]) -> Result<(), FrameDecoderError> {
        let dict = Dictionary::decode_dict(raw_dictionary)?;
        self.add_dict(dict)
    }

    /// Add a pre-parsed dictionary handle for reuse across decoders.
    ///
    /// This API is available on targets with pointer-width atomics
    /// (`target_has_atomic = "ptr"`).
    ///
    /// Returns [`FrameDecoderError::DictAlreadyRegistered`] if the ID is already
    /// registered (either as owned or shared).
    #[cfg(target_has_atomic = "ptr")]
    pub fn add_dict_handle(&mut self, dict: DictionaryHandle) -> Result<(), FrameDecoderError> {
        Self::validate_registered_dictionary(dict.as_dict())?;
        let dict_id = dict.id();
        if self.owned_dicts.contains_key(&dict_id) || self.shared_dicts.contains_key(&dict_id) {
            return Err(FrameDecoderError::DictAlreadyRegistered { dict_id });
        }
        self.shared_dicts.insert(dict_id, dict);
        Ok(())
    }

    pub fn force_dict(&mut self, dict_id: u32) -> Result<(), FrameDecoderError> {
        use FrameDecoderError as err;
        let state = self.state.as_mut().ok_or(err::NotYetInitialized)?;
        let owned_dicts = &self.owned_dicts;
        #[cfg(target_has_atomic = "ptr")]
        let shared_dicts = &self.shared_dicts;

        let dict = owned_dicts
            .get(&dict_id)
            .or_else(|| {
                #[cfg(target_has_atomic = "ptr")]
                {
                    shared_dicts.get(&dict_id).map(DictionaryHandle::as_dict)
                }
                #[cfg(not(target_has_atomic = "ptr"))]
                {
                    None
                }
            })
            .ok_or(err::DictNotProvided { dict_id })?;
        state.decoder_scratch.init_from_dict(dict);
        state.using_dict = Some(dict_id);

        Ok(())
    }

    /// Returns how many bytes the frame contains after decompression
    pub fn content_size(&self) -> u64 {
        match &self.state {
            None => 0,
            Some(s) => s.frame_header.frame_content_size(),
        }
    }

    /// Returns the checksum that was read from the data. Only available after all bytes have been read. It is the last 4 bytes of a zstd-frame
    pub fn get_checksum_from_data(&self) -> Option<u32> {
        let state = self.state.as_ref()?;

        state.check_sum
    }

    /// Returns the checksum that was calculated while decoding.
    /// Only a sensible value after all decoded bytes have been collected/read from the FrameDecoder.
    /// Returns `None` when the frame header has `content_checksum_flag = 0`:
    /// no hash is computed for such frames (the post-decode XXH64 pass was a
    /// 63 % decode-wall hotspot on flag-off frames; skipping it when the
    /// frame format declares no trailing digest avoids that wasted work).
    #[cfg(feature = "hash")]
    pub fn get_calculated_checksum(&self) -> Option<u32> {
        let state = self.state.as_ref()?;
        if !state.frame_header.descriptor.content_checksum_flag() {
            return None;
        }
        let cksum_64bit = state.decoder_scratch.hash_finish();
        //truncate to lower 32bit because reasons...
        Some(cksum_64bit as u32)
    }

    /// Counter for how many bytes have been consumed while decoding the frame
    pub fn bytes_read_from_source(&self) -> u64 {
        let state = match &self.state {
            None => return 0,
            Some(s) => s,
        };
        state.bytes_read_counter
    }

    /// Whether the current frames last block has been decoded yet
    /// If this returns true you can call the drain* functions to get all content
    /// (the read() function will drain automatically if this returns true)
    pub fn is_finished(&self) -> bool {
        let state = match &self.state {
            None => return true,
            Some(s) => s,
        };
        if state.frame_header.descriptor.content_checksum_flag() {
            state.frame_finished && state.check_sum.is_some()
        } else {
            state.frame_finished
        }
    }

    /// Counter for how many blocks have already been decoded
    pub fn blocks_decoded(&self) -> usize {
        let state = match &self.state {
            None => return 0,
            Some(s) => s,
        };
        state.block_counter
    }

    /// Decodes blocks from a reader. It requires that the framedecoder has been initialized first.
    /// The Strategy influences how many blocks will be decoded before the function returns
    /// This is important if you want to manage memory consumption carefully. If you don't care
    /// about that you can just choose the strategy "All" and have all blocks of the frame decoded into the buffer
    pub fn decode_blocks(
        &mut self,
        mut source: impl Read,
        strat: BlockDecodingStrategy,
    ) -> Result<bool, FrameDecoderError> {
        use FrameDecoderError as err;
        let state = self.state.as_mut().ok_or(err::NotYetInitialized)?;

        // Streaming entry point: pre-reserve the backing buffer to
        // `window_size` so multi-block frames don't pay repeated
        // `reserve_amortized` grow steps (128 KiB → 256 KiB → ... →
        // window) as blocks accumulate. `decode_all` does the same up
        // front in `decode_all_impl`; this mirrors it for callers
        // driving `decode_blocks` directly. Idempotent — the
        // backend's `reserve` early-returns when capacity is already
        // sufficient.
        let window_size = state.frame_header.window_size().unwrap_or(0) as usize;
        state.decoder_scratch.reserve_buffer(window_size);

        let mut block_dec = decoding::block_decoder::new();

        let buffer_size_before = state.decoder_scratch.buffer_len();
        let block_counter_before = state.block_counter;
        loop {
            vprintln!("################");
            vprintln!("Next Block: {}", state.block_counter);
            vprintln!("################");
            let (block_header, block_header_size) = block_dec
                .read_block_header(&mut source)
                .map_err(err::FailedToReadBlockHeader)?;
            state.bytes_read_counter += u64::from(block_header_size);

            vprintln!();
            vprintln!(
                "Found {} block with size: {}, which will be of size: {}",
                block_header.block_type,
                block_header.content_size,
                block_header.decompressed_size
            );

            #[cfg(all(feature = "lsm", feature = "hash"))]
            let len_before_block: Option<usize> = if self.per_block_checksums_enabled {
                Some(state.decoder_scratch.buffer_len())
            } else {
                None
            };
            let bytes_read_in_block_body = state
                .decoder_scratch
                .decode_block_content(&mut block_dec, &block_header, &mut source)
                .map_err(err::FailedToReadBlockBody)?;
            state.bytes_read_counter += bytes_read_in_block_body;

            // Per-block XXH64 (low 32 bits) of the just-decompressed
            // bytes. Hashed from `last_n_as_slices` so RingBuffer wrap
            // is handled in-place, no extra copy.
            #[cfg(all(feature = "lsm", feature = "hash"))]
            if let Some(len_before_block) = len_before_block {
                let added = state.decoder_scratch.buffer_len() - len_before_block;
                let (s1, s2) = state.decoder_scratch.last_n_as_slices(added);
                let mut h = twox_hash::XxHash64::with_seed(0);
                use core::hash::Hasher;
                h.write(s1);
                h.write(s2);
                self.computed_block_checksums.push(h.finish() as u32);
            }

            state.block_counter += 1;

            vprintln!("Output: {}", state.decoder_scratch.buffer_len());

            if block_header.last_block {
                state.frame_finished = true;
                if state.frame_header.descriptor.content_checksum_flag() {
                    let mut chksum = [0u8; 4];
                    source
                        .read_exact(&mut chksum)
                        .map_err(err::FailedToReadChecksum)?;
                    state.bytes_read_counter += 4;
                    let chksum = u32::from_le_bytes(chksum);
                    state.check_sum = Some(chksum);
                }
                break;
            }

            match strat {
                BlockDecodingStrategy::All => { /* keep going */ }
                BlockDecodingStrategy::UptoBlocks(n) => {
                    if state.block_counter - block_counter_before >= n {
                        break;
                    }
                }
                BlockDecodingStrategy::UptoBytes(n) => {
                    if state.decoder_scratch.buffer_len() - buffer_size_before >= n {
                        break;
                    }
                }
            }
        }

        Ok(state.frame_finished)
    }

    /// Collect bytes and retain window_size bytes while decoding is still going on.
    /// After decoding of the frame (is_finished() == true) has finished it will collect all remaining bytes
    pub fn collect(&mut self) -> Option<Vec<u8>> {
        let finished = self.is_finished();
        let state = self.state.as_mut()?;
        if finished {
            Some(state.decoder_scratch.buffer_drain())
        } else {
            state.decoder_scratch.buffer_drain_to_window_size()
        }
    }

    /// Collect bytes and retain window_size bytes while decoding is still going on.
    /// After decoding of the frame (is_finished() == true) has finished it will collect all remaining bytes
    pub fn collect_to_writer(&mut self, w: impl Write) -> Result<usize, Error> {
        let finished = self.is_finished();
        let state = match &mut self.state {
            None => return Ok(0),
            Some(s) => s,
        };
        if finished {
            state.decoder_scratch.buffer_drain_to_writer(w)
        } else {
            state.decoder_scratch.buffer_drain_to_window_size_writer(w)
        }
    }

    /// How many bytes can currently be collected from the decodebuffer, while decoding is going on this will be lower than the actual decodbuffer size
    /// because window_size bytes need to be retained for decoding.
    /// After decoding of the frame (is_finished() == true) has finished it will report all remaining bytes
    pub fn can_collect(&self) -> usize {
        let finished = self.is_finished();
        let state = match &self.state {
            None => return 0,
            Some(s) => s,
        };
        if finished {
            state.decoder_scratch.buffer_can_drain()
        } else {
            state
                .decoder_scratch
                .buffer_can_drain_to_window_size()
                .unwrap_or(0)
        }
    }

    /// Decodes as many blocks as possible from the source slice and reads from the decodebuffer into the target slice
    /// The source slice may contain only parts of a frame but must contain at least one full block to make progress
    ///
    /// By all means use decode_blocks if you have a io.Reader available. This is just for compatibility with other decompressors
    /// which try to serve an old-style c api
    ///
    /// Returns (read, written), if read == 0 then the source did not contain a full block and further calls with the same
    /// input will not make any progress!
    ///
    /// Note that no kind of block can be bigger than 128kb.
    /// So to be safe use at least 128*1024 (max block content size) + 3 (block_header size) + 18 (max frame_header size) bytes as your source buffer
    ///
    /// You may call this function with an empty source after all bytes have been decoded. This is equivalent to just call decoder.read(&mut target)
    pub fn decode_from_to(
        &mut self,
        source: &[u8],
        target: &mut [u8],
    ) -> Result<(usize, usize), FrameDecoderError> {
        use FrameDecoderError as err;
        let bytes_read_at_start = match &self.state {
            Some(s) => s.bytes_read_counter,
            None => 0,
        };

        if !self.is_finished() || self.state.is_none() {
            let mut mt_source = source;

            if self.state.is_none() {
                self.init(&mut mt_source)?;
            }

            //pseudo block to scope "state" so we can borrow self again after the block
            {
                let state = match &mut self.state {
                    Some(s) => s,
                    None => panic!("Bug in library"),
                };
                let mut block_dec = decoding::block_decoder::new();

                if state.frame_header.descriptor.content_checksum_flag()
                    && state.frame_finished
                    && state.check_sum.is_none()
                {
                    //this block is needed if the checksum were the only 4 bytes that were not included in the last decode_from_to call for a frame
                    if mt_source.len() >= 4 {
                        let chksum = mt_source[..4].try_into().expect("optimized away");
                        state.bytes_read_counter += 4;
                        let chksum = u32::from_le_bytes(chksum);
                        state.check_sum = Some(chksum);
                    }
                    return Ok((4, 0));
                }

                loop {
                    //check if there are enough bytes for the next header
                    if mt_source.len() < 3 {
                        break;
                    }
                    let (block_header, block_header_size) = block_dec
                        .read_block_header(&mut mt_source)
                        .map_err(err::FailedToReadBlockHeader)?;

                    // check the needed size for the block before updating counters.
                    // If not enough bytes are in the source, the header will have to be read again, so act like we never read it in the first place
                    if mt_source.len() < block_header.content_size as usize {
                        break;
                    }
                    state.bytes_read_counter += u64::from(block_header_size);

                    let bytes_read_in_block_body = state
                        .decoder_scratch
                        .decode_block_content(&mut block_dec, &block_header, &mut mt_source)
                        .map_err(err::FailedToReadBlockBody)?;
                    state.bytes_read_counter += bytes_read_in_block_body;
                    state.block_counter += 1;

                    if block_header.last_block {
                        state.frame_finished = true;
                        if state.frame_header.descriptor.content_checksum_flag() {
                            //if there are enough bytes handle this here. Else the block at the start of this function will handle it at the next call
                            if mt_source.len() >= 4 {
                                let chksum = mt_source[..4].try_into().expect("optimized away");
                                state.bytes_read_counter += 4;
                                let chksum = u32::from_le_bytes(chksum);
                                state.check_sum = Some(chksum);
                            }
                        }
                        break;
                    }
                }
            }
        }

        let result_len = self.read(target).map_err(err::FailedToDrainDecodebuffer)?;
        let bytes_read_at_end = match &mut self.state {
            Some(s) => s.bytes_read_counter,
            None => panic!("Bug in library"),
        };
        let read_len = bytes_read_at_end - bytes_read_at_start;
        Ok((read_len as usize, result_len))
    }

    /// Decode multiple frames into the output slice.
    ///
    /// `input` must contain an exact number of frames. Skippable frames are allowed and will be
    /// skipped during decode.
    ///
    /// `output` must be large enough to hold the decompressed data. If you don't know
    /// how large the output will be, use [`FrameDecoder::decode_blocks`] instead.
    ///
    /// This calls [`FrameDecoder::init`], and all bytes currently in the decoder will be lost.
    ///
    /// Returns the number of bytes written to `output`.
    pub fn decode_all(
        &mut self,
        input: &[u8],
        output: &mut [u8],
    ) -> Result<usize, FrameDecoderError> {
        #[cfg(not(feature = "lsm"))]
        {
            self.decode_all_impl(input, output, |this, src| this.init(src))
        }
        #[cfg(feature = "lsm")]
        {
            self.decode_all_impl(input, output, |this, src| this.init(src), None)
        }
    }

    /// Decode multiple frames into the output slice, invoking `visitor`
    /// for every skippable frame encountered before advancing past it.
    ///
    /// `input` must contain an exact number of frames. Skippable frames
    /// (RFC 8878 §3.1.2 magic numbers `0x184D2A50..=0x184D2A5F`) are
    /// allowed and will be both visited AND skipped: the visitor gets
    /// `(magic_variant, payload)` where `magic_variant` is the low
    /// nibble of the magic (`magic - 0x184D2A50`, range `0..=15`) and
    /// `payload` is a borrowed slice of the on-wire payload bytes (the
    /// skippable frame's `Frame_Size` field worth of data) into
    /// `input` — no allocation.
    ///
    /// The visitor sees skippable frames in stream order; interleaved
    /// regular zstd frames continue to decompress into `output` exactly
    /// as `decode_all` does.
    ///
    /// `output` must be large enough to hold the decompressed data.
    /// Returns the number of bytes written to `output`.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use structured_zstd::decoding::FrameDecoder;
    ///
    /// let mut decoder = FrameDecoder::new();
    /// let mut output = vec![0u8; 1024];
    /// let mut collected: Vec<(u8, Vec<u8>)> = Vec::new();
    /// let n = decoder.decode_all_with_skippable_visitor(
    ///     input,
    ///     &mut output,
    ///     |variant, payload| collected.push((variant, payload.to_vec())),
    /// )?;
    /// ```
    #[cfg(feature = "lsm")]
    #[cfg_attr(docsrs, doc(cfg(feature = "lsm")))]
    pub fn decode_all_with_skippable_visitor<F>(
        &mut self,
        input: &[u8],
        output: &mut [u8],
        mut visitor: F,
    ) -> Result<usize, FrameDecoderError>
    where
        F: FnMut(u8, &[u8]),
    {
        self.decode_all_impl(
            input,
            output,
            |this, src| this.init(src),
            Some(&mut visitor),
        )
    }

    /// Decode multiple frames into the output slice using a pre-parsed dictionary handle.
    ///
    /// `input` must contain an exact number of frames. Skippable frames are allowed and will be
    /// skipped during decode.
    ///
    /// `output` must be large enough to hold the decompressed data. If you don't know
    /// how large the output will be, use [`FrameDecoder::decode_blocks`] instead.
    ///
    /// This calls [`FrameDecoder::init_with_dict_handle`], and all bytes currently in the
    /// decoder will be lost.
    ///
    /// # Warning
    ///
    /// Each decoded frame is initialized with `dict`, even when a frame header
    /// omits the optional dictionary ID. Callers must only use this API when
    /// they already know the input frames were encoded with the provided
    /// dictionary; otherwise decoded output can be silently corrupted.
    pub fn decode_all_with_dict_handle(
        &mut self,
        input: &[u8],
        output: &mut [u8],
        dict: &DictionaryHandle,
    ) -> Result<usize, FrameDecoderError> {
        #[cfg(not(feature = "lsm"))]
        {
            self.decode_all_impl(input, output, |this, src| {
                this.init_with_dict_handle(src, dict)
            })
        }
        #[cfg(feature = "lsm")]
        {
            self.decode_all_impl(
                input,
                output,
                |this, src| this.init_with_dict_handle(src, dict),
                None,
            )
        }
    }

    /// Default-feature decode_all_impl: no visitor parameter so the
    /// no-lsm build's call surface and codegen are byte-identical to
    /// the pre-#172 implementation. Compiles only when `lsm` is OFF.
    #[cfg(not(feature = "lsm"))]
    fn decode_all_impl(
        &mut self,
        mut input: &[u8],
        mut output: &mut [u8],
        mut init_frame: impl FnMut(&mut Self, &mut &[u8]) -> Result<(), FrameDecoderError>,
    ) -> Result<usize, FrameDecoderError> {
        use super::buffer_backend::WILDCOPY_OVERLENGTH;
        let mut total_bytes_written = 0;
        while !input.is_empty() {
            match init_frame(self, &mut input) {
                Ok(_) => {}
                Err(FrameDecoderError::ReadFrameHeaderError(
                    crate::decoding::errors::ReadFrameHeaderError::SkipFrame { length, .. },
                )) => {
                    input = input
                        .get(length as usize..)
                        .ok_or(FrameDecoderError::FailedToSkipFrame)?;
                    continue;
                }
                Err(e) => return Err(e),
            };
            // Per-frame direct-path dispatch. Now safe to route the
            // public `decode_all` here because
            // `UserSliceBackend::exec_sequence_inline` returns
            // `Result<(), ExecuteSequencesError>` instead of
            // panicking on capacity overflow; the error propagates
            // up as `FrameDecoderError`. Eligibility (FCS > 0, no
            // active dict, remaining `output` slice has WILDCOPY
            // slack) puts the frame on the fast path that bypasses
            // the FlatBuf/Ring -> `read()` drain copy. Ineligible
            // frames (no FCS, active dict, output too small for
            // slack) fall through to the legacy `decode_blocks` +
            // `read` drain loop below.
            let (content_size, fcs_declared, dict_active, window_size) = {
                let state_ref = self.state.as_ref().expect("init populated state");
                (
                    state_ref.frame_header.frame_content_size(),
                    state_ref.frame_header.fcs_declared(),
                    state_ref.using_dict.is_some(),
                    state_ref.frame_header.window_size().unwrap_or(0),
                )
            };
            let needed = content_size.saturating_add(WILDCOPY_OVERLENGTH as u64);
            // Per-block checksums collected inside `run_direct_decode`
            // post-loop (over recorded (start, end) ranges of `output`)
            // so the direct path stays eligible AND keeps the
            // window-size cap (`drop_to_window_size`) between blocks
            // that the spec relies on for `offset <= window_size`
            // validation. Path choice no longer alters checksum
            // semantics.
            let direct_eligible =
                content_size > 0 && !dict_active && (output.len() as u64) >= needed;
            if direct_eligible {
                let written = self.run_direct_decode(&mut input, output, content_size)?;
                output = &mut output[written..];
                total_bytes_written += written;
                continue;
            }
            // Non-direct fallback: pre-reserve the backing buffer to
            // `window_size` in a single allocation before block decode
            // starts, so multi-segment frames don't pay repeated
            // `reserve_amortized` grow steps as blocks accumulate (each
            // block only reserves MAX_BLOCK_SIZE = 128 KiB, so a window
            // > 128 KiB otherwise grows through several intermediate
            // sizes with `alloc_zeroed + memcpy` each time).
            if let Some(state) = self.state.as_mut() {
                state.decoder_scratch.reserve_buffer(window_size as usize);
            }
            let frame_start_total = total_bytes_written;
            loop {
                self.decode_blocks(&mut input, BlockDecodingStrategy::UptoBytes(1024 * 1024))?;
                let bytes_written = self
                    .read(output)
                    .map_err(FrameDecoderError::FailedToDrainDecodebuffer)?;
                output = &mut output[bytes_written..];
                total_bytes_written += bytes_written;
                if self.can_collect() != 0 {
                    return Err(FrameDecoderError::TargetTooSmall);
                }
                if self.is_finished() {
                    break;
                }
            }
            // Per-frame FCS validation on the legacy fallback path.
            // Use `fcs_declared()` (NOT `content_size > 0`) so an
            // empty frame with explicit FCS=0 on the wire still gets
            // validated.
            if fcs_declared {
                let produced = (total_bytes_written - frame_start_total) as u64;
                if produced != content_size {
                    return Err(FrameDecoderError::FrameContentSizeMismatch {
                        declared: content_size,
                        produced,
                    });
                }
            }
        }

        Ok(total_bytes_written)
    }

    /// `lsm`-feature decode_all_impl: adds the optional skippable
    /// visitor parameter consumed by
    /// [`Self::decode_all_with_skippable_visitor`]. Mirrors the no-lsm
    /// variant including the direct-path dispatch + FCS-validation
    /// rationale comments, so the two functions stay in sync; the only
    /// behavioral difference is the SkipFrame arm, which uses
    /// `split_at(length)` (single bounds check) instead of two
    /// separate `get(..length)` / `get(length..)` slices and invokes
    /// the visitor (when `Some`) on the borrowed payload before
    /// advancing past it.
    #[cfg(feature = "lsm")]
    #[allow(clippy::type_complexity)]
    fn decode_all_impl(
        &mut self,
        mut input: &[u8],
        mut output: &mut [u8],
        mut init_frame: impl FnMut(&mut Self, &mut &[u8]) -> Result<(), FrameDecoderError>,
        mut skippable_visitor: Option<&mut dyn FnMut(u8, &[u8])>,
    ) -> Result<usize, FrameDecoderError> {
        use super::buffer_backend::WILDCOPY_OVERLENGTH;
        let mut total_bytes_written = 0;
        while !input.is_empty() {
            match init_frame(self, &mut input) {
                Ok(_) => {}
                Err(FrameDecoderError::ReadFrameHeaderError(
                    crate::decoding::errors::ReadFrameHeaderError::SkipFrame {
                        magic_number,
                        length,
                    },
                )) => {
                    let length = length as usize;
                    // Visitor sees the payload slice BEFORE we advance
                    // past it. Borrowed slice — no allocation. The
                    // variant is the low nibble of the magic number
                    // (RFC 8878 §3.1.2). `read_frame_header` only emits
                    // SkipFrame for magic in 0x184D2A50..=0x184D2A5F, so
                    // the subtraction fits in 0..=15.
                    if input.len() < length {
                        return Err(FrameDecoderError::FailedToSkipFrame);
                    }
                    let (payload, rest) = input.split_at(length);
                    if let Some(visitor) = skippable_visitor.as_mut() {
                        let variant = (magic_number - 0x184D2A50) as u8;
                        visitor(variant, payload);
                    }
                    input = rest;
                    continue;
                }
                Err(e) => return Err(e),
            };
            // Per-frame direct-path dispatch. Now safe to route the
            // public `decode_all` here because
            // `UserSliceBackend::exec_sequence_inline` returns
            // `Result<(), ExecuteSequencesError>` instead of
            // panicking on capacity overflow; the error propagates
            // up as `FrameDecoderError`. Eligibility (FCS > 0, no
            // active dict, remaining `output` slice has WILDCOPY
            // slack) puts the frame on the fast path that bypasses
            // the FlatBuf/Ring -> `read()` drain copy. Ineligible
            // frames (no FCS, active dict, output too small for
            // slack) fall through to the legacy `decode_blocks` +
            // `read` drain loop below.
            let (content_size, fcs_declared, dict_active, window_size) = {
                let state_ref = self.state.as_ref().expect("init populated state");
                (
                    state_ref.frame_header.frame_content_size(),
                    state_ref.frame_header.fcs_declared(),
                    state_ref.using_dict.is_some(),
                    state_ref.frame_header.window_size().unwrap_or(0),
                )
            };
            let needed = content_size.saturating_add(WILDCOPY_OVERLENGTH as u64);
            let direct_eligible =
                content_size > 0 && !dict_active && (output.len() as u64) >= needed;
            if direct_eligible {
                let written = self.run_direct_decode(&mut input, output, content_size)?;
                output = &mut output[written..];
                total_bytes_written += written;
                continue;
            }
            // Non-direct fallback: pre-reserve the backing buffer to
            // `window_size` once so the per-block growth cycle is
            // skipped (see same comment on the no-lsm path above).
            if let Some(state) = self.state.as_mut() {
                state.decoder_scratch.reserve_buffer(window_size as usize);
            }
            let frame_start_total = total_bytes_written;
            loop {
                self.decode_blocks(&mut input, BlockDecodingStrategy::UptoBytes(1024 * 1024))?;
                let bytes_written = self
                    .read(output)
                    .map_err(FrameDecoderError::FailedToDrainDecodebuffer)?;
                output = &mut output[bytes_written..];
                total_bytes_written += bytes_written;
                if self.can_collect() != 0 {
                    return Err(FrameDecoderError::TargetTooSmall);
                }
                if self.is_finished() {
                    break;
                }
            }
            // Per-frame FCS validation on the legacy fallback path.
            // Use `fcs_declared()` (NOT `content_size > 0`) so an
            // empty frame with explicit FCS=0 on the wire still gets
            // validated.
            if fcs_declared {
                let produced = (total_bytes_written - frame_start_total) as u64;
                if produced != content_size {
                    return Err(FrameDecoderError::FrameContentSizeMismatch {
                        declared: content_size,
                        produced,
                    });
                }
            }
        }

        Ok(total_bytes_written)
    }

    /// Decode multiple frames into the output slice using a serialized dictionary.
    ///
    /// # Warning
    ///
    /// Each decoded frame is initialized with the parsed dictionary, even when a
    /// frame header omits the optional dictionary ID. Callers must only use this
    /// API when they already know the input frames were encoded with that
    /// dictionary; otherwise decoded output can be silently corrupted.
    pub fn decode_all_with_dict_bytes(
        &mut self,
        input: &[u8],
        output: &mut [u8],
        raw_dictionary: &[u8],
    ) -> Result<usize, FrameDecoderError> {
        let dict = DictionaryHandle::decode_dict(raw_dictionary)?;
        self.decode_all_with_dict_handle(input, output, &dict)
    }

    /// Decode multiple frames into the extra capacity of the output vector.
    ///
    /// `input` must contain an exact number of frames.
    ///
    /// `output` must have enough extra capacity to hold the decompressed data.
    /// This function reserves an additional [`WILDCOPY_OVERLENGTH`]
    /// bytes on top of the caller's capacity so the per-frame direct
    /// decode path stays eligible — that may grow the vector by up
    /// to that fixed amount via `Vec::reserve`. It will NOT grow
    /// further to fit the decompressed payload itself; the caller's
    /// pre-allocated capacity must already cover the data. If you
    /// don't know how large the output will be, use
    /// [`FrameDecoder::decode_blocks`] instead.
    ///
    /// This calls [`FrameDecoder::init`], and all bytes currently in the decoder will be lost.
    ///
    /// The length of the output vector is updated to include the decompressed data.
    /// The length is not changed if an error occurs. The
    /// `WILDCOPY_OVERLENGTH` slack is internal — `output.len()` on
    /// return is the actual decompressed size, NOT the inflated
    /// capacity. Callers who pre-sized the Vec with
    /// `Vec::with_capacity(fcs)` see no functional change beyond
    /// the small one-time capacity bump.
    pub fn decode_all_to_vec(
        &mut self,
        input: &[u8],
        output: &mut Vec<u8>,
    ) -> Result<(), FrameDecoderError> {
        use super::buffer_backend::WILDCOPY_OVERLENGTH;
        let len = output.len();
        // Reserve WILDCOPY slack on top of the caller's capacity so
        // `decode_all` can land on the direct path even when the
        // caller didn't pre-allocate slack themselves.
        output.reserve(WILDCOPY_OVERLENGTH);
        let cap = output.capacity();
        output.resize(cap, 0);
        match self.decode_all(input, &mut output[len..]) {
            Ok(bytes_written) => {
                let new_len = core::cmp::min(len + bytes_written, cap); // Sanitizes `bytes_written`.
                output.resize(new_len, 0);
                Ok(())
            }
            Err(e) => {
                output.resize(len, 0);
                Err(e)
            }
        }
    }

    /// Single-frame direct-decode path. Decodes one zstd frame into
    /// `output[..content_size]` via a stack-local
    /// `DecodeBuffer<UserSliceBackend>`, bypassing the per-block
    /// FlatBuf/Ring -> `read()` drain copy.
    ///
    /// # Preconditions (caller-enforced)
    ///
    /// - `self.init` (or `init_with_dict_handle`) was called for
    ///   this frame so `self.state` is populated.
    /// - `content_size` matches `self.state.frame_header
    ///   .frame_content_size()` and is `> 0` (caller already passed
    ///   the eligibility gate).
    /// - `output.len() >= content_size + WILDCOPY_OVERLENGTH`.
    /// - No active dictionary
    ///   (`self.state.using_dict.is_none()`).
    ///
    /// On return, `input` points at the byte immediately after the
    /// frame's checksum (or after the last block, when the frame
    /// has `content_checksum_flag = 0`). `self.state.frame_finished`
    /// is set so [`Self::is_finished`] reports `true`.
    fn run_direct_decode(
        &mut self,
        input: &mut &[u8],
        output: &mut [u8],
        content_size: u64,
    ) -> Result<usize, FrameDecoderError> {
        use super::block_decoder;
        use super::decode_buffer::DecodeBuffer;
        use super::scratch::DirectScratch;
        use super::user_slice_buf::UserSliceBackend;
        use crate::io::Read;
        use FrameDecoderError as err;

        let state = self
            .state
            .as_mut()
            .expect("caller ensures init populated state");

        // Borrow persistent fields out of whichever scratch variant
        // `init` produced (Flat for single_segment, Ring for
        // multi-segment) — both expose the same HUF/FSE/Vec
        // fields; only `buffer` differs and we don't use that here.
        // Macro-style binding avoids the closure / generic
        // gymnastics of returning multiple `&mut` from a match arm.
        let (huf, fse, offset_hist, literals_buffer, sequences, block_content_buffer, window_size) =
            match &mut state.decoder_scratch {
                DecoderScratchKind::Flat(s) => (
                    &mut s.huf,
                    &mut s.fse,
                    &mut s.offset_hist,
                    &mut s.literals_buffer,
                    &mut s.sequences,
                    &mut s.block_content_buffer,
                    s.buffer.window_size,
                ),
                DecoderScratchKind::Ring(s) => (
                    &mut s.huf,
                    &mut s.fse,
                    &mut s.offset_hist,
                    &mut s.literals_buffer,
                    &mut s.sequences,
                    &mut s.block_content_buffer,
                    s.buffer.window_size,
                ),
            };
        let backend = UserSliceBackend::from_slice(output);
        let buffer = DecodeBuffer::from_backend(backend, window_size);
        let mut direct = DirectScratch {
            huf,
            fse,
            offset_hist,
            literals_buffer,
            sequences,
            block_content_buffer,
            buffer,
        };

        // Block loop. Mirrors `decode_blocks` (without the
        // strategy-bounded early exit — we always decode the whole
        // frame in one shot for the direct path). Keeps
        // `state.bytes_read_counter` / `state.block_counter` in
        // sync with `decode_blocks` so post-call accessors
        // (`bytes_read_from_source`, `blocks_decoded`) return
        // accurate values.
        let mut block_dec = block_decoder::new();
        // Track total output bytes against the declared
        // `frame_content_size` via the buffer's actual write
        // counter — `BlockHeader.decompressed_size` is 0 for
        // Compressed blocks (the header parser can't know the
        // expanded size before decoding the body), so per-header
        // tracking would always count 0 for those blocks and
        // miscount frames that aren't pure Raw/RLE.
        let mut produced: u64 = 0;
        // Per-block output ranges captured during the direct-path
        // loop. After the loop we re-borrow `output` (post-drop of
        // `direct`) and XXH64 each range into
        // `self.computed_block_checksums`, so the digests vector
        // stays consistent with the legacy `decode_blocks` path
        // regardless of which dispatch the frame took.
        // `Vec::new()` does not allocate, so this stays free when
        // `per_block_checksums_enabled` is false: the `push` and the
        // post-loop hashing loop are both gated by the same flag.
        #[cfg(all(feature = "lsm", feature = "hash"))]
        let mut block_ranges: alloc::vec::Vec<(usize, usize)> = alloc::vec::Vec::new();
        loop {
            #[cfg(all(feature = "lsm", feature = "hash"))]
            let produced_before: Option<usize> = if self.per_block_checksums_enabled {
                Some(produced as usize)
            } else {
                None
            };
            let (block_header, hsize) = block_dec
                .read_block_header(&mut *input)
                .map_err(err::FailedToReadBlockHeader)?;
            state.bytes_read_counter += u64::from(hsize);
            // Pre-flight FCS check ONLY for Raw / RLE blocks where
            // `decompressed_size` is the actual block output size.
            // For Compressed blocks the header field is 0; the
            // post-decode check below catches overflow via the
            // backend's actual write counter delta.
            let block_upper = u64::from(block_header.decompressed_size);
            if block_upper > 0 && produced + block_upper > content_size {
                // Frame is corrupt — Raw/RLE block headers claim
                // more output than the FCS allows.
                return Err(err::FrameContentSizeMismatch {
                    declared: content_size,
                    produced: produced + block_upper,
                });
            }
            // Slice-source fast path: consume the block body
            // straight from `input` without copying into the
            // persistent `block_content_buffer`.
            let body_consumed = match block_dec.decode_block_content_from_slice(
                &block_header,
                &mut direct,
                &mut *input,
            ) {
                Ok(n) => n,
                // Defense-in-depth: RLE / Raw block whose declared
                // `decompressed_size` slipped past the per-block
                // pre-flight above and tripped the backend's
                // fallible write surface.
                Err(crate::decoding::errors::DecodeBlockContentError::BackendOverflow {
                    ..
                }) => {
                    // Use saturating_add on the
                    // `produced + decompressed_size` sum. Each block
                    // is bounded by 128 KiB (MAX_BLOCK_SIZE), but
                    // accumulated `produced` can grow toward
                    // u64::MAX across adversarial frames. Saturating
                    // avoids a panic on the error path itself.
                    return Err(err::FrameContentSizeMismatch {
                        declared: content_size,
                        produced: produced
                            .saturating_add(u64::from(block_header.decompressed_size)),
                    });
                }
                Err(e) => return Err(err::FailedToReadBlockBody(e)),
            };
            produced = direct.buffer.buffer_ref().tail() as u64;
            // Post-decode FCS overflow check.
            if produced > content_size {
                return Err(err::FrameContentSizeMismatch {
                    declared: content_size,
                    produced,
                });
            }
            state.bytes_read_counter += body_consumed;
            state.block_counter += 1;
            #[cfg(all(feature = "lsm", feature = "hash"))]
            if let Some(produced_before) = produced_before {
                block_ranges.push((produced_before, produced as usize));
            }
            // Cap the visible buffer at window_size between blocks
            // so the next block's match-offset validation matches
            // the spec's `offset <= window_size` rule.
            direct.buffer.drop_to_window_size();
            if block_header.last_block {
                if state.frame_header.descriptor.content_checksum_flag() {
                    let mut chksum = [0u8; 4];
                    input
                        .read_exact(&mut chksum)
                        .map_err(err::FailedToReadChecksum)?;
                    state.bytes_read_counter += 4;
                    state.check_sum = Some(u32::from_le_bytes(chksum));
                }
                break;
            }
        }
        // Final sanity: blocks summed to exactly `content_size`.
        if produced != content_size {
            return Err(err::FrameContentSizeMismatch {
                declared: content_size,
                produced,
            });
        }

        let written = content_size as usize;
        state.frame_finished = true;
        // Drop the stack-local DirectScratch (and its DecodeBuffer
        // borrow on `output`) so we can re-borrow `output` for the
        // hash pass below.
        drop(direct);
        // Per-block XXH64 (low 32 bits) over the captured ranges.
        // Mirrors `decode_blocks`' per-block hashing so the digests
        // vector stays identical regardless of which dispatch path
        // the frame took. Ranges were recorded inside the loop while
        // `direct` held a mutable borrow on `output`; now that the
        // borrow is dropped we can read the slices directly.
        #[cfg(all(feature = "lsm", feature = "hash"))]
        if self.per_block_checksums_enabled {
            use core::hash::Hasher;
            for (start, end) in &block_ranges {
                let mut h = twox_hash::XxHash64::with_seed(0);
                h.write(&output[*start..*end]);
                self.computed_block_checksums.push(h.finish() as u32);
            }
        }
        #[cfg(feature = "hash")]
        if state.frame_header.descriptor.content_checksum_flag() {
            // Direct path bypasses the per-write hash accounting
            // (DecodeBuffer hashes during drain; the direct path
            // never drains because the user slice IS the buffer).
            // Walk the decoded output once and propagate the
            // resulting hasher state so the frame-tail XXH64 check
            // (in the block loop above) can verify the digest.
            //
            // Gated on `content_checksum_flag`: frames without a
            // trailing checksum byte have nothing to verify, and the
            // 1 MiB+ second-pass scan dominated decode wall time
            // (63 % of total on z000033 L-3 in the standalone perf
            // loop). `get_calculated_checksum()` now returns `None`
            // for flag-off frames, matching this skip.
            use core::hash::Hasher;
            let mut hasher = twox_hash::XxHash64::with_seed(0);
            hasher.write(&output[..written]);
            match &mut state.decoder_scratch {
                DecoderScratchKind::Flat(s) => s.buffer.hash = hasher,
                DecoderScratchKind::Ring(s) => s.buffer.hash = hasher,
            }
        }
        Ok(written)
    }
}

/// Read bytes from the decode_buffer that are no longer needed. While the frame is not yet finished
/// this will retain window_size bytes, else it will drain it completely
impl Read for FrameDecoder {
    fn read(&mut self, target: &mut [u8]) -> Result<usize, Error> {
        let state = match &mut self.state {
            None => return Ok(0),
            Some(s) => s,
        };
        if state.frame_finished {
            state.decoder_scratch.buffer_read_all(target)
        } else {
            state.decoder_scratch.buffer_read(target)
        }
    }
}

#[cfg(test)]
mod tests {
    extern crate std;

    use super::{DictionaryHandle, FrameDecoder};
    use crate::encoding::{CompressionLevel, FrameCompressor};
    use alloc::vec::Vec;

    #[test]
    fn decode_all_legacy_drain_matches_direct_path_on_single_segment_frame() {
        // Roundtrip a small payload through the encoder, then decode
        // it via `decode_all` on two output shapes that select
        // different internal paths:
        //   1. Tight output (no WILDCOPY_OVERLENGTH slack) → legacy
        //      `decode_blocks` + `read()` drain path.
        //   2. Output with WILDCOPY slack → direct
        //      `run_direct_decode` + `UserSliceBackend` path.
        // Both paths must produce identical output bytes — the only
        // difference is the internal buffer/drain shape, not the
        // decoded semantics. This is the regression gate for the
        // direct-decode wiring.
        let payload: Vec<u8> = (0..4096u32).map(|i| (i & 0xFF) as u8).collect();
        let mut compressor = FrameCompressor::new(CompressionLevel::Default);
        compressor.set_source(payload.as_slice());
        let mut compressed = Vec::new();
        compressor.set_drain(&mut compressed);
        compressor.compress();

        // Baseline: tight output → legacy drain path.
        let mut dec_a = FrameDecoder::new();
        let mut out_a = alloc::vec![0u8; payload.len()];
        let n_a = dec_a
            .decode_all(compressed.as_slice(), &mut out_a)
            .expect("decode_all (legacy drain) should succeed");
        assert_eq!(n_a, payload.len());
        assert_eq!(&out_a[..n_a], payload.as_slice());

        // Direct: output with WILDCOPY slack → direct path.
        let slack = super::super::buffer_backend::WILDCOPY_OVERLENGTH;
        let mut dec_b = FrameDecoder::new();
        let mut out_b = alloc::vec![0u8; payload.len() + slack];
        let n_b = dec_b
            .decode_all(compressed.as_slice(), &mut out_b)
            .expect("decode_all (direct path) should succeed");
        assert_eq!(
            n_b,
            payload.len(),
            "direct decode produced wrong byte count"
        );
        assert_eq!(&out_b[..n_b], payload.as_slice());
    }

    #[test]
    fn decode_all_multi_segment_frame_decodes_correctly() {
        // Multi-segment frame: payload large enough that the
        // encoder's default frame layout has `single_segment_flag =
        // false` and `window_size < frame_content_size`. The direct
        // path must cap the visible buffer at window_size after each
        // block (drop_to_window_size) so match-offset validation
        // matches the spec rule `offset <= window_size`, and still
        // produce the same bytes as decode_all on the
        // FlatBuf/Ring-backed path.
        //
        // Make the payload structured so multi-segment behavior
        // actually kicks in: 2 MiB of repeating + random-ish bytes
        // forces window_size lower than content_size at the encoder.
        let mut payload: Vec<u8> = Vec::with_capacity(2 * 1024 * 1024);
        for i in 0..payload.capacity() {
            payload.push((i.wrapping_mul(2_654_435_761) & 0xFF) as u8);
        }
        let mut compressor = FrameCompressor::new(CompressionLevel::Default);
        compressor.set_source(payload.as_slice());
        let mut compressed = Vec::new();
        compressor.set_drain(&mut compressed);
        compressor.compress();

        // Baseline: decode_all through the FlatBuf+drain path.
        let mut dec_a = FrameDecoder::new();
        let mut out_a = alloc::vec![0u8; payload.len()];
        let n_a = dec_a
            .decode_all(compressed.as_slice(), &mut out_a)
            .expect("decode_all should succeed");
        assert_eq!(n_a, payload.len());
        assert_eq!(&out_a[..n_a], payload.as_slice());

        // Direct path: must give identical bytes via UserSliceBackend
        // + per-block drop_to_window_size.
        let slack = super::super::buffer_backend::WILDCOPY_OVERLENGTH;
        let mut dec_b = FrameDecoder::new();
        let mut out_b = alloc::vec![0u8; payload.len() + slack];
        let n_b = dec_b
            .decode_all(compressed.as_slice(), &mut out_b)
            .expect("decode_all should succeed on multi-segment frame");
        assert_eq!(n_b, payload.len(), "wrong byte count on direct path");
        assert_eq!(&out_b[..n_b], payload.as_slice());

        // Sanity-check: confirm the encoded frame really IS
        // multi-segment. If a future encoder default changes,
        // catching the assumption here is better than silently
        // testing single_segment on this name.
        let mut sanity = FrameDecoder::new();
        sanity.init(&mut compressed.as_slice()).unwrap();
        assert!(
            !sanity
                .state
                .as_ref()
                .unwrap()
                .frame_header
                .descriptor
                .single_segment_flag(),
            "test precondition violated: frame is single-segment, rename or resize"
        );
    }

    #[cfg(feature = "hash")]
    #[test]
    fn decode_all_propagates_checksum_into_persistent_scratch() {
        // Direct path on a checksum-flagged frame: the FrameCompressor
        // under `feature = "hash"` sets content_checksum_flag, so the
        // decoded frame has a recorded checksum. After
        // decode_all we must be able to verify it matches via
        // the public get_calculated_checksum() accessor — the digest
        // is computed by walking output at end of decode and stored
        // into the persistent scratch's hasher.
        let payload: Vec<u8> = (0..8192u32).map(|i| (i & 0xFF) as u8).collect();
        let mut compressor = FrameCompressor::new(CompressionLevel::Default);
        compressor.set_source(payload.as_slice());
        let mut compressed = Vec::new();
        compressor.set_drain(&mut compressed);
        compressor.compress();

        let slack = super::super::buffer_backend::WILDCOPY_OVERLENGTH;
        let mut dec = FrameDecoder::new();
        let mut out = alloc::vec![0u8; payload.len() + slack];
        let n = dec
            .decode_all(compressed.as_slice(), &mut out)
            .expect("decode_all with checksum must succeed");
        assert_eq!(n, payload.len());
        assert_eq!(&out[..n], payload.as_slice());

        // Both sides must report the same checksum: the frame header
        // carries the stored u32, and get_calculated_checksum reads
        // the running digest the direct path just propagated.
        let stored = dec.get_checksum_from_data();
        let calculated = dec.get_calculated_checksum();
        assert!(stored.is_some(), "frame must carry stored checksum");
        assert!(
            calculated.is_some(),
            "direct path must propagate calculated checksum"
        );
        assert_eq!(
            stored, calculated,
            "stored vs calculated checksum mismatch on direct path"
        );
    }

    #[test]
    fn decode_all_fcs_overflow_via_corrupt_frame_returns_structured_error() {
        // Hand-build a corrupt frame that declares
        // frame_content_size = 4 but the (last) block carries a
        // larger Raw payload. The pre-flight FCS check inside the
        // direct path's block loop catches this and returns the
        // structured FrameContentSizeMismatch variant — not a
        // panic, not a generic TargetTooSmall.
        //
        // Frame layout (single_segment, FCS=4):
        //   magic            4 bytes  0xFD2FB528
        //   FHD              1 byte   single_segment=1, no checksum,
        //                              FCS field size = 0 (-> 1-byte FCS)
        //   FCS              1 byte   0x04
        //   block_header     3 bytes  last=1, type=Raw, block_size=10
        //   block_payload    10 bytes 0xAA repeated
        let mut frame = alloc::vec::Vec::new();
        // magic
        frame.extend_from_slice(&0xFD2FB528u32.to_le_bytes());
        // FHD: single_segment=1, fcs_flag=0 (1-byte FCS), no checksum,
        // no dict. Bit layout: FCS(7-6)=0, single_segment(5)=1,
        // reserved/uncs(4)=0, content_checksum(2)=0, dict(0-1)=00.
        frame.push(0b0010_0000);
        // FCS: 1 byte
        frame.push(4);
        // Block header: cBlockSize=10, type=Raw (0), last=1
        // 3-byte LE: bit0=last, bits1-2=type(2 bits), bits3-23=size
        let cblock_size: u32 = 10;
        let bh: u32 = 1 | (cblock_size << 3); // last=1, type=Raw=0
        frame.push((bh & 0xFF) as u8);
        frame.push((bh >> 8) as u8);
        frame.push((bh >> 16) as u8);
        // Payload — 10 bytes that, if decoded, would exceed FCS=4.
        frame.extend(core::iter::repeat_n(0xAAu8, 10));

        let slack = super::super::buffer_backend::WILDCOPY_OVERLENGTH;
        let mut dec = FrameDecoder::new();
        let mut out = alloc::vec![0u8; 4 + slack];
        let err = dec
            .decode_all(&frame, &mut out)
            .expect_err("FCS-overflow frame must fail decode");
        assert!(
            matches!(
                err,
                super::FrameDecoderError::FrameContentSizeMismatch { .. }
            ),
            "expected FrameContentSizeMismatch, got {:?}",
            err
        );
    }

    #[test]
    fn decode_all_falls_back_when_output_too_small_for_wildcopy_slack() {
        // Output sized exactly to frame_content_size (no
        // WILDCOPY_OVERLENGTH slack) must NOT trigger the direct
        // path — the burst's `extend_from_within_unchecked` writes
        // past `tail` into the slack region. Direct dispatcher
        // recognises this and falls back to the FlatBuf + drain
        // path which still produces the right output.
        let payload: Vec<u8> = (0..2048u32)
            .map(|i| (i.wrapping_mul(31) & 0xFF) as u8)
            .collect();
        let mut compressor = FrameCompressor::new(CompressionLevel::Default);
        compressor.set_source(payload.as_slice());
        let mut compressed = Vec::new();
        compressor.set_drain(&mut compressed);
        compressor.compress();

        let mut dec = FrameDecoder::new();
        // Exactly payload.len(), no slack — direct path is gated out.
        let mut out = alloc::vec![0u8; payload.len()];
        let n = dec
            .decode_all(compressed.as_slice(), &mut out)
            .expect("decode_all should still succeed via fallback");
        assert_eq!(n, payload.len());
        assert_eq!(&out[..n], payload.as_slice());
    }

    #[test]
    fn decode_all_fallback_validates_fcs_against_total_output() {
        // Synthetic single-segment frame: FCS = 20 bytes, but the
        // last-block flag fires after only 4 bytes of raw payload.
        // On the direct path this would trip the post-block
        // `produced > content_size` check; the fallback path
        // (eligible=false because output is sized exactly to FCS,
        // no WILDCOPY slack) used to silently return Ok(4). With
        // the fix it now surfaces `FrameContentSizeMismatch`
        // matching the direct path.
        //
        // Frame layout: 4 B magic | 1 B FHD (single_segment=1,
        // FCS_flag=3 → 8-byte FCS) | 8 B FCS=20 | block header
        // (Raw, last, size=4) | 4 raw bytes.
        let mut wire = Vec::new();
        wire.extend_from_slice(&0xFD2F_B528u32.to_le_bytes()); // magic
        // FHD: FCS_flag=3 (8-byte FCS) <<6 | single_segment=1 <<5.
        wire.push(0b1110_0000);
        wire.extend_from_slice(&20u64.to_le_bytes()); // declared FCS
        // Block header: (size << 3) | (block_type << 1) | last_block.
        // Raw block (block_type=0), last_block=1, size=4 → 0b00100001 = 0x21.
        wire.push(0x21);
        wire.push(0x00);
        wire.push(0x00);
        wire.extend_from_slice(&[1u8, 2, 3, 4]);

        let mut dec = FrameDecoder::new();
        // Size output exactly at declared FCS (no WILDCOPY slack)
        // so the eligibility check gates the direct path out.
        let mut out = alloc::vec![0u8; 20];
        let err = dec
            .decode_all(wire.as_slice(), &mut out)
            .expect_err("fallback must reject corrupt FCS underflow");
        match err {
            crate::decoding::errors::FrameDecoderError::FrameContentSizeMismatch {
                declared,
                produced,
            } => {
                assert_eq!(declared, 20);
                assert_eq!(produced, 4);
            }
            other => panic!("expected FrameContentSizeMismatch, got {other:?}"),
        }
    }

    #[test]
    fn decode_all_fallback_treats_explicit_fcs_zero_as_declared() {
        // Synthetic multi-segment frame with FCS_flag=2 (4-byte
        // FCS) explicitly set to 0. The header DECLARES zero
        // content, but the body carries a 5-byte raw last-block.
        // `fcs_declared()` must return true (the field is on the
        // wire) so the fallback's post-decode size check sees the
        // mismatch — even though `frame_content_size == 0`. This
        // is exactly the FCS=0 edge case where the previous
        // `content_size > 0` proxy would have silently accepted
        // the corrupt frame.
        //
        // Frame layout:
        //   4 B magic            — 28 B5 2F FD
        //   1 B FHD              — FCS_flag=2 (bits 7-6), no
        //                          single_segment, content_checksum=0,
        //                          dict_id_flag=0 → 0b1000_0000
        //   1 B window_descriptor — exp=10, mantissa=0 → window=1 MiB
        //   4 B FCS              — 0 LE
        //   3 B block header     — raw, last, size=5 → 0x29 0x00 0x00
        //   5 B raw payload      — anything non-empty
        let mut wire = Vec::new();
        wire.extend_from_slice(&0xFD2F_B528u32.to_le_bytes());
        wire.push(0b1000_0000); // FHD: FCS_flag=2, others 0.
        wire.push(0x50); // window_descriptor: exp=10, mantissa=0.
        wire.extend_from_slice(&0u32.to_le_bytes()); // FCS = 0.
        // Block header (24-bit LE): (size << 3) | (block_type << 1) | last_block
        // = (5 << 3) | (0 << 1) | 1 = 0x29.
        wire.push(0x29);
        wire.push(0x00);
        wire.push(0x00);
        wire.extend_from_slice(&[1u8, 2, 3, 4, 5]);

        let mut dec = FrameDecoder::new();
        // FCS=0 declared, so eligibility (`content_size > 0`)
        // false — falls through to the drain loop. Output buffer
        // size doesn't matter for the eligibility check here;
        // give it some room so `read()` can drain the block.
        let mut out = alloc::vec![0u8; 16];
        let err = dec
            .decode_all(wire.as_slice(), &mut out)
            .expect_err("corrupt FCS=0 + 5-byte block must error");
        match err {
            crate::decoding::errors::FrameDecoderError::FrameContentSizeMismatch {
                declared,
                produced,
            } => {
                assert_eq!(declared, 0);
                assert_eq!(produced, 5);
            }
            other => panic!("expected FrameContentSizeMismatch, got {other:?}"),
        }
    }

    #[test]
    fn decode_all_fallback_accepts_honest_explicit_fcs_zero() {
        // Companion to the corrupt-FCS=0 test above: an HONEST
        // empty frame with FCS_flag=2 (4-byte FCS) explicitly set
        // to 0 AND a 0-byte raw last-block. `fcs_declared()`
        // returns true and `content_size == 0 == total_written`,
        // so the fallback validation accepts the frame instead of
        // misreporting a mismatch.
        //
        // (Single-segment FCS=0 would test a similar invariant
        // but trips header-stage validation: `window_size =
        // frame_content_size = 0 < MIN_WINDOW_SIZE` fails the
        // window-size sanity check before decode runs. Use the
        // multi-segment shape where `window_size` comes from
        // `window_descriptor` independently of FCS.)
        //
        // Frame layout:
        //   4 B magic
        //   1 B FHD              — FCS_flag=2, others 0 → 0x80
        //   1 B window_descriptor — exp=10 → 1 MiB window
        //   4 B FCS              — 0 LE
        //   3 B block header     — raw, last, size=0 → 0x01 0x00 0x00
        let mut wire = Vec::new();
        wire.extend_from_slice(&0xFD2F_B528u32.to_le_bytes());
        wire.push(0b1000_0000);
        wire.push(0x50);
        wire.extend_from_slice(&0u32.to_le_bytes());
        // Block header: (0 << 3) | (0 << 1) | 1 = 0x01.
        wire.push(0x01);
        wire.push(0x00);
        wire.push(0x00);

        let mut dec = FrameDecoder::new();
        let mut out = alloc::vec![0u8; 16];
        let n = dec
            .decode_all(wire.as_slice(), &mut out)
            .expect("honest FCS=0 + empty block must succeed");
        assert_eq!(n, 0);
    }

    #[test]
    fn reset_with_dict_handle_applies_dict_when_no_dict_id() {
        let payload = b"reset-without-dict-id";
        let mut compressor = FrameCompressor::new(CompressionLevel::Default);
        compressor.set_source(payload.as_slice());
        let mut compressed = Vec::new();
        compressor.set_drain(&mut compressed);
        compressor.compress();

        let dict_raw = include_bytes!("../../dict_tests/dictionary");
        let handle = DictionaryHandle::decode_dict(dict_raw).expect("dictionary should parse");

        let mut decoder = FrameDecoder::new();
        decoder
            .reset_with_dict_handle(compressed.as_slice(), &handle)
            .expect("reset should succeed");
        let state = decoder.state.as_ref().expect("state should be initialized");
        assert!(state.frame_header.dictionary_id().is_none());
        assert_eq!(state.using_dict, Some(handle.id()));
    }

    #[cfg(feature = "lsm")]
    mod expect_validation {
        use super::*;
        use crate::decoding::errors::FrameDecoderError;

        fn compress(payload: &[u8]) -> Vec<u8> {
            let mut compressor = FrameCompressor::new(CompressionLevel::Default);
            compressor.set_source(payload);
            let mut compressed = Vec::new();
            compressor.set_drain(&mut compressed);
            compressor.compress();
            compressed
        }

        fn compress_with_dict(payload: &[u8], dict_raw: &[u8]) -> Vec<u8> {
            let mut compressor = FrameCompressor::new(CompressionLevel::Default);
            compressor
                .set_dictionary_from_bytes(dict_raw)
                .expect("dict load");
            compressor.set_source(payload);
            let mut compressed = Vec::new();
            compressor.set_drain(&mut compressed);
            compressor.compress();
            compressed
        }

        #[test]
        fn expect_dict_id_none_default_allows_anything() {
            let compressed = compress(b"hello-no-expect");
            let mut decoder = FrameDecoder::new();
            decoder
                .reset(compressed.as_slice())
                .expect("default None passes");
        }

        #[test]
        fn expect_dict_id_zero_matches_frame_without_dict_id() {
            // Default-encoded frame has no dict_id; pinning Some(0)
            // ("no dictionary expected") must accept it.
            let compressed = compress(b"payload");
            let mut decoder = FrameDecoder::new();
            decoder.expect_dict_id(Some(0));
            decoder
                .reset(compressed.as_slice())
                .expect("Some(0) ~ None");
        }

        #[test]
        fn expect_dict_id_matching_value_passes() {
            let dict_raw = include_bytes!("../../dict_tests/dictionary");
            let handle = DictionaryHandle::decode_dict(dict_raw).expect("dict parse");
            let actual_id = handle.id();

            let compressed = compress_with_dict(b"payload-with-dict", dict_raw);

            let mut decoder = FrameDecoder::new();
            decoder.expect_dict_id(Some(actual_id));
            // Decode requires the dict to be registered; using
            // reset_with_dict_handle for that.
            decoder
                .reset_with_dict_handle(compressed.as_slice(), &handle)
                .expect("matching dict_id passes");
        }

        #[test]
        fn expect_dict_id_mismatching_value_fails_before_decode() {
            let dict_raw = include_bytes!("../../dict_tests/dictionary");
            let handle = DictionaryHandle::decode_dict(dict_raw).expect("dict parse");
            let actual_id = handle.id();
            let wrong_id = actual_id.wrapping_add(1);

            let compressed = compress_with_dict(b"payload-with-dict", dict_raw);

            let mut decoder = FrameDecoder::new();
            decoder.expect_dict_id(Some(wrong_id));
            let err = decoder
                .reset_with_dict_handle(compressed.as_slice(), &handle)
                .expect_err("mismatch must fail");
            match err {
                FrameDecoderError::UnexpectedDictId { expected, found } => {
                    assert_eq!(expected, Some(wrong_id));
                    assert_eq!(found, Some(actual_id));
                }
                other => panic!("expected UnexpectedDictId, got {other:?}"),
            }
        }

        #[test]
        fn expect_dict_id_nonzero_fails_on_frame_without_dict_id() {
            // Frame has no dict_id; expecting Some(42) (non-zero)
            // must fail with found = None.
            let compressed = compress(b"no-dict-frame");
            let mut decoder = FrameDecoder::new();
            decoder.expect_dict_id(Some(42));
            let err = decoder
                .reset(compressed.as_slice())
                .expect_err("nonzero expectation on dictless frame must fail");
            match err {
                FrameDecoderError::UnexpectedDictId { expected, found } => {
                    assert_eq!(expected, Some(42));
                    assert_eq!(found, None);
                }
                other => panic!("expected UnexpectedDictId, got {other:?}"),
            }
        }

        #[test]
        fn expect_window_descriptor_none_default_allows_anything() {
            let compressed = compress(b"hello-no-wd-expect");
            let mut decoder = FrameDecoder::new();
            decoder
                .reset(compressed.as_slice())
                .expect("default None passes");
        }

        #[test]
        fn expect_window_descriptor_mismatch_fails_before_decode() {
            // Compress a payload large enough to force a
            // multi-segment frame (window_descriptor on wire).
            // Default compression at >256 KiB produces multi-
            // segment frames with a real window_descriptor byte.
            let payload = alloc::vec![0xABu8; 512 * 1024];
            let compressed = compress(&payload);

            // Read the actual window_descriptor by decoding once
            // without expectations, then pin a wrong value.
            let mut probe_decoder = FrameDecoder::new();
            probe_decoder.reset(compressed.as_slice()).unwrap();
            let probe_state = probe_decoder.state.as_ref().unwrap();
            let actual_wd = probe_state
                .frame_header
                .window_descriptor()
                .expect("multi-segment frame should expose window_descriptor");
            let wrong_wd = actual_wd.wrapping_add(0x10); // bump exponent

            let mut decoder = FrameDecoder::new();
            decoder.expect_window_descriptor(Some(wrong_wd));
            let err = decoder
                .reset(compressed.as_slice())
                .expect_err("wrong window_descriptor must fail");
            match err {
                FrameDecoderError::UnexpectedWindowDescriptor { expected, found } => {
                    assert_eq!(expected, wrong_wd);
                    assert_eq!(found, Some(actual_wd));
                }
                other => panic!("expected UnexpectedWindowDescriptor, got {other:?}"),
            }
        }

        /// Build a minimal synthetic single-segment zstd frame
        /// carrying a 4-byte raw payload. RFC 8878 §3.1.1.1
        /// layout, hand-rolled because our default
        /// `FrameCompressor` settings don't emit
        /// `single_segment_flag` for tiny inputs.
        ///
        /// Wire bytes (13 total for 4-byte payload):
        /// ```text
        /// 28 B5 2F FD       magic
        /// 20                FHD: single_segment=1, FCS_flag=0
        /// 04                FCS (single byte, value = payload.len())
        /// 21 00 00          block header: raw, last, size=4
        /// .. .. .. ..       payload bytes
        /// ```
        fn synth_single_segment_frame(payload: &[u8]) -> Vec<u8> {
            assert!(payload.len() <= 255, "1-byte FCS field caps at 255");
            assert!(payload.len() < (1usize << 21), "block size 21-bit max");
            let mut out = Vec::new();
            // Magic 0xFD2FB528 LE.
            out.extend_from_slice(&0xFD2F_B528u32.to_le_bytes());
            // FHD: single_segment_flag (bit 5) set, everything
            // else zero. With single_segment + FCS_flag=0 the FCS
            // field is 1 byte. No window_descriptor on wire.
            out.push(0b0010_0000);
            // 1-byte FCS = payload length.
            out.push(payload.len() as u8);
            // Block header (3 bytes LE):
            // last_block=1, block_type=0 (Raw), block_size=payload.len().
            // Encoded: (size << 3) | (block_type << 1) | last_block.
            // Block header: last_block flag in bit 0, block_type
            // (0 = Raw) in bits 1-2, block size in bits 3+.
            let bh: u32 = ((payload.len() as u32) << 3) | 1;
            out.push((bh & 0xFF) as u8);
            out.push(((bh >> 8) & 0xFF) as u8);
            out.push(((bh >> 16) & 0xFF) as u8);
            // Raw payload.
            out.extend_from_slice(payload);
            out
        }

        #[test]
        fn expect_window_descriptor_on_single_segment_frame_fails_with_found_none() {
            // Single-segment frames omit the window_descriptor
            // byte from the wire entirely. Setting an expectation
            // here must surface `found: None` so callers
            // distinguish "wrong descriptor" from "no descriptor
            // on the wire" — never silently pass.
            let compressed = synth_single_segment_frame(b"tiny");

            // First sanity-check: the synthetic frame decodes
            // cleanly without any expectation.
            {
                let mut probe = FrameDecoder::new();
                probe
                    .reset(compressed.as_slice())
                    .expect("synth frame parses");
                let probe_state = probe.state.as_ref().unwrap();
                assert!(
                    probe_state.frame_header.window_descriptor().is_none(),
                    "synth frame must be single-segment"
                );
            }

            let mut decoder = FrameDecoder::new();
            decoder.expect_window_descriptor(Some(0x40));
            let err = decoder
                .reset(compressed.as_slice())
                .expect_err("single-segment + expectation must fail");
            match err {
                FrameDecoderError::UnexpectedWindowDescriptor { expected, found } => {
                    assert_eq!(expected, 0x40);
                    assert_eq!(found, None);
                }
                other => panic!("expected UnexpectedWindowDescriptor, got {other:?}"),
            }
        }

        #[test]
        fn validation_failure_leaves_decoder_re_resettable() {
            // After UnexpectedDictId on a wrong-expectation reset,
            // clearing the expectation and re-calling reset must
            // succeed on the same source — no lingering failed
            // state.
            let compressed = compress(b"re-resettable");

            let mut decoder = FrameDecoder::new();
            decoder.expect_dict_id(Some(42));
            let err = decoder
                .reset(compressed.as_slice())
                .expect_err("first reset fails");
            assert!(matches!(err, FrameDecoderError::UnexpectedDictId { .. }));

            // Clear expectation and retry on a fresh source.
            decoder.expect_dict_id(None);
            decoder
                .reset(compressed.as_slice())
                .expect("retry after clearing expectation should succeed");
        }
    }

    /// Build a skippable frame on the wire: 4-byte LE magic + 4-byte LE
    /// length + payload bytes. RFC 8878 §3.1.2 restricts the magic
    /// variant to `0..=15`; assert here so accidental misuse of the
    /// helper can't smuggle a non-skippable magic past the tests.
    #[cfg(feature = "lsm")]
    fn build_skippable_frame(variant: u8, payload: &[u8]) -> Vec<u8> {
        assert!(
            variant <= 15,
            "skippable-frame variant {variant} outside RFC 8878 0..=15 range",
        );
        let mut out = Vec::with_capacity(8 + payload.len());
        let magic: u32 = 0x184D2A50 + u32::from(variant);
        out.extend_from_slice(&magic.to_le_bytes());
        out.extend_from_slice(&u32::try_from(payload.len()).unwrap().to_le_bytes());
        out.extend_from_slice(payload);
        out
    }

    #[cfg(feature = "lsm")]
    #[test]
    fn decode_all_with_skippable_visitor_sees_payloads_in_order() {
        // Build a stream: skippable(v0, "alpha") + zstd_frame +
        // skippable(v3, "beta") + zstd_frame + skippable(v15, "")
        // and verify the visitor is invoked exactly three times with
        // the correct (variant, payload) pairs in stream order while
        // the zstd frames decode normally.
        let payload_a: Vec<u8> = (0..256u16).map(|i| i as u8).collect();
        let payload_b: Vec<u8> = (0..256u16).map(|i| (i ^ 0xAA) as u8).collect();

        let mut comp_a = Vec::new();
        let mut c = FrameCompressor::new(CompressionLevel::Default);
        c.set_source(payload_a.as_slice());
        c.set_drain(&mut comp_a);
        c.compress();

        let mut comp_b = Vec::new();
        let mut c = FrameCompressor::new(CompressionLevel::Default);
        c.set_source(payload_b.as_slice());
        c.set_drain(&mut comp_b);
        c.compress();

        let skip0 = build_skippable_frame(0, b"alpha");
        let skip3 = build_skippable_frame(3, b"beta");
        let skip15 = build_skippable_frame(15, &[]);

        let mut stream = Vec::new();
        stream.extend_from_slice(&skip0);
        stream.extend_from_slice(&comp_a);
        stream.extend_from_slice(&skip3);
        stream.extend_from_slice(&comp_b);
        stream.extend_from_slice(&skip15);

        let mut decoder = FrameDecoder::new();
        let mut out = alloc::vec![0u8; payload_a.len() + payload_b.len()];
        let mut collected: Vec<(u8, Vec<u8>)> = Vec::new();
        let n = decoder
            .decode_all_with_skippable_visitor(stream.as_slice(), &mut out, |variant, payload| {
                collected.push((variant, payload.to_vec()));
            })
            .expect("decode_all_with_skippable_visitor should succeed");

        // All three skippables visited in stream order.
        assert_eq!(collected.len(), 3);
        assert_eq!(collected[0], (0u8, b"alpha".to_vec()));
        assert_eq!(collected[1], (3u8, b"beta".to_vec()));
        assert_eq!(collected[2], (15u8, Vec::<u8>::new()));

        // Both zstd frames decoded into `out` back-to-back.
        assert_eq!(n, payload_a.len() + payload_b.len());
        assert_eq!(&out[..payload_a.len()], payload_a.as_slice());
        assert_eq!(&out[payload_a.len()..n], payload_b.as_slice());
    }

    #[cfg(feature = "lsm")]
    #[test]
    fn decode_all_silently_skips_when_no_visitor() {
        // Regression gate: plain decode_all must still silently skip
        // skippable frames (RFC 8878 mandated behavior) with no
        // behavioral change after the visitor refactor.
        let payload: Vec<u8> = (0..512u16).map(|i| i as u8).collect();
        let mut comp = Vec::new();
        let mut c = FrameCompressor::new(CompressionLevel::Default);
        c.set_source(payload.as_slice());
        c.set_drain(&mut comp);
        c.compress();

        let skip = build_skippable_frame(7, b"ignored sidecar");
        let mut stream = Vec::new();
        stream.extend_from_slice(&skip);
        stream.extend_from_slice(&comp);

        let mut decoder = FrameDecoder::new();
        let mut out = alloc::vec![0u8; payload.len()];
        let n = decoder
            .decode_all(stream.as_slice(), &mut out)
            .expect("decode_all should succeed on skippable + zstd stream");
        assert_eq!(n, payload.len());
        assert_eq!(&out[..n], payload.as_slice());
    }

    #[cfg(feature = "lsm")]
    #[test]
    fn frame_emit_info_describes_emitted_block_layout() {
        // Encode a payload large enough to force >1 block, fetch
        // FrameEmitInfo, walk blocks[] and verify each block's
        // (offset_in_frame, header_size, body_size) matches the bytes
        // actually emitted into the drain buffer.
        let payload: Vec<u8> = (0..200_000u32).map(|i| (i & 0xFF) as u8).collect();
        let mut compressor = FrameCompressor::new(CompressionLevel::Default);
        compressor.set_source(payload.as_slice());
        let mut compressed = Vec::new();
        compressor.set_drain(&mut compressed);
        compressor.compress();

        let info = compressor
            .last_frame_emit_info()
            .expect("last_frame_emit_info populated after compress")
            .clone();
        drop(compressor);

        // Frame header range starts at 0 and is non-empty.
        assert_eq!(info.frame_header_range.start, 0);
        assert!(info.frame_header_range.end > 0);
        // Total size matches what was written to the drain.
        assert_eq!(info.total_size as usize, compressed.len());
        // At least one block, and the last entry has last_block=true.
        assert!(!info.blocks.is_empty());
        assert!(info.blocks.last().unwrap().last_block);
        // All non-final blocks have last_block=false.
        for b in &info.blocks[..info.blocks.len() - 1] {
            assert!(!b.last_block);
        }
        // Walk and verify each block's header bytes match the
        // recorded type / size by re-decoding the 3-byte header.
        // Walking arithmetic: offset_in_frame + header_size + body_size
        // must land exactly on the next block's offset_in_frame (or,
        // for the last block, on the checksum / end of frame).
        for (i, b) in info.blocks.iter().enumerate() {
            let off = b.offset_in_frame as usize;
            assert_eq!(b.header_size, 3);
            let mut hdr = [0u8; 4];
            hdr[..3].copy_from_slice(&compressed[off..off + 3]);
            let raw = u32::from_le_bytes(hdr);
            let last = (raw & 1) != 0;
            let ty = (raw >> 1) & 0b11;
            let sz = raw >> 3;
            assert_eq!(last, b.last_block);
            assert_eq!(sz, b.block_size_field);
            // body_size is the PHYSICAL length on the wire: spec's
            // Block_Size for Raw/Compressed, always 1 for RLE.
            let expected_physical = match b.block_type {
                crate::encoding::frame_emit_info::BlockType::RLE => 1,
                _ => sz,
            };
            assert_eq!(b.body_size, expected_physical);
            let expected_ty = match b.block_type {
                crate::encoding::frame_emit_info::BlockType::Raw => 0,
                crate::encoding::frame_emit_info::BlockType::RLE => 1,
                crate::encoding::frame_emit_info::BlockType::Compressed => 2,
                crate::encoding::frame_emit_info::BlockType::Reserved => 3,
            };
            assert_eq!(ty, expected_ty);
            // Walking-arithmetic invariant.
            let next_off = b.offset_in_frame + b.header_size as u32 + b.body_size;
            if let Some(next) = info.blocks.get(i + 1) {
                assert_eq!(
                    next_off, next.offset_in_frame,
                    "block {i} body_size doesn't reach next block's offset_in_frame",
                );
            } else if let Some(cs) = info.checksum_range.as_ref() {
                assert_eq!(
                    next_off, cs.start,
                    "last block body_size doesn't reach checksum_range.start",
                );
            } else {
                assert_eq!(
                    next_off, info.total_size,
                    "last block body_size doesn't reach total_size",
                );
            }
        }
        // Checksum range present iff `feature = "hash"` is enabled.
        assert_eq!(info.checksum_range.is_some(), cfg!(feature = "hash"));
    }

    #[cfg(all(feature = "lsm", feature = "hash"))]
    #[test]
    fn per_block_checksum_round_trip() {
        // Encode with per-block checksums enabled. Decode with
        // per-block verification. Both sides emit exactly 1
        // checksum per physical block written to / read from the
        // wire (encoder hashes per emission site, including each
        // post-split partition; decoder hashes each decoded block).
        // Cardinality and element-wise contents must match
        // round-trip.
        let payload: Vec<u8> = (0..200_000u32).map(|i| (i & 0xFF) as u8).collect();
        let mut compressor = FrameCompressor::new(CompressionLevel::Default);
        compressor.set_source(payload.as_slice());
        compressor.enable_per_block_checksums();
        let mut compressed = Vec::new();
        compressor.set_drain(&mut compressed);
        compressor.compress();

        let encoder_checksums = compressor
            .last_frame_block_checksums()
            .expect("checksums populated after enable + compress")
            .to_vec();
        drop(compressor);
        assert!(!encoder_checksums.is_empty());

        // Decode side: enable verification, decode, compare.
        let mut decoder = FrameDecoder::new();
        decoder.enable_per_block_checksums();
        let mut output = alloc::vec![0u8; payload.len()];
        let n = decoder
            .decode_all(compressed.as_slice(), &mut output)
            .expect("decode_all should succeed");
        assert_eq!(n, payload.len());
        assert_eq!(&output[..n], payload.as_slice());

        let decoder_checksums = decoder.computed_block_checksums();
        assert_eq!(decoder_checksums, encoder_checksums.as_slice());
    }
}