buffa 0.8.1

//! The core [`Message`] trait and [`DecodeOptions`] builder.
//!
//! Every generated message type implements [`Message`], which provides
//! encode/decode/merge methods and a two-pass serialization model
//! (`compute_size` populates a [`SizeCache`](crate::SizeCache), `write_to`
//! consumes it) that avoids the exponential-time problem affecting naïve
//! length-delimited encoders.

use bytes::{Buf, BufMut};

use crate::error::DecodeError;
use crate::message_field::DefaultInstance;

/// Default recursion depth limit for decoding nested messages.
///
/// Protobuf implementations are required to enforce a recursion limit to
/// prevent stack overflow from deeply nested messages in untrusted input.
/// This value (100) matches the limit used by the official protobuf
/// implementations and the protobuf conformance suite.
///
/// Pass this constant as the depth when constructing a [`DecodeContext`] for
/// a top-level [`Message::merge`] call.  The provided convenience methods
/// ([`Message::decode`], [`Message::decode_from_slice`],
/// [`Message::merge_from_slice`]) use this limit automatically.
pub const RECURSION_LIMIT: u32 = 100;

/// Default limit on unknown fields decoded per top-level decode: 1,000,000.
///
/// Bounds the number of [`UnknownField`](crate::UnknownField) values the
/// decoder will materialize in a single top-level decode, independent of
/// the input size. Without this bound, wire data can force allocation far
/// in excess of its own size: every 2-byte unknown varint field
/// materialises a ~40-byte `UnknownField`, a ~20× amplification, so a
/// 64 MiB payload of unknown fields would otherwise force over 1 GiB of
/// heap. The count limit caps that overhead at roughly `limit × 40` bytes
/// (~40 MB at the default); unknown length-delimited *payload* bytes are
/// not counted against the limit because they are already bounded by the
/// input size, which [`DecodeOptions::with_max_message_size`] governs.
///
/// A million unknown fields is far more than any realistic
/// forward-compatibility scenario needs. Raise the limit with
/// [`DecodeOptions::with_unknown_field_limit`] if you decode trusted
/// messages that legitimately carry more (e.g. a proxy forwarding messages
/// with a huge unpacked repeated field from a much newer schema).
pub const DEFAULT_UNKNOWN_FIELD_LIMIT: usize = 1_000_000;

/// Per-decode limits threaded through every merge call.
///
/// Carries the remaining recursion depth and a shared unknown-field
/// allowance. The context is `Copy` — passing it to a callee hands over the
/// current depth by value, while the unknown-field allowance lives in a
/// [`Cell`](core::cell::Cell) owned by the top-level decode entry point, so
/// every field decoded under one entry point draws from the same
/// allowance.
///
/// Constructed automatically by the [`Message`] convenience methods
/// ([`decode`](Message::decode), [`decode_from_slice`](Message::decode_from_slice),
/// [`merge_from_slice`](Message::merge_from_slice)) and by [`DecodeOptions`].
/// Construct one manually only when calling [`Message::merge`] or the other
/// depth-threading methods directly — and construct a **fresh limit cell
/// per top-level decode**. Reusing one cell across decode calls makes the
/// limit cumulative: each call drains it further until every decode fails
/// with [`DecodeError::UnknownFieldLimitExceeded`].
///
/// ```rust
/// # use buffa::__doctest_fixtures::Person;
/// use core::cell::Cell;
/// use buffa::{DecodeContext, Message, DEFAULT_UNKNOWN_FIELD_LIMIT, RECURSION_LIMIT};
///
/// # fn example(mut bytes: &[u8]) -> Result<(), buffa::DecodeError> {
/// let limit = Cell::new(DEFAULT_UNKNOWN_FIELD_LIMIT);
/// let mut msg = Person::default();
/// msg.merge(&mut bytes, DecodeContext::new(RECURSION_LIMIT, &limit))?;
/// # Ok(())
/// # }
/// ```
#[derive(Clone, Copy, Debug)]
pub struct DecodeContext<'a> {
    depth: u32,
    unknown_fields_remaining: &'a core::cell::Cell<usize>,
}

impl<'a> DecodeContext<'a> {
    /// Create a context with `depth` remaining recursion levels and the
    /// remaining unknown-field allowance stored in `unknown_field_limit`.
    #[must_use]
    pub fn new(depth: u32, unknown_field_limit: &'a core::cell::Cell<usize>) -> Self {
        Self {
            depth,
            unknown_fields_remaining: unknown_field_limit,
        }
    }

    /// The remaining recursion depth.
    #[must_use]
    pub fn depth(&self) -> u32 {
        self.depth
    }

    /// The number of additional unknown fields this decode may materialize.
    #[must_use]
    pub fn remaining_unknown_fields(&self) -> usize {
        self.unknown_fields_remaining.get()
    }

    /// Consume one level of recursion depth.
    ///
    /// # Errors
    ///
    /// Returns [`DecodeError::RecursionLimitExceeded`] when the depth budget
    /// is exhausted.
    pub fn descend(self) -> Result<Self, DecodeError> {
        let depth = self
            .depth
            .checked_sub(1)
            .ok_or(DecodeError::RecursionLimitExceeded)?;
        Ok(Self { depth, ..self })
    }

    /// Consume one slot of the shared unknown-field allowance.
    ///
    /// Call **before** materializing an [`UnknownField`](crate::UnknownField).
    ///
    /// # Errors
    ///
    /// Returns [`DecodeError::UnknownFieldLimitExceeded`] (leaving the
    /// allowance unchanged) when no slots remain.
    pub fn register_unknown_field(&self) -> Result<(), DecodeError> {
        let remaining = self.unknown_fields_remaining.get();
        if remaining == 0 {
            return Err(DecodeError::UnknownFieldLimitExceeded);
        }
        self.unknown_fields_remaining.set(remaining - 1);
        Ok(())
    }
}

/// The core trait implemented by all protobuf message types.
///
/// This trait is implemented by **generated code** — you write a `.proto` file,
/// codegen emits the Rust struct and its `Message` impl. You should almost
/// never implement this trait by hand.
///
/// # Manual implementation is discouraged
///
/// The only reason to implement `Message` yourself is when you need a
/// custom in-memory representation that codegen cannot produce — for
/// example, wrapping a `std::ops::Range<i64>` as a leaf message so the
/// rest of your code uses the natural Rust type. If you just want a message
/// type, **write a `.proto` file instead.**
///
/// Manual implementation is intentionally high-friction:
/// - You must correctly implement the two-pass serialization contract
///   (`compute_size` populates the [`SizeCache`](crate::SizeCache) in the
///   same traversal order that `write_to` consumes it).
/// - You must implement wire-format decoding in `merge_field`.
/// - You must implement the [`DefaultInstance`] supertrait, which provides
///   the lazily-initialized static default that [`MessageField`](crate::MessageField)
///   dereferences to when unset.
///
/// If you still need to do this, see the [custom types section of the
/// user guide](https://github.com/anthropics/buffa/blob/main/docs/guide.md#custom-type-implementations)
/// for a complete worked example.
///
/// # Serialization model
///
/// Serialization is a two-pass process to avoid the exponential-time problem
/// that affects prost with deeply nested messages:
///
/// 1. **`compute_size()`** — walks the message tree and records the encoded
///    size of every length-delimited sub-message in a [`SizeCache`].
/// 2. **`write_to()`** — walks the tree again, writing bytes and consuming
///    cached sizes for length-prefixed sub-messages.
///
/// The provided [`encode`](Self::encode) method performs both passes with a
/// fresh [`SizeCache`] — most callers use that and never touch the cache
/// directly. `compute_size` / `write_to` take the cache explicitly so that
/// manual `Message` implementations can thread it through nested-message
/// recursion.
///
/// # Thread safety
///
/// `Message` requires `Send + Sync`. Generated structs contain no interior
/// mutability — serialization state lives in the external [`SizeCache`], not
/// in the message — so messages can be placed in an `Arc` and shared across
/// threads freely. `merge` requires `&mut self`, so mutation is exclusive.
///
/// [`SizeCache`]: crate::SizeCache
pub trait Message: DefaultInstance + Clone + PartialEq + Send + Sync {
    /// Compute the encoded byte size of this message, recording nested
    /// sub-message sizes in `cache` for `write_to` to consume.
    ///
    /// Most callers should use [`encode`](Self::encode) instead, which runs
    /// both passes with a fresh cache. Manual `Message` implementations call
    /// this recursively on nested message fields, wrapping each call in
    /// [`SizeCache::reserve`] / [`SizeCache::set`] for length-delimited
    /// fields — see the user guide's custom-types section for the pattern.
    ///
    /// # Size limit
    ///
    /// The protobuf specification limits messages to 2 GiB. The return type
    /// is `u32`, so messages whose encoded size exceeds `u32::MAX` (4 GiB)
    /// will produce a wrapped (undefined) size and a truncated encoding.
    /// Stay well within the 2 GiB spec limit.
    ///
    /// [`SizeCache::reserve`]: crate::SizeCache::reserve
    /// [`SizeCache::set`]: crate::SizeCache::set
    fn compute_size(&self, cache: &mut crate::SizeCache) -> u32;

    /// Write this message's encoded bytes to a buffer, consuming
    /// nested-message sizes from `cache` (populated by a prior
    /// `compute_size` call on the same cache).
    ///
    /// Most callers should use [`encode`](Self::encode) instead.
    fn write_to(&self, cache: &mut crate::SizeCache, buf: &mut impl BufMut);

