verit_core/

message.rs

//! Zero-copy message access. [`Message::parse`] only reads the 24-byte
//! header; nothing else is touched until a field is asked for, and every
//! field read is a bounds-checked load straight out of the buffer through a
//! precompiled [`Resolver`] plan. Strings and byte fields are returned as
//! `&str` / `&[u8]` borrowing the message buffer — no allocation, no copy.

use std::cell::Cell;

use crate::encode::{FLAG_INLINE_SCHEMA, HEADER_LEN, MESSAGE_MAGIC};
use crate::error::{Error, Result};
use crate::resolve::{
    ElemPlan, FieldSource, Load, MapPlan, NumKind, Resolver, StructPlan, UnionPlan,
};
use crate::schema::{Default, Schema, StructDef};

/// A per-read **traversal budget** — an opt-in guard against amplification-DoS
/// on untrusted messages (the wire spec §5.2). Veritate's offsets are absolute and
/// may alias, so a small hostile message can point many fields at the same
/// large sub-object and make a naive full read do work super-linear in the
/// message's own size. Memory safety (bounds, depth, allocation) always holds;
/// a `Budget` additionally caps *total work*.
///
/// Create one per message read and open the root with
/// [`Message::root_bounded`]. Every byte a read touches through the buffer is
/// charged against the budget; because aliased sub-objects are charged on every
/// visit, total work is capped regardless of how offsets alias. When exhausted,
/// reads fail with [`Error::TraversalBudgetExceeded`] instead of running away.
/// A good starting limit is [`Message::suggested_budget`].
///
/// ```
/// # use verit_core::*;
/// # fn go(buf: &[u8], resolver: &Resolver) -> Result<()> {
/// let msg = Message::parse(buf)?;
/// let budget = Budget::new(msg.suggested_budget());
/// let root = msg.root_bounded(resolver, &budget)?;
/// // ... traverse `root`; any amplification trips TraversalBudgetExceeded ...
/// # Ok(()) }
/// ```
#[derive(Debug)]
pub struct Budget {
    remaining: Cell<u64>,
}

impl Budget {
    /// A budget that allows `limit` total bytes of buffer access before
    /// tripping.
    pub fn new(limit: u64) -> Self {
        Budget {
            remaining: Cell::new(limit),
        }
    }

    /// Bytes still allowed before the budget trips.
    pub fn remaining(&self) -> u64 {
        self.remaining.get()
    }

    /// Charge `bytes` against the budget, returning
    /// [`Error::TraversalBudgetExceeded`] if it would go negative. Public so
    /// generated code (see [`crate::wire`]) can charge its own buffer accesses.
    #[inline]
    pub fn charge(&self, bytes: u64) -> Result<()> {
        match self.remaining.get().checked_sub(bytes) {
            Some(rem) => {
                self.remaining.set(rem);
                Ok(())
            }
            None => Err(Error::TraversalBudgetExceeded),
        }
    }
}

#[inline]
fn charge(budget: Option<&Budget>, bytes: u64) -> Result<()> {
    match budget {
        Some(b) => b.charge(bytes),
        None => Ok(()),
    }
}

#[derive(Clone, Debug)]
pub struct Message<'b> {
    buf: &'b [u8],
    schema_id: u128,
    root_offset: u32,
    schema_range: Option<(usize, usize)>,
}

impl<'b> Message<'b> {
    pub fn parse(buf: &'b [u8]) -> Result<Message<'b>> {
        if buf.len() < HEADER_LEN {
            return Err(Error::Truncated);
        }
        // Magic is "VRT" + a version digit. A message from a different major
        // version fails with UnsupportedVersion (never a silent misread); a
        // buffer that is not a Veritate message at all fails with BadMagic.
        if &buf[0..4] != MESSAGE_MAGIC {
            if buf[0..3] == MESSAGE_MAGIC[0..3] {
                return Err(Error::UnsupportedVersion {
                    found: buf[3],
                    supported: MESSAGE_MAGIC[3],
                });
            }
            return Err(Error::BadMagic);
        }
        let flags = u16::from_le_bytes(buf[4..6].try_into().unwrap());
        // Reject header bits this version does not define, so a message that
        // relies on semantics we don't implement is refused rather than
        // misinterpreted. Only bit 0 (inline schema) is defined in v2.
        if flags & !FLAG_INLINE_SCHEMA != 0 {
            return Err(Error::MalformedHeader("unknown flag bit set"));
        }
        let reserved = u16::from_le_bytes(buf[6..8].try_into().unwrap());
        if reserved != 0 {
            return Err(Error::MalformedHeader("reserved header field is not zero"));
        }
        let schema_id = u128::from_le_bytes(buf[8..24].try_into().unwrap());
        let root_offset = u32::from_le_bytes(buf[24..28].try_into().unwrap());
        let schema_len = u32::from_le_bytes(buf[28..32].try_into().unwrap()) as usize;
        let schema_range = if flags & FLAG_INLINE_SCHEMA != 0 {
            let end = HEADER_LEN.checked_add(schema_len).ok_or(Error::Truncated)?;
            if end > buf.len() {
                return Err(Error::Truncated);
            }
            Some((HEADER_LEN, end))
        } else {
            None
        };
        Ok(Message {
            buf,
            schema_id,
            root_offset,
            schema_range,
        })
    }

    pub fn buffer(&self) -> &'b [u8] {
        self.buf
    }

