mlt_core/encoder/

writer.rs

use std::collections::HashMap;
use std::{io, mem};

use fsst::Compressor;
use integer_encoding::VarIntWriter as _;

#[cfg(feature = "unstable-v2")]
use crate::decoder::LayerHeader02;
#[cfg(feature = "unstable-v2")]
use crate::decoder::stream::header02::{Count02, Family, WordWidth};
use crate::decoder::{ColumnType, Morton};
#[cfg(feature = "unstable-v2")]
use crate::encoder::model::FloatEncoding;
use crate::encoder::model::{CurveParams, ExplicitEncoder, StrEncoding, StreamCtx};
use crate::encoder::{EncoderConfig, IntEncoder, VertexBufferType};
use crate::utils::BinarySerializer as _;
use crate::{MltError, MltResult};

/// Stateful encoder that accumulates encoded layer bytes.
///
/// Logical temporary buffers live in `Codecs` and are passed alongside
/// the encoder while a stream is being transformed and serialized. Physical
/// encoders live here with their own scratch buffers, then copy complete
/// payloads into [`data`](Encoder::data).
///
/// # Buffer layout
///
/// The MLT layer wire format is:
///
/// ```text
/// [varint(body_len + 1)] [tag = 1]
/// [name: string] [extent: varint] [column_count: varint]   <- hdr
/// [col_type₁] [col_type₂] … [col_typeN]                    <- meta
/// [col₁ stream data] [col₂ stream data] … [colN stream data] <- data
/// ```
///
/// The three sections are accumulated into separate buffers so they can be
/// combined at the end *without* any in-place insertion or extra copies:
///
/// * `hdr` - layer header (name, extent, `column_count`).
/// * [`meta`] - column-type bytes (one byte + optional name per column).
/// * [`data`] - encoded stream data; also the target of [`impl Write`].
///
/// # Sort-strategy trialing
///
/// Create one `Encoder` per sort-strategy trial, encode the layer into it,
/// and keep the one whose `total_len()` is smallest:
///
/// ```rust,ignore
/// let mut codecs = Codecs::default();
/// let mut best: Option<Encoder> = None;
/// for strategy in strategies {
///     let mut enc = Encoder::new(cfg);
///     layer.write_to(&mut enc, &mut codecs)?;
///     if best.as_ref().is_none_or(|b| enc.total_len() < b.total_len()) {
///         best = Some(enc);
///     }
/// }
/// return best.unwrap().into_layer_bytes();
/// ```
///
/// # Stream-level encoding alternatives
///
/// Use [`Encoder::try_alternatives`] to open a competition,
/// then submit each candidate via `AltSession::with`.  The guard's `Drop`
/// impl finalises the competition automatically:
///
/// ```rust,ignore
/// let mut alt = enc.try_alternatives();
/// alt.with(|enc| write_stream_as_varint(data, enc))?;
/// alt.with(|enc| write_stream_as_fastpfor(data, enc))?;
/// // alt drops -> keeps whichever was shorter
/// ```
///
/// [`meta`]: Encoder::meta
/// [`data`]: Encoder::data
/// [`impl Write`]: Encoder#impl-Write
#[derive(Default)]
pub struct Encoder {
    /// Encoding configuration: controls which optimization strategies are tried
    /// (sort orders, compression algorithms, etc.).
    ///
    /// Set once at construction time via [`Encoder::new`]; propagated
    /// automatically to all sub-encoders so individual encode methods do not
    /// need a separate `cfg` argument.
    cfg: EncoderConfig,

    /// When [`Some`], property / ID / geometry encoders use `ExplicitEncoder`
    /// callbacks instead of trying candidate encodings. When [`None`], the
    /// automatic optimization path runs.
    pub(crate) explicit: Option<ExplicitEncoder>,

    /// Layer header bytes: `name`, `extent`, `column_count`.
    ///
    /// This section comes first in the wire format and is never subject to alternatives.
    hdr: Vec<u8>,

    /// Column-type metadata bytes.
    ///
    /// Each column contributes one type byte (plus a name string for property
    /// columns).  Written by the `write_columns_meta_to` methods, which write
    /// directly to `enc.meta`.  This section comes second in the wire format
    /// and is never subject to alternatives (column types are fixed).
    meta: Vec<u8>,