    /// Compute size, then write. This is the primary encoding API.
    fn encode(&self, buf: &mut impl BufMut) {
        let mut cache = crate::SizeCache::new();
        self.compute_size(&mut cache);
        self.write_to(&mut cache, buf);
    }

    /// Encode using a caller-supplied [`SizeCache`](crate::SizeCache), for
    /// reuse across many encodes in a hot loop. Clears the cache first.
    fn encode_with_cache(&self, cache: &mut crate::SizeCache, buf: &mut impl BufMut) {
        cache.clear();
        self.compute_size(cache);
        self.write_to(cache, buf);
    }

    /// Compute the encoded byte size of this message.
    ///
    /// Walks the message tree, discarding the intermediate [`SizeCache`].
    /// If you also intend to encode, prefer [`encode`](Self::encode) or
    /// [`encode_to_vec`](Self::encode_to_vec) — they do a single size pass
    /// and reuse the cache for the write.
    ///
    /// [`SizeCache`]: crate::SizeCache
    #[must_use]
    fn encoded_len(&self) -> u32 {
        self.compute_size(&mut crate::SizeCache::new())
    }

    /// Encode this message as a length-delimited byte sequence.
    fn encode_length_delimited(&self, buf: &mut impl BufMut) {
        let mut cache = crate::SizeCache::new();
        let len = self.compute_size(&mut cache);
        crate::encoding::encode_varint(len as u64, buf);
        self.write_to(&mut cache, buf);
    }

    /// Encode this message to a new `Vec<u8>`.
    #[must_use]
    fn encode_to_vec(&self) -> alloc::vec::Vec<u8> {
        let mut cache = crate::SizeCache::new();
        let size = self.compute_size(&mut cache) as usize;
        let mut buf = alloc::vec::Vec::with_capacity(size);
        self.write_to(&mut cache, &mut buf);
        buf
    }

    /// Encode this message to a new [`bytes::Bytes`].
    ///
    /// Useful when handing off to networking code (hyper, tonic, axum)
    /// that expects `Bytes` frame or body payloads. Works in `no_std`.
    ///
    /// This is equivalent to `Bytes::from(self.encode_to_vec())` — both
    /// are zero-copy with respect to the encoded bytes — but saves readers
    /// from having to know that `From<Vec<u8>> for Bytes` is zero-copy.
    #[must_use]
    fn encode_to_bytes(&self) -> bytes::Bytes {
        let mut cache = crate::SizeCache::new();
        let size = self.compute_size(&mut cache) as usize;
        let mut buf = bytes::BytesMut::with_capacity(size);
        self.write_to(&mut cache, &mut buf);
        buf.freeze()
    }

    /// Decode a message from a buffer.
    fn decode(buf: &mut impl Buf) -> Result<Self, DecodeError>
    where
        Self: Sized,
    {
        let limit = core::cell::Cell::new(DEFAULT_UNKNOWN_FIELD_LIMIT);
        let mut msg = Self::default();
        msg.merge(buf, DecodeContext::new(RECURSION_LIMIT, &limit))?;
        Ok(msg)
    }

    /// Decode a message from a byte slice.
    ///
    /// Convenience wrapper around [`decode`](Self::decode) that avoids the
    /// `&mut bytes.as_slice()` incantation.
    fn decode_from_slice(mut data: &[u8]) -> Result<Self, DecodeError>
    where
        Self: Sized,
    {
        // `mut data` creates a local mutable copy of the fat pointer so that
        // `Buf::advance` can move the read cursor without affecting the caller.
        Self::decode(&mut data)
    }

    /// Decode a length-delimited message from a buffer.
    ///
    /// This is a **top-level** entry point.  It reads a varint length prefix,
    /// then decodes using arithmetic bounds checking, calling
    /// [`merge_to_limit`](Self::merge_to_limit) with a fresh
    /// [`RECURSION_LIMIT`] budget.  Any sub-messages inside are decoded via
    /// [`merge_length_delimited`](Self::merge_length_delimited), which tracks
    /// and decrements the budget.
    ///
    /// Do **not** call this method from within a
    /// [`merge_to_limit`](Self::merge_to_limit) implementation to decode a
    /// nested sub-message field; use
    /// [`merge_length_delimited`](Self::merge_length_delimited) instead so
    /// that the caller's depth budget is propagated correctly.
    fn decode_length_delimited(buf: &mut impl Buf) -> Result<Self, DecodeError>
    where
        Self: Sized,
    {
        // Refuse messages larger than 2 GiB to prevent allocating attacker-
        // controlled amounts of memory from a crafted length prefix.
        const MAX_MESSAGE_BYTES: u64 = 0x7FFF_FFFF;
        let len_u64 = crate::encoding::decode_varint(buf)?;
        if len_u64 > MAX_MESSAGE_BYTES {
            return Err(DecodeError::MessageTooLarge);
        }
        // Safe on 32-bit: len_u64 <= 2 GiB - 1 < u32::MAX, so the cast never truncates.
        let len = usize::try_from(len_u64).map_err(|_| DecodeError::MessageTooLarge)?;
        if buf.remaining() < len {
            return Err(DecodeError::UnexpectedEof);
        }
        // Arithmetic limit: decode `len` bytes from the buffer without
        // wrapping it in `Take`.  This keeps the buffer type `B` unchanged
        // through every recursion level, avoiding E0275 for recursive
        // message types like `google.protobuf.Struct ↔ Value`.
        let limit = buf.remaining() - len;
        let field_limit = core::cell::Cell::new(DEFAULT_UNKNOWN_FIELD_LIMIT);
        let mut msg = Self::default();
        msg.merge_to_limit(
            buf,
            DecodeContext::new(RECURSION_LIMIT, &field_limit),
            limit,
        )?;
        if buf.remaining() != limit {
            let remaining = buf.remaining();
            if remaining > limit {
                buf.advance(remaining - limit);
            } else {
                return Err(DecodeError::UnexpectedEof);
            }
        }
        Ok(msg)
    }

    /// Processes a single already-decoded tag and its associated field data
    /// from `buf`.
    ///
    /// This is the per-field dispatch method generated for each message type.
    /// Both [`merge_to_limit`](Self::merge_to_limit) and
    /// [`merge_group`](Self::merge_group) call this in their respective loops.
    ///
    /// `ctx` carries the remaining nesting depth and the shared allocation
    /// budget.
    ///
    /// # Errors
    ///
    /// Returns a [`DecodeError`] if:
    /// - the buffer is truncated or malformed,
    /// - a wire-type mismatch is detected for a known field,
    /// - the recursion limit is exceeded, or
    /// - the allocation budget is exhausted.
    fn merge_field(
        &mut self,
        tag: crate::encoding::Tag,
        buf: &mut impl Buf,
        ctx: DecodeContext<'_>,
    ) -> Result<(), DecodeError>;

    /// Merge fields from a buffer until `buf.remaining()` reaches `limit`.
    ///
    /// This is the core decode loop.  [`merge`](Self::merge) delegates to this
    /// with `limit = 0` (read until exhausted).
    /// [`merge_length_delimited`](Self::merge_length_delimited) computes
    /// `limit` from the declared sub-message length and calls this directly.
    ///
    /// The caller must ensure `limit <= buf.remaining()`.  The default
    /// implementations of [`merge`](Self::merge) and
    /// [`merge_length_delimited`](Self::merge_length_delimited) uphold this
    /// invariant.
    ///
    /// `ctx` carries the remaining nesting depth and the shared allocation
    /// budget.  Each call to
    /// [`merge_length_delimited`](Self::merge_length_delimited) consumes one
    /// depth level before recursing; when the depth reaches zero the call
    /// returns [`DecodeError::RecursionLimitExceeded`].
    fn merge_to_limit(
        &mut self,
        buf: &mut impl Buf,
        ctx: DecodeContext<'_>,
        limit: usize,
    ) -> Result<(), DecodeError> {
        while buf.remaining() > limit {
            let tag = crate::encoding::Tag::decode(buf)?;
            self.merge_field(tag, buf, ctx)?;
        }
        Ok(())
    }

    /// Merges a group-encoded message from `buf`, reading fields until an
    /// EndGroup tag with the given `field_number` is encountered.
    ///
    /// Proto2 groups use StartGroup/EndGroup wire types instead of
    /// length-delimited encoding. The opening StartGroup tag has already been
    /// consumed by the caller; this method reads the group body and the
    /// closing EndGroup tag.
    ///
    /// # Errors
    ///
    /// Returns a [`DecodeError`] if:
    /// - the buffer is truncated before the EndGroup tag,
    /// - an EndGroup tag is encountered with a mismatched field number,
    /// - a wire-type mismatch is detected for a known field, or
    /// - the recursion limit is exceeded.
    fn merge_group(
        &mut self,
        buf: &mut impl Buf,
        ctx: DecodeContext<'_>,
        field_number: u32,
    ) -> Result<(), DecodeError> {
        let ctx = ctx.descend()?;
        loop {
            if !buf.has_remaining() {
                return Err(DecodeError::UnexpectedEof);
            }
            let tag = crate::encoding::Tag::decode(buf)?;
            if tag.wire_type() == crate::encoding::WireType::EndGroup {
                return if tag.field_number() == field_number {
                    Ok(())
                } else {
                    Err(DecodeError::InvalidEndGroup(tag.field_number()))
                };
            }
            self.merge_field(tag, buf, ctx)?;
        }
    }