    pub fn schema_id(&self) -> u128 {
        self.schema_id
    }

    /// Absolute offset of the root struct block. Exposed for generated code
    /// (the codegen identity fast path reads at constant offsets from here).
    pub fn root_offset(&self) -> u32 {
        self.root_offset
    }

    pub fn has_inline_schema(&self) -> bool {
        self.schema_range.is_some()
    }

    /// Decode the inline writer schema, if the message carries one. The
    /// decoded schema's content hash must match the header's schema id.
    pub fn writer_schema(&self) -> Result<Option<Schema>> {
        match self.schema_range {
            None => Ok(None),
            Some((start, end)) => {
                let schema = Schema::from_canonical(&self.buf[start..end])?;
                if schema.id() != self.schema_id {
                    return Err(Error::SchemaIdMismatch {
                        message: self.schema_id,
                        expected: schema.id(),
                    });
                }
                Ok(Some(schema))
            }
        }
    }

    /// Open the root struct through a resolver whose writer schema matches
    /// this message's schema id. **Unbounded**: reads are memory-safe but do
    /// not cap total traversal work — use this for trusted data or after an
    /// upstream size cap. For untrusted input, prefer [`Self::root_bounded`].
    pub fn root<'r>(&self, resolver: &'r Resolver) -> Result<StructReader<'b, 'r>> {
        self.open_root(resolver, None)
    }

    /// Open the root struct with a [`Budget`] that caps total traversal work,
    /// guarding against amplification-DoS on untrusted input (the wire spec §5.2).
    /// The `budget` outlives the returned readers, which charge every byte they
    /// touch against it.
    pub fn root_bounded<'r>(
        &self,
        resolver: &'r Resolver,
        budget: &'r Budget,
    ) -> Result<StructReader<'b, 'r>> {
        self.open_root(resolver, Some(budget))
    }

    /// Walk the whole message once under a [`Budget`], touching every reachable
    /// field and element, and return `Ok(())` iff the traversal stays within
    /// budget (and depth). This is the guard for the **fast/codegen** path:
    /// verify untrusted bytes *once*, and if it passes, a single subsequent
    /// zero-copy scan — via the generated readers or [`Self::root`] — cannot
    /// amplify beyond the budget, because a full scan touches no more than this
    /// walk did. Trusted data skips it and pays nothing. (Analogous to
    /// FlatBuffers' verified `root`, but Veritate needs no separate encoding —
    /// the same bytes are then read directly.)
    pub fn verify(&self, resolver: &Resolver, budget: &Budget) -> Result<()> {
        let root = self.root_bounded(resolver, budget)?;
        verify_struct(&root, 0)
    }

    /// A sensible default traversal-budget limit for this message:
    /// `max(64 KiB, 64 × message length)`. Scales with legitimate message size
    /// (a well-formed message touches ~its own size once), leaving generous
    /// headroom while capping amplification at ~64× the bytes on the wire.
    pub fn suggested_budget(&self) -> u64 {
        (self.buf.len() as u64).saturating_mul(64).max(64 * 1024)
    }

    fn open_root<'r>(
        &self,
        resolver: &'r Resolver,
        budget: Option<&'r Budget>,
    ) -> Result<StructReader<'b, 'r>> {
        if resolver.writer_id() != self.schema_id {
            return Err(Error::SchemaIdMismatch {
                message: self.schema_id,
                expected: resolver.writer_id(),
            });
        }
        Ok(StructReader {
            buf: self.buf,
            base: self.root_offset,
            plan: resolver.plan(resolver.root_plan_index()),
            resolver,
            budget,
        })
    }
}

