structured_zstd/encoding/

streaming_encoder.rs

use alloc::format;
use alloc::string::{String, ToString};
use alloc::vec::Vec;
use core::borrow::BorrowMut;
use core::marker::PhantomData;

use crate::common::MAX_BLOCK_SIZE;
#[cfg(feature = "hash")]
use core::hash::Hasher;
#[cfg(feature = "hash")]
use twox_hash::XxHash64;

use crate::encoding::levels::compress_block_encoded;
use crate::encoding::{
    CompressionLevel, EncoderDictionary, MatchGeneratorDriver, Matcher, block_header::BlockHeader,
    frame_compressor::CompressState, frame_compressor::FrameTuning, frame_compressor::FseTables,
    frame_compressor::PreviousFseTable, frame_header::FrameHeader,
};
use crate::io::{Error, ErrorKind, Write};

/// Incremental frame encoder that implements [`Write`].
///
/// Data can be provided with multiple `write()` calls. Full blocks are compressed
/// automatically, `flush()` emits the currently buffered partial block as non-last,
/// and `finish()` closes the frame and returns the wrapped writer.
///
/// One encoder writes one frame into the drain it owns, through a
/// [`CompressionContext`] it owns ([`new`](StreamingEncoder::new)) or borrows
/// ([`with_context`](Self::with_context)). Borrowing is how frame after frame
/// is compressed with the same settings, dictionary and match-finder
/// allocations: the context outlives each encoder and is ready for the next
/// frame once [`finish`](Self::finish) returns.
pub struct StreamingEncoder<
    W: Write,
    M: Matcher = MatchGeneratorDriver,
    C: BorrowMut<CompressionContext<M>> = CompressionContext<M>,
> {
    drain: Option<W>,
    context: C,
    matcher: PhantomData<M>,
}

/// A reusable streaming compression context: the settings, the attached
/// dictionary, the match finder and its buffers, kept from one frame to the
/// next, with the output handed in on each call rather than owned. The
/// counterpart of upstream zstd's `ZSTD_CCtx` driven by `ZSTD_compressStream2`.
///
/// Settings apply from the next frame on and must be made before its first
/// [`write`](Self::write); [`finish_frame`](Self::finish_frame) closes the
/// frame and readies the context for another. A pledged size belongs to one
/// frame; every other setting, the dictionary included, stays until replaced.
/// A frame that fails leaves the context failed: every later call reports
/// that failure, and a new context is needed.
///
/// Reusing a context produces the same frames as a fresh
/// [`StreamingEncoder`] per frame, without rebuilding the match finder's
/// tables or re-attaching the dictionary for each of them.
///
/// # Examples
/// ```
/// use structured_zstd::encoding::{CompressionContext, CompressionLevel};
///
/// let mut context = CompressionContext::new(CompressionLevel::Default);
/// let mut frames = Vec::new();
/// for payload in [&b"first frame"[..], b"second frame"] {
///     let mut frame = Vec::new();
///     context.set_pledged_content_size(payload.len() as u64).unwrap();
///     context.write(&mut frame, payload).unwrap();
///     context.finish_frame(&mut frame).unwrap();
///     frames.push(frame);
/// }
/// use std::io::Read;
/// let mut decoder = structured_zstd::decoding::StreamingDecoder::new(&frames[1][..]);
/// let mut decoded = Vec::new();
/// decoder.read_to_end(&mut decoded).unwrap();
/// assert_eq!(decoded, b"second frame");
/// ```
pub struct CompressionContext<M: Matcher = MatchGeneratorDriver> {
    compression_level: CompressionLevel,
    state: CompressState<M>,
    /// The block being written out. An uncompressed frame also buffers its
    /// input here, behind the header, so a raw block is assembled where it
    /// is written from; a compressed frame buffers input in the match
    /// finder's history, uncommitted until its block is cut.
    encoded_scratch: Vec<u8>,
    errored: bool,
    last_error_kind: Option<ErrorKind>,
    last_error_message: Option<String>,
    frame_started: bool,
    /// Upper bound on emitted block sizes (upstream `ZSTD_c_targetCBlockSize`
    /// semantics; see `FrameCompressor::set_target_block_size`). `None` =
    /// the format's 128 KiB ceiling.
    target_block_size: Option<u32>,
    /// Largest block a frame may carry (upstream `ZSTD_c_maxBlockSize`); see
    /// [`Self::set_max_block_size`].
    max_block_size: Option<u32>,
    /// The pledged size of the frame in progress; cleared when it ends.
    pledged_content_size: Option<u64>,
    /// Advisory source-size hint from [`set_source_size_hint`](Self::set_source_size_hint).
    /// Unlike `pledged_content_size` it carries no end-of-frame enforcement, but
    /// it still feeds the small-input gates (matcher sizing AND the Fast HUF
    /// fast-path gate) so `set_source_size_hint(small)` reduces work the same way
    /// a pledge does. The HUF gate reads `pledged_content_size.or(source_size_hint)`.
    /// A parameter like upstream `ZSTD_c_srcSizeHint`, so it outlives a frame.
    source_size_hint: Option<u64>,
    /// Whether a pledged size is written into the header's
    /// `Frame_Content_Size` field (upstream `ZSTD_c_contentSizeFlag`).
    /// Pledge *enforcement* is independent of this flag — upstream
    /// validates consumed bytes against the pledge at frame end even
    /// when the header omits the field. Default `true`.
    content_size_flag: bool,
    bytes_consumed: u64,
    /// Upstream `ZSTD_compress_frameChunk` `savings`: bytes consumed minus
    /// bytes produced so far in this frame; the block pre-splitter only cuts
    /// full blocks once the frame has saved enough.
    savings: i64,
    tuning: FrameTuning,
    /// `ZSTD_f_zstd1_magicless` — omit the 4-byte magic number prefix.
    /// Default false. See [`Self::set_magicless`].
    magicless: bool,
    /// Whether to emit a trailing XXH64 content checksum and set the frame
    /// header's `Content_Checksum_flag` (upstream `ZSTD_c_checksumFlag`).
    /// Default `false`, matching the upstream library default; combined with
    /// the `hash` feature, so without `hash` no checksum is emitted
    /// regardless. See [`Self::set_content_checksum`].
    content_checksum: bool,
    /// Dictionary applied to each frame (upstream zstd `ZSTD_CCtx_loadDictionary`
    /// on a streaming context), with the entropy tables it seeds. `None` = no
    /// dictionary. Set before a frame's first write.
    dictionary: Option<EncoderDictionary>,
    /// Whether the frame header records the attached dictionary's ID
    /// (upstream `ZSTD_c_dictIDFlag`). Default `true`. Raw-content
    /// dictionaries (upstream `ZSTD_CCtx_refPrefix`) carry a synthetic
    /// non-zero ID that must not reach the wire, so their attach path
    /// turns this off. See [`Self::set_dictionary_id_flag`].
    dictionary_id_flag: bool,
    #[cfg(feature = "hash")]
    hasher: XxHash64,
}

impl<W: Write> StreamingEncoder<W, MatchGeneratorDriver> {
    /// Creates a streaming encoder backed by the default match generator.
    ///
    /// The encoder writes compressed bytes into `drain` and applies `compression_level`
    /// to all subsequently written blocks.
    pub fn new(drain: W, compression_level: CompressionLevel) -> Self {
        Self::with_context(drain, CompressionContext::new(compression_level))
    }
}