    /// Encoded stream data.
    ///
    /// All stream counts, per-stream encoding-metadata bytes, and encoded
    /// data bytes land here via [`impl Write`].  This section comes last in
    /// the wire format and is where stream-level alternatives compete.
    ///
    /// [`impl Write`]: Encoder#impl-Write
    data: Vec<u8>,

    /// Morton parameters for this layer's vertex set; `None` if the extent
    /// exceeds 16 bits per axis (Morton encoding is unusable in that case).
    /// Pre-populated by [`StagedLayer::encode_into`](crate::encoder::StagedLayer::encode_into).
    pub(crate) morton_cache: Option<Morton>,

    /// Hilbert curve parameters for this layer's vertex set. Pre-populated by
    /// [`StagedLayer::encode_into`](crate::encoder::StagedLayer::encode_into).
    pub(crate) hilbert_cache: Option<CurveParams>,

    /// Cached FSST compressor per string column, keyed by column name.
    /// `None` means training found FSST not viable for that column.
    /// Trained on deduplicated values on the first sort trial, reused on subsequent trials.
    pub(crate) fsst_cache: HashMap<String, Option<Compressor>>,

    /// What a v2 decoder will read the stream being written against: the layer's
    /// `feature_count`, or the presence popcount while an optional column's data
    /// stream is being written.
    ///
    /// [`Count02::Explicit`] while an m-value column or a shared dictionary's
    /// corpus is being written, whose counts context gives no way to infer, so
    /// every one of those streams writes its own.
    ///
    /// Read by the v2 stream-header codec to decide whether an explicit count
    /// varint must be emitted; ignored entirely for v1 layers.
    #[cfg(feature = "unstable-v2")]
    pub(crate) count_context: Count02,

    /// The family the stream being written is numbered in, set by the v2 writers alongside [`Self::count_context`].
    /// Ignored for v1 layers.
    #[cfg(feature = "unstable-v2")]
    pub(crate) family_context: Family,

    // -----------------------------------------------------------------------
    // Alternatives state - a stack that supports nested competitions.
    //
    // Invariant between candidates at any level:
    //   data.len() == level.data_start + level.best_data_size.unwrap_or(0)
    //   meta.len() == level.meta_start + level.best_meta_size.unwrap_or(0)
    //
    // Empty stack <-> no competition in progress.
    // -----------------------------------------------------------------------
    /// Stack of active encoding competitions, innermost last.
    ///
    /// Empty while no [`Encoder::try_alternatives`] session
    /// is in progress.
    alt_stack: Vec<AltLevel>,
}

impl Encoder {
    /// Create a new encoder with the given [`EncoderConfig`].
    ///
    /// Use [`Encoder::default()`] when the default configuration is sufficient.
    #[inline]
    #[must_use]
    pub fn new(cfg: EncoderConfig) -> Self {
        Self {
            cfg,
            ..Self::default()
        }
    }

    /// Like [`Self::new`] but with the explicit encoder set for deterministic encoding
    /// (tests, synthetics). Use with `StagedLayer::encode_explicit`.
    #[inline]
    #[must_use]
    pub fn with_explicit(cfg: EncoderConfig, explicit: ExplicitEncoder) -> Self {
        Self {
            cfg,
            explicit: Some(explicit),
            ..Self::default()
        }
    }

    /// Ensure this encoder is in the good state, and moves results to a new instance.
    /// This allows current instance to be reused for other experiment, avoiding repeat of some operations.
    #[must_use]
    pub(crate) fn preserve_results(&mut self) -> Self {
        assert_eq!(self.alt_stack.len(), 0, "Alternatives stack is not empty");
        Self {
            // Keep the config: the archived result still needs it to frame the
            // layer with the right tag in `into_layer_bytes`.
            cfg: self.cfg,
            explicit: None,
            hdr: mem::take(&mut self.hdr),
            meta: mem::take(&mut self.meta),
            data: mem::take(&mut self.data),
            morton_cache: None,
            hilbert_cache: None,
            fsst_cache: HashMap::new(),
            #[cfg(feature = "unstable-v2")]
            count_context: Count02::Explicit,
            #[cfg(feature = "unstable-v2")]
            family_context: Family::Int(WordWidth::W32),
            alt_stack: vec![],
        }
    }