/// A field value read from the buffer. Scalars are by value; strings, bytes,
/// structs, and lists borrow the message buffer (`'b`).
#[derive(Clone, Debug)]
pub enum Ref<'b, 'r> {
    Bool(bool),
    U8(u8),
    U16(u16),
    U32(u32),
    U64(u64),
    I8(i8),
    I16(i16),
    I32(i32),
    I64(i64),
    F32(f32),
    F64(f64),
    Str(&'b str),
    Bytes(&'b [u8]),
    Enum(u32),
    Struct(StructReader<'b, 'r>),
    List(ListReader<'b, 'r>),
    Map(MapReader<'b, 'r>),
    Union(UnionReader<'b, 'r>),
}

impl<'b, 'r> Ref<'b, 'r> {
    pub fn kind(&self) -> &'static str {
        match self {
            Ref::Bool(_) => "bool",
            Ref::U8(_) => "u8",
            Ref::U16(_) => "u16",
            Ref::U32(_) => "u32",
            Ref::U64(_) => "u64",
            Ref::I8(_) => "i8",
            Ref::I16(_) => "i16",
            Ref::I32(_) => "i32",
            Ref::I64(_) => "i64",
            Ref::F32(_) => "f32",
            Ref::F64(_) => "f64",
            Ref::Str(_) => "string",
            Ref::Bytes(_) => "bytes",
            Ref::Enum(_) => "enum",
            Ref::Struct(_) => "struct",
            Ref::List(_) => "list",
            Ref::Map(_) => "map",
            Ref::Union(_) => "union",
        }
    }
}

#[derive(Clone, Debug)]
pub struct StructReader<'b, 'r> {
    buf: &'b [u8],
    base: u32,
    plan: &'r StructPlan,
    resolver: &'r Resolver,
    /// `Some` on a bounded read (via [`Message::root_bounded`]); charged for
    /// every byte this reader and its children touch. `None` = unbounded.
    budget: Option<&'r Budget>,
}

macro_rules! typed_getter {
    ($doc:literal, $name:ident, $variant:ident, $ret:ty) => {
        #[doc = $doc]
        pub fn $name(&self, id: u16) -> Result<Option<$ret>> {
            match self.get(id)? {
                None => Ok(None),
                Some(Ref::$variant(x)) => Ok(Some(x)),
                Some(other) => Err(Error::TypeMismatch {
                    expected: stringify!($variant).to_lowercase(),
                    got: other.kind().into(),
                }),
            }
        }
    };
}

impl<'b, 'r> StructReader<'b, 'r> {
    /// Read a field by its stable ID. `Ok(None)` means absent — either the
    /// writer didn't set it, or the writer's schema doesn't have it at all.
    pub fn get(&self, id: u16) -> Result<Option<Ref<'b, 'r>>> {
        let pos = match self.plan.fields.binary_search_by_key(&id, |f| f.id) {
            Ok(pos) => pos,
            Err(_) => return Err(Error::UnknownFieldId(id)),
        };
        match &self.plan.fields[pos].source {
            FieldSource::Absent => Ok(None),
            FieldSource::Slot {
                offset,
                presence_byte,
                presence_mask,
                load,
            } => {
                // Mask 0 = dense writer struct: the field is always present.
                if *presence_mask != 0 {
                    let pbyte = read_u8(
                        self.buf,
                        self.base as u64 + *presence_byte as u64,
                        self.budget,
                    )?;
                    if pbyte & presence_mask == 0 {
                        return Ok(None);
                    }
                }
                let at = self.base as u64 + *offset as u64;
                load_at(self.buf, self.resolver, load, at, self.budget).map(Some)
            }
            FieldSource::Packed { writer_pos, load } => {
                // Packed writer struct: recover presence and offset from the
                // per-message bitmap via the writer's popcount layout.
                let lay = self
                    .resolver
                    .writer_schema()
                    .packed_layout_unchecked(self.plan.writer_type);
                let bitmap =
                    read_bitmap(self.buf, self.base as u64, lay.bitmap_bytes, self.budget)?;
                if bitmap & (1u64 << writer_pos) == 0 {
                    return Ok(None);
                }
                let at = self.base as u64 + lay.field_offset(bitmap, *writer_pos as usize) as u64;
                load_at(self.buf, self.resolver, load, at, self.budget).map(Some)
            }
        }
    }

    /// Like [`get`](Self::get), but if the field is absent and the **reader
    /// schema** gives it a custom default, return that default value instead of
    /// `None`. Presence is unchanged — [`get`](Self::get) still reports the raw
    /// wire state — so this is a read-time convenience layered on top.
    pub fn get_or_default(&self, id: u16) -> Result<Option<Ref<'b, 'r>>> {
        if let Some(v) = self.get(id)? {
            return Ok(Some(v));
        }
        Ok(self
            .struct_def()
            .fields
            .iter()
            .find(|f| f.id == id)
            .and_then(|f| f.default)
            .map(default_to_ref))
    }