impl<W: Write, C: BorrowMut<CompressionContext>> StreamingEncoder<W, MatchGeneratorDriver, C> {
    /// Configure fine-grained compression parameters; see
    /// [`CompressionContext::set_parameters`]. Must be called before the first
    /// [`write`](Write::write).
    pub fn set_parameters(
        &mut self,
        params: &crate::encoding::CompressionParameters,
    ) -> Result<(), Error> {
        self.context.borrow_mut().set_parameters(params)
    }
}

impl<W: Write, M: Matcher> StreamingEncoder<W, M> {
    /// Creates a streaming encoder with an explicitly provided matcher implementation.
    ///
    /// This constructor is primarily intended for tests and advanced callers that need
    /// custom match-window behavior.
    pub fn new_with_matcher(matcher: M, drain: W, compression_level: CompressionLevel) -> Self {
        Self::with_context(
            drain,
            CompressionContext::new_with_matcher(matcher, compression_level),
        )
    }
}

impl<W: Write, M: Matcher, C: BorrowMut<CompressionContext<M>>> StreamingEncoder<W, M, C> {
    /// Write one frame into `drain` through `context`: owned, or borrowed from
    /// a caller that keeps it for the next frame with every setting, the
    /// dictionary and the allocations it has.
    ///
    /// # Examples
    /// ```
    /// use std::io::Write;
    /// use structured_zstd::encoding::{CompressionContext, CompressionLevel, StreamingEncoder};
    ///
    /// let mut context = CompressionContext::new(CompressionLevel::Default);
    /// for payload in [&b"first frame"[..], b"second frame"] {
    ///     let mut encoder = StreamingEncoder::with_context(Vec::new(), &mut context);
    ///     encoder.write_all(payload).unwrap();
    ///     let frame = encoder.finish().unwrap();
    ///     assert!(!frame.is_empty());
    /// }
    /// ```
    pub fn with_context(drain: W, context: C) -> Self {
        Self {
            drain: Some(drain),
            context,
            matcher: PhantomData,
        }
    }

    /// Bound each block's payload; see
    /// [`CompressionContext::set_target_block_size`]. Must be set before the
    /// first write.
    pub fn set_target_block_size(&mut self, target: Option<u32>) -> Result<(), Error> {
        self.context.borrow_mut().set_target_block_size(target)
    }

    /// Cap each block's size; see [`CompressionContext::set_max_block_size`].
    /// Must be set before the first write.
    pub fn set_max_block_size(&mut self, size: Option<u32>) -> Result<(), Error> {
        self.context.borrow_mut().set_max_block_size(size)
    }

    /// Enable or disable the trailing XXH64 content checksum; see
    /// [`CompressionContext::set_content_checksum`]. Must be called before the
    /// first write.
    pub fn set_content_checksum(&mut self, emit: bool) -> Result<(), Error> {
        self.context.borrow_mut().set_content_checksum(emit)
    }

    /// Enable or disable the magicless frame format; see
    /// [`CompressionContext::set_magicless`]. Must be called before the first
    /// write.
    pub fn set_magicless(&mut self, magicless: bool) -> Result<(), Error> {
        self.context.borrow_mut().set_magicless(magicless)
    }

    /// Pledge the total uncompressed content size of the frame; see
    /// [`CompressionContext::set_pledged_content_size`]. Must be called before
    /// the first write.
    pub fn set_pledged_content_size(&mut self, size: u64) -> Result<(), Error> {
        self.context.borrow_mut().set_pledged_content_size(size)
    }

    /// Control whether a pledged size reaches the header; see
    /// [`CompressionContext::set_content_size_flag`]. Must be called before
    /// the first write.
    pub fn set_content_size_flag(&mut self, emit: bool) -> Result<(), Error> {
        self.context.borrow_mut().set_content_size_flag(emit)
    }

    /// Provide an advisory size for the frame; see
    /// [`CompressionContext::set_source_size_hint`]. Must be called before the
    /// first write.
    pub fn set_source_size_hint(&mut self, size: u64) -> Result<(), Error> {
        self.context.borrow_mut().set_source_size_hint(size)
    }

    /// Attach a dictionary blob to the frame; see
    /// [`CompressionContext::set_dictionary_from_bytes`]. Must be called before
    /// the first write.
    pub fn set_dictionary_from_bytes(&mut self, raw_dictionary: &[u8]) -> Result<(), Error> {
        self.context
            .borrow_mut()
            .set_dictionary_from_bytes(raw_dictionary)
    }

    /// Whether the header records the dictionary ID; see
    /// [`CompressionContext::set_dictionary_id_flag`]. Must be set before the
    /// first write.
    pub fn set_dictionary_id_flag(&mut self, emit: bool) -> Result<(), Error> {
        self.context.borrow_mut().set_dictionary_id_flag(emit)
    }

    /// Attach an already-parsed [`EncoderDictionary`] to the frame; see
    /// [`CompressionContext::set_encoder_dictionary`]. Must be called before
    /// the first write.
    pub fn set_encoder_dictionary(&mut self, dict: EncoderDictionary) -> Result<(), Error> {
        self.context.borrow_mut().set_encoder_dictionary(dict)
    }

    /// Returns an immutable reference to the wrapped output drain.
    ///
    /// The drain remains available for the encoder lifetime; [`finish`](Self::finish)
    /// consumes the encoder and returns ownership of the drain.
    pub fn get_ref(&self) -> &W {
        self.drain
            .as_ref()
            .expect("streaming encoder drain is present until finish consumes self")
    }

    /// Total heap bytes this encoder's allocations hold, excluding the inline
    /// struct and the drain `W` (whose footprint the owner can measure through
    /// [`get_ref`](Self::get_ref)); see [`CompressionContext::heap_size`].
    pub fn heap_size(&self) -> usize {
        self.context.borrow().heap_size()
    }

    /// Returns a mutable reference to the wrapped output drain.
    ///
    /// It is inadvisable to directly write to the underlying writer, as doing
    /// so would corrupt the zstd frame being assembled by the encoder.
    ///
    /// The drain remains available for the encoder lifetime; [`finish`](Self::finish)
    /// consumes the encoder and returns ownership of the drain.
    pub fn get_mut(&mut self) -> &mut W {
        self.drain
            .as_mut()
            .expect("streaming encoder drain is present until finish consumes self")
    }

    /// Finalizes the current zstd frame and returns the wrapped output drain.
    ///
    /// If no payload was written yet, this still emits a valid empty frame.
    /// Calling this method consumes the encoder; a borrowed context is then
    /// ready for the next frame, also when this fails: the frame goes with the
    /// drain (see [`CompressionContext::abandon_frame`]).
    pub fn finish(mut self) -> Result<W, Error> {
        let mut drain = self
            .drain
            .take()
            .expect("streaming encoder drain must be present when finishing");
        self.context.borrow_mut().finish_frame(&mut drain)?;
        Ok(drain)
    }

    fn drain_mut(&mut self) -> (&mut W, &mut CompressionContext<M>) {
        let drain = self
            .drain
            .as_mut()
            .expect("streaming encoder drain is present until finish consumes self");
        (drain, self.context.borrow_mut())
    }
}

/// The frame belongs to the drain this encoder writes into: an encoder that
/// goes without finishing it (dropped mid-frame, or a `finish` that failed)
/// takes it along, so a borrowed context starts the next encoder's frame
/// afresh instead of continuing this one into another drain.
impl<W: Write, M: Matcher, C: BorrowMut<CompressionContext<M>>> Drop for StreamingEncoder<W, M, C> {
    fn drop(&mut self) {
        self.context.borrow_mut().abandon_frame();
    }
}