    /// Merge fields from a buffer into this message.
    ///
    /// Fields that are already set will be overwritten for singular fields,
    /// or appended for repeated fields, following standard protobuf merge
    /// semantics.
    ///
    /// `ctx` carries the remaining nesting depth and the shared allocation
    /// budget.  Each call to
    /// [`merge_length_delimited`](Self::merge_length_delimited) consumes one
    /// depth level before recursing; when the depth reaches zero the call
    /// returns [`DecodeError::RecursionLimitExceeded`].  Construct a fresh
    /// [`DecodeContext`] at the outermost call site, or use the convenience
    /// methods ([`decode`](Self::decode),
    /// [`merge_from_slice`](Self::merge_from_slice)) which do this
    /// automatically.
    fn merge(&mut self, buf: &mut impl Buf, ctx: DecodeContext<'_>) -> Result<(), DecodeError> {
        self.merge_to_limit(buf, ctx, 0)
    }

    /// Merge fields from a byte slice into this message.
    ///
    /// Convenience wrapper around [`merge`](Self::merge) that avoids the
    /// `&mut bytes.as_slice()` incantation.
    fn merge_from_slice(&mut self, mut data: &[u8]) -> Result<(), DecodeError> {
        let limit = core::cell::Cell::new(DEFAULT_UNKNOWN_FIELD_LIMIT);
        self.merge(&mut data, DecodeContext::new(RECURSION_LIMIT, &limit))
    }

    /// Merge fields from a length-delimited sub-message payload into this message.
    ///
    /// Reads a varint length prefix, then calls [`merge_to_limit`](Self::merge_to_limit)
    /// with an arithmetic bound derived from the declared sub-message length.
    /// The buffer type `B` passes through unchanged at every recursion level,
    /// avoiding the `E0275` trait-solver recursion limit that occurs with
    /// `Take<&mut Take<&mut T>>` type growth.
    ///
    /// Used by generated code when decoding singular `MessageField<T>` fields
    /// — the sub-message is merged into the existing value rather than
    /// replaced, per protobuf merge semantics.
    ///
    /// `ctx` carries the remaining nesting depth and the shared allocation
    /// budget passed down from the enclosing
    /// [`merge_to_limit`](Self::merge_to_limit) call.  This method consumes
    /// one depth level before calling the inner `merge_to_limit`; when the
    /// depth reaches zero it returns
    /// [`DecodeError::RecursionLimitExceeded`].
    ///
    /// Enforces the same 2 GiB safety limit as [`decode_length_delimited`](Self::decode_length_delimited).
    ///
    /// # Errors
    ///
    /// Returns an error if the buffer is too short, if the declared length
    /// exceeds 2 GiB, if the recursion limit is reached, or if the inner
    /// `merge_to_limit` call fails.
    fn merge_length_delimited(
        &mut self,
        buf: &mut impl Buf,
        ctx: DecodeContext<'_>,
    ) -> Result<(), DecodeError> {
        let ctx = ctx.descend()?;
        const MAX_SUB_MESSAGE_BYTES: u64 = 0x7FFF_FFFF;
        let len_u64 = crate::encoding::decode_varint(buf)?;
        if len_u64 > MAX_SUB_MESSAGE_BYTES {
            return Err(DecodeError::MessageTooLarge);
        }
        let len = usize::try_from(len_u64).map_err(|_| DecodeError::MessageTooLarge)?;
        if buf.remaining() < len {
            return Err(DecodeError::UnexpectedEof);
        }
        // Arithmetic limit: the sub-message occupies `len` bytes, so the
        // decode loop should stop when `buf.remaining()` drops to
        // `remaining - len`.  This avoids wrapping the buffer in `Take`,
        // which would grow the type at each recursion level and trigger
        // E0275 for recursive message types like `Struct ↔ Value`.
        let limit = buf.remaining() - len;
        self.merge_to_limit(buf, ctx, limit)?;
        if buf.remaining() != limit {
            let remaining = buf.remaining();
            if remaining > limit {
                // Sub-message consumed fewer bytes than declared; skip the rest.
                buf.advance(remaining - limit);
            } else {
                return Err(DecodeError::UnexpectedEof);
            }
        }
        Ok(())
    }

    /// Clear all fields to their default values.
    fn clear(&mut self);
}

/// Compile-time access to a generated message's protobuf identifiers.
///
/// Generic code that needs to *name* a message type — type-erased event
/// registries, structured logging, `Any` packing, schema lookups — can
/// bound on `T: MessageName` and read [`PACKAGE`], [`NAME`],
/// [`FULL_NAME`], or [`TYPE_URL`] without descriptor machinery or runtime
/// reflection. All four are `&'static str` literals computed at codegen
/// time, so there's no allocation or concatenation at runtime — unlike
/// `prost::Name`, whose `full_name()` and `type_url()` are runtime
/// `format!` calls.
///
/// Bring `buffa::MessageName` into scope to use `MyMessage::FULL_NAME`.
/// Without the trait in scope, use
/// `<MyMessage as buffa::MessageName>::FULL_NAME`.
///
/// Codegen implements `MessageName` for both the owned message type and
/// its zero-copy view type (`MyMessageView<'a>`), so the same generic
/// bound dispatches either. The trait has **no** [`Message`] supertrait —
/// it doesn't reach into the wire codec and a name-keyed registry should
/// be able to register a type without proving it can encode.
///
/// Hand-written [`Message`] implementations can opt in by also
/// implementing `MessageName`; it is a separate trait specifically so
/// that omitting it stays non-breaking. For messages that also implement
/// [`ExtensionSet`](crate::ExtensionSet), [`FULL_NAME`] is guaranteed
/// equal to [`ExtensionSet::PROTO_FQN`](crate::ExtensionSet::PROTO_FQN);
/// the inherent `MyMessage::TYPE_URL` const is equal to [`TYPE_URL`].
/// All derive from the same `proto_fqn` source in codegen.
///
/// Because the only items are associated `const`s, this trait is **not**
/// object-safe (`dyn MessageName` does not compile). Use it as a generic
/// bound (`fn foo<T: MessageName>()`), not a trait object.
///
/// ```
/// # use buffa::MessageName;
/// /// A name-keyed registry can register any `MessageName` type — owned
/// /// or view — without proving it can encode.
/// fn registry_key<T: MessageName>() -> &'static str {
///     T::FULL_NAME
/// }
/// # // No generated types in `buffa` itself; just check it monomorphises.
/// # struct Demo;
/// # impl MessageName for Demo {
/// #     const PACKAGE: &'static str = "demo";
/// #     const NAME: &'static str = "Demo";
/// #     const FULL_NAME: &'static str = "demo.Demo";
/// #     const TYPE_URL: &'static str = "type.googleapis.com/demo.Demo";
/// # }
/// assert_eq!(registry_key::<Demo>(), "demo.Demo");
/// ```
///
/// [`PACKAGE`]: Self::PACKAGE
/// [`NAME`]: Self::NAME
/// [`FULL_NAME`]: Self::FULL_NAME
/// [`TYPE_URL`]: Self::TYPE_URL
pub trait MessageName {
    /// The protobuf package the message is declared in.
    ///
    /// `"my.pkg"` for `package my.pkg;`. Empty string for the unnamed
    /// root package. Does not include a leading or trailing dot.
    const PACKAGE: &'static str;

    /// The unqualified message name, with `.` between nesting levels.
    ///
    /// `"Foo"` for a top-level message; `"Outer.Inner"` for a message
    /// nested inside `Outer`. This is the "type name relative to the
    /// package" — what `prost::Name::NAME` calls the same thing — *not*
    /// `DescriptorProto.name`, which is only the leaf segment (`"Inner"`)
    /// for nested types.
    const NAME: &'static str;

    /// The fully-qualified protobuf type name with no leading dot.
    ///
    /// `"my.pkg.Outer.Inner"` for a nested message in package `my.pkg`,
    /// or just `"Foo"` for a top-level message in the unnamed root
    /// package. Equal to `PACKAGE` + `"."` + `NAME` (with the joining
    /// dot omitted when `PACKAGE` is empty); shipped as its own const
    /// because consumers almost always want the joined form and the
    /// dotted string can't be re-split unambiguously
    /// (`foo.Bar.Baz` could be package `foo.Bar` + message `Baz`, or
    /// package `foo` + nested `Bar.Baz`).
    const FULL_NAME: &'static str;

    /// The `google.protobuf.Any.type_url` form for this message.
    ///
    /// `"type.googleapis.com/" + FULL_NAME`. This is the value the
    /// runtime stores in [`Any::type_url`] when packing a message, and
    /// the one a generic `Any` registry should key on.
    ///
    /// [`Any::type_url`]: https://protobuf.dev/programming-guides/proto3/#any
    const TYPE_URL: &'static str;
}

/// Options for configuring message decoding behavior.
///
/// Use this to set custom recursion depth limits or maximum message sizes
/// when decoding from untrusted input.
///
/// # Examples
///
/// ```no_run
/// # use buffa::__doctest_fixtures::Person;
/// use buffa::DecodeOptions;
///
/// # fn example(bytes: &[u8]) -> Result<(), buffa::DecodeError> {
/// // Restrict recursion depth to 50 and message size to 1 MiB:
/// let msg: Person = DecodeOptions::new()
///     .with_recursion_limit(50)
///     .with_max_message_size(1024 * 1024)
///     .decode_from_slice(bytes)?;
/// # Ok(())
/// # }
/// ```
#[derive(Debug, Clone)]
pub struct DecodeOptions {
    recursion_limit: u32,
    max_message_size: usize,
    unbounded_reader_size: bool,
    unknown_field_limit: usize,
}