    /// The reader-schema definition of this struct (field names for
    /// self-description).
    pub fn struct_def(&self) -> &'r StructDef {
        self.resolver
            .reader_schema()
            .struct_def_unchecked(self.plan.reader_type)
    }

    typed_getter!("Typed getter for `bool` fields.", get_bool, Bool, bool);
    typed_getter!("Typed getter for `u8` fields.", get_u8, U8, u8);
    typed_getter!("Typed getter for `u16` fields.", get_u16, U16, u16);
    typed_getter!("Typed getter for `u32` fields.", get_u32, U32, u32);
    typed_getter!("Typed getter for `u64` fields.", get_u64, U64, u64);
    typed_getter!("Typed getter for `i8` fields.", get_i8, I8, i8);
    typed_getter!("Typed getter for `i16` fields.", get_i16, I16, i16);
    typed_getter!("Typed getter for `i32` fields.", get_i32, I32, i32);
    typed_getter!("Typed getter for `i64` fields.", get_i64, I64, i64);
    typed_getter!("Typed getter for `f32` fields.", get_f32, F32, f32);
    typed_getter!("Typed getter for `f64` fields.", get_f64, F64, f64);
    typed_getter!(
        "Typed getter for string fields (borrows the buffer).",
        get_str,
        Str,
        &'b str
    );
    typed_getter!(
        "Typed getter for bytes fields (borrows the buffer).",
        get_bytes,
        Bytes,
        &'b [u8]
    );
    typed_getter!(
        "Typed getter for enum fields (raw open value).",
        get_enum,
        Enum,
        u32
    );
    typed_getter!(
        "Typed getter for nested struct fields.",
        get_struct,
        Struct,
        StructReader<'b, 'r>
    );
    typed_getter!(
        "Typed getter for list fields.",
        get_list,
        List,
        ListReader<'b, 'r>
    );
    typed_getter!(
        "Typed getter for map fields.",
        get_map,
        Map,
        MapReader<'b, 'r>
    );
    typed_getter!(
        "Typed getter for union fields.",
        get_union,
        Union,
        UnionReader<'b, 'r>
    );
}

#[derive(Clone, Debug)]
pub struct ListReader<'b, 'r> {
    buf: &'b [u8],
    resolver: &'r Resolver,
    elem: &'r ElemPlan,
    elems_base: u64,
    count: u32,
    budget: Option<&'r Budget>,
}

impl<'b, 'r> ListReader<'b, 'r> {
    pub fn len(&self) -> u32 {
        self.count
    }

    pub fn is_empty(&self) -> bool {
        self.count == 0
    }

    pub fn get(&self, index: u32) -> Result<Ref<'b, 'r>> {
        if index >= self.count {
            return Err(Error::IndexOutOfBounds);
        }
        let at = self.elems_base + index as u64 * self.elem.stride as u64;
        match &self.elem.load {
            // Fixed struct elements are stored inline: the element position IS
            // the block. Packed struct elements are variable-size and stored
            // by u32 offset (struct_inline = false), so they fall through to
            // load_at, which follows the offset.
            Load::Struct(plan_idx) if self.elem.struct_inline => {
                let base = u32::try_from(at).map_err(|_| Error::OutOfBounds)?;
                Ok(Ref::Struct(StructReader {
                    buf: self.buf,
                    base,
                    plan: self.resolver.plan(*plan_idx),
                    resolver: self.resolver,
                    budget: self.budget,
                }))
            }
            other => load_at(self.buf, self.resolver, other, at, self.budget),
        }
    }

    pub fn iter(&self) -> impl Iterator<Item = Result<Ref<'b, 'r>>> + '_ {
        (0..self.count).map(move |i| self.get(i))
    }

    /// Borrow a `list<u8>` as a slice of the message buffer — no copy, no
    /// conversion.
    ///
    /// `u8` is the one numeric element type with no endianness and no
    /// alignment to satisfy, so it is the only one that can be handed back as a
    /// borrow. Every wider type needs a little-endian decode, which safe Rust
    /// cannot do in place; use the `copy_*` / `to_vec_*` methods for those.
    pub fn as_u8_slice(&self) -> Result<&'b [u8]> {
        match &self.elem.load {
            Load::Num {
                from: NumKind::U8,
                to: NumKind::U8,
            } if self.elem.stride == 1 => {
                get_slice(self.buf, self.elems_base, self.count as u64, self.budget)
            }
            other => Err(Error::TypeMismatch {
                expected: "list<u8>".into(),
                got: elem_kind(other).into(),
            }),
        }
    }
}