impl<W: Write, M: Matcher, C: BorrowMut<CompressionContext<M>>> Write
    for StreamingEncoder<W, M, C>
{
    fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
        let (drain, context) = self.drain_mut();
        context.write(drain, buf)
    }

    fn flush(&mut self) -> Result<(), Error> {
        let (drain, context) = self.drain_mut();
        context.flush(drain)
    }
}

impl CompressionContext<MatchGeneratorDriver> {
    /// Creates a context backed by the default match generator, compressing
    /// at `compression_level`.
    pub fn new(compression_level: CompressionLevel) -> Self {
        Self::new_with_matcher(
            MatchGeneratorDriver::new(MAX_BLOCK_SIZE as usize, 1),
            compression_level,
        )
    }

    /// Configure fine-grained compression parameters (#27): resets the level to
    /// the parameters' level and installs the per-knob overrides (window / hash
    /// / chain / search logs, strategy, long-distance matching) applied at the
    /// next frame. Mirrors [`FrameCompressor::set_parameters`](crate::encoding::FrameCompressor::set_parameters).
    /// Must be called before the frame's first [`write`](Self::write). Only the
    /// built-in `MatchGeneratorDriver` exposes the override knobs, so this
    /// lives on the default-matcher impl.
    pub fn set_parameters(
        &mut self,
        params: &crate::encoding::CompressionParameters,
    ) -> Result<(), Error> {
        self.ensure_settable("compression parameters must be set before the first write")?;
        self.compression_level = params.level();
        let overrides = params.overrides();
        // Persist the strategy override so `ensure_frame_started`'s level-based
        // resync does not discard it (matching `FrameCompressor::set_parameters`).
        self.tuning = FrameTuning::from_overrides(&overrides);
        self.state.strategy_tag = self.tuning.strategy.map_or_else(
            || {
                crate::encoding::strategy::StrategyTag::for_compression_level(
                    self.compression_level,
                )
            },
            |(tag, _)| tag,
        );
        self.state.huf_optimal_search = crate::encoding::frame_compressor::huf_search_enabled(
            self.state.strategy_tag,
            self.pledged_content_size.or(self.source_size_hint),
        );
        self.state.matcher.set_param_overrides(Some(overrides));
        Ok(())
    }

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

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

impl<M: Matcher> CompressionContext<M> {
    /// Creates a context with an explicitly provided matcher implementation.
    ///
    /// This constructor is primarily intended for tests and advanced callers that need
    /// custom match-window behavior.
    pub fn new_with_matcher(matcher: M, compression_level: CompressionLevel) -> Self {
        Self {
            compression_level,
            state: CompressState {
                matcher,
                sequence_capture: None,
                copy_kernel: crate::encoding::fastpath::select_kernel(),
                last_huff_table: None,
                huff_table_spare: None,
                huff_weights: Default::default(),
                seen_content: Default::default(),
                fse_tables: FseTables::new(),
                block_scratch: crate::encoding::blocks::CompressedBlockScratch::new(),
                workspace: crate::encoding::workspace::Workspace::new(),
                offset_hist: [1, 4, 8],
                strategy_tag: crate::encoding::strategy::StrategyTag::for_compression_level(
                    compression_level,
                ),
                pre_split: crate::encoding::levels::config::level_pre_split(compression_level)
                    .map(|tier| tier as u8),
                huf_optimal_search: true,
                literal_compression_disabled: matches!(
                    compression_level,
                    CompressionLevel::Level(n) if n < 0
                ),
                post_split_allowed: true,
            },
            encoded_scratch: Vec::new(),
            errored: false,
            last_error_kind: None,
            last_error_message: None,
            frame_started: false,
            target_block_size: None,
            max_block_size: None,
            pledged_content_size: None,
            source_size_hint: None,
            content_size_flag: true,
            bytes_consumed: 0,
            savings: 0,
            tuning: FrameTuning::default(),
            magicless: false,
            content_checksum: false,
            dictionary: None,
            dictionary_id_flag: true,
            #[cfg(feature = "hash")]
            hasher: XxHash64::with_seed(0),
        }
    }

    /// Compress the next frames at `level`, with the level's own tuning: any
    /// parameter override installed by
    /// [`set_parameters`](CompressionContext::set_parameters) is dropped, as
    /// [`FrameCompressor::set_compression_level`](crate::encoding::FrameCompressor::set_compression_level)
    /// drops it. Must be called before the frame's first [`write`](Self::write).
    ///
    /// # Examples
    /// ```
    /// use structured_zstd::encoding::{CompressionContext, CompressionLevel};
    ///
    /// let mut context = CompressionContext::new(CompressionLevel::Fastest);
    /// context.set_compression_level(CompressionLevel::Better).unwrap();
    /// let mut frame = Vec::new();
    /// context.write(&mut frame, b"compressed at the new level").unwrap();
    /// context.finish_frame(&mut frame).unwrap();
    /// ```
    pub fn set_compression_level(&mut self, level: CompressionLevel) -> Result<(), Error> {
        self.ensure_settable("the compression level must be set before the first write")?;
        self.compression_level = level;
        self.tuning = FrameTuning::default();
        self.state.matcher.clear_param_overrides();
        Ok(())
    }

    /// Set an upper bound on each physical block's payload (semantics of
    /// upstream `ZSTD_c_targetCBlockSize`): every block carries at most
    /// `target` payload bytes, +3-byte block header on the wire — the
    /// upstream knob is likewise a convergence target for block sizing,
    /// not a cap on header-inclusive wire bytes. Clamped to
    /// `[MIN_TARGET_BLOCK_SIZE, MAX_BLOCK_SIZE]`; mirrors
    /// `FrameCompressor::set_target_block_size`. Must be set before the
    /// frame's first write.
    pub fn set_target_block_size(&mut self, target: Option<u32>) -> Result<(), Error> {
        self.ensure_settable("the block-size target must be set before the first write")?;
        self.target_block_size = target.map(|t| {
            t.clamp(
                crate::common::MIN_TARGET_BLOCK_SIZE,
                crate::common::MAX_BLOCK_SIZE,
            )
        });
        Ok(())
    }

    /// Cap every block at `size` bytes (upstream `ZSTD_c_maxBlockSize`),
    /// clamped to `[MIN_MAX_BLOCK_SIZE, MAX_BLOCK_SIZE]`; mirrors
    /// `FrameCompressor::set_max_block_size`. Must be set before the frame's
    /// first write.
    pub fn set_max_block_size(&mut self, size: Option<u32>) -> Result<(), Error> {
        self.ensure_settable("the maximum block size must be set before the first write")?;
        self.max_block_size = size.map(|s| {
            s.clamp(
                crate::common::MIN_MAX_BLOCK_SIZE,
                crate::common::MAX_BLOCK_SIZE,
            )
        });
        Ok(())
    }

    /// Enable or disable the trailing XXH64 content checksum
    /// (upstream `ZSTD_c_checksumFlag`). Default `false`, matching the
    /// upstream library default (`ZSTD_c_checksumFlag = 0`). Must be called
    /// before the frame's first [`write`](Self::write); once the frame header
    /// is emitted the flag is fixed, so a late change returns an error rather
    /// than producing a header/trailer mismatch. Without the `hash` feature
    /// no checksum is emitted regardless.
    pub fn set_content_checksum(&mut self, emit: bool) -> Result<(), Error> {
        self.ensure_settable("content checksum must be set before the first write")?;
        self.content_checksum = emit;
        Ok(())
    }