/// Default maximum message size: 2 GiB - 1 (matches the internal sub-message
/// limit in `merge_length_delimited`).
const DEFAULT_MAX_MESSAGE_SIZE: usize = 0x7FFF_FFFF;

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

impl DecodeOptions {
    /// Create new decode options with defaults.
    ///
    /// Defaults:
    /// - `recursion_limit`: 100 (same as [`RECURSION_LIMIT`])
    /// - `max_message_size`: 2 GiB - 1
    /// - `unknown_field_limit`: 1,000,000 (same as [`DEFAULT_UNKNOWN_FIELD_LIMIT`])
    pub fn new() -> Self {
        Self {
            recursion_limit: RECURSION_LIMIT,
            max_message_size: DEFAULT_MAX_MESSAGE_SIZE,
            unbounded_reader_size: false,
            unknown_field_limit: DEFAULT_UNKNOWN_FIELD_LIMIT,
        }
    }

    /// Set the maximum recursion depth for nested messages.
    ///
    /// Each nested sub-message consumes one level of depth budget. When
    /// the budget reaches zero, decoding returns
    /// [`DecodeError::RecursionLimitExceeded`].
    ///
    /// Default: 100.
    #[must_use]
    pub fn with_recursion_limit(mut self, limit: u32) -> Self {
        self.recursion_limit = limit;
        self
    }

    /// Set the maximum total message size in bytes.
    ///
    /// If the input buffer or length-delimited payload exceeds this size,
    /// decoding returns [`DecodeError::MessageTooLarge`].
    ///
    /// Values above the protobuf message-size limit (2 GiB - 1) are clamped to
    /// that limit. Debug builds assert on out-of-range values so accidental
    /// `usize::MAX` sentinels are caught during development. On `std` builds,
    /// use `without_reader_size_limit` when EOF-bounded `decode_reader` input
    /// intentionally has no byte cap.
    ///
    /// Calling this re-enables the reader byte cap, overriding any prior
    /// `without_reader_size_limit`.
    ///
    /// This is checked at the top-level decode entry point. Individual
    /// sub-messages are still bounded by the internal 2 GiB limit
    /// regardless of this setting.
    ///
    /// Default: 2 GiB - 1 (0x7FFF_FFFF).
    #[must_use]
    pub fn with_max_message_size(mut self, max_bytes: usize) -> Self {
        debug_assert!(
            max_bytes <= DEFAULT_MAX_MESSAGE_SIZE,
            "DecodeOptions::with_max_message_size clamps values above the protobuf 2 GiB limit; \
             on std builds, use DecodeOptions::without_reader_size_limit for intentionally \
             unbounded reader input — there is no unbounded slice/Buf path"
        );
        self.max_message_size = max_bytes.min(DEFAULT_MAX_MESSAGE_SIZE);
        self.unbounded_reader_size = false;
        self
    }

    /// Remove the byte cap for EOF-bounded [`decode_reader`](Self::decode_reader)
    /// input.
    ///
    /// This is only used by the `std::io::Read` entry point that reads until
    /// EOF. Slice-, [`Buf`]-, and view-based entry points remain bounded by
    /// the configured [`with_max_message_size`](Self::with_max_message_size)
    /// (itself capped at the protobuf 2 GiB - 1 maximum), and length-delimited
    /// paths keep the same hard cap because their declared length is
    /// attacker-controlled.
    ///
    /// An unbounded reader can exhaust memory if the source does not end or is
    /// larger than available allocation capacity. Prefer
    /// [`with_max_message_size`](Self::with_max_message_size) for untrusted
    /// input.
    #[cfg(feature = "std")]
    #[must_use]
    pub fn without_reader_size_limit(mut self) -> Self {
        self.unbounded_reader_size = true;
        self
    }

    /// Set the maximum number of unknown fields decoded per decode call.
    ///
    /// Each decoded unknown field occupies a ~40-byte
    /// [`UnknownField`](crate::UnknownField) slot regardless of its wire
    /// size (a minimal field is 2 wire bytes — a ~20× amplification), so an
    /// input-size cap alone does not bound decoder memory; this limit does,
    /// at roughly `limit × 40` bytes of slot overhead. Unknown
    /// length-delimited *payload* bytes are not counted — they are bounded
    /// by the input size, which
    /// [`with_max_message_size`](Self::with_max_message_size) governs. When
    /// the limit is exceeded, decoding returns
    /// [`DecodeError::UnknownFieldLimitExceeded`].
    ///
    /// Zero-copy view decoding ([`decode_view`](Self::decode_view)) charges
    /// one slot per unknown field — including fields nested inside unknown
    /// groups — even though views store unknown fields as coalesced spans
    /// (~16 bytes per contiguous run): coalescing bounds view memory, while
    /// this limit bounds what converting the view to an owned message would
    /// materialize. Conversion replays under exactly the budget decoding
    /// charged, so a view that decodes within this limit always converts.
    ///
    /// Default: 1,000,000 ([`DEFAULT_UNKNOWN_FIELD_LIMIT`]).
    #[must_use]
    pub fn with_unknown_field_limit(mut self, count: usize) -> Self {
        self.unknown_field_limit = count;
        self
    }

    /// Returns the configured recursion depth limit.
    pub fn recursion_limit(&self) -> u32 {
        self.recursion_limit
    }

    /// Returns the configured unknown-field limit.
    pub fn unknown_field_limit(&self) -> usize {
        self.unknown_field_limit
    }

    /// Returns the configured maximum message size in bytes for bounded decode
    /// entry points.
    ///
    /// This returns the configured (clamped) value even when
    /// `without_reader_size_limit` is enabled for EOF-bounded reader input —
    /// the slice/`Buf`/view paths still honor it. Use
    /// `is_reader_size_unbounded` (std only) to inspect the reader flag.
    pub fn max_message_size(&self) -> usize {
        self.max_message_size
    }

    /// Returns whether EOF-bounded reader input has no byte cap.
    #[cfg(feature = "std")]
    pub fn is_reader_size_unbounded(&self) -> bool {
        self.unbounded_reader_size
    }

    /// Decode a message from a buffer.
    pub fn decode<M: Message>(&self, buf: &mut impl Buf) -> Result<M, DecodeError> {
        if buf.remaining() > self.max_message_size {
            return Err(DecodeError::MessageTooLarge);
        }
        let limit = core::cell::Cell::new(self.unknown_field_limit);
        let mut msg = M::default();
        msg.merge(buf, DecodeContext::new(self.recursion_limit, &limit))?;
        Ok(msg)
    }

    /// Decode a message from a byte slice.
    pub fn decode_from_slice<M: Message>(&self, data: &[u8]) -> Result<M, DecodeError> {
        if data.len() > self.max_message_size {
            return Err(DecodeError::MessageTooLarge);
        }
        self.decode_from_slice_unchecked_size(data)
    }

    fn decode_from_slice_unchecked_size<M: Message>(&self, data: &[u8]) -> Result<M, DecodeError> {
        let limit = core::cell::Cell::new(self.unknown_field_limit);
        let mut msg = M::default();
        msg.merge(
            &mut &*data,
            DecodeContext::new(self.recursion_limit, &limit),
        )?;
        Ok(msg)
    }

    /// Decode a length-delimited message from a buffer.
    pub fn decode_length_delimited<M: Message>(
        &self,
        buf: &mut impl Buf,
    ) -> Result<M, DecodeError> {
        // Enforce the 2 GiB internal safety cap even if the user sets a
        // larger max_message_size, to prevent allocating attacker-controlled
        // amounts of memory from a crafted length prefix.
        let max = core::cmp::min(
            self.max_message_size as u64,
            DEFAULT_MAX_MESSAGE_SIZE as u64,
        );
        let len_u64 = crate::encoding::decode_varint(buf)?;
        if len_u64 > max {
            return Err(DecodeError::MessageTooLarge);
        }
        let len = usize::try_from(len_u64).map_err(|_| DecodeError::MessageTooLarge)?;
        if buf.remaining() < len {
            return Err(DecodeError::UnexpectedEof);
        }
        let limit = buf.remaining() - len;
        let field_limit = core::cell::Cell::new(self.unknown_field_limit);
        let mut msg = M::default();
        msg.merge_to_limit(
            buf,
            DecodeContext::new(self.recursion_limit, &field_limit),
            limit,
        )?;
        if buf.remaining() != limit {
            let remaining = buf.remaining();
            if remaining > limit {
                buf.advance(remaining - limit);
            } else {
                return Err(DecodeError::UnexpectedEof);
            }
        }
        Ok(msg)
    }

    /// Merge fields from a buffer into an existing message.
    pub fn merge<M: Message>(&self, msg: &mut M, buf: &mut impl Buf) -> Result<(), DecodeError> {
        if buf.remaining() > self.max_message_size {
            return Err(DecodeError::MessageTooLarge);
        }
        let limit = core::cell::Cell::new(self.unknown_field_limit);
        msg.merge(buf, DecodeContext::new(self.recursion_limit, &limit))
    }

    /// Merge fields from a byte slice into an existing message.
    pub fn merge_from_slice<M: Message>(
        &self,
        msg: &mut M,
        data: &[u8],
    ) -> Result<(), DecodeError> {
        if data.len() > self.max_message_size {
            return Err(DecodeError::MessageTooLarge);
        }
        let limit = core::cell::Cell::new(self.unknown_field_limit);
        msg.merge(
            &mut &*data,
            DecodeContext::new(self.recursion_limit, &limit),
        )
    }