/// Name an element plan for a type-mismatch error.
fn elem_kind(load: &Load) -> &'static str {
    match load {
        Load::Bool => "list<bool>",
        Load::Num { to, .. } => match to {
            NumKind::U8 => "list<u8>",
            NumKind::U16 => "list<u16>",
            NumKind::U32 => "list<u32>",
            NumKind::U64 => "list<u64>",
            NumKind::I8 => "list<i8>",
            NumKind::I16 => "list<i16>",
            NumKind::I32 => "list<i32>",
            NumKind::I64 => "list<i64>",
            NumKind::F32 => "list<f32>",
            NumKind::F64 => "list<f64>",
        },
        Load::Enum => "list<enum>",
        Load::Str => "list<string>",
        Load::Bytes => "list<bytes>",
        Load::Struct(_) => "list<struct>",
        Load::List(_) => "list<list>",
        Load::Map(_) => "list<map>",
        Load::Union(_) => "list<union>",
    }
}

/// Bulk readers for numeric lists — the read-side peer of the bulk list
/// *writer* fast path (ADR-0010).
///
/// A `list<f32>` of 1,536 elements is the shape that matters here (an embedding
/// vector), and reading it element-at-a-time through [`Ref`] pays a match and a
/// bounds check per element. These methods take one bounds check and one budget
/// charge for the whole run, then decode in a tight loop the optimiser can
/// vectorize.
///
/// Widening still works: a `list<u16>` written by an older schema reads through
/// `copy_u32` into a reader that widened the field, at the cost of falling back
/// to the per-element path.
macro_rules! bulk_num {
    ($ty:ty, $kind:ident, $variant:ident, $copy:ident, $to_vec:ident, $name:literal) => {
        impl<'b, 'r> ListReader<'b, 'r> {
            #[doc = concat!("Bulk-copy a `", $name, "` list into `out`, returning how many elements were written.")]
            ///
            /// Copies `min(len(), out.len())` elements — a short `out` is not an
            /// error, so a caller can read a prefix deliberately.
            pub fn $copy(&self, out: &mut [$ty]) -> Result<usize> {
                const WIDTH: usize = std::mem::size_of::<$ty>();
                let Load::Num { from, to } = &self.elem.load else {
                    return Err(Error::TypeMismatch {
                        expected: concat!("list<", $name, ">").into(),
                        got: elem_kind(&self.elem.load).into(),
                    });
                };
                if *to != NumKind::$kind {
                    return Err(Error::TypeMismatch {
                        expected: concat!("list<", $name, ">").into(),
                        got: elem_kind(&self.elem.load).into(),
                    });
                }
                let n = (self.count as usize).min(out.len());
                if n == 0 {
                    return Ok(0);
                }

                // Fast path: the writer stored exactly this type, packed at its
                // natural stride. One range check and one budget charge for the
                // whole run, then a straight little-endian decode.
                if *from == NumKind::$kind && self.elem.stride as usize == WIDTH {
                    let span = (n as u64) * WIDTH as u64;
                    let bytes = get_slice(self.buf, self.elems_base, span, self.budget)?;
                    for (slot, chunk) in out.iter_mut().zip(bytes.chunks_exact(WIDTH)) {
                        *slot = <$ty>::from_le_bytes(chunk.try_into().unwrap());
                    }
                    return Ok(n);
                }

                // Slow path: the writer used a narrower type and the reader
                // widened it, so each element needs converting.
                for (i, slot) in out.iter_mut().enumerate().take(n) {
                    let at = self.elems_base + i as u64 * self.elem.stride as u64;
                    *slot = match num_ref(*to, read_wide(self.buf, at, *from, self.budget)?)? {
                        Ref::$variant(x) => x,
                        _ => return Err(Error::Internal("num kind mismatch in bulk list read")),
                    };
                }
                Ok(n)
            }

            #[doc = concat!("Read a whole `", $name, "` list into a new `Vec`.")]
            ///
            /// Allocates once, at the exact length. Prefer
            #[doc = concat!("[`", stringify!($copy), "`](Self::", stringify!($copy), ")")]
            /// with a reused buffer on a hot path.
            pub fn $to_vec(&self) -> Result<Vec<$ty>> {
                // `count` is attacker-controlled, so the elements are proven to
                // be inside the buffer *before* anything is sized by it — a
                // forged count must fail on arithmetic, not on an OOM kill
                // (File Format Specification §10, wire spec §5).
                let span = (self.count as u64)
                    .checked_mul(self.elem.stride as u64)
                    .ok_or(Error::OutOfBounds)?;
                get_slice(self.buf, self.elems_base, span, None)?;

                let mut out: Vec<$ty> = Vec::new();
                out.try_reserve_exact(self.count as usize)
                    .map_err(|_| Error::OutOfBounds)?;
                out.resize(self.count as usize, <$ty>::default());
                let n = self.$copy(&mut out)?;
                out.truncate(n);
                Ok(out)
            }
        }
    };
}