    #[inline]
    pub(crate) fn write_column_type(&mut self, column_type: ColumnType) -> MltResult<()> {
        column_type.write_to(&mut self.meta).map_err(MltError::from)
    }

    #[inline]
    pub(crate) fn write_column_name(&mut self, name: &str) -> MltResult<()> {
        self.meta.write_string(name).map_err(MltError::from)
    }

    #[inline]
    #[must_use]
    pub fn config(&self) -> EncoderConfig {
        self.cfg
    }

    #[inline]
    #[must_use]
    pub fn data(&self) -> &[u8] {
        &self.data
    }

    #[inline]
    pub(crate) fn data_mut(&mut self) -> &mut Vec<u8> {
        &mut self.data
    }

    #[inline]
    #[must_use]
    pub fn meta(&self) -> &[u8] {
        &self.meta
    }

    #[inline]
    pub(crate) fn meta_mut(&mut self) -> &mut Vec<u8> {
        &mut self.meta
    }

    #[inline]
    #[must_use]
    pub fn section_lens(&self) -> (usize, usize, usize) {
        (self.hdr.len(), self.meta.len(), self.data.len())
    }

    #[inline]
    pub(crate) fn write_column_header(
        &mut self,
        column_type: ColumnType,
        name: &str,
    ) -> MltResult<()> {
        self.write_column_type(column_type)?;
        self.write_column_name(name)
    }

    /// Write the v1 layer header (`name`, `extent`, `column_count`) to `hdr`.
    ///
    /// Must be called exactly once per layer, after all column meta and data.
    #[hotpath::measure]
    pub fn write_header01(
        &mut self,
        name: &str,
        extent: u32,
        column_count: usize,
    ) -> MltResult<()> {
        if name.is_empty() {
            return Err(MltError::MissingLayerName);
        }
        debug_assert!(
            self.alt_stack.is_empty(),
            "write_header called with an open alternatives session"
        );
        let name_len = u32::try_from(name.len())?;
        let column_count = u32::try_from(column_count)?;
        self.hdr.write_varint(name_len).map_err(MltError::from)?;
        self.hdr.extend_from_slice(name.as_bytes());
        self.hdr.write_varint(extent).map_err(MltError::from)?;
        self.hdr
            .write_varint(column_count)
            .map_err(MltError::from)?;
        Ok(())
    }

    /// Write the v2 layer header (`name`, `extent`, `feature_count`) to `hdr`.
    ///
    /// Unlike v1, `column_count` is not part of the header - it precedes the
    /// counted columns in the data section, after the geometry section.
    #[cfg(feature = "unstable-v2")]
    #[hotpath::measure]
    pub(crate) fn write_header02(
        &mut self,
        name: &str,
        header: LayerHeader02,
        feature_count: u32,
    ) -> MltResult<()> {
        if name.is_empty() {
            return Err(MltError::MissingLayerName);
        }
        debug_assert!(
            self.alt_stack.is_empty(),
            "write_header02 called with an open alternatives session"
        );
        self.hdr.write_string(name).map_err(MltError::from)?;
        self.hdr.push(header.to_byte());
        self.hdr
            .write_varint(feature_count)
            .map_err(MltError::from)?;
        Ok(())
    }

    /// When [`Self::explicit`] is [`Some`], returns the callback-chosen [`IntEncoder`].
    /// [`None`] means run automatic candidate selection for that stream.
    #[inline]
    pub(crate) fn override_int_enc(&self, ctx: &StreamCtx<'_>) -> Option<IntEncoder> {
        self.explicit.as_ref().map(|e| (e.get_int_encoder)(ctx))
    }

    /// When [`Self::explicit`] is [`Some`], returns the callback-chosen [`StrEncoding`].
    /// [`None`] means run automatic string / shared-dict corpus selection.
    #[inline]
    pub(crate) fn override_str_enc(&self, name: &str) -> Option<StrEncoding> {
        self.explicit.as_ref().map(|e| (e.get_str_encoding)(name))
    }

    /// When [`Self::explicit`] is [`Some`], returns the callback-chosen [`FloatEncoding`].
    /// [`None`] means cost the float encodings the config allows against each other.
    #[inline]
    #[cfg(feature = "unstable-v2")]
    pub(crate) fn override_float_enc(&self, name: &str) -> Option<FloatEncoding> {
        self.explicit.as_ref().map(|e| (e.get_float_encoding)(name))
    }