    /// Decode a zero-copy view from a byte slice.
    ///
    /// Enforces `max_message_size` on the input, and passes the recursion
    /// limit and unknown-field limit to the view decoder (views charge the
    /// unknown-field limit per field, including fields nested in unknown
    /// groups — see
    /// [`with_unknown_field_limit`](Self::with_unknown_field_limit)).
    ///
    /// # Errors
    ///
    /// Returns [`DecodeError::MessageTooLarge`] for oversized input, or any
    /// error from the view decoder (malformed wire data, recursion limit,
    /// unknown-field limit).
    pub fn decode_view<'a, V: crate::view::MessageView<'a>>(
        &self,
        buf: &'a [u8],
    ) -> Result<V, DecodeError> {
        if buf.len() > self.max_message_size {
            return Err(DecodeError::MessageTooLarge);
        }
        let limit = core::cell::Cell::new(self.unknown_field_limit);
        V::decode_view_with_ctx(buf, DecodeContext::new(self.recursion_limit, &limit))
    }

    /// Decode a lazy view from a byte slice (see
    /// [`LazyMessageView`](crate::view::LazyMessageView)).
    ///
    /// The budgets remaining at each deferred field's position are recorded
    /// and charged when that field is accessed, so the configured limits
    /// flow through deferred decoding. Unlike
    /// [`decode_view`](Self::decode_view), the unknown-field limit is not
    /// enforced globally across the message tree at decode time: each
    /// deferred subtree independently replays the allowance recorded at its
    /// position, so a full traversal can materialize unknown-field records
    /// proportional to input size. Prefer `decode_view` for untrusted input
    /// if the global bound matters.
    ///
    /// # Errors
    ///
    /// Returns [`DecodeError::MessageTooLarge`] for oversized input, or any
    /// error from decoding the message's own fields — including
    /// [`DecodeError::RecursionLimitExceeded`] /
    /// [`DecodeError::UnknownFieldLimitExceeded`] when the configured limits
    /// are exhausted by them. Deferred sub-message bytes surface errors on
    /// access instead.
    pub fn decode_lazy_view<'a, L: crate::view::LazyMessageView<'a>>(
        &self,
        buf: &'a [u8],
    ) -> Result<L, DecodeError> {
        if buf.len() > self.max_message_size {
            return Err(DecodeError::MessageTooLarge);
        }
        let limit = core::cell::Cell::new(self.unknown_field_limit);
        L::decode_lazy_with_ctx(buf, DecodeContext::new(self.recursion_limit, &limit))
    }

    /// Decode a message by reading all bytes from a [`std::io::Read`] source.
    ///
    /// Reads until EOF, enforces the configured reader size limit unless
    /// [`without_reader_size_limit`](Self::without_reader_size_limit) was
    /// selected, then decodes the buffered bytes. Returns `std::io::Error` to
    /// be compatible with `Read`-based error handling.
    #[cfg(feature = "std")]
    pub fn decode_reader<M: Message>(
        &self,
        reader: &mut impl std::io::Read,
    ) -> Result<M, std::io::Error> {
        let bytes = self.read_limited(reader)?;
        self.decode_from_slice_unchecked_size::<M>(&bytes)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
    }

    /// Decode a length-delimited message from a [`std::io::Read`] source.
    ///
    /// Reads a varint length prefix, enforces `max_message_size`, reads
    /// exactly that many bytes, then decodes. Useful for reading sequential
    /// length-delimited messages from a file or stream.
    ///
    /// The declared length is treated as untrusted: the read buffer grows
    /// as bytes actually arrive rather than being allocated up front, so a
    /// source that declares a large length but never delivers the bytes
    /// cannot force a large allocation.
    #[cfg(feature = "std")]
    pub fn decode_length_delimited_reader<M: Message>(
        &self,
        reader: &mut impl std::io::Read,
    ) -> Result<M, std::io::Error> {
        use std::io::Read as _;
        let len = read_varint(reader)?;
        let max = core::cmp::min(
            self.max_message_size as u64,
            DEFAULT_MAX_MESSAGE_SIZE as u64,
        );
        if len > max {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                DecodeError::MessageTooLarge,
            ));
        }
        let len = usize::try_from(len).map_err(|_| {
            std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                DecodeError::MessageTooLarge,
            )
        })?;
        // Pre-size only up to a small bound; `read_to_end` grows the buffer
        // geometrically as data is actually delivered, so peak allocation
        // tracks delivered bytes, not the wire-declared length.
        const INITIAL_CAPACITY_CAP: usize = 64 * 1024;
        let mut buf = alloc::vec::Vec::with_capacity(len.min(INITIAL_CAPACITY_CAP));
        reader.take(len as u64).read_to_end(&mut buf)?;
        if buf.len() < len {
            return Err(std::io::Error::new(
                std::io::ErrorKind::UnexpectedEof,
                DecodeError::UnexpectedEof,
            ));
        }
        self.decode_from_slice::<M>(&buf)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
    }

    /// Read all bytes from a reader up to the configured reader limit.
    #[cfg(feature = "std")]
    fn read_limited(
        &self,
        reader: &mut impl std::io::Read,
    ) -> Result<alloc::vec::Vec<u8>, std::io::Error> {
        use std::io::Read as _;
        let mut buf = alloc::vec::Vec::new();
        if self.unbounded_reader_size {
            reader.read_to_end(&mut buf)?;
            return Ok(buf);
        }
        reader
            .take((self.max_message_size as u64).saturating_add(1))
            .read_to_end(&mut buf)?;
        if buf.len() > self.max_message_size {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                DecodeError::MessageTooLarge,
            ));
        }
        Ok(buf)
    }
}