    /// Enable or disable magicless frame format (`ZSTD_f_zstd1_magicless`).
    ///
    /// When set to `true`, the frame header omits the 4-byte magic number
    /// prefix. Must be called BEFORE the frame's first [`write`](Self::write)
    /// call; calling it after the frame header has already been emitted
    /// returns an error so the caller can't be misled into thinking they
    /// produced a magicless stream.
    pub fn set_magicless(&mut self, magicless: bool) -> Result<(), Error> {
        self.ensure_settable("magicless format must be set before the first write")?;
        self.magicless = magicless;
        Ok(())
    }

    /// Pledge the total uncompressed content size of the next frame.
    ///
    /// When set, the frame header will include a `Frame_Content_Size` field.
    /// This enables decoders to pre-allocate output buffers.
    /// The pledged size is also forwarded as a source-size hint to the
    /// matcher so small inputs can use smaller matching tables.
    ///
    /// Must be called **before** the frame's first [`write`](Self::write);
    /// calling it after the frame header has already been emitted returns an
    /// error. The pledge ends with the frame.
    pub fn set_pledged_content_size(&mut self, size: u64) -> Result<(), Error> {
        self.ensure_settable("pledged content size must be set before the first write")?;
        // The matcher is handed it as its size hint when the frame starts
        // (`ensure_frame_started`), not here: a pledge that goes unused must
        // not stay behind in the matcher for a later frame.
        self.pledged_content_size = Some(size);
        Ok(())
    }

    /// Control whether the pledged size is written into the header's
    /// `Frame_Content_Size` field (upstream `ZSTD_c_contentSizeFlag`,
    /// default on). With the flag off the header omits the field, but a
    /// pledge set via [`set_pledged_content_size`](Self::set_pledged_content_size)
    /// is still enforced against the bytes actually written. Must be
    /// called before the frame's first [`write`](Self::write).
    pub fn set_content_size_flag(&mut self, emit: bool) -> Result<(), Error> {
        self.ensure_settable("content size flag must be set before the first write")?;
        self.content_size_flag = emit;
        Ok(())
    }

    /// Provide a hint about the total uncompressed size of each frame.
    ///
    /// Unlike [`set_pledged_content_size`](Self::set_pledged_content_size),
    /// this does **not** enforce that exactly `size` bytes are written; it
    /// may reduce matcher tables, advertised frame window, and block sizing
    /// for small inputs. A parameter, like upstream `ZSTD_c_srcSizeHint`: it
    /// applies to every frame until replaced. Must be called before the
    /// frame's first [`write`](Self::write).
    pub fn set_source_size_hint(&mut self, size: u64) -> Result<(), Error> {
        self.ensure_settable("source size hint must be set before the first write")?;
        // Read at each frame start as `pledged_content_size.or(source_size_hint)`,
        // by the matcher's sizing and the Fast HUF fast-path gate alike
        // (`ensure_frame_started`), so a small advisory size also lifts Fast
        // streams off the expensive optimal-HUF search.
        self.source_size_hint = Some(size);
        Ok(())
    }

    /// Drop the hint [`set_source_size_hint`](Self::set_source_size_hint)
    /// set, so later frames are sized by their pledge alone. Must be called
    /// before the frame's first [`write`](Self::write).
    pub fn clear_source_size_hint(&mut self) -> Result<(), Error> {
        self.ensure_settable("source size hint must be cleared before the first write")?;
        self.source_size_hint = None;
        Ok(())
    }

    /// Attach a dictionary blob to each frame (upstream zstd
    /// `ZSTD_CCtx_loadDictionary` on a streaming context, which loads in
    /// `ZSTD_dct_auto` mode): a
    /// [serialized](crate::decoding::Dictionary::is_serialized) dictionary,
    /// or raw content. The dictionary primes the
    /// match-finder and seeds the first block's entropy tables + repeat
    /// offsets; a serialized one's ID is written into the frame header, while
    /// raw content has none to write, so the decoder must be given the same
    /// bytes explicitly. Must be called before the frame's first
    /// [`write`](Self::write); repeat offsets must be non-zero.
    pub fn set_dictionary_from_bytes(&mut self, raw_dictionary: &[u8]) -> Result<(), Error> {
        if raw_dictionary.is_empty() {
            // An empty buffer is how the same upstream entry point is told
            // there is no dictionary: it clears and succeeds. Still refused
            // once the frame is open, like any other attach.
            self.ensure_settable("dictionary must be attached before the first write")?;
            // What the match finder kept of it across frames (its primed
            // snapshot, the copy left resident) goes with it, entropy included.
            self.state.matcher.invalidate_primed_dictionary();
            self.dictionary = None;
            return Ok(());
        }
        let dict = EncoderDictionary::from_serialized_or_raw_content(raw_dictionary)
            .map_err(|err| invalid_input_error(&alloc::format!("invalid dictionary: {err:?}")))?;
        self.set_encoder_dictionary(dict)
    }

    /// Whether the frame header records the attached dictionary's ID
    /// (upstream `ZSTD_c_dictIDFlag` semantics; default `true`).
    /// Mirrors [`FrameCompressor::set_dictionary_id_flag`](crate::encoding::FrameCompressor::set_dictionary_id_flag).
    /// Decoders can still decode such frames by supplying the dictionary
    /// explicitly.
    pub fn set_dictionary_id_flag(&mut self, emit: bool) -> Result<(), Error> {
        self.ensure_settable("dictionary ID flag must be set before the first write")?;
        self.dictionary_id_flag = emit;
        Ok(())
    }

    /// Attach an already-parsed [`EncoderDictionary`] to each frame. See
    /// [`set_dictionary_from_bytes`](Self::set_dictionary_from_bytes); must be
    /// called before the frame's first write. The entropy tables it seeds were
    /// built when it was prepared, so attaching builds nothing.
    pub fn set_encoder_dictionary(&mut self, dict: EncoderDictionary) -> Result<(), Error> {
        self.ensure_settable("dictionary must be attached before the first write")?;
        // A zero id marks a raw-content dictionary, which carries no header to
        // hold one; the frame then records no dictionary ID and the decoder
        // must be given the same bytes explicitly.
        if dict.inner.offset_hist.contains(&0) {
            return Err(invalid_input_error(
                "dictionary carries a zero repeat offset",
            ));
        }
        // The match finder's primed snapshot and resident copy belong to the
        // dictionary being replaced; the next frame primes the new one.
        self.state.matcher.invalidate_primed_dictionary();
        // One under 8 bytes is attached as none; see
        // `EncoderDictionary::is_used_for_compression`.
        self.dictionary = dict.is_used_for_compression().then_some(dict);
        Ok(())
    }

    /// The dictionary each frame is compressed with, if one is attached.
    ///
    /// # Examples
    /// ```
    /// use structured_zstd::encoding::{CompressionContext, CompressionLevel, EncoderDictionary};
    ///
    /// let dictionary = EncoderDictionary::from_serialized_or_raw_content(b"some shared history").unwrap();
    /// let mut context = CompressionContext::new(CompressionLevel::Default);
    /// assert!(context.dictionary().is_none());
    /// context.set_encoder_dictionary(dictionary).unwrap();
    /// assert!(context.dictionary().is_some());
    /// ```
    pub fn dictionary(&self) -> Option<&EncoderDictionary> {
        self.dictionary.as_ref()
    }