    /// Pinned vertex layout when an explicit encoder is active.
    #[inline]
    #[allow(clippy::unused_self)]
    pub(crate) fn override_vertex_buffer_type(&self) -> Option<VertexBufferType> {
        self.explicit.as_ref().map(|e| e.vertex_buffer_type)
    }

    /// Whether to force writing a geometry stream even when its data is empty.
    ///
    /// Delegates to [`ExplicitEncoder::force_stream`]; returns `false` when no explicit
    /// encoder is active (the default "skip empty streams" behavior).
    #[inline]
    pub(crate) fn force_stream(&self, ctx: &StreamCtx<'_>) -> bool {
        self.explicit
            .as_ref()
            .is_some_and(|e| (e.force_stream)(ctx))
    }

    /// Total encoded bytes across all three sections (`hdr + meta + data`).
    #[inline]
    #[must_use]
    pub fn total_len(&self) -> usize {
        self.hdr.len() + self.meta.len() + self.data.len()
    }

    /// Empty the output buffers so this encoder can be reused for the next sort trial.
    /// Keeps allocated capacity and the seeded curve/FSST caches.
    /// Used when a trial loses; [`Self::preserve_results`] handles the winning case instead.
    pub(crate) fn clear_results(&mut self) {
        debug_assert!(self.alt_stack.is_empty(), "Alternatives stack is not empty");
        self.hdr.clear();
        self.meta.clear();
        self.data.clear();
    }

    /// Concatenate `hdr + meta + data` into a single buffer **without** a
    /// tag/size prefix.
    ///
    /// Use this when the caller expects raw layer body bytes (without the size/tag framing)
    /// rather than a complete framed wire record - see [`Self::into_layer_bytes`] for the framed form.
    #[must_use]
    pub fn into_raw_bytes(mut self) -> Vec<u8> {
        if self.hdr.is_empty() && self.meta.is_empty() {
            return self.data;
        }
        let mut out = Vec::with_capacity(self.hdr.len() + self.meta.len() + self.data.len());
        out.append(&mut self.hdr);
        out.append(&mut self.meta);
        out.append(&mut self.data);
        out
    }

    /// Assemble the complete layer record.
    pub fn into_layer_bytes(self) -> MltResult<Vec<u8>> {
        #[cfg(feature = "unstable-v2")]
        let tag = self.cfg.wire_version().tag();
        // v1 is the only format this build writes, and `0x01` is its layer tag.
        #[cfg(not(feature = "unstable-v2"))]
        let tag = 1;
        self.into_layer_bytes_with_tag(tag)
    }

    /// Assemble a complete layer record for the given `tag`:
    /// `[varint(body_len + 1)][tag][hdr][meta][data]`.
    fn into_layer_bytes_with_tag(mut self, tag: u8) -> MltResult<Vec<u8>> {
        debug_assert!(
            self.alt_stack.is_empty(),
            "into_layer_bytes_with_tag called with an open alternatives session"
        );
        let body_len = self.hdr.len() + self.meta.len() + self.data.len();
        let size = u32::try_from(body_len + 1)?; // +1 for the tag byte
        let mut out = Vec::with_capacity(5 + 1 + body_len);
        out.write_varint(size).map_err(MltError::from)?;
        out.push(tag);
        out.append(&mut self.hdr);
        out.append(&mut self.meta);
        out.append(&mut self.data);
        Ok(out)
    }