/// Read a varint from a `std::io::Read` source, one byte at a time.
///
/// Mirrors the validation in [`decode_varint_slow`](crate::encoding): a
/// 10th byte > `0x01` (overflow bits set, or continuation bit implying an
/// 11th byte) is rejected.
#[cfg(feature = "std")]
fn read_varint(reader: &mut impl std::io::Read) -> Result<u64, std::io::Error> {
    let mut value: u64 = 0;
    let mut shift: u32 = 0;
    loop {
        let mut byte = [0u8; 1];
        reader.read_exact(&mut byte)?;
        let b = byte[0];
        if shift < 63 {
            value |= ((b & 0x7F) as u64) << shift;
            if b < 0x80 {
                return Ok(value);
            }
            shift += 7;
        } else {
            // 10th byte: only bit 0 maps to bit 63 of the result. A byte
            // > 0x01 means either data overflow (bits 1-6 set) or an 11th
            // byte (continuation bit 0x80 set).
            if b > 0x01 {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    DecodeError::VarintTooLong,
                ));
            }
            value |= (b as u64) << 63;
            return Ok(value);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::encoding::encode_varint;
    use crate::error::DecodeError;
    use crate::message_field::DefaultInstance;
    use crate::SizeCache;

    // Minimal hand-written Message for testing merge_length_delimited.
    #[derive(Clone, Debug, Default, PartialEq)]
    struct FlatMsg {
        value: i32,
    }

    impl DefaultInstance for FlatMsg {
        fn default_instance() -> &'static Self {
            static INST: crate::__private::OnceBox<FlatMsg> = crate::__private::OnceBox::new();
            INST.get_or_init(|| alloc::boxed::Box::new(FlatMsg::default()))
        }
    }

    impl Message for FlatMsg {
        fn compute_size(&self, _cache: &mut SizeCache) -> u32 {
            if self.value != 0 {
                1 + crate::types::int32_encoded_len(self.value) as u32
            } else {
                0
            }
        }

        fn write_to(&self, _cache: &mut SizeCache, buf: &mut impl BufMut) {
            if self.value != 0 {
                crate::encoding::Tag::new(1, crate::encoding::WireType::Varint).encode(buf);
                crate::types::encode_int32(self.value, buf);
            }
        }

        fn merge_field(
            &mut self,
            tag: crate::encoding::Tag,
            buf: &mut impl Buf,
            _ctx: DecodeContext<'_>,
        ) -> Result<(), DecodeError> {
            match tag.field_number() {
                1 => {
                    self.value = crate::types::decode_int32(buf)?;
                }
                _ => {
                    crate::encoding::skip_field(tag, buf)?;
                }
            }
            Ok(())
        }

        fn clear(&mut self) {
            *self = Self::default();
        }
    }

    fn wire_bytes(msg: &FlatMsg) -> alloc::vec::Vec<u8> {
        let mut buf = alloc::vec::Vec::new();
        msg.encode_length_delimited(&mut buf);
        buf
    }

    #[test]
    fn test_merge_length_delimited_basic() {
        let src = FlatMsg { value: 42 };
        let mut dst = FlatMsg::default();
        dst.merge_length_delimited(
            &mut wire_bytes(&src).as_slice(),
            crate::test_ctx(RECURSION_LIMIT),
        )
        .unwrap();
        assert_eq!(dst.value, 42);
    }

    #[test]
    fn test_merge_length_delimited_merges_into_existing() {
        // Second merge overwrites (proto3 last-wins for scalar fields).
        let mut dst = FlatMsg::default();
        dst.merge_length_delimited(
            &mut wire_bytes(&FlatMsg { value: 1 }).as_slice(),
            crate::test_ctx(RECURSION_LIMIT),
        )
        .unwrap();
        assert_eq!(dst.value, 1);
        dst.merge_length_delimited(
            &mut wire_bytes(&FlatMsg { value: 2 }).as_slice(),
            crate::test_ctx(RECURSION_LIMIT),
        )
        .unwrap();
        assert_eq!(dst.value, 2);
    }

    #[test]
    fn test_merge_length_delimited_truncated() {
        // Length prefix says 10 bytes but buffer contains only 2.
        let mut buf = alloc::vec::Vec::new();
        encode_varint(10, &mut buf);
        buf.extend_from_slice(&[0x01, 0x01]);
        let mut dst = FlatMsg::default();
        assert_eq!(
            dst.merge_length_delimited(&mut buf.as_slice(), crate::test_ctx(RECURSION_LIMIT)),
            Err(DecodeError::UnexpectedEof)
        );
    }

    #[test]
    fn test_merge_length_delimited_oversized() {
        // Length prefix exceeds the 2 GiB safety limit.
        let mut buf = alloc::vec::Vec::new();
        encode_varint(0x8000_0000u64, &mut buf); // 2 GiB + 1
        let mut dst = FlatMsg::default();
        assert_eq!(
            dst.merge_length_delimited(&mut buf.as_slice(), crate::test_ctx(RECURSION_LIMIT)),
            Err(DecodeError::MessageTooLarge)
        );
    }

    #[test]
    fn test_merge_length_delimited_recursion_limit() {
        // depth=1 means merge_length_delimited will decrement to 0, then any
        // nested call would return RecursionLimitExceeded.  Passing depth=1
        // to merge_length_delimited itself is the boundary: it decrements to
        // 0 and calls merge with depth=0, which for FlatMsg (a leaf) succeeds.
        // Passing depth=0 directly must return RecursionLimitExceeded.
        let src = FlatMsg { value: 7 };
        let mut dst = FlatMsg::default();
        assert_eq!(
            dst.merge_length_delimited(&mut wire_bytes(&src).as_slice(), crate::test_ctx(0)),
            Err(DecodeError::RecursionLimitExceeded)
        );
        // depth=1 succeeds: exactly one level is consumed.
        dst.merge_length_delimited(&mut wire_bytes(&src).as_slice(), crate::test_ctx(1))
            .unwrap();
        assert_eq!(dst.value, 7);
    }

    #[test]
    fn test_decode_from_slice_basic() {
        let src = FlatMsg { value: 42 };
        let bytes = src.encode_to_vec();
        let dst = FlatMsg::decode_from_slice(&bytes).unwrap();
        assert_eq!(dst.value, 42);
    }

    #[test]
    fn test_encode_to_bytes_matches_encode_to_vec() {
        let src = FlatMsg { value: 42 };
        let vec = src.encode_to_vec();
        let bytes = src.encode_to_bytes();
        assert_eq!(vec.as_slice(), bytes.as_ref());
        // Round-trip through the Bytes variant.
        let dst = FlatMsg::decode_from_slice(&bytes).unwrap();
        assert_eq!(dst.value, 42);
        // Empty-message case: zero-length buffer is well-defined.
        assert!(FlatMsg::default().encode_to_bytes().is_empty());
    }

    #[test]
    fn test_decode_from_slice_empty() {
        let dst = FlatMsg::decode_from_slice(&[]).unwrap();
        assert_eq!(dst.value, 0);
    }

    #[test]
    fn test_decode_from_slice_invalid_returns_error() {
        // A lone 0xFF byte is not a valid varint tag.
        let result = FlatMsg::decode_from_slice(&[0xFF]);
        assert!(result.is_err());
    }

    #[test]
    fn test_merge_from_slice_basic() {
        let src = FlatMsg { value: 7 };
        let bytes = src.encode_to_vec();
        let mut dst = FlatMsg::default();
        dst.merge_from_slice(&bytes).unwrap();
        assert_eq!(dst.value, 7);
    }

    #[test]
    fn test_merge_from_slice_last_wins() {
        let src1 = FlatMsg { value: 1 };
        let src2 = FlatMsg { value: 2 };
        let mut dst = FlatMsg::default();
        dst.merge_from_slice(&src1.encode_to_vec()).unwrap();
        dst.merge_from_slice(&src2.encode_to_vec()).unwrap();
        // Proto3 last-wins semantics for scalar fields.
        assert_eq!(dst.value, 2);
    }

    // ── DecodeOptions tests ──────────────────────────────────────────

    #[test]
    fn test_decode_options_default_works() {
        let src = FlatMsg { value: 99 };
        let bytes = src.encode_to_vec();
        let msg: FlatMsg = DecodeOptions::new().decode_from_slice(&bytes).unwrap();
        assert_eq!(msg.value, 99);
    }

    #[test]
    fn test_decode_options_max_message_size_rejects() {
        let src = FlatMsg { value: 42 };
        let bytes = src.encode_to_vec();
        // Set max size to 1 byte — smaller than the encoded message.
        let result: Result<FlatMsg, _> = DecodeOptions::new()
            .with_max_message_size(1)
            .decode_from_slice(&bytes);
        assert_eq!(result, Err(DecodeError::MessageTooLarge));
    }

    #[test]
    fn test_decode_options_max_message_size_exact_boundary() {
        let src = FlatMsg { value: 42 };
        let bytes = src.encode_to_vec();
        // Exact size should succeed.
        let msg: FlatMsg = DecodeOptions::new()
            .with_max_message_size(bytes.len())
            .decode_from_slice(&bytes)
            .unwrap();
        assert_eq!(msg.value, 42);
        // One byte less should fail.
        let result: Result<FlatMsg, _> = DecodeOptions::new()
            .with_max_message_size(bytes.len() - 1)
            .decode_from_slice(&bytes);
        assert_eq!(result, Err(DecodeError::MessageTooLarge));
    }

    #[test]
    fn test_decode_options_custom_recursion_limit() {
        // FlatMsg has no nested messages, so any recursion limit >= 0 works.
        // Just verify the API compiles and runs.
        let src = FlatMsg { value: 7 };
        let bytes = src.encode_to_vec();
        let msg: FlatMsg = DecodeOptions::new()
            .with_recursion_limit(1)
            .decode_from_slice(&bytes)
            .unwrap();
        assert_eq!(msg.value, 7);
    }

    #[test]
    fn test_decode_options_merge() {
        let src = FlatMsg { value: 55 };
        let bytes = src.encode_to_vec();
        let mut msg = FlatMsg::default();
        DecodeOptions::new()
            .merge_from_slice(&mut msg, &bytes)
            .unwrap();
        assert_eq!(msg.value, 55);
    }

    #[test]
    fn test_decode_options_merge_rejects_oversize() {
        let src = FlatMsg { value: 55 };
        let bytes = src.encode_to_vec();
        let mut msg = FlatMsg::default();
        let result = DecodeOptions::new()
            .with_max_message_size(1)
            .merge_from_slice(&mut msg, &bytes);
        assert_eq!(result, Err(DecodeError::MessageTooLarge));
    }

    #[test]
    fn test_decode_options_length_delimited() {
        let src = FlatMsg { value: 42 };
        let mut ld_bytes = alloc::vec::Vec::new();
        src.encode_length_delimited(&mut ld_bytes);
        let msg: FlatMsg = DecodeOptions::new()
            .decode_length_delimited(&mut ld_bytes.as_slice())
            .unwrap();
        assert_eq!(msg.value, 42);
    }

    #[test]
    fn test_decode_options_length_delimited_rejects_oversize() {
        let src = FlatMsg { value: 42 };
        let mut ld_bytes = alloc::vec::Vec::new();
        src.encode_length_delimited(&mut ld_bytes);
        let result: Result<FlatMsg, _> = DecodeOptions::new()
            .with_max_message_size(1)
            .decode_length_delimited(&mut ld_bytes.as_slice());
        assert_eq!(result, Err(DecodeError::MessageTooLarge));
    }

    #[test]
    fn decode_options_getters_return_defaults() {
        let opts = DecodeOptions::new();
        assert_eq!(opts.recursion_limit(), RECURSION_LIMIT);
        assert_eq!(opts.max_message_size(), 0x7FFF_FFFF);
        assert_eq!(opts.unknown_field_limit(), DEFAULT_UNKNOWN_FIELD_LIMIT);
    }

    #[test]
    fn decode_options_getters_return_custom_values() {
        let opts = DecodeOptions::new()
            .with_recursion_limit(42)
            .with_max_message_size(1024)
            .with_unknown_field_limit(2048);
        assert_eq!(opts.recursion_limit(), 42);
        assert_eq!(opts.max_message_size(), 1024);
        assert_eq!(opts.unknown_field_limit(), 2048);
    }

    #[cfg(debug_assertions)]
    #[test]
    #[should_panic(expected = "protobuf 2 GiB limit")]
    fn decode_options_max_message_size_above_protobuf_limit_debug_asserts() {
        let _ = DecodeOptions::new().with_max_message_size(DEFAULT_MAX_MESSAGE_SIZE + 1);
    }

    #[cfg(not(debug_assertions))]
    #[test]
    fn decode_options_max_message_size_above_protobuf_limit_saturates() {
        let opts = DecodeOptions::new().with_max_message_size(DEFAULT_MAX_MESSAGE_SIZE + 1);
        assert_eq!(opts.max_message_size(), DEFAULT_MAX_MESSAGE_SIZE);
    }

    #[test]
    fn test_decode_options_default_impl() {
        // DecodeOptions::default() ≡ DecodeOptions::new().
        let opts = DecodeOptions::default();
        assert_eq!(opts.recursion_limit(), RECURSION_LIMIT);
        assert_eq!(opts.max_message_size(), 0x7FFF_FFFF);
        assert_eq!(opts.unknown_field_limit(), DEFAULT_UNKNOWN_FIELD_LIMIT);
    }

    #[test]
    fn test_decode_options_decode_buf() {
        // The Buf-taking decode() variant (vs decode_from_slice).
        let src = FlatMsg { value: 123 };
        let bytes = src.encode_to_vec();
        let msg: FlatMsg = DecodeOptions::new().decode(&mut bytes.as_slice()).unwrap();
        assert_eq!(msg.value, 123);
        // Oversize check on the Buf variant.
        let result: Result<FlatMsg, _> = DecodeOptions::new()
            .with_max_message_size(1)
            .decode(&mut bytes.as_slice());
        assert_eq!(result, Err(DecodeError::MessageTooLarge));
    }

    #[test]
    fn test_decode_options_merge_buf() {
        // The Buf-taking merge() variant (vs merge_from_slice).
        let src = FlatMsg { value: 77 };
        let bytes = src.encode_to_vec();
        let mut msg = FlatMsg::default();
        DecodeOptions::new()
            .merge(&mut msg, &mut bytes.as_slice())
            .unwrap();
        assert_eq!(msg.value, 77);
        // Oversize check.
        let mut msg = FlatMsg::default();
        let result = DecodeOptions::new()
            .with_max_message_size(1)
            .merge(&mut msg, &mut bytes.as_slice());
        assert_eq!(result, Err(DecodeError::MessageTooLarge));
    }

    // ── Message trait default methods ─────────────────────────────────

    #[test]
    fn test_message_encode_trait_default() {
        // Message::encode(buf) ≡ compute_size() then write_to(buf).
        let src = FlatMsg { value: 42 };
        let mut buf = alloc::vec::Vec::new();
        src.encode(&mut buf);
        assert_eq!(buf, src.encode_to_vec());
    }

    #[test]
    fn test_message_decode_length_delimited_trait_default() {
        // The trait-level decode_length_delimited (distinct from
        // DecodeOptions::decode_length_delimited).
        let src = FlatMsg { value: 42 };
        let mut ld = alloc::vec::Vec::new();
        src.encode_length_delimited(&mut ld);
        let got = FlatMsg::decode_length_delimited(&mut ld.as_slice()).unwrap();
        assert_eq!(got.value, 42);
    }

    #[test]
    fn test_message_decode_length_delimited_oversize() {
        // Length prefix > 2 GiB → MessageTooLarge.
        let mut buf = alloc::vec::Vec::new();
        encode_varint(0x8000_0000u64, &mut buf);
        let result = FlatMsg::decode_length_delimited(&mut buf.as_slice());
        assert_eq!(result, Err(DecodeError::MessageTooLarge));
    }

    #[test]
    fn test_message_decode_length_delimited_truncated() {
        // Length prefix says 10 bytes, buffer has 2.
        let mut buf = alloc::vec::Vec::new();
        encode_varint(10, &mut buf);
        buf.push(0x08);
        buf.push(0x01);
        let result = FlatMsg::decode_length_delimited(&mut buf.as_slice());
        assert_eq!(result, Err(DecodeError::UnexpectedEof));
    }

    #[test]
    fn test_message_decode_length_delimited_with_trailing() {
        // Buffer has two back-to-back length-delimited messages.
        // decode_length_delimited should consume exactly the first one
        // and leave the buffer positioned at the second.
        let a = FlatMsg { value: 1 };
        let b = FlatMsg { value: 2 };
        let mut buf = alloc::vec::Vec::new();
        a.encode_length_delimited(&mut buf);
        b.encode_length_delimited(&mut buf);

        let mut cur = buf.as_slice();
        let first = FlatMsg::decode_length_delimited(&mut cur).unwrap();
        assert_eq!(first.value, 1);
        let second = FlatMsg::decode_length_delimited(&mut cur).unwrap();
        assert_eq!(second.value, 2);
        assert!(cur.is_empty());
    }

    // ── merge_group tests ─────────────────────────────────────────────

    /// Build a group body for FlatMsg (field 1 = value) terminated by
    /// EndGroup with the given field number.
    fn group_bytes(value: i32, group_field_number: u32) -> alloc::vec::Vec<u8> {
        use crate::encoding::{Tag, WireType};
        let mut buf = alloc::vec::Vec::new();
        if value != 0 {
            Tag::new(1, WireType::Varint).encode(&mut buf);
            crate::types::encode_int32(value, &mut buf);
        }
        Tag::new(group_field_number, WireType::EndGroup).encode(&mut buf);
        buf
    }

    #[test]
    fn test_merge_group_basic() {
        let data = group_bytes(42, 5);
        let mut dst = FlatMsg::default();
        dst.merge_group(&mut data.as_slice(), crate::test_ctx(RECURSION_LIMIT), 5)
            .unwrap();
        assert_eq!(dst.value, 42);
    }

    #[test]
    fn test_merge_group_empty() {
        // Group with no fields — just EndGroup.
        let data = group_bytes(0, 3);
        let mut dst = FlatMsg::default();
        dst.merge_group(&mut data.as_slice(), crate::test_ctx(RECURSION_LIMIT), 3)
            .unwrap();
        assert_eq!(dst.value, 0);
    }

    #[test]
    fn test_merge_group_merges_into_existing() {
        let data1 = group_bytes(1, 5);
        let data2 = group_bytes(2, 5);
        let mut dst = FlatMsg::default();
        dst.merge_group(&mut data1.as_slice(), crate::test_ctx(RECURSION_LIMIT), 5)
            .unwrap();
        assert_eq!(dst.value, 1);
        dst.merge_group(&mut data2.as_slice(), crate::test_ctx(RECURSION_LIMIT), 5)
            .unwrap();
        assert_eq!(dst.value, 2);
    }

    #[test]
    fn test_merge_group_recursion_limit_zero() {
        // depth=0 should immediately fail with RecursionLimitExceeded
        // because merge_group decrements before entering the loop.
        let data = group_bytes(42, 5);
        let mut dst = FlatMsg::default();
        assert_eq!(
            dst.merge_group(&mut data.as_slice(), crate::test_ctx(0), 5),
            Err(DecodeError::RecursionLimitExceeded)
        );
    }

    #[test]
    fn test_merge_group_recursion_limit_one_succeeds() {
        // depth=1 succeeds: merge_group decrements to 0, but FlatMsg's
        // merge_field doesn't recurse further.
        let data = group_bytes(7, 5);
        let mut dst = FlatMsg::default();
        dst.merge_group(&mut data.as_slice(), crate::test_ctx(1), 5)
            .unwrap();
        assert_eq!(dst.value, 7);
    }

    #[test]
    fn test_merge_group_mismatched_end() {
        // EndGroup with wrong field number.
        use crate::encoding::{Tag, WireType};
        let mut data = alloc::vec::Vec::new();
        Tag::new(99, WireType::EndGroup).encode(&mut data);

        let mut dst = FlatMsg::default();
        assert_eq!(
            dst.merge_group(&mut data.as_slice(), crate::test_ctx(RECURSION_LIMIT), 5),
            Err(DecodeError::InvalidEndGroup(99))
        );
    }

    #[test]
    fn test_merge_group_truncated() {
        // Buffer ends without EndGroup tag.
        use crate::encoding::{Tag, WireType};
        let mut data = alloc::vec::Vec::new();
        Tag::new(1, WireType::Varint).encode(&mut data);
        crate::types::encode_int32(42, &mut data);
        // No EndGroup.

        let mut dst = FlatMsg::default();
        assert_eq!(
            dst.merge_group(&mut data.as_slice(), crate::test_ctx(RECURSION_LIMIT), 5),
            Err(DecodeError::UnexpectedEof)
        );
    }

    #[test]
    fn test_merge_group_empty_buffer() {
        let mut dst = FlatMsg::default();
        assert_eq!(
            dst.merge_group(&mut [].as_slice(), crate::test_ctx(RECURSION_LIMIT), 5),
            Err(DecodeError::UnexpectedEof)
        );
    }

    #[test]
    fn test_merge_group_unknown_fields_skipped() {
        // Group body contains an unknown field (field 99) which FlatMsg
        // routes to skip_field; the known field (field 1) should still
        // be decoded.
        use crate::encoding::{Tag, WireType};
        let mut data = alloc::vec::Vec::new();
        // Unknown varint field 99 = 0
        Tag::new(99, WireType::Varint).encode(&mut data);
        crate::encoding::encode_varint(0, &mut data);
        // Known field 1 = 99
        Tag::new(1, WireType::Varint).encode(&mut data);
        crate::types::encode_int32(99, &mut data);
        // EndGroup(5)
        Tag::new(5, WireType::EndGroup).encode(&mut data);

        let mut dst = FlatMsg::default();
        dst.merge_group(&mut data.as_slice(), crate::test_ctx(RECURSION_LIMIT), 5)
            .unwrap();
        assert_eq!(dst.value, 99);
    }

    #[test]
    fn test_merge_group_trailing_data_preserved() {
        // After the EndGroup tag, trailing data should remain in the buffer.
        let mut data = group_bytes(42, 5);
        data.extend_from_slice(&[0xDE, 0xAD]);

        let mut cur = data.as_slice();
        let mut dst = FlatMsg::default();
        dst.merge_group(&mut cur, crate::test_ctx(RECURSION_LIMIT), 5)
            .unwrap();
        assert_eq!(dst.value, 42);
        assert_eq!(cur, &[0xDE, 0xAD]);
    }

    // ── read_varint (std::io::Read) tests ──────────────────────────────

    #[cfg(feature = "std")]
    mod read_varint_tests {
        use super::super::read_varint;
        use crate::encoding::encode_varint;

        #[test]
        fn roundtrip_values() {
            let cases: &[u64] = &[0, 1, 127, 128, 300, 1 << 14, 1 << 35, 1 << 63, u64::MAX];
            for &v in cases {
                let mut buf = Vec::new();
                encode_varint(v, &mut buf);
                let got = read_varint(&mut buf.as_slice()).unwrap();
                assert_eq!(got, v, "roundtrip failed for {v}");
            }
        }

        #[test]
        fn rejects_10th_byte_overflow() {
            // 9 continuation bytes + 10th byte with overflow bit (0x02).
            // Mirrors encoding::tests::test_varint_10th_byte_overflow_rejected.
            let mut bad: Vec<u8> = vec![0xFF; 9];
            bad.push(0x02);
            let err = read_varint(&mut bad.as_slice()).unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        }

        #[test]
        fn rejects_11th_byte() {
            // 10 continuation bytes — implies an 11th byte is needed.
            let bad: &[u8] = &[0xFF; 10];
            let err = read_varint(&mut &bad[..]).unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        }

        #[test]
        fn u64_max_roundtrips() {
            // u64::MAX requires 10 bytes with the 10th byte == 0x01.
            let mut buf = Vec::new();
            encode_varint(u64::MAX, &mut buf);
            assert_eq!(buf.len(), 10);
            assert_eq!(buf[9], 0x01);
            let got = read_varint(&mut buf.as_slice()).unwrap();
            assert_eq!(got, u64::MAX);
        }

        #[test]
        fn eof_before_terminator_is_error() {
            // Single continuation byte with no follow-up.
            let bad: &[u8] = &[0x80];
            let err = read_varint(&mut &bad[..]).unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
        }

        #[test]
        fn empty_input_is_error() {
            let err = read_varint(&mut &[][..]).unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
        }
    }

    // ── DecodeOptions std::io::Read tests ─────────────────────────────

    #[cfg(feature = "std")]
    mod reader_tests {
        use super::*;

        #[test]
        fn decode_reader_basic() {
            let src = FlatMsg { value: 42 };
            let bytes = src.encode_to_vec();
            let msg: FlatMsg = DecodeOptions::new()
                .decode_reader(&mut bytes.as_slice())
                .unwrap();
            assert_eq!(msg.value, 42);
        }

        #[test]
        fn decode_reader_rejects_oversize() {
            let src = FlatMsg { value: 42 };
            let bytes = src.encode_to_vec();
            let err = DecodeOptions::new()
                .with_max_message_size(1)
                .decode_reader::<FlatMsg>(&mut bytes.as_slice())
                .unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        }

        #[test]
        fn decode_reader_exact_boundary() {
            // max_message_size == encoded length → success.
            let src = FlatMsg { value: 42 };
            let bytes = src.encode_to_vec();
            let msg: FlatMsg = DecodeOptions::new()
                .with_max_message_size(bytes.len())
                .decode_reader(&mut bytes.as_slice())
                .unwrap();
            assert_eq!(msg.value, 42);
        }

        #[test]
        fn decode_options_reader_size_unbounded_getter_tracks_builder_order() {
            let opts = DecodeOptions::new();
            assert!(!opts.is_reader_size_unbounded());

            let opts = opts.without_reader_size_limit();
            assert!(opts.is_reader_size_unbounded());

            let opts = opts.with_max_message_size(1024);
            assert!(!opts.is_reader_size_unbounded());
        }

        #[test]
        fn decode_reader_without_size_limit_does_not_overflow() {
            let src = FlatMsg { value: 42 };
            let bytes = src.encode_to_vec();
            let msg: FlatMsg = DecodeOptions::new()
                .without_reader_size_limit()
                .decode_reader(&mut bytes.as_slice())
                .unwrap();
            assert_eq!(msg.value, 42);
        }

        #[test]
        fn decode_reader_with_max_message_size_after_without_limit_reenables_limit() {
            let src = FlatMsg { value: 42 };
            let bytes = src.encode_to_vec();
            let opts = DecodeOptions::new()
                .without_reader_size_limit()
                .with_max_message_size(1);
            assert!(!opts.is_reader_size_unbounded());

            let err = opts
                .decode_reader::<FlatMsg>(&mut bytes.as_slice())
                .unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        }

        #[test]
        fn decode_reader_without_size_limit_keeps_slice_limit() {
            let src = FlatMsg { value: 42 };
            let bytes = src.encode_to_vec();
            let opts = DecodeOptions::new()
                .with_max_message_size(1)
                .without_reader_size_limit();
            assert!(opts.is_reader_size_unbounded());

            let slice_result: Result<FlatMsg, _> = opts.decode_from_slice(&bytes);
            assert_eq!(slice_result, Err(DecodeError::MessageTooLarge));

            let msg: FlatMsg = opts.decode_reader(&mut bytes.as_slice()).unwrap();
            assert_eq!(msg.value, 42);
        }

        #[test]
        fn decode_reader_propagates_read_error() {
            // A reader that errors immediately.
            struct ErrReader;
            impl std::io::Read for ErrReader {
                fn read(&mut self, _: &mut [u8]) -> std::io::Result<usize> {
                    Err(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"))
                }
            }
            let err = DecodeOptions::new()
                .decode_reader::<FlatMsg>(&mut ErrReader)
                .unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe);
        }

        #[test]
        fn decode_length_delimited_reader_basic() {
            let src = FlatMsg { value: 99 };
            let mut ld = Vec::new();
            src.encode_length_delimited(&mut ld);
            let msg: FlatMsg = DecodeOptions::new()
                .decode_length_delimited_reader(&mut ld.as_slice())
                .unwrap();
            assert_eq!(msg.value, 99);
        }

        #[test]
        fn decode_length_delimited_reader_rejects_oversize_prefix() {
            // Length prefix claims more bytes than max_message_size allows.
            let src = FlatMsg { value: 99 };
            let mut ld = Vec::new();
            src.encode_length_delimited(&mut ld);
            let err = DecodeOptions::new()
                .with_max_message_size(1)
                .decode_length_delimited_reader::<FlatMsg>(&mut ld.as_slice())
                .unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        }

        #[test]
        fn decode_length_delimited_reader_zero_length() {
            // A zero length prefix decodes to the default message and
            // consumes only the prefix byte.
            let stream = [0x00u8, 0xFF]; // len=0, then unrelated trailing byte
            let mut cursor = std::io::Cursor::new(stream);
            let msg: FlatMsg = DecodeOptions::new()
                .decode_length_delimited_reader(&mut cursor)
                .unwrap();
            assert_eq!(msg, FlatMsg::default());
            assert_eq!(cursor.position(), 1);
        }

        #[test]
        fn decode_length_delimited_reader_truncated_reports_eof() {
            // Length prefix declares more bytes than the stream delivers.
            let src = FlatMsg { value: 99 };
            let mut ld = Vec::new();
            src.encode_length_delimited(&mut ld);
            ld.truncate(ld.len() - 1);
            let err = DecodeOptions::new()
                .decode_length_delimited_reader::<FlatMsg>(&mut ld.as_slice())
                .unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
        }

        #[test]
        fn decode_length_delimited_reader_huge_claim_without_delivery() {
            // A 5-byte prefix declaring ~2 GiB followed by EOF must fail with
            // UnexpectedEof without allocating the declared length up front —
            // the buffer only grows as bytes are actually delivered. (This
            // test completes instantly; before the incremental-read fix it
            // zero-filled a ~2 GiB buffer first.)
            let mut stream = Vec::new();
            encode_varint(0x7FFF_FFF0, &mut stream); // just under the cap
            let err = DecodeOptions::new()
                .decode_length_delimited_reader::<FlatMsg>(&mut stream.as_slice())
                .unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
        }

        #[test]
        fn decode_length_delimited_without_reader_size_limit_keeps_protobuf_cap() {
            let mut stream = Vec::new();
            encode_varint(0x8000_0000, &mut stream);
            let result: Result<FlatMsg, _> = DecodeOptions::new()
                .without_reader_size_limit()
                .decode_length_delimited(&mut stream.as_slice());
            assert_eq!(result, Err(DecodeError::MessageTooLarge));
        }

        #[test]
        fn decode_length_delimited_reader_without_size_limit_keeps_protobuf_cap() {
            let mut stream = Vec::new();
            encode_varint(0x8000_0000, &mut stream);
            let err = DecodeOptions::new()
                .without_reader_size_limit()
                .decode_length_delimited_reader::<FlatMsg>(&mut stream.as_slice())
                .unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        }

        #[test]
        fn decode_length_delimited_reader_sequential() {
            // Two messages in a stream — typical log-file use case.
            let a = FlatMsg { value: 10 };
            let b = FlatMsg { value: 20 };
            let mut stream = Vec::new();
            a.encode_length_delimited(&mut stream);
            b.encode_length_delimited(&mut stream);

            let mut cursor = std::io::Cursor::new(stream);
            let first: FlatMsg = DecodeOptions::new()
                .decode_length_delimited_reader(&mut cursor)
                .unwrap();
            assert_eq!(first.value, 10);
            let second: FlatMsg = DecodeOptions::new()
                .decode_length_delimited_reader(&mut cursor)
                .unwrap();
            assert_eq!(second.value, 20);
        }

        #[test]
        fn decode_length_delimited_reader_truncated_body() {
            // Length prefix says N bytes but reader EOFs early.
            let mut buf = Vec::new();
            crate::encoding::encode_varint(100, &mut buf);
            buf.push(0x08);
            let err = DecodeOptions::new()
                .decode_length_delimited_reader::<FlatMsg>(&mut buf.as_slice())
                .unwrap_err();
            assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
        }
    }
}