bulk_num!(u8, U8, U8, copy_u8, to_vec_u8, "u8");
bulk_num!(u16, U16, U16, copy_u16, to_vec_u16, "u16");
bulk_num!(u32, U32, U32, copy_u32, to_vec_u32, "u32");
bulk_num!(u64, U64, U64, copy_u64, to_vec_u64, "u64");
bulk_num!(i8, I8, I8, copy_i8, to_vec_i8, "i8");
bulk_num!(i16, I16, I16, copy_i16, to_vec_i16, "i16");
bulk_num!(i32, I32, I32, copy_i32, to_vec_i32, "i32");
bulk_num!(i64, I64, I64, copy_i64, to_vec_i64, "i64");
bulk_num!(f32, F32, F32, copy_f32, to_vec_f32, "f32");
bulk_num!(f64, F64, F64, copy_f64, to_vec_f64, "f64");

/// A `map<K, V>` read lazily from the buffer. Entries are stored sorted by key
/// (canonical order), so [`get`](MapReader::get) yields them in that order.
#[derive(Clone, Debug)]
pub struct MapReader<'b, 'r> {
    buf: &'b [u8],
    resolver: &'r Resolver,
    plan: &'r MapPlan,
    entries_base: u64,
    count: u32,
    budget: Option<&'r Budget>,
}

impl<'b, 'r> MapReader<'b, 'r> {
    pub fn len(&self) -> u32 {
        self.count
    }

    pub fn is_empty(&self) -> bool {
        self.count == 0
    }

    /// The `(key, value)` of entry `index` (in canonical key order).
    pub fn get(&self, index: u32) -> Result<(Ref<'b, 'r>, Ref<'b, 'r>)> {
        if index >= self.count {
            return Err(Error::IndexOutOfBounds);
        }
        let entry = self.entries_base + index as u64 * self.plan.stride as u64;
        let key = load_at(
            self.buf,
            self.resolver,
            &self.plan.key,
            entry + self.plan.key_off as u64,
            self.budget,
        )?;
        let value = load_at(
            self.buf,
            self.resolver,
            &self.plan.value,
            entry + self.plan.value_off as u64,
            self.budget,
        )?;
        Ok((key, value))
    }

    pub fn iter(&self) -> impl Iterator<Item = Result<(Ref<'b, 'r>, Ref<'b, 'r>)>> + '_ {
        (0..self.count).map(move |i| self.get(i))
    }
}

/// A `union<…>` value read lazily: a variant [`tag`](UnionReader::tag) and the
/// selected variant's [`value`](UnionReader::value).
#[derive(Clone, Debug)]
pub struct UnionReader<'b, 'r> {
    buf: &'b [u8],
    resolver: &'r Resolver,
    plan: &'r UnionPlan,
    base: u64,
    tag: u32,
    budget: Option<&'r Budget>,
}

impl<'b, 'r> UnionReader<'b, 'r> {
    /// The variant tag (index into the union's variant list).
    pub fn tag(&self) -> u32 {
        self.tag
    }

    /// The selected variant's value. Errors with [`Error::BadUnionTag`] if the
    /// wire tag is out of range for the reader schema's variant list.
    pub fn value(&self) -> Result<Ref<'b, 'r>> {
        let vp = self
            .plan
            .variants
            .get(self.tag as usize)
            .ok_or(Error::BadUnionTag(self.tag))?;
        load_at(
            self.buf,
            self.resolver,
            &vp.load,
            self.base + vp.payload_off as u64,
            self.budget,
        )
    }
}

/// Materialize a scalar [`Default`] into a [`Ref`]. Defaults are by-value
/// scalars, so the returned `Ref` is valid for any lifetimes.
fn default_to_ref<'b, 'r>(d: Default) -> Ref<'b, 'r> {
    match d {
        Default::Bool(x) => Ref::Bool(x),
        Default::U8(x) => Ref::U8(x),
        Default::U16(x) => Ref::U16(x),
        Default::U32(x) => Ref::U32(x),
        Default::U64(x) => Ref::U64(x),
        Default::I8(x) => Ref::I8(x),
        Default::I16(x) => Ref::I16(x),
        Default::I32(x) => Ref::I32(x),
        Default::I64(x) => Ref::I64(x),
        Default::F32(bits) => Ref::F32(f32::from_bits(bits)),
        Default::F64(bits) => Ref::F64(f64::from_bits(bits)),
        Default::Enum(x) => Ref::Enum(x),
    }
}