    /// Begin a new encoding competition.
    ///
    /// Returns an `AltSession` guard.  Submit each candidate via
    /// `AltSession::with`; the guard's `Drop` impl finalises
    /// the competition and retains the shortest candidate automatically.
    ///
    /// Nesting is supported: calling `try_alternatives` inside a
    /// `with` closure opens an inner competition on the same stack,
    /// resolved before the outer candidate is committed.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let mut alt = enc.try_alternatives();
    /// for cand in candidates {
    ///     alt.with(|enc| write_candidate(cand, enc))?;
    /// }
    /// // alt drops -> finalises the competition
    /// ```
    pub fn try_alternatives(&mut self) -> AltSession<'_> {
        self.alt_stack.push(AltLevel {
            data_start: self.data.len(),
            meta_start: self.meta.len(),
            best_data: None,
            best_meta: None,
        });
        AltSession { enc: self }
    }

    /// Commit the current candidate at the innermost competition level.
    ///
    /// Compares bytes written since the last commit against the running best
    /// by **total** (`data + meta`) size; keeps the shorter one.
    ///
    /// Called internally by `AltSession::with` on `Ok`.
    fn alt_commit(&mut self) {
        debug_assert!(
            !self.alt_stack.is_empty(),
            "alt_commit called outside an active AltSession"
        );
        let (data, meta, stack) = (&mut self.data, &mut self.meta, &mut self.alt_stack);
        let level = stack.last_mut().unwrap();
        Self::close_candidate(data, meta, level);
    }

    /// Finalize the innermost competition and pop it from the stack.
    ///
    /// Any bytes written since the last `alt_commit` are evaluated as a
    /// final candidate; if no pending bytes exist and a best is already
    /// recorded this is a cheap stack-pop.
    fn alt_pop(&mut self) {
        debug_assert!(
            !self.alt_stack.is_empty(),
            "alt_pop called outside an active AltSession"
        );
        {
            let (data, meta, stack) = (&mut self.data, &mut self.meta, &mut self.alt_stack);
            let level = stack.last_mut().unwrap();
            let data_pending = data.len() - (level.data_start + level.best_data.unwrap_or(0));
            let meta_pending = meta.len() - (level.meta_start + level.best_meta.unwrap_or(0));
            if data_pending > 0 || meta_pending > 0 || level.best_data.is_none() {
                Self::close_candidate(data, meta, level);
            }
        }
        self.alt_stack.pop();
    }

    /// Shared compare-and-keep logic used by both `alt_commit` and `alt_pop`.
    ///
    /// Compares the bytes written since the last committed candidate against
    /// the current best by **total** (`data + meta`) size.
    /// Keeps the shorter one; ties preserve the existing best.
    fn close_candidate(data: &mut Vec<u8>, meta: &mut Vec<u8>, level: &mut AltLevel) {
        let best_data_end = level.data_start + level.best_data.unwrap_or(0);
        let best_meta_end = level.meta_start + level.best_meta.unwrap_or(0);
        let cand_data = data.len() - best_data_end;
        let cand_meta = meta.len() - best_meta_end;
        let cand_total = cand_data + cand_meta;
        let best_total = level.best_data.unwrap_or(0) + level.best_meta.unwrap_or(0);
        if level.best_data.is_none_or(|_| cand_total < best_total) {
            // New best: shift data candidate bytes to data_start.
            if level.best_data.is_some() {
                data.copy_within(best_data_end..best_data_end + cand_data, level.data_start);
                meta.copy_within(best_meta_end..best_meta_end + cand_meta, level.meta_start);
            }
            data.truncate(level.data_start + cand_data);
            meta.truncate(level.meta_start + cand_meta);
            level.best_data = Some(cand_data);
            level.best_meta = Some(cand_meta);
        } else {
            // Not an improvement: discard.
            data.truncate(best_data_end);
            meta.truncate(best_meta_end);
        }
    }
}

/// State for one level of an encoding competition.
///
/// Tracks the starting position in both the [`data`](Encoder::data) and
/// [`meta`](Encoder::meta) buffers, and the byte count of the best candidate
/// committed so far.
///
/// Candidates are compared by **total** bytes (`data + meta`); the shorter one
/// wins, with ties resolved in favor of the earlier candidate.
#[derive(Debug, Default, Clone)]
struct AltLevel {
    data_start: usize,
    meta_start: usize,
    /// Byte count appended to `data` by the current best candidate.
    best_data: Option<usize>,
    /// Byte count appended to `meta` by the current best candidate.
    best_meta: Option<usize>,
}

/// RAII guard for a stream-encoding competition opened by [`Encoder::try_alternatives`].
///
/// Submit each candidate via [`with`](AltSession::with); on `Ok` the candidate is
/// committed (compared against the running best and kept if shorter); on `Err`
/// the partial write is rolled back and the error propagates.  The guard's
/// `Drop` impl finalises the competition automatically, so the [`Encoder`] is
/// always left in a consistent state even when an error exits the loop early.
///
/// Nesting is allowed: calling [`Encoder::try_alternatives`] inside a
/// `with` closure opens an inner competition that is fully
/// resolved before the outer candidate is committed.
#[must_use = "AltSession must be used; drop it to finalise the competition"]
pub struct AltSession<'a> {
    enc: &'a mut Encoder,
}