    /// Total heap bytes this context's allocations hold, excluding the inline
    /// struct: match-finder tables / history, retained Huffman tables, the
    /// `encoded_scratch` block buffer, the retained dictionary content, and
    /// its entropy tables. Mirrors
    /// `FrameCompressor::heap_size` so a context can report its true
    /// footprint through `ZSTD_sizeof_CCtx`.
    pub fn heap_size(&self) -> usize {
        let mut total = self.state.matcher.heap_size();
        total += self
            .state
            .last_huff_table
            .as_ref()
            .map_or(0, |table| table.heap_size());
        total += self
            .state
            .huff_table_spare
            .as_ref()
            .map_or(0, |table| table.heap_size());
        // Kept between blocks and frames; see `FrameCompressor::heap_size`.
        total += self.state.huff_weights.heap_size();
        total += self.state.retained_scratch_heap_size();
        total += self.state.seen_content.heap_size();
        total += self.state.captured_parse_heap_size();
        total += self.encoded_scratch.capacity();
        total += self
            .dictionary
            .as_ref()
            .map_or(0, EncoderDictionary::heap_size);
        total
    }

    /// Compress `buf` into the frame in progress, starting one (and writing
    /// its header to `drain`) if none is. Full blocks are compressed as they
    /// fill and written to `drain`; the rest stays buffered for the next call,
    /// [`flush`](Self::flush) or [`finish_frame`](Self::finish_frame).
    ///
    /// Returns how much of `buf` was taken, which is all of it unless a pledge
    /// set with [`set_pledged_content_size`](Self::set_pledged_content_size)
    /// allows less.
    pub fn write<D: Write + ?Sized>(&mut self, drain: &mut D, buf: &[u8]) -> Result<usize, Error> {
        self.ensure_open()?;
        if buf.is_empty() {
            return Ok(0);
        }

        // Check pledge before emitting the frame header so that a misuse
        // like set_pledged_content_size(0) + write(non_empty) doesn't leave
        // a partially-written header in the drain.
        if let Some(pledged) = self.pledged_content_size
            && self.bytes_consumed >= pledged
        {
            return Err(invalid_input_error(
                "write would exceed pledged content size",
            ));
        }

        self.ensure_frame_started(drain)?;

        // Enforce pledged upper bound: truncate the accepted slice to the
        // remaining allowance so that partial-write semantics are honored
        // (return Ok(n) with n < buf.len()) instead of failing the full call.
        // The check above leaves at least one byte of it.
        let buf = if let Some(pledged) = self.pledged_content_size {
            debug_assert!(self.bytes_consumed < pledged);
            let remaining_allowed = pledged - self.bytes_consumed;
            let accepted = core::cmp::min(
                buf.len(),
                usize::try_from(remaining_allowed).unwrap_or(usize::MAX),
            );
            &buf[..accepted]
        } else {
            buf
        };

        let block_capacity = self.block_capacity();
        let mut remaining = buf;
        let mut consumed = 0usize;

        while !remaining.is_empty() {
            // A block is emitted the moment it fills, below, and the
            // capacity is fixed for the frame, so there is always room here.
            let pending = self.pending_len();
            debug_assert!(pending < block_capacity);
            let to_take = core::cmp::min(remaining.len(), block_capacity - pending);
            self.buffer_input(&remaining[..to_take]);
            remaining = &remaining[to_take..];
            consumed += to_take;

            if self.emit_full_pending_block(drain, block_capacity).is_err() {
                // The bytes this call took are reported; the failure, sticky
                // on the context now, is what the next call returns.
                break;
            }
        }
        self.bytes_consumed += consumed as u64;
        Ok(consumed)
    }

    /// Emit the buffered partial block as a non-last block and flush `drain`.
    pub fn flush<D: Write + ?Sized>(&mut self, drain: &mut D) -> Result<(), Error> {
        self.ensure_open()?;
        if self.pending_len() == 0 {
            return drain.flush().map_err(|err| self.fail(err));
        }
        self.ensure_frame_started(drain)?;
        self.emit_pending_block(drain, false)?;
        drain.flush().map_err(|err| self.fail(err))
    }

    /// Close the frame in progress into `drain`: its last block, then its
    /// checksum when enabled. A frame nothing was written to is still a valid
    /// empty frame. The context is then ready for the next frame, with every
    /// setting and the dictionary as they were and the pledge cleared.
    pub fn finish_frame<D: Write + ?Sized>(&mut self, drain: &mut D) -> Result<(), Error> {
        self.ensure_open()?;

        // Validate the pledge before finalizing the frame. If this is called
        // before any writes, this also avoids emitting a header with an
        // incorrect FCS into the drain on mismatch.
        if let Some(pledged) = self.pledged_content_size
            && self.bytes_consumed != pledged
        {
            return Err(invalid_input_error(
                "pledged content size does not match bytes consumed",
            ));
        }

        self.ensure_frame_started(drain)?;
        // An empty frame, or one whose input ended on a block boundary, closes
        // with an empty last block.
        self.emit_pending_block(drain, true)?;

        #[cfg(feature = "hash")]
        if self.content_checksum {
            let checksum = self.hasher.finish() as u32;
            drain
                .write_all(&checksum.to_le_bytes())
                .map_err(|err| self.fail(err))?;
        }

        drain.flush().map_err(|err| self.fail(err))?;
        self.abandon_frame();
        Ok(())
    }

    /// Drop the frame in progress without closing it (upstream
    /// `ZSTD_CCtx_reset(ZSTD_reset_session_only)`): its buffered input and its
    /// pledge go, and the next [`write`](Self::write) starts a new frame. The
    /// settings, the dictionary and every allocation stay. What the frame
    /// already wrote to its drain stays there, an unfinished frame.
    ///
    /// This is the way on from a [`finish_frame`](Self::finish_frame) refused
    /// for a pledge the frame did not meet, when the rest of the input is not
    /// coming; writing it and finishing again completes the frame instead.
    ///
    /// # Examples
    /// ```
    /// use structured_zstd::encoding::{CompressionContext, CompressionLevel};
    ///
    /// let mut context = CompressionContext::new(CompressionLevel::Default);
    /// let mut unfinished = Vec::new();
    /// context.set_pledged_content_size(100).unwrap();
    /// context.write(&mut unfinished, b"only part of it").unwrap();
    /// assert!(context.finish_frame(&mut unfinished).is_err());
    /// context.abandon_frame();
    ///
    /// let mut frame = Vec::new();
    /// context.write(&mut frame, b"a frame of its own").unwrap();
    /// context.finish_frame(&mut frame).unwrap();
    /// ```
    pub fn abandon_frame(&mut self) {
        // What belongs to the frame goes with it; the settings, the dictionary
        // and every allocation stay for the next one.
        self.frame_started = false;
        self.bytes_consumed = 0;
        self.pledged_content_size = None;
        // A compressed frame's buffered input is uncommitted in the match
        // finder, which the next frame's reset drops.
        self.encoded_scratch.clear();
    }

    /// Start the next frame now, writing its header to `drain`, rather than at
    /// its first [`write`](Self::write). Settings are fixed from here on, as
    /// after a write; a frame already open is left as it is.
    ///
    /// ```
    /// use structured_zstd::encoding::{CompressionContext, CompressionLevel};
    ///
    /// let mut context = CompressionContext::new(CompressionLevel::Default);
    /// let mut frame = Vec::new();
    /// context.start_frame(&mut frame).unwrap();
    /// assert!(!frame.is_empty());
    /// assert!(context.frame_block_size().is_some());
    /// ```
    pub fn start_frame<D: Write + ?Sized>(&mut self, drain: &mut D) -> Result<(), Error> {
        self.ensure_open()?;
        self.ensure_frame_started(drain)
    }