/// Depth ceiling for [`Message::verify`], matching `dump_json`. Bounds forged
/// offset *cycles* (which the byte budget alone would only stop after burning
/// the whole budget); far deeper than any real data.
const MAX_VERIFY_DEPTH: u32 = 128;

fn verify_struct(sr: &StructReader, depth: u32) -> Result<()> {
    if depth > MAX_VERIFY_DEPTH {
        return Err(Error::DepthLimitExceeded);
    }
    // `struct_def` comes from the reader schema; every present field is read
    // (charging the budget via the bounded reader) and recursed into.
    let ids: Vec<u16> = sr.struct_def().fields.iter().map(|f| f.id).collect();
    for id in ids {
        if let Some(v) = sr.get(id)? {
            verify_ref(&v, depth)?;
        }
    }
    Ok(())
}

fn verify_ref(v: &Ref, depth: u32) -> Result<()> {
    match v {
        Ref::Struct(s) => verify_struct(s, depth + 1),
        Ref::List(l) => {
            for i in 0..l.len() {
                verify_ref(&l.get(i)?, depth + 1)?;
            }
            Ok(())
        }
        Ref::Map(m) => {
            for i in 0..m.len() {
                let (k, v) = m.get(i)?;
                verify_ref(&k, depth + 1)?;
                verify_ref(&v, depth + 1)?;
            }
            Ok(())
        }
        Ref::Union(u) => verify_ref(&u.value()?, depth + 1),
        _ => Ok(()),
    }
}

// ---------------------------------------------------------------------------
// Raw bounds-checked loads
// ---------------------------------------------------------------------------

/// Bounds-checked slice fetch. On a bounded read (`budget = Some`) it also
/// charges the touched bytes against the budget — since *every* buffer access
/// funnels through here, that alone caps total traversal work: an aliased
/// sub-object re-read costs its bytes again each time (the wire spec §5.2).
fn get_slice<'b>(buf: &'b [u8], off: u64, len: u64, budget: Option<&Budget>) -> Result<&'b [u8]> {
    charge(budget, len)?;
    let start = usize::try_from(off).map_err(|_| Error::OutOfBounds)?;
    let len = usize::try_from(len).map_err(|_| Error::OutOfBounds)?;
    let end = start.checked_add(len).ok_or(Error::OutOfBounds)?;
    buf.get(start..end).ok_or(Error::OutOfBounds)
}

fn read_u8(buf: &[u8], off: u64, budget: Option<&Budget>) -> Result<u8> {
    Ok(get_slice(buf, off, 1, budget)?[0])
}

/// Load a packed struct's presence bitmap (1..=8 bytes, little-endian) into a
/// u64 for popcount rank queries. A validated schema guarantees
/// `bitmap_bytes <= 8` (packed structs cap at 64 fields), but we check anyway
/// so a corrupt layout can never index past the word or panic.
fn read_bitmap(buf: &[u8], base: u64, bitmap_bytes: u32, budget: Option<&Budget>) -> Result<u64> {
    if bitmap_bytes > 8 {
        return Err(Error::Internal("packed bitmap wider than 8 bytes"));
    }
    let bytes = get_slice(buf, base, bitmap_bytes as u64, budget)?;
    let mut word = [0u8; 8];
    word[..bytes.len()].copy_from_slice(bytes);
    Ok(u64::from_le_bytes(word))
}

fn read_u32(buf: &[u8], off: u64, budget: Option<&Budget>) -> Result<u32> {
    Ok(u32::from_le_bytes(
        get_slice(buf, off, 4, budget)?.try_into().unwrap(),
    ))
}

enum Wide {
    U(u64),
    I(i64),
    F(f64),
}

