Skip to main content

sedsnet/
wire_format.rs

1//! Wire-format packing and unpacking of telemetry packets.
2//!
3//! This module defines the compact v2 wire format used to send and receive
4//! [`Packet`]s, along with:
5//! - [`pack_packet`] / [`unpack_packet`] for full packets.
6//! - [`peek_envelope`] for header-only inspection without touching the payload.
7//! - Size helpers like [`header_size_bytes`] and [`packet_wire_size`].
8//!
9//! The core public type here is [`TelemetryEnvelope`], a lightweight view of
10//! the header fields used by `peek_envelope`.
11
12use crate::MessageElement;
13use crate::{
14    DataEndpoint, TelemetryError, TelemetryResult, config::DataType, is_reliable_type,
15    message_meta, packet::Packet,
16};
17
18use crate::config::StandardSmallPayload;
19use crate::packet::{hash_bytes_u64, sender_address_u32};
20#[cfg(feature = "std")]
21use alloc::collections::BTreeMap;
22#[cfg(feature = "cryptography")]
23use alloc::vec;
24use alloc::{sync::Arc, vec::Vec};
25use crc32fast::Hasher as Crc32Hasher;
26#[cfg(feature = "std")]
27use std::sync::{Mutex, OnceLock};
28
29/// Lightweight header-only view of a packed [`Packet`].
30///
31/// Produced by [`peek_envelope`] without allocating or copying the payload.
32#[derive(Clone, Debug, PartialEq, Eq)]
33pub struct TelemetryEnvelope {
34    /// Telemetry [`DataType`] discriminant.
35    pub ty: DataType,
36    /// All endpoints this packet is destined for (set bits in the bitmap).
37    pub endpoints: Arc<[DataEndpoint]>,
38    /// Sender identity resolved from discovery/config address state.
39    pub sender: Arc<str>,
40    /// Compact source address carried by the packet header.
41    pub source_address: u32,
42    /// Timestamp in milliseconds (as stored on the wire).
43    pub timestamp_ms: u64,
44    /// Inline wire-format payload shape, if present.
45    pub wire_shape: Option<MessageElement>,
46    /// Frozen destination sender hashes, if present.
47    pub target_senders: Arc<[u64]>,
48}
49
50/// Reliable header included for data types marked `reliable` in the schema.
51#[derive(Clone, Copy, Debug, PartialEq, Eq)]
52pub struct ReliableHeader {
53    pub flags: u8,
54    pub seq: u32,
55    pub ack: u32,
56}
57
58/// Reliable header flag: frame is ACK-only control (no payload).
59pub const RELIABLE_FLAG_ACK_ONLY: u8 = 0x01;
60
61/// Reliable header flag: frame is reliable but unordered (ACK/retransmit without ordering).
62pub const RELIABLE_FLAG_UNORDERED: u8 = 0x02;
63
64/// Reliable header flag: frame is unsequenced best-effort (no ordering/ack).
65pub const RELIABLE_FLAG_UNSEQUENCED: u8 = 0x80;
66
67/// Fixed size of the reliable header on the wire.
68pub const RELIABLE_HEADER_BYTES: usize = 1 + 4 + 4;
69/// Fixed size of the CRC32 trailer on the wire.
70pub const CRC32_BYTES: usize = 4;
71
72// packet Layout:
73//
74//   [FLAGS: u8]
75//       Bit 0: payload compressed flag (1 = compressed)
76//       Bit 1: reliable header present
77//       Bit 7: explicit endpoint ID list present (mutually exclusive with bit5)
78//       Bit 5: endpoint bitmap present (0 = derive endpoints from data type metadata)
79//       Bit 6: compact reliable header present (if a reliable header exists)
80//
81//   [NEP: u8]
82//       Number of selected endpoints.
83//
84//   VARINT(ty: u32 as u64)           -- ULEB128
85//   VARINT(data_size: u64)           -- ULEB128   (LOGICAL payload size, uncompressed)
86//   VARINT(timestamp: u64)           -- ULEB128
87//   [VARINT(nonce: u16 as u64)]      -- ULEB128   (ONLY if bit3 set)
88//   VARINT(src_addr: u32 as u64)     -- ULEB128
89//
90//   [ENDPOINTS_BITMAP]               -- ONLY if bit5 set; 1 bit per possible endpoint
91//   [ENDPOINT_ID_LIST]               -- NEP sorted unique ULEB128 IDs if bit7 set
92//   [RELIABLE HEADER]                -- declared by bit1 or the wire contract
93//       [REL_FLAGS: u8]
94//       [SEQ: u32 LE]
95//       [ACK: u32 LE]
96//   PAYLOAD BYTES                    -- raw or compressed payload bytes
97//   [CRC32: u32 LE]                  -- checksum of all prior bytes
98
99// ===========================================================================
100// ULEB128 (varint) encoding helpers
101// ===========================================================================
102const FLAG_COMPRESSED_PAYLOAD: u8 = 0x01;
103const FLAG_RELIABLE_HEADER_PRESENT: u8 = 0x02;
104const FLAG_WIRE_CONTRACT: u8 = 0x04;
105const FLAG_PACKET_NONCE: u8 = 0x08;
106#[cfg(feature = "cryptography")]
107const FLAG_E2E_ENCRYPTED_PAYLOAD: u8 = 0x10;
108const FLAG_ENDPOINT_BITMAP_PRESENT: u8 = 0x20;
109const FLAG_ENDPOINT_LIST_PRESENT: u8 = 0x80;
110const FLAG_COMPACT_RELIABLE_HEADER: u8 = 0x40;
111const CONTRACT_FLAG_TARGETS: u8 = 0x01;
112const CONTRACT_FLAG_SHAPE: u8 = 0x02;
113const CONTRACT_FLAG_RELIABLE_HEADER: u8 = 0x04;
114const RELIABLE_WIRE_FLAG_SEQ_PRESENT: u8 = 0x04;
115const RELIABLE_WIRE_FLAG_ACK_PRESENT: u8 = 0x08;
116const RELIABLE_PUBLIC_FLAGS_MASK: u8 =
117    RELIABLE_FLAG_ACK_ONLY | RELIABLE_FLAG_UNORDERED | RELIABLE_FLAG_UNSEQUENCED;
118#[cfg(feature = "cryptography")]
119const E2E_NONCE_LEN: usize = 12;
120#[cfg(feature = "cryptography")]
121const E2E_TAG_CAP: usize = 32;
122
123#[derive(Clone, Debug, PartialEq, Eq)]
124struct WireContract {
125    shape: Option<MessageElement>,
126    target_senders: Option<Arc<[u64]>>,
127    has_reliable_header: bool,
128}
129
130/// Encode a `u64` as ULEB128 and append it to `out`.
131#[inline]
132fn write_uleb128<T>(mut v: u64, out: &mut Vec<T>)
133where
134    T: From<u8>,
135{
136    loop {
137        let mut byte = (v & 0x7F) as u8;
138        v >>= 7;
139        if v != 0 {
140            byte |= 0x80;
141        }
142        out.push(T::from(byte));
143        if v == 0 {
144            break;
145        }
146    }
147}
148
149/// Decode a ULEB128-encoded `u64` from the given reader.
150#[inline]
151fn read_uleb128(r: &mut ByteReader) -> Result<u64, TelemetryError> {
152    crate::varint::read_u64(r.buf, &mut r.off)
153}
154
155/// Compute the encoded length (in bytes) of a ULEB128-encoded `u64`.
156#[inline]
157fn uleb128_size(mut v: u64) -> usize {
158    let mut n = 1;
159    while v >= 0x80 {
160        v >>= 7;
161        n += 1;
162    }
163    n
164}
165
166/// Count the total number of bits set across all bytes of the bitmap.
167#[inline]
168fn bitmap_popcount(bm: &[u8]) -> usize {
169    bm.iter().map(|b| b.count_ones() as usize).sum()
170}
171
172// ===========================================================================
173// ByteReader: tiny cursor over a byte slice
174// ===========================================================================
175
176#[derive(Clone, Copy)]
177struct ByteReader<'a> {
178    buf: &'a [u8],
179    off: usize,
180}
181
182impl<'a> ByteReader<'a> {
183    /// Create a new reader over the given buffer starting at offset 0.
184    fn new(buf: &'a [u8]) -> Self {
185        Self { buf, off: 0 }
186    }
187
188    /// Remaining bytes that can still be read.
189    fn remaining(&self) -> usize {
190        self.buf.len().saturating_sub(self.off)
191    }
192
193    /// Read exactly `n` bytes, advancing the internal offset.
194    #[inline(always)]
195    fn read_bytes(&mut self, n: usize) -> Result<&'a [u8], TelemetryError> {
196        if self.remaining() < n {
197            return Err(TelemetryError::Unpack("short read"));
198        }
199        let s = &self.buf[self.off..self.off + n];
200        self.off += n;
201        Ok(s)
202    }
203}
204
205#[inline]
206fn write_u32_le(v: u32, out: &mut Vec<u8>) {
207    out.extend_from_slice(&v.to_le_bytes());
208}
209
210#[inline]
211fn read_u32_le(r: &mut ByteReader) -> Result<u32, TelemetryError> {
212    let b = r.read_bytes(4)?;
213    Ok(u32::from_le_bytes([b[0], b[1], b[2], b[3]]))
214}
215
216#[inline]
217fn write_wire_shape(shape: MessageElement, out: &mut Vec<u8>) -> Result<(), TelemetryError> {
218    let dt = crate::config::message_data_type_code(shape.data_type());
219    let class = crate::config::message_class_code(shape.message_type());
220    let mut packed = dt | (class << 4);
221    if matches!(shape, MessageElement::Static(_, _, _)) {
222        packed |= 1 << 6;
223    }
224    out.push(packed);
225    if let MessageElement::Static(count, _, _) = shape {
226        let count = u64::try_from(count).map_err(|_| TelemetryError::Pack("wire shape count"))?;
227        write_uleb128(count, out);
228    }
229    Ok(())
230}
231
232#[inline]
233fn decode_wire_shape(r: &mut ByteReader) -> Result<MessageElement, TelemetryError> {
234    let packed = r.read_bytes(1)?[0];
235    let dt = crate::config::message_data_type_from_code(packed & 0x0F)
236        .ok_or(TelemetryError::Unpack("wire shape type"))?;
237    let class = crate::config::message_class_from_code((packed >> 4) & 0x03)
238        .ok_or(TelemetryError::Unpack("wire shape class"))?;
239    if (packed & (1 << 6)) != 0 {
240        let count = usize::try_from(read_uleb128(r)?)
241            .map_err(|_| TelemetryError::Unpack("wire shape count"))?;
242        Ok(MessageElement::Static(count, dt, class))
243    } else {
244        Ok(MessageElement::Dynamic(dt, class))
245    }
246}
247
248#[inline]
249fn write_wire_contract(
250    shape: Option<MessageElement>,
251    target_senders: &[u64],
252    has_reliable_header: bool,
253    out: &mut Vec<u8>,
254) -> Result<(), TelemetryError> {
255    // Keep the contract self-contained and compact. It exists only to preserve
256    // delivery/decode intent for packets that are already in flight while
257    // schema or topology updates are still converging.
258    let mut flags = 0u8;
259    if !target_senders.is_empty() {
260        flags |= CONTRACT_FLAG_TARGETS;
261    }
262    if shape.is_some() {
263        flags |= CONTRACT_FLAG_SHAPE;
264    }
265    if has_reliable_header {
266        flags |= CONTRACT_FLAG_RELIABLE_HEADER;
267    }
268    out.push(flags);
269    if let Some(shape) = shape {
270        write_wire_shape(shape, out)?;
271    }
272    if !target_senders.is_empty() {
273        write_uleb128(target_senders.len() as u64, out);
274        for hash in target_senders {
275            out.extend_from_slice(&hash.to_le_bytes());
276        }
277    }
278    Ok(())
279}
280
281/// Size the frozen contract without constructing temporary shape/target buffers.
282fn wire_contract_size(shape: Option<MessageElement>, target_senders: &[u64]) -> usize {
283    1 + shape.map_or(0, |shape| match shape {
284        MessageElement::Static(count, _, _) => 1 + uleb128_size(count as u64),
285        _ => 1,
286    }) + if target_senders.is_empty() {
287        0
288    } else {
289        uleb128_size(target_senders.len() as u64) + target_senders.len() * 8
290    }
291}
292
293#[inline]
294fn decode_wire_contract(
295    r: &mut ByteReader,
296    has_contract: bool,
297) -> Result<WireContract, TelemetryError> {
298    if !has_contract {
299        return Ok(WireContract {
300            shape: None,
301            target_senders: None,
302            has_reliable_header: false,
303        });
304    }
305    let contract_len = usize::try_from(read_uleb128(r)?)
306        .map_err(|_| TelemetryError::Unpack("wire contract length"))?;
307    let contract_bytes = r.read_bytes(contract_len)?;
308    let mut cr = ByteReader::new(contract_bytes);
309    // Parsing through a bounded sub-reader lets us reject trailing garbage in
310    // the contract cleanly instead of accidentally treating it as payload or
311    // reliable-header bytes.
312    let flags = cr.read_bytes(1)?[0];
313    let shape = if (flags & CONTRACT_FLAG_SHAPE) != 0 {
314        Some(decode_wire_shape(&mut cr)?)
315    } else {
316        None
317    };
318    let target_senders: Option<Arc<[u64]>> = if (flags & CONTRACT_FLAG_TARGETS) != 0 {
319        let count = usize::try_from(read_uleb128(&mut cr)?)
320            .map_err(|_| TelemetryError::Unpack("wire contract target count"))?;
321        if count > cr.remaining() / 8 {
322            return Err(TelemetryError::Unpack("wire contract target count"));
323        }
324        if count == 0 {
325            None
326        } else {
327            let mut targets = Arc::<[u64]>::new_uninit_slice(count);
328            for slot in Arc::get_mut(&mut targets).expect("new target storage is unique") {
329                let bytes = cr.read_bytes(8)?;
330                slot.write(u64::from_le_bytes([
331                    bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
332                ]));
333            }
334            // The bounded target count guarantees every slot was initialized.
335            Some(unsafe { targets.assume_init() })
336        }
337    } else {
338        None
339    };
340    if cr.remaining() != 0 {
341        return Err(TelemetryError::Unpack("wire contract trailing bytes"));
342    }
343    Ok(WireContract {
344        shape,
345        target_senders,
346        has_reliable_header: (flags & CONTRACT_FLAG_RELIABLE_HEADER) != 0,
347    })
348}
349
350#[inline]
351fn crc32_bytes(data: &[u8]) -> u32 {
352    let mut hasher = Crc32Hasher::new();
353    hasher.update(data);
354    hasher.finalize()
355}
356
357#[inline]
358fn append_crc32(out: &mut Vec<u8>) {
359    let crc = crc32_bytes(out);
360    out.extend_from_slice(&crc.to_le_bytes());
361}
362
363// Initialize every byte before publishing the shared frame. Keeping the
364// payload separate avoids a second full-size allocation during Arc creation.
365fn join_frame_with_crc(header: &[u8], payload: &[u8]) -> Arc<[u8]> {
366    let mut hasher = Crc32Hasher::new();
367    hasher.update(header);
368    hasher.update(payload);
369    let crc = hasher.finalize().to_le_bytes();
370    let mut frame = Arc::<[u8]>::new_uninit_slice(header.len() + payload.len() + crc.len());
371    let bytes = Arc::get_mut(&mut frame).unwrap();
372    for (slot, value) in bytes
373        .iter_mut()
374        .zip(header.iter().chain(payload).chain(&crc))
375    {
376        slot.write(*value);
377    }
378    // SAFETY: the iterator above writes exactly header + payload + CRC bytes,
379    // which is the full allocation length. No uninitialized byte escapes.
380    unsafe { frame.assume_init() }
381}
382
383#[inline]
384fn split_crc32(buf: &[u8]) -> Result<(&[u8], u32), TelemetryError> {
385    if buf.len() < CRC32_BYTES {
386        return Err(TelemetryError::Unpack("short buffer"));
387    }
388    let data_len = buf.len() - CRC32_BYTES;
389    let crc = u32::from_le_bytes([
390        buf[data_len],
391        buf[data_len + 1],
392        buf[data_len + 2],
393        buf[data_len + 3],
394    ]);
395    Ok((&buf[..data_len], crc))
396}
397
398#[cfg(all(test, feature = "std", feature = "discovery"))]
399std::thread_local! {
400    static CRC_VERIFICATIONS: core::cell::Cell<usize> = const { core::cell::Cell::new(0) };
401}
402#[cfg(all(test, feature = "std", feature = "discovery"))]
403pub(crate) fn crc_verification_count() -> usize {
404    CRC_VERIFICATIONS.with(core::cell::Cell::get)
405}
406
407#[inline]
408fn verify_crc32(buf: &[u8]) -> Result<&[u8], TelemetryError> {
409    #[cfg(all(test, feature = "std", feature = "discovery"))]
410    CRC_VERIFICATIONS.with(|count| count.set(count.get().wrapping_add(1)));
411    let (data, expected) = split_crc32(buf)?;
412    let actual = crc32_bytes(data);
413    if actual != expected {
414        return Err(TelemetryError::Unpack("crc32 mismatch"));
415    }
416    Ok(data)
417}
418
419#[cfg(feature = "cryptography")]
420#[inline]
421fn e2e_nonce_for_packet(pkt: &Packet) -> [u8; E2E_NONCE_LEN] {
422    let mut nonce = [0u8; E2E_NONCE_LEN];
423    nonce[..4].copy_from_slice(&pkt.data_type().as_u32().to_le_bytes());
424    nonce[4..10].copy_from_slice(&(pkt.timestamp() & 0x0000_FFFF_FFFF_FFFF).to_le_bytes()[..6]);
425    nonce[10..].copy_from_slice(&pkt.nonce().to_le_bytes());
426    nonce
427}
428
429#[cfg(feature = "cryptography")]
430fn write_encrypted_payload(
431    pkt: &Packet,
432    key_id: u32,
433    plaintext_wire_payload: &[u8],
434    out: &mut Vec<u8>,
435) -> TelemetryResult<()> {
436    let aad_end = out.len();
437    let nonce = e2e_nonce_for_packet(pkt);
438    let mut ciphertext = vec![0u8; plaintext_wire_payload.len()];
439    let mut tag = [0u8; E2E_TAG_CAP];
440    let (ciphertext_len, tag_len) = crate::crypto::seal_with_registered_crypto(
441        key_id,
442        &nonce,
443        &out[..aad_end],
444        plaintext_wire_payload,
445        &mut ciphertext,
446        &mut tag,
447    )?;
448    if ciphertext_len > ciphertext.len() || tag_len > tag.len() {
449        return Err(TelemetryError::SizeMismatchError);
450    }
451    write_uleb128(u64::from(key_id), out);
452    write_uleb128(plaintext_wire_payload.len() as u64, out);
453    write_uleb128(nonce.len() as u64, out);
454    out.extend_from_slice(&nonce);
455    write_uleb128(tag_len as u64, out);
456    out.extend_from_slice(&tag[..tag_len]);
457    out.extend_from_slice(&ciphertext[..ciphertext_len]);
458    Ok(())
459}
460
461#[cfg(feature = "cryptography")]
462fn read_encrypted_payload(
463    r: &mut ByteReader,
464    aad: &[u8],
465    plaintext_len: usize,
466) -> TelemetryResult<Vec<u8>> {
467    let key_id = u32::try_from(read_uleb128(r)?)
468        .map_err(|_| TelemetryError::Unpack("e2e key id too large"))?;
469    let wire_payload_len = usize::try_from(read_uleb128(r)?)
470        .map_err(|_| TelemetryError::Unpack("e2e payload length"))?;
471    if wire_payload_len > plaintext_len {
472        return Err(TelemetryError::Unpack("bad e2e payload length"));
473    }
474    let nonce_len = usize::try_from(read_uleb128(r)?)
475        .map_err(|_| TelemetryError::Unpack("e2e nonce length"))?;
476    if nonce_len == 0 || nonce_len > 64 {
477        return Err(TelemetryError::Unpack("bad e2e nonce length"));
478    }
479    let nonce = r.read_bytes(nonce_len)?;
480    let tag_len =
481        usize::try_from(read_uleb128(r)?).map_err(|_| TelemetryError::Unpack("e2e tag length"))?;
482    if tag_len == 0 || tag_len > E2E_TAG_CAP {
483        return Err(TelemetryError::Unpack("bad e2e tag length"));
484    }
485    let tag = r.read_bytes(tag_len)?;
486    let ciphertext_len = r.remaining();
487    let ciphertext = r.read_bytes(ciphertext_len)?;
488    let mut plaintext = vec![0u8; wire_payload_len];
489    let opened_len = crate::crypto::open_with_registered_crypto(
490        key_id,
491        nonce,
492        aad,
493        ciphertext,
494        tag,
495        &mut plaintext,
496    )?;
497    if opened_len != wire_payload_len {
498        return Err(TelemetryError::SizeMismatchError);
499    }
500    Ok(plaintext)
501}
502
503#[inline]
504fn write_reliable_header(h: ReliableHeader, out: &mut Vec<u8>) {
505    out.push(h.flags);
506    write_u32_le(h.seq, out);
507    write_u32_le(h.ack, out);
508}
509
510#[inline]
511fn reliable_compact_size(h: ReliableHeader) -> usize {
512    let seq_present = (h.flags & RELIABLE_FLAG_ACK_ONLY) == 0 || h.seq != 0;
513    let ack_present = h.ack != 0 || (h.flags & RELIABLE_FLAG_ACK_ONLY) != 0;
514    1 + if seq_present {
515        uleb128_size(h.seq as u64)
516    } else {
517        0
518    } + if ack_present {
519        uleb128_size(h.ack as u64)
520    } else {
521        0
522    }
523}
524
525#[inline]
526fn should_compact_reliable_header(h: ReliableHeader) -> bool {
527    reliable_compact_size(h) < RELIABLE_HEADER_BYTES
528}
529
530#[inline]
531fn reliable_wire_size(h: ReliableHeader, compact: bool) -> usize {
532    if compact {
533        reliable_compact_size(h)
534    } else {
535        RELIABLE_HEADER_BYTES
536    }
537}
538
539pub(crate) fn write_reliable_header_encoded(h: ReliableHeader, compact: bool, out: &mut Vec<u8>) {
540    if !compact {
541        write_reliable_header(h, out);
542        return;
543    }
544
545    let seq_present = (h.flags & RELIABLE_FLAG_ACK_ONLY) == 0 || h.seq != 0;
546    let ack_present = h.ack != 0 || (h.flags & RELIABLE_FLAG_ACK_ONLY) != 0;
547    let mut wire_flags = h.flags & RELIABLE_PUBLIC_FLAGS_MASK;
548    if seq_present {
549        wire_flags |= RELIABLE_WIRE_FLAG_SEQ_PRESENT;
550    }
551    if ack_present {
552        wire_flags |= RELIABLE_WIRE_FLAG_ACK_PRESENT;
553    }
554    out.push(wire_flags);
555    if seq_present {
556        write_uleb128(h.seq as u64, out);
557    }
558    if ack_present {
559        write_uleb128(h.ack as u64, out);
560    }
561}
562
563#[inline]
564fn read_reliable_header(r: &mut ByteReader) -> Result<ReliableHeader, TelemetryError> {
565    let flags = r.read_bytes(1)?[0];
566    let seq = read_u32_le(r)?;
567    let ack = read_u32_le(r)?;
568    Ok(ReliableHeader { flags, seq, ack })
569}
570
571fn read_reliable_header_compact(r: &mut ByteReader) -> Result<ReliableHeader, TelemetryError> {
572    let wire_flags = r.read_bytes(1)?[0];
573    let seq = if (wire_flags & RELIABLE_WIRE_FLAG_SEQ_PRESENT) != 0 {
574        u32::try_from(read_uleb128(r)?)
575            .map_err(|_| TelemetryError::Unpack("reliable seq too large"))?
576    } else {
577        0
578    };
579    let ack = if (wire_flags & RELIABLE_WIRE_FLAG_ACK_PRESENT) != 0 {
580        u32::try_from(read_uleb128(r)?)
581            .map_err(|_| TelemetryError::Unpack("reliable ack too large"))?
582    } else {
583        0
584    };
585    Ok(ReliableHeader {
586        flags: wire_flags & RELIABLE_PUBLIC_FLAGS_MASK,
587        seq,
588        ack,
589    })
590}
591
592#[inline]
593fn read_reliable_header_encoded(
594    r: &mut ByteReader,
595    compact: bool,
596) -> Result<ReliableHeader, TelemetryError> {
597    if compact {
598        read_reliable_header_compact(r)
599    } else {
600        read_reliable_header(r)
601    }
602}
603
604#[cfg(feature = "std")]
605static ADDRESS_BOOK: OnceLock<Mutex<BTreeMap<u32, Arc<str>>>> = OnceLock::new();
606
607#[cfg(feature = "std")]
608fn address_book() -> &'static Mutex<BTreeMap<u32, Arc<str>>> {
609    ADDRESS_BOOK.get_or_init(|| Mutex::new(BTreeMap::new()))
610}
611
612#[inline]
613pub(crate) fn source_address_for_sender(sender: &str) -> u32 {
614    let addr = sender_address_u32(sender);
615    remember_source_address(addr, sender);
616    addr
617}
618
619#[cfg(feature = "std")]
620pub(crate) fn remember_source_address(addr: u32, sender: &str) {
621    address_book()
622        .lock()
623        .expect("wire address book poisoned")
624        .entry(addr)
625        .or_insert_with(|| Arc::<str>::from(sender));
626}
627
628#[cfg(not(feature = "std"))]
629pub(crate) fn remember_source_address(_addr: u32, _sender: &str) {}
630
631#[cfg(feature = "std")]
632fn sender_name_for_address(addr: u32) -> Arc<str> {
633    address_book()
634        .lock()
635        .expect("wire address book poisoned")
636        .get(&addr)
637        .cloned()
638        .unwrap_or_else(|| format_sender_address(addr))
639}
640
641#[cfg(not(feature = "std"))]
642fn sender_name_for_address(addr: u32) -> Arc<str> {
643    format_sender_address(addr)
644}
645
646fn format_sender_address(addr: u32) -> Arc<str> {
647    // Format the numeric alias on the stack; only the retained Arc allocates.
648    let mut bytes = [0u8; 16];
649    let mut start = bytes.len();
650    let mut remaining = addr;
651    loop {
652        start -= 1;
653        bytes[start] = b'0' + (remaining % 10) as u8;
654        remaining /= 10;
655        if remaining == 0 {
656            break;
657        }
658    }
659    start -= 6;
660    bytes[start..start + 6].copy_from_slice(b"@addr:");
661    Arc::from(core::str::from_utf8(&bytes[start..]).expect("numeric alias is ASCII"))
662}
663
664// ===========================================================================
665// Endpoint bitmap constants and helpers
666// ===========================================================================
667
668/// Number of bits needed to cover all possible `DataEndpoint` discriminants.
669const EP_BITMAP_BITS: usize = 256;
670#[cfg(test)]
671use crate::MAX_VALUE_DATA_ENDPOINT;
672
673/// Number of bytes required to store [`EP_BITMAP_BITS`] bits.
674const EP_BITMAP_BYTES: usize = EP_BITMAP_BITS.div_ceil(8);
675
676/// Build a compact endpoint bitmap from the provided list of endpoints.
677///
678/// Each endpoint `ep` sets the bit at position `ep.as_u32()` in the bitmap.
679/// Bits are packed LSB-first within each byte.
680#[inline]
681fn build_endpoint_bitmap(eps: &[DataEndpoint]) -> [u8; EP_BITMAP_BYTES] {
682    let mut bm = [0u8; EP_BITMAP_BYTES];
683    for &ep in eps {
684        let idx = ep.as_u32() as usize;
685        debug_assert!(idx < EP_BITMAP_BITS, "endpoint discriminant out of range");
686        if idx < EP_BITMAP_BITS {
687            let byte = idx / 8;
688            let bit = idx % 8;
689            bm[byte] |= 1u8 << bit;
690        }
691    }
692    bm
693}
694
695/// Expand a bitmap into registered endpoints in ascending discriminant order.
696fn expand_endpoint_bitmap(bm: &[u8]) -> TelemetryResult<Arc<[DataEndpoint]>> {
697    if bm.len() != EP_BITMAP_BYTES {
698        return Err(TelemetryError::Unpack("bad endpoint bitmap size"));
699    }
700    // Only visit set bits. Sparse telemetry destinations should not initialize
701    // and copy a 256-element endpoint array on every header inspection.
702    let count = bitmap_popcount(bm);
703    let mut endpoints = Arc::<[DataEndpoint]>::new_uninit_slice(count);
704    let slots = Arc::get_mut(&mut endpoints).expect("new endpoint storage is unique");
705    let mut written = 0;
706    for (byte_index, &byte) in bm.iter().enumerate() {
707        let mut bits = byte;
708        while bits != 0 {
709            let idx = byte_index * 8 + bits.trailing_zeros() as usize;
710            slots[written].write(
711                DataEndpoint::try_from_u32(idx as u32)
712                    .ok_or(TelemetryError::Unpack("bad endpoint bit set"))?,
713            );
714            written += 1;
715            bits &= bits - 1;
716        }
717    }
718    // Every set bit initializes exactly one slot; errors drop uninitialized storage.
719    debug_assert_eq!(written, count);
720    Ok(unsafe { endpoints.assume_init() })
721}
722
723pub(crate) fn unique_sorted_endpoints(
724    endpoints: &[DataEndpoint],
725) -> impl Iterator<Item = DataEndpoint> + '_ {
726    let mut previous = None;
727    core::iter::from_fn(move || {
728        let next = endpoints
729            .iter()
730            .copied()
731            .filter(|endpoint| previous.is_none_or(|previous| *endpoint > previous))
732            .min();
733        // Retain the last maximum after exhaustion; subsequent calls remain
734        // exhausted without a separate state field or restarting the sequence.
735        previous = next.or(previous);
736        next
737    })
738}
739
740fn endpoint_arc_from_iter(
741    count: usize,
742    endpoints: impl Iterator<Item = DataEndpoint>,
743) -> TelemetryResult<Arc<[DataEndpoint]>> {
744    let mut output = Arc::<[DataEndpoint]>::new_uninit_slice(count);
745    let slots = Arc::get_mut(&mut output).unwrap();
746    let mut written = 0;
747    for endpoint in endpoints {
748        if written == count {
749            return Err(TelemetryError::Unpack("endpoint count mismatch"));
750        }
751        slots[written].write(endpoint);
752        written += 1;
753    }
754    if written != count {
755        return Err(TelemetryError::Unpack("endpoint count mismatch"));
756    }
757    // SAFETY: every slot was initialized exactly once above.
758    Ok(unsafe { output.assume_init() })
759}
760fn endpoint_list_size(eps: &[DataEndpoint]) -> usize {
761    unique_sorted_endpoints(eps)
762        .map(|ep| uleb128_size(u64::from(ep.0)))
763        .sum()
764}
765fn explicit_endpoint_list(eps: &[DataEndpoint]) -> bool {
766    eps.iter().any(|ep| ep.0 >= EP_BITMAP_BITS as u32) || endpoint_list_size(eps) < EP_BITMAP_BYTES
767}
768fn endpoints_match_schema(ty: DataType, eps: &[DataEndpoint]) -> bool {
769    let meta = message_meta(ty);
770    unique_sorted_endpoints(eps).eq(unique_sorted_endpoints(meta.endpoints_ref()))
771}
772fn read_endpoint_list<'a>(r: &mut ByteReader<'a>, count: usize) -> TelemetryResult<&'a [u8]> {
773    let start = r.off;
774    let mut previous = None;
775    for _ in 0..count {
776        let id = u32::try_from(read_uleb128(r)?)
777            .map_err(|_| TelemetryError::Unpack("endpoint too large"))?;
778        if previous.is_some_and(|previous| id <= previous) {
779            return Err(TelemetryError::Unpack(
780                "endpoint list is not unique and sorted",
781            ));
782        }
783        previous = Some(id);
784    }
785    Ok(&r.buf[start..r.off])
786}
787fn endpoint_list_iter(bytes: &[u8]) -> impl Iterator<Item = DataEndpoint> + '_ {
788    let mut r = ByteReader::new(bytes);
789    core::iter::from_fn(move || {
790        if r.remaining() == 0 {
791            return None;
792        }
793        read_uleb128(&mut r)
794            .ok()
795            .and_then(|id| u32::try_from(id).ok())
796            .map(DataEndpoint)
797    })
798}
799
800#[inline]
801fn schema_endpoints_from_type(ty: DataType, nep: usize) -> TelemetryResult<Arc<[DataEndpoint]>> {
802    // Match bitmap ordering and duplicate removal without expanding all 256
803    // possible destinations for the usual one-endpoint schema.
804    let meta = message_meta(ty);
805    let schema = meta.endpoints_ref();
806    if schema.len() == nep
807        && schema
808            .windows(2)
809            .all(|pair| pair[0].as_u32() < pair[1].as_u32())
810        && schema
811            .iter()
812            .all(|ep| DataEndpoint::try_from_u32(ep.as_u32()).is_some())
813    {
814        // Compiled/registered metadata already owns sorted endpoints on hosts;
815        // embedded metadata is static. Avoid a temporary Vec in either case.
816        #[cfg(feature = "std")]
817        return Ok(meta.endpoints);
818        #[cfg(not(feature = "std"))]
819        return Ok(Arc::from(schema));
820    }
821    endpoint_arc_from_iter(nep, unique_sorted_endpoints(schema))
822}
823
824#[inline]
825fn endpoints_from_wire_or_schema(
826    r: &mut ByteReader<'_>,
827    bitmap_present: bool,
828    list_present: bool,
829    ty: Option<DataType>,
830    nep: usize,
831) -> TelemetryResult<Arc<[DataEndpoint]>> {
832    if list_present {
833        if bitmap_present {
834            return Err(TelemetryError::Unpack("ambiguous endpoint encoding"));
835        }
836        let bytes = read_endpoint_list(r, nep)?;
837        #[cfg(feature = "std")]
838        if let Some(meta) = ty.and_then(crate::config::try_get_message_meta)
839            && endpoint_list_iter(bytes).eq(meta.endpoints.iter().copied())
840        {
841            return Ok(meta.endpoints);
842        }
843        return endpoint_arc_from_iter(nep, endpoint_list_iter(bytes));
844    }
845    if bitmap_present {
846        let bm = r.read_bytes(EP_BITMAP_BYTES)?;
847        let endpoints = expand_endpoint_bitmap(bm)?;
848        if endpoints.len() != nep {
849            return Err(TelemetryError::Unpack("endpoint count mismatch"));
850        }
851        Ok(endpoints)
852    } else {
853        let ty = ty.ok_or(TelemetryError::InvalidType)?;
854        schema_endpoints_from_type(ty, nep)
855    }
856}
857
858#[inline]
859fn data_type_id_from_wire(ty_v: u64) -> TelemetryResult<u32> {
860    let ty_u32 = u32::try_from(ty_v).map_err(|_| TelemetryError::Unpack("type too large"))?;
861    Ok(ty_u32)
862}
863
864// ===========================================================================
865// Packing
866// ===========================================================================
867
868/// Pack a [`Packet`] into the compact v2 wire format.
869///
870/// The returned `Arc<[u8]>` owns the packed bytes and can be shared cheaply.
871/// # Arguments
872/// - `pkt`: Telemetry packet to pack.
873///
874/// # Returns
875/// - `Arc<[u8]>`: Packed packet in compact v2 wire format.
876pub fn pack_packet(pkt: &Packet) -> Arc<[u8]> {
877    if is_reliable_type(pkt.data_type()) {
878        // Default to an unsequenced reliable header to keep the wire format consistent.
879        // Use `pack_packet_with_reliable` for ordered/retransmitted delivery.
880        let hdr = ReliableHeader {
881            flags: RELIABLE_FLAG_UNSEQUENCED,
882            seq: 0,
883            ack: 0,
884        };
885        return pack_packet_with_reliable(pkt, hdr);
886    }
887    pack_packet_inner(pkt, None)
888}
889
890/// Pack a [`Packet`] with an explicit reliable header.
891///
892/// This should be used for data types configured as `reliable` in the schema.
893pub fn pack_packet_with_reliable(pkt: &Packet, header: ReliableHeader) -> Arc<[u8]> {
894    pack_packet_inner(pkt, Some(header))
895}
896
897/// Pack a reliable ACK-only control frame for the given data type.
898///
899/// The resulting bytes are not a valid `Packet` and should be handled
900/// by the router's reliable layer.
901pub fn pack_reliable_ack(sender: &str, ty: DataType, timestamp_ms: u64, ack: u32) -> Arc<[u8]> {
902    let source_address = source_address_for_sender(sender);
903
904    // No payload for ACK-only control frames.
905    let mut out = Vec::with_capacity(32 + EP_BITMAP_BYTES + CRC32_BYTES);
906
907    let flags: u8 = FLAG_ENDPOINT_LIST_PRESENT | FLAG_RELIABLE_HEADER_PRESENT;
908    out.push(flags);
909    out.push(0u8); // NEP = 0
910
911    write_uleb128(ty.as_u32() as u64, &mut out);
912    write_uleb128(0u64, &mut out); // payload size
913    write_uleb128(timestamp_ms, &mut out);
914    write_uleb128(source_address as u64, &mut out);
915
916    let reliable = ReliableHeader {
917        flags: RELIABLE_FLAG_ACK_ONLY,
918        seq: 0,
919        ack,
920    };
921    if should_compact_reliable_header(reliable) {
922        out[0] |= FLAG_COMPACT_RELIABLE_HEADER;
923        write_reliable_header_encoded(reliable, true, &mut out);
924    } else {
925        write_reliable_header_encoded(reliable, false, &mut out);
926    }
927    append_crc32(&mut out);
928
929    Arc::<[u8]>::from(out)
930}
931
932// Discovery must bootstrap across peers before optional codec support is known.
933// A host with compression enabled must still seed an empty no-compression MCU.
934fn payload_for_wire(pkt: &Packet) -> (bool, alloc::borrow::Cow<'_, [u8]>) {
935    payload_for_encoding(pkt, true)
936}
937
938fn payload_for_encoding(
939    pkt: &Packet,
940    #[cfg_attr(not(feature = "compression"), allow(unused_variables))] compression_allowed: bool,
941) -> (bool, alloc::borrow::Cow<'_, [u8]>) {
942    #[cfg(feature = "compression")]
943    if !compression_allowed {
944        return (false, alloc::borrow::Cow::Borrowed(pkt.payload()));
945    }
946    #[cfg(feature = "discovery")]
947    if crate::discovery::is_discovery_type(pkt.data_type()) {
948        return (false, alloc::borrow::Cow::Borrowed(pkt.payload()));
949    }
950    payload_compression::compress_if_beneficial(pkt.payload())
951}
952
953fn pack_packet_inner(pkt: &Packet, reliable: Option<ReliableHeader>) -> Arc<[u8]> {
954    pack_packet_inner_with_contract(
955        pkt,
956        reliable,
957        pkt.wire_shape(),
958        pkt.wire_target_senders(),
959        None,
960    )
961    .expect("plaintext packet packing failed")
962}
963
964/// Pack `pkt` while explicitly controlling the wire-contract metadata.
965///
966/// Router and relay forwarding paths use this helper when they need to attach
967/// or preserve a migration-safe contract instead of simply serializing the
968/// packet against the current runtime schema/topology view.
969///
970/// # Parameters
971/// - `pkt`: Logical packet to pack.
972/// - `reliable`: Optional hop-level reliable header to append after the
973///   contract.
974/// - `shape`: Optional inline payload shape to encode into the contract so
975///   downstream unpackers can validate/decode the payload against the
976///   original shape.
977/// - `target_senders`: Frozen destination-holder sender hashes that keep
978///   in-flight routing bound to the intended holders.
979///
980/// # Returns
981/// - `Ok(Arc<[u8]>)` containing the packed frame bytes.
982/// - `Err(TelemetryError)` if contract encoding fails.
983#[cfg(any(test, not(feature = "compression")))]
984pub(crate) fn pack_packet_with_wire_contract(
985    pkt: &Packet,
986    reliable: Option<ReliableHeader>,
987    shape: Option<MessageElement>,
988    target_senders: &[u64],
989) -> TelemetryResult<Arc<[u8]>> {
990    pack_packet_inner_with_contract(pkt, reliable, shape, target_senders, None)
991}
992
993#[derive(Clone, Copy, Debug)]
994#[cfg_attr(not(feature = "cryptography"), allow(dead_code))]
995pub(crate) struct E2eSealConfig {
996    pub key_id: u32,
997}
998
999#[cfg(all(feature = "cryptography", not(feature = "compression")))]
1000pub(crate) fn pack_packet_with_wire_contract_e2e(
1001    pkt: &Packet,
1002    reliable: Option<ReliableHeader>,
1003    shape: Option<MessageElement>,
1004    target_senders: &[u64],
1005    e2e: E2eSealConfig,
1006) -> TelemetryResult<Arc<[u8]>> {
1007    pack_packet_inner_with_contract(pkt, reliable, shape, target_senders, Some(e2e))
1008}
1009
1010fn pack_packet_inner_with_contract(
1011    pkt: &Packet,
1012    reliable: Option<ReliableHeader>,
1013    shape: Option<MessageElement>,
1014    target_senders: &[u64],
1015    e2e: Option<E2eSealConfig>,
1016) -> TelemetryResult<Arc<[u8]>> {
1017    with_packet_frame_parts(
1018        pkt,
1019        reliable,
1020        shape,
1021        target_senders,
1022        e2e,
1023        true,
1024        |header, payload| Ok(join_frame_with_crc(header, payload)),
1025    )
1026}
1027
1028/// Router/relay replay basis is independent of the current outgoing peer set.
1029/// Sealed packets also keep their inner payload codec-free: transit nodes cannot
1030/// rewrite authenticated ciphertext without the recipient's key.
1031#[cfg(feature = "compression")]
1032pub(crate) fn pack_routed_packet(
1033    pkt: &Packet,
1034    reliable: Option<ReliableHeader>,
1035    shape: Option<MessageElement>,
1036    target_senders: &[u64],
1037    e2e: Option<E2eSealConfig>,
1038) -> TelemetryResult<Arc<[u8]>> {
1039    with_packet_frame_parts(
1040        pkt,
1041        reliable,
1042        shape,
1043        target_senders,
1044        e2e,
1045        false,
1046        |header, payload| Ok(join_frame_with_crc(header, payload)),
1047    )
1048}
1049
1050/// Re-encode only an ordinary plaintext payload, preserving its complete header,
1051/// frozen delivery contract, nonce and reliable sequence. No schema registration
1052/// or application decoding is needed. `None` retains the existing shared frame.
1053#[cfg(feature = "compression")]
1054pub(crate) fn payload_encoding_for_side(
1055    raw: &[u8],
1056    allowed: bool,
1057    max_decoded_bytes: usize,
1058) -> TelemetryResult<Option<Arc<[u8]>>> {
1059    let flags = *raw.first().ok_or(TelemetryError::Unpack("short prelude"))?;
1060    let compressed = flags & FLAG_COMPRESSED_PAYLOAD != 0;
1061    if flags & 0x10 != 0 {
1062        if compressed && !allowed {
1063            return Err(TelemetryError::Io(
1064                "opaque compressed payload requires codec-capable peers",
1065            ));
1066        }
1067        return Ok(None);
1068    }
1069    if !compressed && !allowed {
1070        return Ok(None);
1071    }
1072    // Small plaintext samples need neither a second CRC pass nor codec scratch.
1073    // This is an optional transformation; ordinary ingress validation remains
1074    // authoritative when the original shared bytes are retained unchanged.
1075    if !compressed {
1076        let mut prelude = ByteReader::new(raw);
1077        prelude.read_bytes(2)?;
1078        read_uleb128(&mut prelude)?;
1079        if read_uleb128(&mut prelude)? < crate::config::runtime_payload_compress_threshold() as u64
1080        {
1081            return Ok(None);
1082        }
1083    }
1084    let data = verify_crc32(raw)?;
1085    let mut reader = ByteReader::new(data);
1086    let envelope = read_routing_envelope(&mut reader)?;
1087    #[cfg(feature = "discovery")]
1088    let allowed = allowed && !crate::discovery::is_discovery_type(envelope.ty);
1089    if envelope.is_reliable || envelope.has_reliable_header {
1090        read_reliable_header_encoded(&mut reader, flags & FLAG_COMPACT_RELIABLE_HEADER != 0)?;
1091    }
1092    let header = &data[..reader.off];
1093    let payload = &data[reader.off..];
1094    if compressed && allowed {
1095        return Ok(None);
1096    }
1097    if !compressed && !allowed {
1098        return Ok(None);
1099    }
1100    let mut prelude = ByteReader::new(data);
1101    prelude.read_bytes(2)?;
1102    read_uleb128(&mut prelude)?;
1103    let logical = usize::try_from(read_uleb128(&mut prelude)?)
1104        .map_err(|_| TelemetryError::Unpack("payload too large"))?;
1105    if logical > max_decoded_bytes || logical > isize::MAX as usize - header.len() - CRC32_BYTES {
1106        if !compressed {
1107            return Ok(None);
1108        }
1109        return Err(TelemetryError::Io("payload exceeds routing memory budget"));
1110    }
1111    let bytes = logical
1112        .saturating_mul(2)
1113        .saturating_add(header.len().saturating_mul(2))
1114        .saturating_add(64);
1115    let largest = logical
1116        .saturating_add(header.len())
1117        .saturating_add(CRC32_BYTES)
1118        .saturating_add(32);
1119    if let Err(error) = crate::memory_admission::check(bytes, largest) {
1120        // Compression is optional; required decompression must retain ownership
1121        // and backpressure rather than emitting an undecodable frame.
1122        if !compressed {
1123            return Ok(None);
1124        }
1125        return Err(error);
1126    }
1127    let (new_flag, new_payload) = if compressed {
1128        (
1129            flags & !FLAG_COMPRESSED_PAYLOAD,
1130            alloc::borrow::Cow::Owned(payload_compression::decompress(payload, logical)?),
1131        )
1132    } else {
1133        if payload.len() != logical {
1134            return Err(TelemetryError::Unpack("payload length mismatch"));
1135        }
1136        let (beneficial, value) = payload_compression::compress_if_beneficial(payload);
1137        if !beneficial {
1138            return Ok(None);
1139        }
1140        (flags | FLAG_COMPRESSED_PAYLOAD, value)
1141    };
1142    let mut prefix = header.to_vec();
1143    prefix[0] = new_flag;
1144    Ok(Some(join_frame_with_crc(&prefix, &new_payload)))
1145}
1146
1147/// Conservative size credit before retaining a compressed-to-raw conversion.
1148/// Includes the complete original header/CRC and the logical payload, so the
1149/// bounded queue reservation never depends on optimistic compression ratios.
1150#[cfg(feature = "compression")]
1151pub(crate) fn decoded_frame_upper_bound(raw: &[u8]) -> TelemetryResult<usize> {
1152    let mut reader = ByteReader::new(raw);
1153    reader.read_bytes(2)?;
1154    read_uleb128(&mut reader)?;
1155    let logical = usize::try_from(read_uleb128(&mut reader)?)
1156        .map_err(|_| TelemetryError::Unpack("payload too large"))?;
1157    raw.len()
1158        .checked_add(logical)
1159        .ok_or(TelemetryError::Io("memory pressure"))
1160}
1161
1162/// Keep replay ownership independent of a transient side codec choice. Sealed
1163/// frames cannot be rewritten without their key; their side compatibility is
1164/// checked before dispatch instead.
1165#[cfg(feature = "compression")]
1166pub(crate) fn payload_encoding_for_retention(
1167    raw: &[u8],
1168    budget: usize,
1169) -> TelemetryResult<Option<Arc<[u8]>>> {
1170    if raw.first().is_some_and(|flags| flags & 0x10 != 0) {
1171        return Ok(None);
1172    }
1173    payload_encoding_for_side(raw, false, budget)
1174}
1175
1176/// Serialize the normal plaintext header while retaining the existing payload.
1177/// Used by chunked internal discovery to avoid a second schema-sized buffer.
1178pub(crate) fn with_plain_packet_parts<T>(
1179    pkt: &Packet,
1180    consume: impl FnOnce(&[u8], &[u8]) -> TelemetryResult<T>,
1181) -> TelemetryResult<T> {
1182    let reliable = is_reliable_type(pkt.data_type()).then_some(ReliableHeader {
1183        flags: RELIABLE_FLAG_UNSEQUENCED,
1184        seq: 0,
1185        ack: 0,
1186    });
1187    with_packet_frame_parts(
1188        pkt,
1189        reliable,
1190        pkt.wire_shape(),
1191        pkt.wire_target_senders(),
1192        None,
1193        true,
1194        consume,
1195    )
1196}
1197
1198fn packet_endpoint_bitmap_present(
1199    pkt: &Packet,
1200    shape: Option<MessageElement>,
1201    target_senders: &[u64],
1202) -> bool {
1203    let carries_wire_contract = shape.is_some() || !target_senders.is_empty();
1204    let endpoints_are_schema_default = endpoints_match_schema(pkt.data_type(), pkt.endpoints());
1205    // Reliability controls have a protocol-defined endpoint on every peer;
1206    // their receipt contract does not require a second, 32-byte copy of it.
1207    // Runtime application types retain explicit endpoints with a frozen
1208    // contract, since their schemas may differ during discovery/rejoin.
1209    let fixed_control_endpoints = matches!(
1210        pkt.data_type(),
1211        DataType::ReliableAck | DataType::ReliablePartialAck | DataType::ReliablePacketRequest
1212    );
1213    // Application catalogs may be complementary or still converging. A short
1214    // explicit list preserves this packet's actual destinations in that case.
1215    pkt.data_type().0 > DataType::P2pMessage.0
1216        || !endpoints_are_schema_default
1217        || (carries_wire_contract && !fixed_control_endpoints)
1218}
1219
1220fn with_packet_frame_parts<T>(
1221    pkt: &Packet,
1222    reliable: Option<ReliableHeader>,
1223    shape: Option<MessageElement>,
1224    target_senders: &[u64],
1225    #[cfg_attr(not(feature = "cryptography"), allow(unused_variables))] e2e: Option<E2eSealConfig>,
1226    compression_allowed: bool,
1227    consume: impl FnOnce(&[u8], &[u8]) -> TelemetryResult<T>,
1228) -> TelemetryResult<T> {
1229    let carries_wire_contract = shape.is_some() || !target_senders.is_empty();
1230    let endpoint_bitmap_present = packet_endpoint_bitmap_present(pkt, shape, target_senders);
1231    let endpoint_list_present = endpoint_bitmap_present && explicit_endpoint_list(pkt.endpoints());
1232    let nep_unique = unique_sorted_endpoints(pkt.endpoints()).count();
1233    let endpoint_bytes = if endpoint_list_present {
1234        endpoint_list_size(pkt.endpoints())
1235    } else if endpoint_bitmap_present {
1236        EP_BITMAP_BYTES
1237    } else {
1238        0
1239    };
1240
1241    let source_address = source_address_for_sender(pkt.sender());
1242
1243    // Decide whether to compress the payload.
1244    let (payload_compressed, payload_wire) = payload_for_encoding(pkt, compression_allowed);
1245
1246    // Size the variable-length prefix exactly. A 16-byte estimate can be
1247    // too small for real timestamps/nonces, doubling a multi-KiB schema
1248    // buffer while its payload and the old allocation are still retained.
1249    let reliable_is_compact = reliable.is_some_and(should_compact_reliable_header);
1250    let reliable_len = if let Some(hdr) = reliable {
1251        reliable_wire_size(hdr, reliable_is_compact)
1252    } else {
1253        0
1254    };
1255    let contract_bytes = wire_contract_size(shape, target_senders);
1256    let contract_len = if carries_wire_contract {
1257        uleb128_size(contract_bytes as u64) + contract_bytes
1258    } else {
1259        0
1260    };
1261    let capacity = header_size_bytes(pkt)
1262        + endpoint_bytes
1263        + contract_len
1264        + reliable_len
1265        + payload_wire.len()
1266        + CRC32_BYTES;
1267    #[cfg(feature = "cryptography")]
1268    let capacity = capacity
1269        + e2e.map_or(0, |seal| {
1270            uleb128_size(u64::from(seal.key_id))
1271                + uleb128_size(payload_wire.len() as u64)
1272                + uleb128_size(E2E_NONCE_LEN as u64)
1273                + E2E_NONCE_LEN
1274                + uleb128_size(E2E_TAG_CAP as u64)
1275                + E2E_TAG_CAP
1276        });
1277    // Plain frames need only a small header scratch buffer. Construct the
1278    // final Arc directly: Vec -> Arc otherwise retains two frame-sized
1279    // allocations alongside the schema payload on embedded routers.
1280    #[cfg(feature = "cryptography")]
1281    let encrypted = e2e.is_some();
1282    #[cfg(not(feature = "cryptography"))]
1283    let encrypted = false;
1284    let scratch_capacity = if encrypted {
1285        capacity
1286    } else {
1287        capacity - payload_wire.len() - CRC32_BYTES
1288    };
1289    let mut out = Vec::with_capacity(scratch_capacity);
1290    #[cfg(debug_assertions)]
1291    let initial_capacity = out.capacity();
1292
1293    // FLAGS byte
1294    let mut flags: u8 = 0;
1295    if payload_compressed {
1296        flags |= FLAG_COMPRESSED_PAYLOAD;
1297    }
1298    if carries_wire_contract {
1299        flags |= FLAG_WIRE_CONTRACT;
1300    }
1301    if pkt.nonce() != 0 {
1302        flags |= FLAG_PACKET_NONCE;
1303    }
1304    if endpoint_list_present {
1305        flags |= FLAG_ENDPOINT_LIST_PRESENT;
1306    } else if endpoint_bitmap_present {
1307        flags |= FLAG_ENDPOINT_BITMAP_PRESENT;
1308    }
1309    if reliable.is_some() {
1310        flags |= FLAG_RELIABLE_HEADER_PRESENT;
1311    }
1312    if reliable_is_compact {
1313        flags |= FLAG_COMPACT_RELIABLE_HEADER;
1314    }
1315    #[cfg(feature = "cryptography")]
1316    if e2e.is_some() {
1317        flags |= FLAG_E2E_ENCRYPTED_PAYLOAD;
1318    }
1319    out.push(flags);
1320
1321    assert!(
1322        nep_unique <= u8::MAX as usize,
1323        "too many endpoints selected to fit in NEP u8"
1324    );
1325    out.push(nep_unique as u8);
1326
1327    // NOTE: data_size is the *logical* (uncompressed) payload size.
1328    write_uleb128(pkt.data_type().as_u32() as u64, &mut out);
1329    write_uleb128(pkt.data_size() as u64, &mut out);
1330    write_uleb128(pkt.timestamp(), &mut out);
1331    if pkt.nonce() != 0 {
1332        write_uleb128(pkt.nonce() as u64, &mut out);
1333    }
1334
1335    write_uleb128(source_address as u64, &mut out);
1336
1337    if endpoint_list_present {
1338        for endpoint in unique_sorted_endpoints(pkt.endpoints()) {
1339            write_uleb128(u64::from(endpoint.0), &mut out);
1340        }
1341    } else if endpoint_bitmap_present {
1342        out.extend_from_slice(&build_endpoint_bitmap(pkt.endpoints()));
1343    }
1344    if (flags & FLAG_WIRE_CONTRACT) != 0 {
1345        // The contract must appear before the reliable header because it may
1346        // carry the "reliable header present" bit for packets whose current
1347        // runtime schema lookup no longer reflects the original wire semantics.
1348        write_uleb128(contract_bytes as u64, &mut out);
1349        write_wire_contract(shape, target_senders, reliable.is_some(), &mut out)?;
1350    }
1351    if let Some(hdr) = reliable {
1352        write_reliable_header_encoded(hdr, reliable_is_compact, &mut out);
1353    }
1354    #[cfg(feature = "cryptography")]
1355    if let Some(e2e) = e2e {
1356        write_encrypted_payload(pkt, e2e.key_id, &payload_wire, &mut out)?;
1357        return consume(&out, &[]);
1358    }
1359    #[cfg(debug_assertions)]
1360    debug_assert_eq!(
1361        out.capacity(),
1362        initial_capacity,
1363        "packet header grew its buffer"
1364    );
1365    consume(&out, &payload_wire)
1366}
1367
1368// ===========================================================================
1369// Unpacking (full packet)
1370// ===========================================================================
1371
1372/// Unpack a full [`Packet`] from a packed v2 wire frame.
1373///
1374/// This validates the frame CRC, resolves the endpoint set, decodes the
1375/// optional migration-safe wire contract, parses the optional reliable header,
1376/// and reconstructs the logical `Packet`. When the frame carries an inline wire
1377/// shape, the returned packet preserves that shape so payload validation and
1378/// formatting remain stable even if the local runtime schema has changed since
1379/// the packet was originally packed.
1380///
1381/// # Parameters
1382/// - `buf`: Complete packed frame bytes, including the CRC32 trailer.
1383///
1384/// # Returns
1385/// - `Ok(Packet)` when the frame is well-formed and contains a payload-bearing
1386///   telemetry packet.
1387///
1388/// # Errors
1389/// - [`TelemetryError::Unpack`] if the frame is malformed, truncated, or
1390///   fails CRC validation.
1391/// - [`TelemetryError::InvalidType`] if the type ID is not valid and the frame
1392///   does not carry enough inline shape information to keep decoding it.
1393pub fn unpack_packet(buf: &[u8]) -> Result<Packet, TelemetryError> {
1394    unpack_packet_with_role(buf, false)
1395}
1396
1397pub(crate) fn unpack_packet_with_role(buf: &[u8], managed: bool) -> Result<Packet, TelemetryError> {
1398    let data = verify_crc32(buf)?;
1399    if data.is_empty() {
1400        return Err(TelemetryError::Unpack("short prelude"));
1401    }
1402    let mut r = ByteReader::new(data);
1403
1404    let flags = r.read_bytes(1)?[0];
1405    let payload_is_compressed = (flags & FLAG_COMPRESSED_PAYLOAD) != 0;
1406    let endpoint_bitmap_present = (flags & FLAG_ENDPOINT_BITMAP_PRESENT) != 0;
1407    let compact_reliable_header = (flags & FLAG_COMPACT_RELIABLE_HEADER) != 0;
1408    #[cfg(feature = "cryptography")]
1409    let payload_is_encrypted = (flags & FLAG_E2E_ENCRYPTED_PAYLOAD) != 0;
1410    #[cfg(not(feature = "cryptography"))]
1411    if (flags & 0x10) != 0 {
1412        return Err(TelemetryError::Unpack("e2e crypto unsupported"));
1413    }
1414
1415    let nep = r.read_bytes(1)?[0] as usize;
1416
1417    let ty_v = read_uleb128(&mut r)?;
1418    let dsz = read_uleb128(&mut r)? as usize; // logical (uncompressed) payload size
1419    let ts_v = read_uleb128(&mut r)?;
1420    let nonce = if (flags & FLAG_PACKET_NONCE) != 0 {
1421        u16::try_from(read_uleb128(&mut r)?)
1422            .map_err(|_| TelemetryError::Unpack("packet nonce too large"))?
1423    } else {
1424        0
1425    };
1426    let source_address = u32::try_from(read_uleb128(&mut r)?)
1427        .map_err(|_| TelemetryError::Unpack("source address too large"))?;
1428    let sender_str = sender_name_for_address(source_address);
1429    let ty_u32 = data_type_id_from_wire(ty_v)?;
1430    let known_ty = DataType::try_from_u32(ty_u32);
1431    let endpoint_bytes = if endpoint_bitmap_present {
1432        EP_BITMAP_BYTES
1433    } else {
1434        0
1435    };
1436
1437    // For uncompressed payload: bitmap + [contract] + [reliable] + payload(dsz)
1438    // For compressed payload: bitmap + [contract] + [reliable] + at least 1 byte.
1439    if !payload_is_compressed {
1440        if r.remaining() < endpoint_bytes + dsz {
1441            return Err(TelemetryError::Unpack("short buffer"));
1442        }
1443    } else if r.remaining() < endpoint_bytes + 1 {
1444        return Err(TelemetryError::Unpack("short buffer"));
1445    }
1446
1447    let eps = endpoints_from_wire_or_schema(
1448        &mut r,
1449        endpoint_bitmap_present,
1450        flags & FLAG_ENDPOINT_LIST_PRESENT != 0,
1451        known_ty,
1452        nep,
1453    )?;
1454
1455    let contract = decode_wire_contract(&mut r, (flags & FLAG_WIRE_CONTRACT) != 0)?;
1456    let ty = known_ty
1457        .or_else(|| contract.shape.map(|_| DataType(ty_u32)))
1458        .ok_or(TelemetryError::InvalidType)?;
1459
1460    // ----- Reliable header (optional) -----
1461    let mut reliable_hdr: Option<ReliableHeader> = None;
1462    if flags & FLAG_RELIABLE_HEADER_PRESENT != 0 || contract.has_reliable_header {
1463        let hdr = read_reliable_header_encoded(&mut r, compact_reliable_header)?;
1464        if (hdr.flags & RELIABLE_FLAG_ACK_ONLY) != 0 {
1465            return Err(TelemetryError::Unpack("reliable control frame"));
1466        }
1467        reliable_hdr = Some(hdr);
1468    }
1469
1470    // Reassembled side frames bypass the small-fragment ingress estimate.
1471    // Check again immediately before allocating their logical payload, including
1472    // decompression/decryption scratch when enabled.
1473    let payload_copies = if payload_is_compressed { 3 } else { 1 };
1474    #[cfg(feature = "cryptography")]
1475    let payload_copies = payload_copies + usize::from(payload_is_encrypted);
1476    let small_ack = matches!(ty, DataType::ReliableAck | DataType::ReliablePartialAck)
1477        && dsz <= 32
1478        && payload_copies == 1;
1479    if crate::memory_admission::supports_work_kinds()
1480        && !small_ack
1481        && buf.len() <= 128
1482        && flags & 0x11 == 0
1483        && (crate::memory_admission::protected_discovery_request_shape(
1484            ty,
1485            dsz == 0 && r.remaining() == 0,
1486            eps.as_ref() == [DataEndpoint::Discovery],
1487            contract
1488                .target_senders
1489                .as_ref()
1490                .map_or(0, |owners| owners.len()),
1491            contract.shape,
1492        ) || crate::memory_admission::protected_control_shape(
1493            ty,
1494            nep,
1495            contract
1496                .target_senders
1497                .as_ref()
1498                .map_or(0, |owners| owners.len()),
1499            contract.shape,
1500            reliable_hdr.is_some(),
1501            managed,
1502        ))
1503    {
1504        crate::memory_admission::check_small_protected_decode(buf.len())?;
1505    } else {
1506        crate::memory_admission::check(
1507            if small_ack {
1508                512
1509            } else {
1510                dsz.saturating_mul(payload_copies).saturating_add(512)
1511            },
1512            dsz.saturating_add(32),
1513        )?;
1514    }
1515    // ----- Payload handling -----
1516    let payload: StandardSmallPayload = {
1517        #[cfg(feature = "cryptography")]
1518        let payload_wire_owned;
1519        #[cfg(feature = "cryptography")]
1520        let payload_wire: &[u8] = if payload_is_encrypted {
1521            let aad_end = r.off;
1522            payload_wire_owned = read_encrypted_payload(&mut r, &data[..aad_end], dsz)?;
1523            &payload_wire_owned
1524        } else if !payload_is_compressed {
1525            r.read_bytes(dsz)?
1526        } else {
1527            let comp_len = r.remaining();
1528            r.read_bytes(comp_len)?
1529        };
1530
1531        #[cfg(not(feature = "cryptography"))]
1532        let payload_wire: &[u8] = if !payload_is_compressed {
1533            r.read_bytes(dsz)?
1534        } else {
1535            let comp_len = r.remaining();
1536            r.read_bytes(comp_len)?
1537        };
1538
1539        if payload_is_compressed {
1540            let decompressed = payload_compression::decompress(payload_wire, dsz)?;
1541            crate::memory_admission::check(dsz.saturating_add(512), dsz.saturating_add(32))?;
1542            StandardSmallPayload::new(&decompressed)
1543        } else {
1544            if payload_wire.len() != dsz {
1545                return Err(TelemetryError::Unpack("payload length mismatch"));
1546            }
1547            StandardSmallPayload::new(payload_wire)
1548        }
1549    };
1550
1551    // `Packet` preserves logical payload data plus wire-contract metadata, but
1552    // not hop-level reliable transport state. The router/relay reliable layer
1553    // consumes that header before handing the logical packet onward.
1554    let _ = reliable_hdr;
1555    Packet::new_with_owned_wire_contract(
1556        ty,
1557        eps,
1558        sender_str,
1559        ts_v,
1560        nonce,
1561        payload,
1562        contract.shape,
1563        contract.target_senders,
1564    )
1565}
1566
1567// ===========================================================================
1568// Peek / envelope-only decode
1569// ===========================================================================
1570
1571/// Decode only the routing-relevant envelope of a packed packet.
1572///
1573/// This reads enough of the frame to expose type, endpoints, sender, timestamp,
1574/// and any migration-safe contract metadata without touching payload bytes.
1575/// That makes it the fast path for routing and other header-only inspection.
1576///
1577/// # Parameters
1578/// - `buf`: Complete packed frame bytes, including the CRC32 trailer.
1579///
1580/// # Returns
1581/// - `Ok(TelemetryEnvelope)` containing the logical packet envelope plus any
1582///   inline wire-shape and target-sender metadata.
1583///
1584/// # Errors
1585/// - [`TelemetryError::Unpack`] if the frame is malformed or fails CRC.
1586/// - [`TelemetryError::InvalidType`] if the type ID cannot be resolved.
1587pub fn peek_envelope(buf: &[u8]) -> TelemetryResult<TelemetryEnvelope> {
1588    Ok(peek_routing_envelope(buf)?.into_owned())
1589}
1590
1591/// Validated routing metadata without owned strings or heap-backed arrays.
1592/// The fixed bitmap is normalized against the current schema on every parse;
1593/// nothing is cached across schema updates or packet lifetimes.
1594#[derive(Clone, Debug)]
1595pub(crate) struct RoutingEnvelope<'a> {
1596    pub ty: DataType,
1597    bitmap: [u64; EP_BITMAP_BYTES / 8],
1598    endpoint_list: Option<&'a [u8]>,
1599    #[cfg(feature = "std")]
1600    schema_endpoints: Option<Arc<[DataEndpoint]>>,
1601    #[cfg(not(feature = "std"))]
1602    schema_endpoints: Option<&'static [DataEndpoint]>,
1603    pub source_address: u32,
1604    pub timestamp_ms: u64,
1605    pub wire_shape: Option<MessageElement>,
1606    pub is_reliable: bool,
1607    payload_is_empty: bool,
1608    targets: &'a [u8],
1609    has_reliable_header: bool,
1610}
1611#[derive(Clone)]
1612pub(crate) struct RoutingEndpointIter<'a> {
1613    envelope: &'a RoutingEnvelope<'a>,
1614    word: usize,
1615    bits: u64,
1616    list_offset: usize,
1617    previous: Option<DataEndpoint>,
1618    finished: bool,
1619}
1620impl Iterator for RoutingEndpointIter<'_> {
1621    type Item = DataEndpoint;
1622    #[inline(never)]
1623    fn next(&mut self) -> Option<DataEndpoint> {
1624        if self.finished {
1625            return None;
1626        }
1627        if let Some(bytes) = self.envelope.endpoint_list {
1628            let mut reader = ByteReader {
1629                buf: bytes,
1630                off: self.list_offset,
1631            };
1632            if reader.remaining() == 0 {
1633                self.finished = true;
1634                return None;
1635            }
1636            let endpoint = read_uleb128(&mut reader)
1637                .ok()
1638                .and_then(|id| u32::try_from(id).ok())
1639                .map(DataEndpoint);
1640            self.list_offset = reader.off;
1641            return endpoint;
1642        }
1643        if let Some(endpoints) = &self.envelope.schema_endpoints {
1644            let next = endpoints
1645                .iter()
1646                .copied()
1647                .filter(|ep| self.previous.is_none_or(|previous| *ep > previous))
1648                .min();
1649            self.previous = next;
1650            self.finished = next.is_none();
1651            return next;
1652        }
1653        while self.bits == 0 {
1654            self.word += 1;
1655            if self.word == self.envelope.bitmap.len() {
1656                self.finished = true;
1657                return None;
1658            }
1659            self.bits = self.envelope.bitmap[self.word];
1660        }
1661        let bit = self.bits.trailing_zeros();
1662        self.bits &= self.bits - 1;
1663        Some(DataEndpoint((self.word * 64 + bit as usize) as u32))
1664    }
1665}
1666
1667impl RoutingEnvelope<'_> {
1668    pub fn endpoints(&self) -> RoutingEndpointIter<'_> {
1669        RoutingEndpointIter {
1670            envelope: self,
1671            word: 0,
1672            bits: self.bitmap[0],
1673            list_offset: 0,
1674            previous: None,
1675            finished: false,
1676        }
1677    }
1678    pub fn target_senders(&self) -> impl ExactSizeIterator<Item = u64> + '_ {
1679        self.targets
1680            .as_chunks::<8>()
1681            .0
1682            .iter()
1683            .map(|bytes| u64::from_le_bytes(*bytes))
1684    }
1685    fn into_owned(self) -> TelemetryEnvelope {
1686        TelemetryEnvelope {
1687            ty: self.ty,
1688            endpoints: self.endpoints().collect::<Arc<[_]>>(),
1689            sender: sender_name_for_address(self.source_address),
1690            source_address: self.source_address,
1691            timestamp_ms: self.timestamp_ms,
1692            wire_shape: self.wire_shape,
1693            target_senders: self.target_senders().collect::<Arc<[_]>>(),
1694        }
1695    }
1696}
1697
1698fn read_routing_envelope<'a>(r: &mut ByteReader<'a>) -> TelemetryResult<RoutingEnvelope<'a>> {
1699    let flags = r.read_bytes(1)?[0];
1700    let nep = r.read_bytes(1)?[0] as usize;
1701    let ty_v = read_uleb128(r)?;
1702    let payload_is_empty = read_uleb128(r)? == 0;
1703    let timestamp_ms = read_uleb128(r)?;
1704    if flags & FLAG_PACKET_NONCE != 0 {
1705        let _ = read_uleb128(r)?;
1706    }
1707    let source_address = u32::try_from(read_uleb128(r)?)
1708        .map_err(|_| TelemetryError::Unpack("source address too large"))?;
1709    let ty_u32 = data_type_id_from_wire(ty_v)?;
1710    let is_reliable = flags & FLAG_RELIABLE_HEADER_PRESENT != 0;
1711    // Word-sized bitmaps avoid repeatedly scanning 32 individual bytes for
1712    // the usual one-destination telemetry frame on size-optimized MCUs.
1713    let mut bitmap = [0u64; EP_BITMAP_BYTES / 8];
1714    let mut endpoint_list = None;
1715    let mut schema_endpoints = None;
1716    if flags & FLAG_ENDPOINT_LIST_PRESENT != 0 {
1717        if flags & FLAG_ENDPOINT_BITMAP_PRESENT != 0 {
1718            return Err(TelemetryError::Unpack("ambiguous endpoint encoding"));
1719        }
1720        let bytes = read_endpoint_list(r, nep)?;
1721        endpoint_list = Some(bytes);
1722    } else if flags & FLAG_ENDPOINT_BITMAP_PRESENT != 0 {
1723        let bytes = r.read_bytes(EP_BITMAP_BYTES)?;
1724        for (word, chunk) in bitmap.iter_mut().zip(bytes.as_chunks::<8>().0) {
1725            *word = u64::from_le_bytes(*chunk);
1726        }
1727        if bitmap
1728            .iter()
1729            .map(|word| word.count_ones() as usize)
1730            .sum::<usize>()
1731            != nep
1732        {
1733            return Err(TelemetryError::Unpack("endpoint count mismatch"));
1734        }
1735        for (index, &word) in bitmap.iter().enumerate() {
1736            let mut bits = word;
1737            while bits != 0 {
1738                let id = index * 64 + bits.trailing_zeros() as usize;
1739                if !crate::config::endpoint_exists(DataEndpoint(id as u32)) {
1740                    return Err(TelemetryError::Unpack("bad endpoint bit set"));
1741                }
1742                bits &= bits - 1;
1743            }
1744        }
1745    } else {
1746        let meta = crate::config::try_get_message_meta(DataType(ty_u32))
1747            .ok_or(TelemetryError::InvalidType)?;
1748        if unique_sorted_endpoints(meta.endpoints_ref()).count() != nep {
1749            return Err(TelemetryError::Unpack("endpoint count mismatch"));
1750        }
1751        schema_endpoints = Some(meta.endpoints);
1752    }
1753    let mut shape = None;
1754    let mut targets: &'a [u8] = &[];
1755    let mut has_reliable_header = false;
1756    if flags & FLAG_WIRE_CONTRACT != 0 {
1757        let len = usize::try_from(read_uleb128(r)?)
1758            .map_err(|_| TelemetryError::Unpack("wire contract length"))?;
1759        let mut cr = ByteReader::new(r.read_bytes(len)?);
1760        let contract_flags = cr.read_bytes(1)?[0];
1761        if contract_flags & CONTRACT_FLAG_SHAPE != 0 {
1762            shape = Some(decode_wire_shape(&mut cr)?);
1763        }
1764        if contract_flags & CONTRACT_FLAG_TARGETS != 0 {
1765            let count = usize::try_from(read_uleb128(&mut cr)?)
1766                .map_err(|_| TelemetryError::Unpack("wire contract target count"))?;
1767            let bytes = count
1768                .checked_mul(8)
1769                .ok_or(TelemetryError::Unpack("wire contract target count"))?;
1770            targets = cr.read_bytes(bytes)?;
1771        }
1772        if cr.remaining() != 0 {
1773            return Err(TelemetryError::Unpack("wire contract trailing bytes"));
1774        }
1775        has_reliable_header = contract_flags & CONTRACT_FLAG_RELIABLE_HEADER != 0;
1776    }
1777    if nep == 0 {
1778        if !is_reliable && !has_reliable_header {
1779            return Err(TelemetryError::EmptyEndpoints);
1780        }
1781        let mut header = ByteReader {
1782            buf: r.buf,
1783            off: r.off,
1784        };
1785        let reliable =
1786            read_reliable_header_encoded(&mut header, flags & FLAG_COMPACT_RELIABLE_HEADER != 0)?;
1787        if reliable.flags & RELIABLE_FLAG_ACK_ONLY == 0 {
1788            return Err(TelemetryError::EmptyEndpoints);
1789        }
1790    }
1791    let ty = DataType(ty_u32);
1792    Ok(RoutingEnvelope {
1793        ty,
1794        bitmap,
1795        endpoint_list,
1796        schema_endpoints,
1797        source_address,
1798        timestamp_ms,
1799        wire_shape: shape,
1800        is_reliable,
1801        payload_is_empty,
1802        targets,
1803        has_reliable_header,
1804    })
1805}
1806
1807/// Read only the bounded type prefix for queue scheduling. This is deliberately
1808/// not frame validation: dispatch must still verify CRC and the full envelope.
1809/// Scanning an RX queue must not checksum every retained payload on each pass.
1810pub(crate) fn scheduling_type(buf: &[u8]) -> Option<DataType> {
1811    let mut reader = ByteReader::new(buf.get(..buf.len().min(12))?);
1812    reader.read_bytes(2).ok()?;
1813    let id = data_type_id_from_wire(read_uleb128(&mut reader).ok()?).ok()?;
1814    // Scheduling reads identity, not registry membership. Dynamic/frozen wire
1815    // contracts may precede their schema, and authoritative parsing follows.
1816    // Avoid a redundant schema lookup for every retained queue item.
1817    Some(DataType(id))
1818}
1819
1820/// Identify only a schema sender in a retained, CRC-validated SDT first chunk.
1821/// This borrowed prefix inspection neither validates a complete canonical CRC
1822/// nor admits a packet, route, or schema. Incomplete headers give no credit.
1823#[cfg(feature = "discovery")]
1824pub(crate) fn chunked_schema_source(prefix: &[u8]) -> Option<u32> {
1825    let envelope = read_routing_envelope(&mut ByteReader::new(prefix)).ok()?;
1826    (envelope.ty == DataType::DiscoverySchema).then_some(envelope.source_address)
1827}
1828
1829pub(crate) fn peek_routing_envelope(buf: &[u8]) -> TelemetryResult<RoutingEnvelope<'_>> {
1830    let data = verify_crc32(buf)?;
1831    read_routing_envelope(&mut ByteReader::new(data))
1832}
1833
1834/// Decode the header/envelope and optional reliable header without touching the payload.
1835pub struct TelemetryFrameInfo {
1836    pub envelope: TelemetryEnvelope,
1837    pub reliable: Option<ReliableHeader>,
1838}
1839
1840impl TelemetryFrameInfo {
1841    #[inline]
1842    /// Returns `true` when the frame carries only a reliable-delivery acknowledgment.
1843    pub fn ack_only(&self) -> bool {
1844        self.reliable
1845            .map(|h| (h.flags & RELIABLE_FLAG_ACK_ONLY) != 0)
1846            .unwrap_or(false)
1847    }
1848}
1849
1850/// Borrow frozen target hashes from a Packet or its little-endian wire header.
1851/// Membership never allocates an owner array or changes the frozen cohort.
1852#[cfg(feature = "discovery")]
1853#[derive(Clone, Copy)]
1854pub(crate) enum TargetSenderSet<'a> {
1855    Hashes(&'a [u64]),
1856    Encoded(&'a [u8]),
1857}
1858#[cfg(feature = "discovery")]
1859impl<'a> TargetSenderSet<'a> {
1860    pub(crate) fn skip(self, count: usize) -> Self {
1861        match self {
1862            Self::Hashes(v) => Self::Hashes(v.get(count..).unwrap_or(&[])),
1863            Self::Encoded(v) => Self::Encoded(v.get(count.saturating_mul(8)..).unwrap_or(&[])),
1864        }
1865    }
1866    pub(crate) fn iter(self) -> impl Iterator<Item = u64> + 'a {
1867        let (hashes, encoded) = match self {
1868            Self::Hashes(v) => (v, &[][..]),
1869            Self::Encoded(v) => (&[][..], v),
1870        };
1871        hashes.iter().copied().chain(
1872            encoded
1873                .as_chunks::<8>()
1874                .0
1875                .iter()
1876                .map(|v| u64::from_le_bytes(*v)),
1877        )
1878    }
1879    pub(crate) fn is_empty(&self) -> bool {
1880        match self {
1881            Self::Hashes(v) => v.is_empty(),
1882            Self::Encoded(v) => v.is_empty(),
1883        }
1884    }
1885    pub(crate) fn contains(&self, target: u64) -> bool {
1886        match self {
1887            Self::Hashes(v) => v.contains(&target),
1888            Self::Encoded(v) => v
1889                .as_chunks::<8>()
1890                .0
1891                .iter()
1892                .any(|bytes| u64::from_le_bytes(*bytes) == target),
1893        }
1894    }
1895}
1896#[cfg(feature = "discovery")]
1897impl<'a> RoutingEnvelope<'a> {
1898    pub(crate) fn target_sender_set(&self) -> TargetSenderSet<'a> {
1899        TargetSenderSet::Encoded(self.targets)
1900    }
1901}
1902
1903pub(crate) struct RoutingFrameInfo<'a> {
1904    pub envelope: RoutingEnvelope<'a>,
1905    pub reliable: Option<ReliableHeader>,
1906    pub empty_payload: bool,
1907}
1908impl RoutingFrameInfo<'_> {
1909    pub fn ack_only(&self) -> bool {
1910        self.reliable
1911            .is_some_and(|header| header.flags & RELIABLE_FLAG_ACK_ONLY != 0)
1912    }
1913}
1914/// Borrow an uncompressed discovery payload without owning its envelope or bytes.
1915/// Sequenced, piggyback-ACK, compressed, targeted and encrypted frames retain normal processing.
1916#[cfg(feature = "discovery")]
1917pub(crate) fn peek_plain_discovery_payload(
1918    buf: &[u8],
1919    expected: DataType,
1920) -> TelemetryResult<Option<(u32, &[u8])>> {
1921    // Avoid a second CRC pass for ordinary telemetry. Its existing receive
1922    // and decode path remains responsible for complete frame validation.
1923    let mut prelude = ByteReader::new(buf);
1924    prelude.read_bytes(2)?;
1925    if data_type_id_from_wire(read_uleb128(&mut prelude)?)? != expected.as_u32() {
1926        return Ok(None);
1927    }
1928    let data = verify_crc32(buf)?;
1929    let mut reader = ByteReader::new(data);
1930    let envelope = read_routing_envelope(&mut reader)?;
1931    if envelope.ty != expected
1932        || data[0] & (FLAG_COMPRESSED_PAYLOAD | 0x10) != 0
1933        || envelope.target_senders().len() != 0
1934        || envelope
1935            .wire_shape
1936            .is_some_and(|shape| shape != crate::message_meta(expected).element)
1937    {
1938        return Ok(None);
1939    }
1940    if envelope.is_reliable || envelope.has_reliable_header {
1941        let reliable =
1942            read_reliable_header_encoded(&mut reader, data[0] & FLAG_COMPACT_RELIABLE_HEADER != 0)?;
1943        // Only the normal unsequenced broadcast snapshot is a no-op. A
1944        // sequenced snapshot or piggyback ACK still needs reliable processing.
1945        if reliable.flags != RELIABLE_FLAG_UNSEQUENCED || reliable.seq != 0 || reliable.ack != 0 {
1946            return Ok(None);
1947        }
1948    }
1949    // The routing parser already validated endpoints and any inline contract.
1950    // Recover the logical size from the short prelude without owning arrays.
1951    let mut prelude = ByteReader::new(data);
1952    prelude.read_bytes(2)?;
1953    read_uleb128(&mut prelude)?;
1954    let size = usize::try_from(read_uleb128(&mut prelude)?)
1955        .map_err(|_| TelemetryError::Unpack("discovery payload too large"))?;
1956    if reader.remaining() != size {
1957        return Err(TelemetryError::Unpack("discovery payload length mismatch"));
1958    }
1959    Ok(Some((envelope.source_address, reader.read_bytes(size)?)))
1960}
1961
1962/// Fully validate a replaceable plaintext sample without owning its payload.
1963/// Contract/codec/reliable frames never enter freshness coalescing.
1964#[cfg(feature = "latest-value-egress")]
1965pub(crate) fn peek_latest_plain_frame_info(buf: &[u8]) -> TelemetryResult<RoutingFrameInfo<'_>> {
1966    if buf.first().is_none_or(|flags| flags & 0x17 != 0) {
1967        return Err(TelemetryError::BadArg);
1968    }
1969    let data = verify_crc32(buf)?;
1970    let mut reader = ByteReader::new(data);
1971    let envelope = read_routing_envelope(&mut reader)?;
1972    if envelope.has_reliable_header || envelope.endpoints().next().is_none() {
1973        return Err(TelemetryError::BadArg);
1974    }
1975    let mut prelude = ByteReader::new(data);
1976    prelude.read_bytes(2)?;
1977    read_uleb128(&mut prelude)?;
1978    let size = usize::try_from(read_uleb128(&mut prelude)?).map_err(|_| TelemetryError::BadArg)?;
1979    if reader.remaining() != size {
1980        return Err(TelemetryError::SizeMismatch {
1981            expected: size,
1982            got: reader.remaining(),
1983        });
1984    }
1985    let meta =
1986        crate::config::try_get_message_meta(envelope.ty).ok_or(TelemetryError::InvalidType)?;
1987    Packet::validate_payload_against_element(meta.element, reader.read_bytes(size)?)?;
1988    Ok(RoutingFrameInfo {
1989        envelope,
1990        reliable: None,
1991        empty_payload: size == 0,
1992    })
1993}
1994
1995pub(crate) fn peek_routing_frame_info(buf: &[u8]) -> TelemetryResult<RoutingFrameInfo<'_>> {
1996    let data = verify_crc32(buf)?;
1997    let mut reader = ByteReader::new(data);
1998    let envelope = read_routing_envelope(&mut reader)?;
1999    let reliable = if envelope.is_reliable || envelope.has_reliable_header {
2000        Some(read_reliable_header_encoded(
2001            &mut reader,
2002            data[0] & FLAG_COMPACT_RELIABLE_HEADER != 0,
2003        )?)
2004    } else {
2005        None
2006    };
2007    let empty_payload = envelope.payload_is_empty && reader.remaining() == 0;
2008    Ok(RoutingFrameInfo {
2009        envelope,
2010        reliable,
2011        empty_payload,
2012    })
2013}
2014
2015fn peek_frame_info_inner(buf: &[u8]) -> TelemetryResult<TelemetryFrameInfo> {
2016    let mut r = ByteReader::new(buf);
2017    let header = read_routing_envelope(&mut r)?;
2018    let reliable = if header.is_reliable || header.has_reliable_header {
2019        Some(read_reliable_header_encoded(
2020            &mut r,
2021            buf[0] & FLAG_COMPACT_RELIABLE_HEADER != 0,
2022        )?)
2023    } else {
2024        None
2025    };
2026    Ok(TelemetryFrameInfo {
2027        envelope: header.into_owned(),
2028        reliable,
2029    })
2030}
2031
2032/// Peek the envelope plus reliable header (if present) without decoding payload bytes.
2033///
2034/// This is the primary router/relay fast path for reliable-layer decisions on
2035/// packed traffic. It still validates the frame CRC before exposing any
2036/// information.
2037pub fn peek_frame_info(buf: &[u8]) -> TelemetryResult<TelemetryFrameInfo> {
2038    let data = verify_crc32(buf)?;
2039    peek_frame_info_inner(data)
2040}
2041
2042/// Peek the envelope plus reliable header (if present) without validating CRC32.
2043///
2044/// This is intended only for internal call sites that have already validated
2045/// frame integrity or intentionally want best-effort inspection of partially
2046/// trusted bytes.
2047pub fn peek_frame_info_unchecked(buf: &[u8]) -> TelemetryResult<TelemetryFrameInfo> {
2048    let (data, _crc) = split_crc32(buf)?;
2049    peek_frame_info_inner(data)
2050}
2051
2052/// Locate the reliable-header byte offset within a packed frame.
2053///
2054/// The offset is computed after walking the source address and optional wire
2055/// contract. This matters because the contract can explicitly state that a
2056/// reliable header is present even if the current runtime schema no longer
2057/// marks the type as reliable.
2058///
2059/// # Parameters
2060/// - `buf`: Packed frame bytes, including the CRC32 trailer.
2061///
2062/// # Returns
2063/// - `Ok(Some(offset))` when a reliable header is present.
2064/// - `Ok(None)` when the frame carries no reliable header.
2065/// - `Err(TelemetryError)` when the frame is malformed.
2066pub fn reliable_header_offset(buf: &[u8]) -> TelemetryResult<Option<usize>> {
2067    Ok(reliable_header_span(buf)?.map(|(off, _, _)| off))
2068}
2069
2070pub(crate) fn reliable_header_span(
2071    buf: &[u8],
2072) -> TelemetryResult<Option<(usize, usize, ReliableHeader)>> {
2073    if buf.len() < CRC32_BYTES + 1 {
2074        return Err(TelemetryError::Unpack("short prelude"));
2075    }
2076    let data_len = buf.len().saturating_sub(CRC32_BYTES);
2077    let mut r = ByteReader::new(&buf[..data_len]);
2078
2079    let flags = r.read_bytes(1)?[0];
2080    let endpoint_bitmap_present = (flags & FLAG_ENDPOINT_BITMAP_PRESENT) != 0;
2081    let compact_reliable_header = (flags & FLAG_COMPACT_RELIABLE_HEADER) != 0;
2082
2083    let nep = r.read_bytes(1)?[0] as usize;
2084
2085    let ty_v = read_uleb128(&mut r)?;
2086    let _dsz = read_uleb128(&mut r)? as usize;
2087    let _ts_v = read_uleb128(&mut r)?;
2088    if (flags & FLAG_PACKET_NONCE) != 0 {
2089        let _ = read_uleb128(&mut r)?;
2090    }
2091    let _source_address = u32::try_from(read_uleb128(&mut r)?)
2092        .map_err(|_| TelemetryError::Unpack("source address too large"))?;
2093    let endpoint_bytes = if endpoint_bitmap_present {
2094        EP_BITMAP_BYTES
2095    } else {
2096        0
2097    };
2098
2099    if r.remaining() < endpoint_bytes {
2100        return Err(TelemetryError::Unpack("short buffer"));
2101    }
2102
2103    if flags & FLAG_ENDPOINT_LIST_PRESENT != 0 {
2104        if endpoint_bitmap_present {
2105            return Err(TelemetryError::Unpack("ambiguous endpoint encoding"));
2106        }
2107        read_endpoint_list(&mut r, nep)?;
2108    } else if endpoint_bitmap_present {
2109        r.read_bytes(EP_BITMAP_BYTES)?;
2110    }
2111    let contract = decode_wire_contract(&mut r, (flags & FLAG_WIRE_CONTRACT) != 0)?;
2112    let ty_u32 = data_type_id_from_wire(ty_v)?;
2113    let _ty = DataType(ty_u32);
2114    if flags & FLAG_RELIABLE_HEADER_PRESENT == 0 && !contract.has_reliable_header {
2115        return Ok(None);
2116    }
2117
2118    let off = r.off;
2119    let hdr = read_reliable_header_encoded(&mut r, compact_reliable_header)?;
2120    Ok(Some((off, r.off - off, hdr)))
2121}
2122
2123/// Rewrite the reliable header in-place and refresh the frame CRC32.
2124///
2125/// This avoids reserializing the entire packet when reliable transport code
2126/// only needs to change sequence/ACK state in an already-built frame.
2127///
2128/// # Parameters
2129/// - `buf`: Mutable packed frame bytes.
2130/// - `flags`: Replacement reliable-header flags byte.
2131/// - `seq`: Replacement sequence number.
2132/// - `ack`: Replacement cumulative ACK value.
2133///
2134/// # Returns
2135/// - `Ok(true)` if the frame carried a reliable header and it was rewritten.
2136/// - `Ok(false)` if no reliable header is present.
2137/// - `Err(TelemetryError)` if the frame is malformed.
2138pub fn rewrite_reliable_header(
2139    buf: &mut [u8],
2140    flags: u8,
2141    seq: u32,
2142    ack: u32,
2143) -> TelemetryResult<bool> {
2144    let Some((off, old_len, _)) = reliable_header_span(buf)? else {
2145        return Ok(false);
2146    };
2147    let hdr = ReliableHeader { flags, seq, ack };
2148    let compact = should_compact_reliable_header(hdr);
2149    if reliable_wire_size(hdr, compact) != old_len {
2150        return Err(TelemetryError::Unpack(
2151            "reliable header rewrite changes wire size",
2152        ));
2153    }
2154    let data_len = buf.len().saturating_sub(CRC32_BYTES);
2155    if data_len.saturating_sub(off) < old_len {
2156        return Err(TelemetryError::Unpack("short buffer"));
2157    }
2158    if compact {
2159        buf[0] |= FLAG_COMPACT_RELIABLE_HEADER;
2160    } else {
2161        buf[0] &= !FLAG_COMPACT_RELIABLE_HEADER;
2162    }
2163    let mut encoded = Vec::with_capacity(old_len);
2164    write_reliable_header_encoded(hdr, compact, &mut encoded);
2165    buf[off..off + old_len].copy_from_slice(&encoded);
2166    if buf.len() < CRC32_BYTES {
2167        return Err(TelemetryError::Unpack("short buffer"));
2168    }
2169    let crc = crc32_bytes(&buf[..data_len]);
2170    buf[data_len..data_len + CRC32_BYTES].copy_from_slice(&crc.to_le_bytes());
2171    Ok(true)
2172}
2173
2174pub(crate) fn rewrite_reliable_header_owned(
2175    buf: &[u8],
2176    flags: u8,
2177    seq: u32,
2178    ack: u32,
2179) -> TelemetryResult<Option<Arc<[u8]>>> {
2180    let Some((off, old_len, _)) = reliable_header_span(buf)? else {
2181        return Ok(None);
2182    };
2183    let data_len = buf.len().saturating_sub(CRC32_BYTES);
2184    if data_len < off + old_len {
2185        return Err(TelemetryError::Unpack("short buffer"));
2186    }
2187    let hdr = ReliableHeader { flags, seq, ack };
2188    let compact = should_compact_reliable_header(hdr);
2189    let mut encoded = Vec::with_capacity(reliable_wire_size(hdr, compact));
2190    write_reliable_header_encoded(hdr, compact, &mut encoded);
2191
2192    let mut out = Vec::with_capacity(data_len - old_len + encoded.len() + CRC32_BYTES);
2193    out.extend_from_slice(&buf[..off]);
2194    if compact {
2195        out[0] |= FLAG_COMPACT_RELIABLE_HEADER;
2196    } else {
2197        out[0] &= !FLAG_COMPACT_RELIABLE_HEADER;
2198    }
2199    out.extend_from_slice(&encoded);
2200    out.extend_from_slice(&buf[off + old_len..data_len]);
2201    let crc = crc32_bytes(&out);
2202    out.extend_from_slice(&crc.to_le_bytes());
2203    Ok(Some(Arc::from(out)))
2204}
2205
2206// ===========================================================================
2207// Size helpers
2208// ===========================================================================
2209
2210/// Compute the encoded metadata-prefix size for `pkt`.
2211///
2212/// This includes the fixed prelude and top-level varints, but excludes the
2213/// optional endpoint bitmap, optional wire contract, optional reliable header,
2214/// payload bytes, and CRC32 trailer.
2215///
2216/// # Parameters
2217/// - `pkt`: Packet to size.
2218///
2219/// # Returns
2220/// - Byte count of the packed metadata prefix.
2221pub fn header_size_bytes(pkt: &Packet) -> usize {
2222    let prelude = 2; // FLAGS (u8) + NEP (u8)
2223
2224    let source_address = sender_address_u32(pkt.sender());
2225
2226    prelude
2227        + uleb128_size(pkt.data_type().as_u32() as u64)
2228        + uleb128_size(pkt.data_size() as u64)
2229        + uleb128_size(pkt.timestamp())
2230        + if pkt.nonce() != 0 {
2231            uleb128_size(pkt.nonce() as u64)
2232        } else {
2233            0
2234        }
2235        + uleb128_size(source_address as u64)
2236}
2237
2238/// Compute the full packed wire size for `pkt`.
2239///
2240/// This applies the same sender/payload compression heuristics as normal
2241/// packing, then sums the metadata prefix, optional endpoint bitmap and frozen
2242/// wire contract, payload bytes, and CRC32 trailer. It matches [`pack_packet`];
2243/// its default unsequenced reliable header is included.
2244///
2245/// # Parameters
2246/// - `pkt`: Packet to size.
2247///
2248/// # Returns
2249/// - Total encoded frame size in bytes.
2250pub fn packet_wire_size(pkt: &Packet) -> usize {
2251    let shape = pkt.wire_shape();
2252    let targets = pkt.wire_target_senders();
2253    let contract_bytes = wire_contract_size(shape, targets);
2254    let contract_len = if shape.is_some() || !targets.is_empty() {
2255        uleb128_size(contract_bytes as u64) + contract_bytes
2256    } else {
2257        0
2258    };
2259    let endpoint_len = if packet_endpoint_bitmap_present(pkt, shape, targets) {
2260        if explicit_endpoint_list(pkt.endpoints()) {
2261            endpoint_list_size(pkt.endpoints())
2262        } else {
2263            EP_BITMAP_BYTES
2264        }
2265    } else {
2266        0
2267    };
2268    let (_, payload_wire) = payload_for_wire(pkt);
2269    let reliable_len = if is_reliable_type(pkt.data_type()) {
2270        let header = ReliableHeader {
2271            flags: RELIABLE_FLAG_UNSEQUENCED,
2272            seq: 0,
2273            ack: 0,
2274        };
2275        reliable_wire_size(header, should_compact_reliable_header(header))
2276    } else {
2277        0
2278    };
2279    // Match pack_packet, including its default unsequenced reliable header.
2280    // No frame/header allocation, payload copy or CRC scan is needed for sizing.
2281    header_size_bytes(pkt)
2282        + endpoint_len
2283        + contract_len
2284        + reliable_len
2285        + payload_wire.len()
2286        + CRC32_BYTES
2287}
2288
2289#[inline]
2290/// Computes the same packet ID as [`Packet::packet_id`] directly from a packed wire frame.
2291pub fn packet_id_from_wire(buf: &[u8]) -> Result<u64, TelemetryError> {
2292    let data = verify_crc32(buf)?;
2293    if data.len() < 2 {
2294        return Err(TelemetryError::Unpack("short prelude"));
2295    }
2296
2297    let mut r = ByteReader::new(data);
2298
2299    let flags = r.read_bytes(1)?[0];
2300    let payload_is_compressed = (flags & FLAG_COMPRESSED_PAYLOAD) != 0;
2301    let compact_reliable_header = (flags & FLAG_COMPACT_RELIABLE_HEADER) != 0;
2302    #[cfg(feature = "cryptography")]
2303    let payload_is_encrypted = (flags & FLAG_E2E_ENCRYPTED_PAYLOAD) != 0;
2304    #[cfg(not(feature = "cryptography"))]
2305    if (flags & 0x10) != 0 {
2306        return Err(TelemetryError::Unpack("e2e crypto unsupported"));
2307    }
2308
2309    let _nep = r.read_bytes(1)?[0] as usize;
2310
2311    let _ = read_uleb128(&mut r)?;
2312    let dsz = read_uleb128(&mut r)? as usize; // logical payload size (uncompressed)
2313    let ts_v = read_uleb128(&mut r)?;
2314    let nonce = if (flags & FLAG_PACKET_NONCE) != 0 {
2315        u16::try_from(read_uleb128(&mut r)?)
2316            .map_err(|_| TelemetryError::Unpack("packet nonce too large"))?
2317    } else {
2318        0
2319    };
2320    let source_address = u32::try_from(read_uleb128(&mut r)?)
2321        .map_err(|_| TelemetryError::Unpack("source address too large"))?;
2322    // Reuse the validated borrowed envelope. Computing an ID must not build
2323    // owned endpoint/target arrays merely to hash or inspect their contents.
2324    let mut envelope_reader = ByteReader::new(data);
2325    let envelope = read_routing_envelope(&mut envelope_reader)?;
2326    r = envelope_reader;
2327    let ty = envelope.ty;
2328
2329    // ---- reliable header (optional) ----
2330    if envelope.is_reliable || envelope.has_reliable_header {
2331        let hdr = read_reliable_header_encoded(&mut r, compact_reliable_header)?;
2332        if (hdr.flags & RELIABLE_FLAG_ACK_ONLY) != 0 {
2333            return Err(TelemetryError::Unpack("reliable control frame"));
2334        }
2335    }
2336
2337    // ---- payload bytes (must hash *logical* payload bytes) ----
2338    #[cfg(feature = "cryptography")]
2339    let payload_wire_owned;
2340    #[cfg(feature = "cryptography")]
2341    let payload_wire: &[u8] = if payload_is_encrypted {
2342        let aad_end = r.off;
2343        payload_wire_owned =
2344            read_encrypted_payload(&mut r, data.get(..aad_end).unwrap_or(&[]), dsz)?;
2345        &payload_wire_owned
2346    } else if !payload_is_compressed {
2347        if r.remaining() < dsz {
2348            return Err(TelemetryError::Unpack("short buffer"));
2349        }
2350        r.read_bytes(dsz)?
2351    } else {
2352        let comp_len = r.remaining();
2353        if comp_len < 1 {
2354            return Err(TelemetryError::Unpack("short buffer"));
2355        }
2356        r.read_bytes(comp_len)?
2357    };
2358    #[cfg(not(feature = "cryptography"))]
2359    let payload_wire: &[u8] = if !payload_is_compressed {
2360        if r.remaining() < dsz {
2361            return Err(TelemetryError::Unpack("short buffer"));
2362        }
2363        r.read_bytes(dsz)?
2364    } else {
2365        let comp_len = r.remaining();
2366        if comp_len < 1 {
2367            return Err(TelemetryError::Unpack("short buffer"));
2368        }
2369        r.read_bytes(comp_len)?
2370    };
2371    let payload_decompressed;
2372    let payload_bytes: &[u8] = if payload_is_compressed {
2373        payload_decompressed = payload_compression::decompress(payload_wire, dsz)?;
2374        &payload_decompressed
2375    } else {
2376        if payload_wire.len() != dsz {
2377            return Err(TelemetryError::Unpack("payload length mismatch"));
2378        }
2379        payload_wire
2380    };
2381
2382    // ---- hash exactly like Packet::packet_id() ----
2383    let mut h: u64 = 0x9E37_79B9_7F4A_7C15;
2384
2385    // Compact source address.
2386    h = hash_bytes_u64(h, &source_address.to_le_bytes());
2387
2388    // Stable numeric identity is independent of when schema names are learned.
2389    h = hash_bytes_u64(h, &ty.0.to_le_bytes());
2390
2391    // Endpoints as string bytes, in ascending discriminant order.
2392    for ep in envelope.endpoints() {
2393        h = hash_bytes_u64(h, &ep.0.to_le_bytes());
2394    }
2395
2396    // Timestamp + data_size as bytes
2397    h = hash_bytes_u64(h, &ts_v.to_le_bytes());
2398    h = hash_bytes_u64(h, &nonce.to_le_bytes());
2399    h = hash_bytes_u64(h, &(dsz as u64).to_le_bytes());
2400
2401    // Payload bytes (logical payload)
2402    h = hash_bytes_u64(h, payload_bytes);
2403    Ok(h)
2404}
2405
2406mod payload_compression {
2407    use crate::TelemetryError;
2408    use alloc::borrow::Cow;
2409    #[cfg(feature = "compression")]
2410    use alloc::vec;
2411    use alloc::vec::Vec;
2412
2413    #[cfg(feature = "compression")]
2414    use crate::config::runtime_payload_compress_threshold;
2415    #[cfg(feature = "compression")]
2416    use zstd_safe::CompressionLevel;
2417
2418    /// Compress the given payload if it is beneficial to do so.
2419    /// # Arguments
2420    /// - `payload`: Original uncompressed payload bytes.
2421    /// # Returns
2422    /// - `(bool, Cow<[u8]>)`: Tuple where the first element indicates whether
2423    ///   compression was applied, and the second element is the resulting
2424    ///   payload bytes (compressed or original).
2425    #[cfg(feature = "compression")]
2426    pub fn compress_if_beneficial(payload: &'_ [u8]) -> (bool, Cow<'_, [u8]>) {
2427        if payload.len() < runtime_payload_compress_threshold() {
2428            return (false, Cow::Borrowed(payload));
2429        }
2430
2431        // Bound output and avoid growth beyond useful threshold.
2432        let Some(compressed) = compress_to_vec_bounded(payload, payload.len().saturating_sub(2))
2433        else {
2434            return (false, Cow::Borrowed(payload));
2435        };
2436
2437        // Only use compressed form if it actually saves space.
2438        if compressed.len() + 1 >= payload.len() {
2439            (false, Cow::Borrowed(payload))
2440        } else {
2441            (true, Cow::Owned(compressed))
2442        }
2443    }
2444
2445    #[cfg(feature = "compression")]
2446    fn compress_to_vec_bounded(input: &[u8], max_output: usize) -> Option<Vec<u8>> {
2447        if input.is_empty() || max_output == 0 {
2448            return None;
2449        }
2450
2451        let mut out = vec![0u8; max_output];
2452        // Use default-level behavior for better compression ratio on typical telemetry payloads.
2453        let level: CompressionLevel = 1;
2454        let written = zstd_safe::compress(&mut out[..], input, level).ok()?;
2455        out.truncate(written);
2456        Some(out)
2457    }
2458
2459    /// Decompress the given compressed payload.
2460    /// # Arguments
2461    /// - `compressed`: Compressed payload bytes.
2462    /// - `expected_len`: Expected length of the decompressed payload.
2463    /// # Returns
2464    /// - `Vec<u8>`: Decompressed payload bytes.
2465    /// # Errors
2466    /// - `TelemetryError::Unpack` if decompression fails or the size
2467    ///   does not match `expected_len`.
2468    #[cfg(feature = "compression")]
2469    pub fn decompress(compressed: &[u8], expected_len: usize) -> Result<Vec<u8>, TelemetryError> {
2470        let mut out = vec![0u8; expected_len];
2471        let written = zstd_safe::decompress(&mut out[..], compressed)
2472            .map_err(|_| TelemetryError::Unpack("decompression failed"))?;
2473        if written != expected_len {
2474            return Err(TelemetryError::Unpack("decompressed size mismatch"));
2475        }
2476        Ok(out)
2477    }
2478
2479    // Stub when compression is disabled (never actually produces compressed payloads).
2480    #[cfg(not(feature = "compression"))]
2481    /// Returns the original payload unchanged when compression support is disabled.
2482    pub fn compress_if_beneficial<'a>(payload: &'a [u8]) -> (bool, Cow<'a, [u8]>) {
2483        (false, Cow::Borrowed(payload))
2484    }
2485
2486    #[cfg(not(feature = "compression"))]
2487    /// Reports that compressed payloads cannot be decoded when compression support is disabled.
2488    pub fn decompress(_compressed: &[u8], _expected_len: usize) -> Result<Vec<u8>, TelemetryError> {
2489        Err(TelemetryError::Unpack(
2490            "compressed payloads not supported (compression feature disabled)",
2491        ))
2492    }
2493}
2494
2495#[cfg(test)]
2496mod endpoint_decode_tests {
2497    use super::*;
2498
2499    #[test]
2500    fn numeric_sender_alias_keeps_packet_wire_and_receipt_identity() {
2501        let ty = DataType::ReliableAck;
2502        let make = |sender: &str| {
2503            Packet::new_with_wire_contract(
2504                ty,
2505                message_meta(ty).endpoints_ref(),
2506                sender,
2507                200,
2508                9,
2509                Arc::from(42u64.to_le_bytes()),
2510                Some(MessageElement::Static(
2511                    2,
2512                    crate::MessageDataType::UInt32,
2513                    crate::MessageClass::Data,
2514                )),
2515                Arc::from([123u64]),
2516            )
2517            .unwrap()
2518        };
2519        let named = make("FC");
2520        let alias = alloc::format!("@addr:{}", sender_address_u32("FC"));
2521        let unresolved = make(&alias);
2522        assert_eq!(unresolved.packet_id(), named.packet_id());
2523        let named_wire = pack_packet(&named);
2524        let alias_wire = pack_packet(&unresolved);
2525        assert_eq!(alias_wire, named_wire);
2526        assert_eq!(packet_id_from_wire(&alias_wire).unwrap(), named.packet_id());
2527        assert_eq!(
2528            unpack_packet(&alias_wire).unwrap().packet_id(),
2529            named.packet_id()
2530        );
2531    }
2532
2533    #[test]
2534    fn scheduling_type_is_bounded_and_never_replaces_frame_validation() {
2535        let pkt = Packet::new_with_wire_contract(
2536            DataType::ReliableAck,
2537            message_meta(DataType::ReliableAck).endpoints_ref(),
2538            "E2EACK:GS",
2539            100,
2540            1,
2541            Arc::from(42u64.to_le_bytes()),
2542            Some(MessageElement::Static(
2543                2,
2544                crate::MessageDataType::UInt32,
2545                crate::MessageClass::Data,
2546            )),
2547            Arc::from([123u64]),
2548        )
2549        .unwrap();
2550        let mut bytes = pack_packet(&pkt).to_vec();
2551        assert_eq!(scheduling_type(&bytes), Some(DataType::ReliableAck));
2552        let last = bytes.len() - 1;
2553        bytes[last] ^= 1;
2554        assert_eq!(scheduling_type(&bytes), Some(DataType::ReliableAck));
2555        assert!(peek_routing_envelope(&bytes).is_err());
2556        assert!(unpack_packet(&bytes).is_err());
2557        assert_eq!(scheduling_type(&[0, 0, 0x80]), None);
2558        assert_eq!(scheduling_type(&[0xff; 1024]), None);
2559        assert_eq!(scheduling_type(&[]), None);
2560        let unknown_prefix = [0, 0, 0xff, 0xff, 0xff, 0xff, 0x0f];
2561        assert_eq!(scheduling_type(&unknown_prefix), Some(DataType(u32::MAX)));
2562        assert!(peek_routing_envelope(&unknown_prefix).is_err());
2563        assert!(unpack_packet(&unknown_prefix).is_err());
2564    }
2565
2566    #[test]
2567    fn receipt_contract_omits_redundant_builtin_endpoint_bitmap() {
2568        for ty in [
2569            DataType::ReliableAck,
2570            DataType::ReliablePartialAck,
2571            DataType::ReliablePacketRequest,
2572        ] {
2573            let shape = Some(MessageElement::Static(
2574                2,
2575                crate::MessageDataType::UInt32,
2576                crate::MessageClass::Data,
2577            ));
2578            let targets: Arc<[u64]> = Arc::from([0xa7fcbb129e92435f, 0x41426f62de53eb28]);
2579            let payload: Arc<[u8]> = Arc::from(0x9c2a63d6d7d44100u64.to_le_bytes());
2580            let make = |eps: &[DataEndpoint]| {
2581                Packet::new_with_wire_contract(
2582                    ty,
2583                    eps,
2584                    "E2EACK:GS",
2585                    1791737696510,
2586                    1234,
2587                    payload.clone(),
2588                    shape,
2589                    targets.clone(),
2590                )
2591                .unwrap()
2592            };
2593            let packet = make(message_meta(ty).endpoints_ref());
2594            let packed = pack_packet(&packet);
2595            assert_eq!(packed[0] & FLAG_ENDPOINT_BITMAP_PRESENT, 0);
2596            assert_ne!(packed[0] & FLAG_WIRE_CONTRACT, 0);
2597            let routing = peek_routing_envelope(&packed).unwrap();
2598            assert_eq!(routing.endpoints().collect::<Vec<_>>(), packet.endpoints());
2599            assert_eq!(
2600                routing.target_senders().collect::<Vec<_>>(),
2601                targets.as_ref()
2602            );
2603            let decoded = unpack_packet(&packed).unwrap();
2604            assert_eq!(decoded.packet_id(), packet.packet_id());
2605            assert_eq!(decoded.payload(), packet.payload());
2606            assert_eq!(decoded.wire_target_senders(), targets.as_ref());
2607            // Sparse explicit endpoints use a short sorted list.
2608            let explicit = pack_packet(&make(&[DataEndpoint::Discovery]));
2609            assert_ne!(explicit[0] & FLAG_ENDPOINT_LIST_PRESENT, 0);
2610            assert_eq!(
2611                explicit.len(),
2612                packed.len() + uleb128_size(u64::from(DataEndpoint::Discovery.0))
2613            );
2614            assert_eq!(
2615                unpack_packet(&explicit).unwrap().endpoints(),
2616                &[DataEndpoint::Discovery]
2617            );
2618        }
2619    }
2620
2621    #[test]
2622    fn borrowed_routing_header_preserves_targets_and_validates_truncated_contracts() {
2623        let targets: Arc<[u64]> = Arc::from([1, u64::MAX, 42]);
2624        let packet = Packet::new_with_wire_contract(
2625            DataType::DiscoverySchema,
2626            &[DataEndpoint::Discovery],
2627            "PEER",
2628            1234,
2629            7,
2630            Arc::from([1u8, 2, 3]),
2631            Some(MessageElement::Dynamic(
2632                crate::MessageDataType::Binary,
2633                crate::MessageClass::Data,
2634            )),
2635            targets.clone(),
2636        )
2637        .unwrap();
2638        let packed = pack_packet(&packet);
2639        let routing = peek_routing_envelope(&packed).unwrap();
2640        let full = unpack_packet(&packed).unwrap();
2641        assert_eq!(routing.ty, full.data_type());
2642        assert_eq!(routing.endpoints().collect::<Vec<_>>(), full.endpoints());
2643        assert_eq!(
2644            routing.target_senders().collect::<Vec<_>>(),
2645            targets.as_ref()
2646        );
2647        assert_eq!(routing.timestamp_ms, full.timestamp());
2648        // Cut through the header, recomputing CRC so truncation validation runs.
2649        let data = verify_crc32(&packed).unwrap();
2650        let mut reader = ByteReader::new(data);
2651        read_routing_envelope(&mut reader).unwrap();
2652        for end in 0..reader.off {
2653            let mut truncated = data[..end].to_vec();
2654            append_crc32(&mut truncated);
2655            assert!(
2656                peek_routing_envelope(&truncated).is_err(),
2657                "accepted cut at {end}"
2658            );
2659        }
2660    }
2661
2662    #[test]
2663    fn large_packet_with_long_prefix_does_not_grow_packing_buffer() {
2664        let payload: Vec<u8> = (0..3500)
2665            .map(|i| ((i * 73 + i / 251) % 256) as u8)
2666            .collect();
2667        let packet = Packet::new(
2668            DataType::DiscoverySchema,
2669            &[DataEndpoint::Discovery],
2670            "LONG_PREFIX_SENDER",
2671            u64::MAX,
2672            payload.into(),
2673        )
2674        .unwrap()
2675        .with_nonce(u16::MAX);
2676        assert!(header_size_bytes(&packet) > 16);
2677        let packed = pack_packet(&packet);
2678        let unpacked = unpack_packet(&packed).unwrap();
2679        assert_eq!(unpacked.payload(), packet.payload());
2680        assert_eq!(unpacked.timestamp(), u64::MAX);
2681        assert_eq!(unpacked.nonce(), u16::MAX);
2682    }
2683
2684    #[test]
2685    fn bitmap_decode_preserves_sorted_unique_destinations() {
2686        let input = [
2687            DataEndpoint::TelemetryError,
2688            DataEndpoint::TimeSync,
2689            DataEndpoint::Discovery,
2690            DataEndpoint::TimeSync,
2691        ];
2692        let decoded = expand_endpoint_bitmap(&build_endpoint_bitmap(&input)).unwrap();
2693        assert_eq!(
2694            &*decoded,
2695            &[
2696                DataEndpoint::TimeSync,
2697                DataEndpoint::Discovery,
2698                DataEndpoint::TelemetryError
2699            ]
2700        );
2701        assert!(
2702            expand_endpoint_bitmap(&[0; EP_BITMAP_BYTES])
2703                .unwrap()
2704                .is_empty()
2705        );
2706        assert!(expand_endpoint_bitmap(&[0; EP_BITMAP_BYTES - 1]).is_err());
2707    }
2708
2709    #[test]
2710    fn bitmap_decode_rejects_unregistered_endpoint_bits() {
2711        if let Some(id) =
2712            (0..=MAX_VALUE_DATA_ENDPOINT).find(|id| DataEndpoint::try_from_u32(*id).is_none())
2713        {
2714            let mut bitmap = [0; EP_BITMAP_BYTES];
2715            bitmap[id as usize / 8] = 1 << (id % 8);
2716            assert!(expand_endpoint_bitmap(&bitmap).is_err());
2717        }
2718    }
2719}
2720
2721#[cfg(all(test, feature = "discovery", feature = "compression"))]
2722mod bootstrap_compression_tests {
2723    use super::*;
2724
2725    #[test]
2726    fn bootstrap_payloads_remain_decodable_without_optional_compression() {
2727        let payload: Arc<[u8]> = Arc::from(alloc::vec![b'A'; 4096]);
2728        assert!(payload_compression::compress_if_beneficial(&payload).0);
2729        for ty in [
2730            DataType::DiscoverySchema,
2731            DataType::DiscoveryTopology,
2732            DataType::ManagedVariableValue,
2733        ] {
2734            let packet =
2735                Packet::new(ty, &[DataEndpoint::Discovery], "HOST", 123, payload.clone()).unwrap();
2736            let wire = pack_packet(&packet);
2737            assert_eq!(wire[0] & FLAG_COMPRESSED_PAYLOAD, 0, "{ty:?}");
2738            assert_eq!(packet_wire_size(&packet), wire.len());
2739            assert_eq!(unpack_packet(&wire).unwrap().payload(), payload.as_ref());
2740        }
2741    }
2742
2743    #[test]
2744    fn ordinary_data_keeps_optional_payload_compression() {
2745        let payload: Arc<[u8]> = Arc::from(alloc::vec![b'A'; 4096]);
2746        let packet = Packet::new(
2747            DataType::TelemetryError,
2748            &[DataEndpoint::TelemetryError],
2749            "HOST",
2750            123,
2751            payload.clone(),
2752        )
2753        .unwrap();
2754        let wire = pack_packet(&packet);
2755        assert_ne!(wire[0] & FLAG_COMPRESSED_PAYLOAD, 0);
2756        assert_eq!(packet_wire_size(&packet), wire.len());
2757        assert_eq!(unpack_packet(&wire).unwrap().payload(), payload.as_ref());
2758    }
2759}