    /// The largest block the open frame cuts: the block size limits held to
    /// the window (upstream `ZSTD_getBlockSize`). `None` between frames.
    pub fn frame_block_size(&self) -> Option<usize> {
        self.frame_started.then(|| self.block_capacity())
    }

    /// Compress `block` as one block of the open frame and append its content,
    /// without the block header, to `payload` (upstream `ZSTD_compressBlock`).
    /// Returns whether it compressed: `false` appends nothing, and the caller
    /// stores the block raw, as a decoder then takes it into its history.
    /// Either way the block joins the history later blocks match against.
    ///
    /// The frame must be open ([`start_frame`](Self::start_frame)), hold no
    /// input still waiting for its block, and `block` must fit
    /// [`frame_block_size`](Self::frame_block_size).
    ///
    /// ```
    /// use structured_zstd::encoding::{CompressionContext, CompressionLevel};
    ///
    /// let mut context = CompressionContext::new(CompressionLevel::Default);
    /// context.start_frame(&mut Vec::new()).unwrap();
    /// let mut payload = Vec::new();
    /// let compressed = context.compress_block(&b"blocks".repeat(100), &mut payload).unwrap();
    /// assert!(compressed && payload.len() < 600);
    /// ```
    pub fn compress_block(&mut self, block: &[u8], payload: &mut Vec<u8>) -> Result<bool, Error> {
        self.ensure_open()?;
        if !self.frame_started {
            return Err(invalid_input_error(
                "the frame must be started before a block of it is compressed",
            ));
        }
        if self.pending_len() != 0 {
            return Err(invalid_input_error(
                "a block cannot follow input still waiting for its block",
            ));
        }
        if block.len() > self.block_capacity() {
            return Err(invalid_input_error(
                "the block is larger than the frame's block size",
            ));
        }
        if block.is_empty() {
            return Ok(false);
        }
        self.buffer_input(block);
        let start = payload.len();
        // One block in, one block out: the caller writes its single header,
        // so the post-split, which would emit several physical blocks, stays
        // off (upstream `ZSTD_compressBlock` never splits either).
        let post_split = core::mem::replace(&mut self.state.post_split_allowed, false);
        let encoded = self.encode_block(payload, block.len(), false);
        self.state.post_split_allowed = post_split;
        encoded.map_err(|err| self.fail(err))?;
        self.bytes_consumed += block.len() as u64;
        // RFC 8878 3.1.1.2.2: Block_Type 2 is Compressed. A Raw or RLE block
        // changes no entropy table and no repeat offset, exactly as a block
        // the caller stores raw does on the decoding side.
        let compressed = (payload[start] >> 1) & 3 == 2;
        if compressed {
            payload.drain(start..start + BLOCK_HEADER_LEN);
        } else {
            payload.truncate(start);
        }
        Ok(compressed)
    }

    /// Input written to the open frame that no block has compressed yet: it
    /// waits for its block to fill, or for a flush. Zero between frames.
    /// Backs the progress the C `ZSTD_getFrameProgression` reports.
    ///
    /// ```
    /// use structured_zstd::encoding::{CompressionContext, CompressionLevel};
    ///
    /// let mut context = CompressionContext::new(CompressionLevel::Default);
    /// let mut frame = Vec::new();
    /// context.write(&mut frame, b"less than a block").unwrap();
    /// assert_eq!(context.buffered_input_len(), 17);
    /// context.flush(&mut frame).unwrap();
    /// assert_eq!(context.buffered_input_len(), 0);
    /// ```
    pub fn buffered_input_len(&self) -> usize {
        self.pending_len()
    }

    /// How much more input completes the block being filled, or a whole block
    /// when none is partly filled: the read size that lets the next write
    /// compress a block straight away (upstream zstd `ZSTD_nextInputSizeHint`).
    ///
    /// ```
    /// use structured_zstd::encoding::{CompressionContext, CompressionLevel};
    ///
    /// let mut context = CompressionContext::new(CompressionLevel::Default);
    /// let mut frame = Vec::new();
    /// context.write(&mut frame, &[7u8; 100]).unwrap();
    /// assert_eq!(context.next_input_size_hint(), 128 * 1024 - 100);
    /// ```
    pub fn next_input_size_hint(&self) -> usize {
        let block = self.block_capacity();
        block - self.pending_len() % block
    }

    fn ensure_open(&self) -> Result<(), Error> {
        if self.errored {
            return Err(self.sticky_error());
        }
        Ok(())
    }

    /// Refuse a setting once the frame it would change is under way.
    fn ensure_settable(&self, too_late: &str) -> Result<(), Error> {
        self.ensure_open()?;
        if self.frame_started {
            return Err(invalid_input_error(too_late));
        }
        Ok(())
    }

    // Cold path (only reached after poisoning). The format!() calls still allocate
    // in no_std even though error_with_kind_message/other_error_owned drop the
    // message; this is acceptable on an error recovery path to keep match arms simple.
    fn sticky_error(&self) -> Error {
        match (self.last_error_kind, self.last_error_message.as_deref()) {
            (Some(kind), Some(message)) => error_with_kind_message(
                kind,
                format!(
                    "streaming encoder is in an errored state due to previous {kind:?} failure: {message}"
                ),
            ),
            (Some(kind), None) => error_from_kind(kind),
            (None, Some(message)) => other_error_owned(format!(
                "streaming encoder is in an errored state: {message}"
            )),
            (None, None) => other_error("streaming encoder is in an errored state"),
        }
    }