fn read_wide(buf: &[u8], at: u64, kind: NumKind, budget: Option<&Budget>) -> Result<Wide> {
    Ok(match kind {
        NumKind::U8 => Wide::U(read_u8(buf, at, budget)? as u64),
        NumKind::U16 => {
            Wide::U(u16::from_le_bytes(get_slice(buf, at, 2, budget)?.try_into().unwrap()) as u64)
        }
        NumKind::U32 => Wide::U(read_u32(buf, at, budget)? as u64),
        NumKind::U64 => Wide::U(u64::from_le_bytes(
            get_slice(buf, at, 8, budget)?.try_into().unwrap(),
        )),
        NumKind::I8 => Wide::I(read_u8(buf, at, budget)? as i8 as i64),
        NumKind::I16 => {
            Wide::I(i16::from_le_bytes(get_slice(buf, at, 2, budget)?.try_into().unwrap()) as i64)
        }
        NumKind::I32 => {
            Wide::I(i32::from_le_bytes(get_slice(buf, at, 4, budget)?.try_into().unwrap()) as i64)
        }
        NumKind::I64 => Wide::I(i64::from_le_bytes(
            get_slice(buf, at, 8, budget)?.try_into().unwrap(),
        )),
        NumKind::F32 => {
            Wide::F(f32::from_le_bytes(get_slice(buf, at, 4, budget)?.try_into().unwrap()) as f64)
        }
        NumKind::F64 => Wide::F(f64::from_le_bytes(
            get_slice(buf, at, 8, budget)?.try_into().unwrap(),
        )),
    })
}

fn num_ref<'b, 'r>(to: NumKind, wide: Wide) -> Result<Ref<'b, 'r>> {
    Ok(match (to, wide) {
        (NumKind::U8, Wide::U(x)) => Ref::U8(x as u8),
        (NumKind::U16, Wide::U(x)) => Ref::U16(x as u16),
        (NumKind::U32, Wide::U(x)) => Ref::U32(x as u32),
        (NumKind::U64, Wide::U(x)) => Ref::U64(x),
        (NumKind::I8, Wide::I(x)) => Ref::I8(x as i8),
        (NumKind::I16, Wide::I(x)) => Ref::I16(x as i16),
        (NumKind::I32, Wide::I(x)) => Ref::I32(x as i32),
        (NumKind::I64, Wide::I(x)) => Ref::I64(x),
        (NumKind::F32, Wide::F(x)) => Ref::F32(x as f32),
        (NumKind::F64, Wide::F(x)) => Ref::F64(x),
        _ => return Err(Error::Internal("num kind mismatch in access plan")),
    })
}

/// Materialize a value whose slot (or list element position) is at `at`.
/// For heap loads, `at` holds a u32 absolute offset to the object.
fn load_at<'b, 'r>(
    buf: &'b [u8],
    resolver: &'r Resolver,
    load: &'r Load,
    at: u64,
    budget: Option<&'r Budget>,
) -> Result<Ref<'b, 'r>> {
    match load {
        Load::Bool => Ok(Ref::Bool(read_u8(buf, at, budget)? != 0)),
        Load::Num { from, to } => num_ref(*to, read_wide(buf, at, *from, budget)?),
        Load::Enum => Ok(Ref::Enum(read_u32(buf, at, budget)?)),
        Load::Str => {
            let off = read_u32(buf, at, budget)? as u64;
            let len = read_u32(buf, off, budget)? as u64;
            let bytes = get_slice(buf, off + 4, len, budget)?;
            let s = std::str::from_utf8(bytes).map_err(|_| Error::BadUtf8)?;
            Ok(Ref::Str(s))
        }
        Load::Bytes => {
            let off = read_u32(buf, at, budget)? as u64;
            let len = read_u32(buf, off, budget)? as u64;
            Ok(Ref::Bytes(get_slice(buf, off + 4, len, budget)?))
        }
        Load::Struct(plan_idx) => {
            let off = read_u32(buf, at, budget)?;
            Ok(Ref::Struct(StructReader {
                buf,
                base: off,
                plan: resolver.plan(*plan_idx),
                resolver,
                budget,
            }))
        }
        Load::List(elem) => {
            let off = read_u32(buf, at, budget)?;
            let count = read_u32(buf, off as u64, budget)?;
            // u64 arithmetic: off + 4 rounded up to the element alignment
            // cannot overflow here even at the 4 GiB message limit.
            let x = off as u64 + 4;
            let a = elem.align as u64;
            let elems_base = (x + a - 1) & !(a - 1);
            Ok(Ref::List(ListReader {
                buf,
                resolver,
                elem,
                elems_base,
                count,
                budget,
            }))
        }
        Load::Map(plan) => {
            let off = read_u32(buf, at, budget)?;
            let count = read_u32(buf, off as u64, budget)?;
            let x = off as u64 + 4;
            let a = plan.align as u64;
            let entries_base = (x + a - 1) & !(a - 1);
            Ok(Ref::Map(MapReader {
                buf,
                resolver,
                plan,
                entries_base,
                count,
                budget,
            }))
        }
        Load::Union(plan) => {
            let off = read_u32(buf, at, budget)? as u64;
            let tag = read_u32(buf, off, budget)?;
            Ok(Ref::Union(UnionReader {
                buf,
                resolver,
                plan,
                base: off,
                tag,
                budget,
            }))
        }
    }
}