impl AltSession<'_> {
    /// Encode one candidate.
    ///
    /// - **`Ok`** - commits the candidate; replaces the running best if shorter.
    /// - **`Err`** - truncates the partial write back to the pre-call checkpoint
    ///   and returns the error.  The guard's `Drop` still finalises the
    ///   competition cleanly using whichever candidates succeeded so far.
    #[hotpath::measure]
    pub fn with<F>(&mut self, f: F) -> MltResult<()>
    where
        F: FnOnce(&mut Encoder) -> MltResult<()>,
    {
        let data_cp = self.enc.data.len();
        let meta_cp = self.enc.meta.len();
        match f(self.enc) {
            Ok(()) => {
                self.enc.alt_commit();
                Ok(())
            }
            Err(e) => {
                self.enc.data.truncate(data_cp);
                self.enc.meta.truncate(meta_cp);
                Err(e)
            }
        }
    }
}

impl Drop for AltSession<'_> {
    fn drop(&mut self) {
        self.enc.alt_pop();
    }
}

/// Writes bytes to [`Encoder::data`].
///
/// This blanket implementation makes `Encoder` compatible with all
/// `BinarySerializer`, `VarIntWriter`, and other `Write`-based utilities so that
/// stream-data methods do not need a separate code path.
impl io::Write for Encoder {
    #[inline]
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.data.write(buf)
    }

    #[inline]
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }

    #[inline]
    fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
        self.data.write_all(buf)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Helper: directly extend `enc.data` with raw bytes (simulates a stream write).
    fn push(enc: &mut Encoder, bytes: &[u8]) {
        enc.data.extend_from_slice(bytes);
    }

    // ── basic single-level behavior ──────────────────────────────────────

    /// The shortest candidate wins.
    #[test]
    fn alternatives_keeps_shortest() {
        let mut enc = Encoder::default();
        push(&mut enc, b"prefix");

        let mut alt = enc.try_alternatives();
        alt.with(|enc| {
            push(enc, b"longer");
            Ok(())
        })
        .unwrap(); // 6 bytes
        alt.with(|enc| {
            push(enc, b"ab");
            Ok(())
        })
        .unwrap(); // 2 bytes - shortest
        alt.with(|enc| {
            push(enc, b"xyz");
            Ok(())
        })
        .unwrap(); // 3 bytes
        drop(alt);

        assert_eq!(enc.data, b"prefixab");
    }

    /// On a tie the first candidate is kept (strict `<`, not `<=`).
    #[test]
    fn alternatives_tie_keeps_first() {
        let mut enc = Encoder::default();

        let mut alt = enc.try_alternatives();
        alt.with(|enc| {
            push(enc, b"aaa");
            Ok(())
        })
        .unwrap(); // 3 bytes
        alt.with(|enc| {
            push(enc, b"bbb");
            Ok(())
        })
        .unwrap(); // 3 bytes - equal
        drop(alt);

        assert_eq!(enc.data, b"aaa");
    }

    /// A single candidate is unconditionally the winner.
    #[test]
    fn alternatives_single_candidate() {
        let mut enc = Encoder::default();

        let mut alt = enc.try_alternatives();
        alt.with(|enc| {
            push(enc, b"only");
            Ok(())
        })
        .unwrap();
        drop(alt);

        assert_eq!(enc.data, b"only");
    }

    /// Bytes written before `try_alternatives` are left intact throughout.
    #[test]
    fn prefix_bytes_are_preserved() {
        let mut enc = Encoder::default();
        push(&mut enc, b"HDR");

        let mut alt = enc.try_alternatives();
        alt.with(|enc| {
            push(enc, b"long_encoding");
            Ok(())
        })
        .unwrap(); // 13 bytes
        alt.with(|enc| {
            push(enc, b"short");
            Ok(())
        })
        .unwrap(); // 5 bytes - winner
        drop(alt);

        assert_eq!(&enc.data[..3], b"HDR");
        assert_eq!(&enc.data[3..], b"short");
    }

    /// Dropping the guard after all candidates are committed is a cheap stack-pop.
    #[test]
    fn drop_after_all_committed_is_noop() {
        let mut enc = Encoder::default();

        let mut alt = enc.try_alternatives();
        alt.with(|enc| {
            push(enc, b"best");
            Ok(())
        })
        .unwrap();
        drop(alt); // all candidates committed; drop just pops the stack

        assert!(enc.alt_stack.is_empty(), "stack empty after drop");
        assert_eq!(enc.data, b"best");
    }

    // ── nesting ───────────────────────────────────────────────────────────

    /// An inner competition is resolved before the outer candidate is committed.
    #[test]
    fn nested_alternatives() {
        let mut enc = Encoder::default();

        let mut outer = enc.try_alternatives();

        // Outer candidate A: header bytes + inner competition.
        outer
            .with(|enc| {
                push(enc, b"A:");
                let mut inner = enc.try_alternatives(); // inner level pushed
                inner.with(|enc| {
                    push(enc, b"long_inner");
                    Ok(())
                })?; // 10 bytes
                inner.with(|enc| {
                    push(enc, b"in");
                    Ok(())
                })?; // 2 bytes - inner winner
                drop(inner); // inner done; enc = b"A:in"
                push(enc, b"!");
                Ok(())
            })
            .unwrap(); // outer candidate A = b"A:in!" (5 bytes)

        // Outer candidate B: shorter overall.
        outer
            .with(|enc| {
                push(enc, b"B");
                Ok(())
            })
            .unwrap(); // 1 byte - winner
        drop(outer);

        assert_eq!(enc.data, b"B");
    }

    /// Stack depth tracks nesting level; inner guard drops before outer closure returns.
    #[test]
    fn nesting_depth_reflected_in_stack() {
        let mut enc = Encoder::default();

        assert_eq!(enc.alt_stack.len(), 0);
        let mut outer = enc.try_alternatives();

        outer
            .with(|enc| {
                assert_eq!(enc.alt_stack.len(), 1); // outer level on stack
                let mut inner = enc.try_alternatives();
                inner.with(|enc| {
                    assert_eq!(enc.alt_stack.len(), 2); // both levels on stack
                    push(enc, b"x");
                    Ok(())
                })?;
                drop(inner); // inner popped
                assert_eq!(enc.alt_stack.len(), 1);
                push(enc, b"y");
                Ok(())
            })
            .unwrap();

        drop(outer); // outer popped
        assert_eq!(enc.alt_stack.len(), 0);
    }

    // ── meta buffer tracking ──────────────────────────────────────────────

    /// Writes to both `data` and `meta` are rolled back for the losing
    /// candidate and kept for the winner, measured by total bytes.
    #[test]
    fn alternatives_tracks_meta_and_data() {
        let mut enc = Encoder::default();
        enc.data.extend_from_slice(b"D");
        enc.meta.extend_from_slice(b"M");

        let mut alt = enc.try_alternatives();
        // Candidate A: 4 data + 2 meta = 6 total
        alt.with(|enc| {
            push(enc, b"DDDD");
            enc.meta.extend_from_slice(b"mm");
            Ok(())
        })
        .unwrap();
        // Candidate B: 1 data + 1 meta = 2 total - winner
        alt.with(|enc| {
            push(enc, b"d");
            enc.meta.extend_from_slice(b"n");
            Ok(())
        })
        .unwrap();
        drop(alt);

        assert_eq!(enc.data, b"Dd");
        assert_eq!(enc.meta, b"Mn");
    }

    // ── error rollback ────────────────────────────────────────────────────

    /// A failing candidate is rolled back; prior best is preserved.
    #[test]
    fn error_candidate_is_rolled_back() {
        let mut enc = Encoder::default();

        let mut alt = enc.try_alternatives();
        alt.with(|enc| {
            push(enc, b"ok");
            Ok(())
        })
        .unwrap();
        let _ = alt.with(|enc| {
            push(enc, b"partial");
            Err(MltError::IntegerOverflow) // simulated failure
        });
        drop(alt);

        assert_eq!(enc.data, b"ok"); // "partial" was rolled back; "ok" kept
    }
}