    fn ensure_frame_started<D: Write + ?Sized>(&mut self, drain: &mut D) -> Result<(), Error> {
        if self.frame_started {
            return Ok(());
        }

        // Frames are independent, so the raw-skip's memory of emitted content
        // starts empty; the allocation is kept across frames.
        self.state.seen_content.reset_for_frame();
        // Same reason as the frame compressor's start: what the last frame
        // ended on is about to be replaced, and it is exactly the buffer this
        // frame wants to build into.
        self.state.fse_tables.park_previous_before_frame();
        self.ensure_level_supported()?;
        // A dictionary is only active when it can actually be primed: the level
        // compresses (not `Uncompressed`) AND the matcher supports priming AND a
        // dictionary is attached. Mirrors `FrameCompressor`'s `use_dictionary_state`
        // so a streaming frame never advertises a `Dictionary_ID` or seeds dict
        // entropy/offsets unless the dictionary is genuinely in play (otherwise
        // it would emit frames that needlessly require a dictionary at decode
        // time).
        let use_dictionary_state =
            !matches!(self.compression_level, CompressionLevel::Uncompressed)
                && self.state.matcher.supports_dictionary_priming()
                && self.dictionary.is_some();
        // The dictionary sizes select the CDict cParams tier (consumed inside
        // `reset`), so hand them over BEFORE reset.
        if use_dictionary_state && let Some(dict) = self.dictionary.as_ref() {
            self.state.matcher.set_dictionary_size_hint(dict.sizes());
        }
        // The matcher resolves the frame from the LAST hint it was handed, so
        // re-forward the authoritative size (`pledge.or(advisory)`, the same
        // value the gates below read) right before the reset: without this a
        // pledge followed by a different advisory hint (or vice versa) left
        // the matcher and the frame gates on different size tiers.
        if let Some(size) = self.pledged_content_size.or(self.source_size_hint) {
            self.state.matcher.set_source_size_hint(size);
        }
        let block_target = crate::encoding::frame_compressor::block_ceiling(
            self.target_block_size,
            self.max_block_size,
        );
        // Raw frames keep no history; everything else is read block by block,
        // and a pledge makes its length exact. One past what the address space
        // holds is laid out as a stream, which sizes to the window anyway.
        let ingest = if matches!(self.compression_level, CompressionLevel::Uncompressed) {
            crate::encoding::workspace::IngestPlan::Raw
        } else if let Some(pledged) = self
            .pledged_content_size
            .and_then(|pledged| usize::try_from(pledged).ok())
        {
            crate::encoding::workspace::IngestPlan::PledgedStream(pledged)
        } else {
            crate::encoding::workspace::IngestPlan::Stream
        };
        self.state
            .reset_for_frame(self.compression_level, block_target, ingest);
        let block_capacity = self.block_capacity();
        self.state.finish_layout(block_capacity);
        // Sync `state.strategy_tag` / `state.pre_split` to the strategy the
        // matcher's reset resolved (size- and dictionary-adaptive; a public
        // strategy override wins) so the literal-compression gates, the block
        // pre-splitter and the dictionary load below agree with the parse.
        // Mirrors `FrameCompressor::compress` and keeps both entry points
        // byte-equivalent.
        let hint = self.pledged_content_size.or(self.source_size_hint);
        let params = crate::encoding::frame_compressor::resolve_frame_params(
            self.compression_level,
            hint,
            self.dictionary.as_ref().filter(|_| use_dictionary_state),
        );
        crate::encoding::frame_compressor::sync_effective_strategy(
            &mut self.state,
            self.compression_level,
            &params,
            self.tuning.strategy,
        );
        self.state.huf_optimal_search =
            crate::encoding::frame_compressor::huf_search_enabled(self.state.strategy_tag, hint);
        self.state.literal_compression_disabled =
            crate::encoding::frame_compressor::literal_compression_disabled(
                self.state.strategy_tag,
                self.compression_level,
                crate::encoding::frame_compressor::gate_target_length(
                    self.compression_level,
                    &self.tuning,
                    self.dictionary.as_ref().filter(|_| use_dictionary_state),
                ),
                self.tuning.literal_compression,
            );
        // Seed the repeat-offset history from the dictionary (upstream zstd
        // `ZSTD_compress_insertDictionary`), or the default rep codes
        // otherwise, and load the dictionary into the match finder: primed,
        // restored from a snapshot, or, on a reused context whose reset kept
        // it resident, left in place with its offsets reapplied.
        // `dict` borrows `self.dictionary`; `self.state` is a disjoint field.
        self.state.offset_hist = [1, 4, 8];
        if use_dictionary_state && let Some(dict) = self.dictionary.as_ref() {
            self.state.offset_hist = dict.inner.offset_hist;
            crate::encoding::frame_compressor::load_frame_dictionary(
                &mut self.state,
                self.compression_level,
                dict,
                hint,
            );
        }
        // Seed the first block's entropy from the dictionary's encoder tables
        // (upstream zstd `cdict->cBlockState`), or clear to defaults.
        if use_dictionary_state && let Some(dict) = self.dictionary.as_ref() {
            let cache = &dict.inner.entropy;
            self.state.last_huff_table.clone_from(&cache.huff);
            self.state
                .fse_tables
                .ll_previous
                .clone_from(&cache.ll_previous);
            self.state
                .fse_tables
                .ml_previous
                .clone_from(&cache.ml_previous);
            self.state
                .fse_tables
                .of_previous
                .clone_from(&cache.of_previous);
            let ll_entropy = match cache.ll_previous.as_ref() {
                Some(PreviousFseTable::Custom(table)) => Some(table.as_ref()),
                _ => None,
            };
            let ml_entropy = match cache.ml_previous.as_ref() {
                Some(PreviousFseTable::Custom(table)) => Some(table.as_ref()),
                _ => None,
            };
            let of_entropy = match cache.of_previous.as_ref() {
                Some(PreviousFseTable::Custom(table)) => Some(table.as_ref()),
                _ => None,
            };
            self.state.matcher.seed_dictionary_entropy(
                self.state.last_huff_table.as_ref(),
                ll_entropy,
                ml_entropy,
                of_entropy,
            );
        } else {
            self.state.last_huff_table = None;
            self.state.fse_tables.ll_previous = None;
            self.state.fse_tables.ml_previous = None;
            self.state.fse_tables.of_previous = None;
        }
        self.savings = 0;
        #[cfg(feature = "hash")]
        {
            self.hasher = XxHash64::with_seed(0);
        }

        let window_size = self.state.matcher.window_size();
        if window_size == 0 {
            return Err(invalid_input_error(
                "matcher reported window_size == 0, which is invalid",
            ));
        }

        let frame_content_size = self.pledged_content_size.filter(|_| self.content_size_flag);
        let single_segment =
            crate::encoding::frame_header::single_segment(frame_content_size, window_size);

        let header = FrameHeader {
            frame_content_size,
            single_segment,
            content_checksum: cfg!(feature = "hash") && self.content_checksum,
            dictionary_id: if use_dictionary_state && self.dictionary_id_flag {
                // Id 0 is a raw-content dictionary: RFC 8878 spells "no
                // dictionary ID" as an absent field, not as a stored zero.
                self.dictionary
                    .as_ref()
                    .map(|dict| dict.inner.id)
                    .filter(|id| *id != 0)
                    .map(u64::from)
            } else {
                None
            },
            window_size: if single_segment {
                None
            } else {
                Some(window_size)
            },
            magicless: self.magicless,
        };
        let mut encoded_header = Vec::new();
        header.serialize(&mut encoded_header);
        drain
            .write_all(&encoded_header)
            .map_err(|err| self.fail(err))?;

        if matches!(self.compression_level, CompressionLevel::Uncompressed) {
            // The raw block's header goes in front of its input once the
            // block's length is known.
            self.encoded_scratch.clear();
            self.encoded_scratch
                .reserve(BLOCK_HEADER_LEN + self.block_capacity());
            self.encoded_scratch
                .extend_from_slice(&[0; BLOCK_HEADER_LEN]);
        }
        self.frame_started = true;
        Ok(())
    }

    fn block_capacity(&self) -> usize {
        let matcher_window = self.state.matcher.window_size() as usize;
        let ceiling = crate::encoding::frame_compressor::block_ceiling(
            self.target_block_size,
            self.max_block_size,
        );
        core::cmp::max(1, core::cmp::min(matcher_window, ceiling))
    }

    /// Input taken for the frame that no block has carried out yet.
    fn pending_len(&self) -> usize {
        if !self.frame_started {
            return 0;
        }
        match self.compression_level {
            CompressionLevel::Uncompressed => self.encoded_scratch.len() - BLOCK_HEADER_LEN,
            _ => self.state.matcher.uncommitted_input().len(),
        }
    }

    /// Buffers `bytes` for the block being filled: behind the raw block's
    /// header, or uncommitted in the match finder's history, where the block
    /// is later matched without another copy.
    fn buffer_input(&mut self, bytes: &[u8]) {
        match self.compression_level {
            CompressionLevel::Uncompressed => self.encoded_scratch.extend_from_slice(bytes),
            _ => {
                self.state
                    .matcher
                    .fill_in_place(bytes.len(), &mut |history| {
                        history.extend_from_slice(bytes);
                        (bytes.len(), false)
                    });
            }
        }
    }

    /// Where the full pending block is cut (upstream `ZSTD_compress_frameChunk`
    /// sizing every block with `ZSTD_optimalBlockSize`): the pre-splitter's
    /// boundary once the frame has saved enough, else the whole block.
    /// `remaining` is the input still to come as far as the splitter knows:
    /// a full block again while writes continue, the buffered bytes at the
    /// end of the frame.
    fn pre_split_len(&mut self, block_capacity: usize, remaining: usize) -> usize {
        if matches!(self.compression_level, CompressionLevel::Uncompressed) {
            return self.pending_len();
        }
        let pending = self.state.matcher.uncommitted_input();
        let window = pending.len();
        let cut = crate::encoding::frame_compressor::optimal_block_size_with(
            self.state.pre_split.map(usize::from),
            pending,
            remaining,
            block_capacity,
            self.savings,
        )
        .min(window);
        // The splitter sees byte statistics only; a cut it proposes inside a
        // match the matcher already knows of is moved out.
        if cut < window {
            self.state.matcher.settle_block_cut(window, cut)
        } else {
            cut
        }
    }

    /// Emit the pending block once it is full, cut where the pre-splitter
    /// says; the suffix stays pending and heads the next block. A failure
    /// leaves the context failed.
    fn emit_full_pending_block<D: Write + ?Sized>(
        &mut self,
        drain: &mut D,
        block_capacity: usize,
    ) -> Result<(), Error> {
        if self.pending_len() != block_capacity {
            return Ok(());
        }
        let block_len = self.pre_split_len(block_capacity, block_capacity);
        self.encode_block(drain, block_len, false)
            .map_err(|err| self.fail(err))
    }

    fn emit_pending_block<D: Write + ?Sized>(
        &mut self,
        drain: &mut D,
        last_block: bool,
    ) -> Result<(), Error> {
        let block_capacity = self.block_capacity();
        if last_block {
            // A full final buffer is cut like any other block (the reader
            // path splits it with `remaining = len`); each cut prefix goes
            // out as a non-last block and the suffix is re-examined.
            while self.pending_len() == block_capacity {
                let block_len = self.pre_split_len(block_capacity, block_capacity);
                if block_len == block_capacity {
                    break;
                }
                self.encode_block(drain, block_len, false)
                    .map_err(|err| self.fail(err))?;
            }
        }
        let block_len = self.pending_len();
        self.encode_block(drain, block_len, last_block)
            .map_err(|err| self.fail(err))
    }

    // Exhaustive match kept intentionally: adding a new CompressionLevel
    // variant will produce a compile error here, forcing the developer to
    // decide whether the streaming encoder supports it before shipping.
    fn ensure_level_supported(&self) -> Result<(), Error> {
        match self.compression_level {
            CompressionLevel::Uncompressed
            | CompressionLevel::Fastest
            | CompressionLevel::Default
            | CompressionLevel::Better
            | CompressionLevel::Best
            | CompressionLevel::Level(_) => Ok(()),
        }
    }

    /// Writes the first `block_len` pending bytes to `drain` as one block,
    /// `last_block` if it closes the frame.
    fn encode_block<D: Write + ?Sized>(
        &mut self,
        drain: &mut D,
        block_len: usize,
        last_block: bool,
    ) -> Result<(), Error> {
        if matches!(self.compression_level, CompressionLevel::Uncompressed) {
            // The block was assembled behind its header as it was written.
            debug_assert_eq!(block_len, self.pending_len());
            let header = BlockHeader {
                last_block,
                block_type: crate::blocks::block::BlockType::Raw,
                // At most one block, which a block header's 21-bit size holds.
                block_size: block_len as u32,
            };
            self.encoded_scratch[..BLOCK_HEADER_LEN].copy_from_slice(&header.to_le_bytes());
            drain.write_all(&self.encoded_scratch)?;
            #[cfg(feature = "hash")]
            if self.content_checksum {
                self.hasher.write(&self.encoded_scratch[BLOCK_HEADER_LEN..]);
            }
            self.encoded_scratch.truncate(BLOCK_HEADER_LEN);
            return Ok(());
        }
        self.encoded_scratch.clear();
        if block_len == 0 {
            // Only the last block of a frame is ever empty.
            let header = BlockHeader {
                last_block,
                block_type: crate::blocks::block::BlockType::Raw,
                block_size: 0,
            };
            header.serialize(&mut self.encoded_scratch);
            return drain.write_all(&self.encoded_scratch);
        }
        let dict_active =
            self.dictionary.is_some() && self.state.matcher.supports_dictionary_priming();
        compress_block_encoded(
            &mut self.state,
            self.compression_level,
            last_block,
            block_len,
            &mut self.encoded_scratch,
            dict_active,
            // No FrameEmitInfo on the streaming encoder path: it does not
            // surface per-block layout, so no sidecar.
            #[cfg(feature = "lsm")]
            None,
            #[cfg(all(feature = "lsm", feature = "hash"))]
            None,
        );
        // The block is committed to the match finder whether or not the drain
        // takes it; a drain that fails leaves the context failed.
        drain.write_all(&self.encoded_scratch)?;
        // `savings` counts the block header too, as upstream's
        // `ZSTD_compress_frameChunk` does (`cSize` includes it).
        self.savings += block_len as i64 - self.encoded_scratch.len() as i64;
        #[cfg(feature = "hash")]
        if self.content_checksum {
            self.hasher.write(self.state.matcher.get_last_space());
        }
        Ok(())
    }

    fn fail(&mut self, err: Error) -> Error {
        self.errored = true;
        if self.last_error_kind.is_none() {
            self.last_error_kind = Some(err.kind());
        }
        if self.last_error_message.is_none() {
            self.last_error_message = Some(err.to_string());
        }
        err
    }
}

/// Bytes of a block header (RFC 8878 3.1.1.2).
const BLOCK_HEADER_LEN: usize = 3;

fn error_from_kind(kind: ErrorKind) -> Error {
    Error::from(kind)
}

fn error_with_kind_message(kind: ErrorKind, message: String) -> Error {
    #[cfg(feature = "std")]
    {
        Error::new(kind, message)
    }
    #[cfg(not(feature = "std"))]
    {
        Error::new(kind, alloc::boxed::Box::new(message))
    }
}

fn invalid_input_error(message: &str) -> Error {
    #[cfg(feature = "std")]
    {
        Error::new(ErrorKind::InvalidInput, message)
    }
    #[cfg(not(feature = "std"))]
    {
        Error::new(
            ErrorKind::Other,
            alloc::boxed::Box::new(alloc::string::String::from(message)),
        )
    }
}

fn other_error_owned(message: String) -> Error {
    #[cfg(feature = "std")]
    {
        Error::other(message)
    }
    #[cfg(not(feature = "std"))]
    {
        Error::new(ErrorKind::Other, alloc::boxed::Box::new(message))
    }
}

fn other_error(message: &str) -> Error {
    #[cfg(feature = "std")]
    {
        Error::other(message)
    }
    #[cfg(not(feature = "std"))]
    {
        Error::new(
            ErrorKind::Other,
            alloc::boxed::Box::new(alloc::string::String::from(message)),
        )
    }
}

#[cfg(test)]
mod tests;