libdd_trace_utils/msgpack_decoder/v1/
mod.rs

1// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/
2// SPDX-License-Identifier: Apache-2.0
3
4pub(super) mod span;
5
6use crate::msgpack_decoder::decode::buffer::Buffer;
7use crate::msgpack_decoder::decode::error::DecodeError;
8use crate::span::v1::{TraceChunk, TracerPayload, TracerPayloadBytes, TracerPayloadSlice};
9use crate::span::DeserializableTraceData;
10use rmp::decode;
11use std::borrow::Borrow;
12
13// Integer keys used by the V1 wire format. Kept in sync with the encoder side
14// (`msgpack_encoder::v1::{trace_key, chunk_key, SpanKey, SpanLinkKey, SpanEventKey, AnyValueKey}`).
15
16pub(super) mod trace_key {
17    pub const CONTAINER_ID: u8 = 2;
18    pub const LANGUAGE_NAME: u8 = 3;
19    pub const LANGUAGE_VERSION: u8 = 4;
20    pub const TRACER_VERSION: u8 = 5;
21    pub const RUNTIME_ID: u8 = 6;
22    pub const ENV_REF: u8 = 7;
23    pub const HOSTNAME_REF: u8 = 8;
24    pub const APP_VERSION_REF: u8 = 9;
25    pub const ATTRIBUTES: u8 = 10;
26    pub const CHUNKS: u8 = 11;
27}
28
29pub(super) mod chunk_key {
30    pub const PRIORITY: u8 = 1;
31    pub const ORIGIN: u8 = 2;
32    pub const ATTRIBUTES: u8 = 3;
33    pub const SPANS: u8 = 4;
34    pub const DROPPED_TRACE: u8 = 5;
35    pub const TRACE_ID: u8 = 6;
36    pub const SAMPLING_MECHANISM: u8 = 7;
37}
38
39pub(super) mod span_key {
40    pub const SERVICE: u8 = 1;
41    pub const NAME: u8 = 2;
42    pub const RESOURCE: u8 = 3;
43    pub const SPAN_ID: u8 = 4;
44    pub const PARENT_ID: u8 = 5;
45    pub const START: u8 = 6;
46    pub const DURATION: u8 = 7;
47    pub const ERROR: u8 = 8;
48    pub const ATTRIBUTES: u8 = 9;
49    pub const TYPE: u8 = 10;
50    pub const SPAN_LINKS: u8 = 11;
51    pub const SPAN_EVENTS: u8 = 12;
52    pub const ENV: u8 = 13;
53    pub const VERSION: u8 = 14;
54    pub const COMPONENT: u8 = 15;
55    pub const KIND: u8 = 16;
56}
57
58pub(super) mod span_link_key {
59    pub const TRACE_ID: u8 = 1;
60    pub const SPAN_ID: u8 = 2;
61    pub const ATTRIBUTES: u8 = 3;
62    pub const TRACE_STATE: u8 = 4;
63    pub const FLAGS: u8 = 5;
64}
65
66pub(super) mod span_event_key {
67    pub const TIME: u8 = 1;
68    pub const NAME: u8 = 2;
69    pub const ATTRIBUTES: u8 = 3;
70}
71
72pub(super) const ANY_VALUE_KEY_STRING: u8 = 1;
73pub(super) const ANY_VALUE_KEY_BOOL: u8 = 2;
74pub(super) const ANY_VALUE_KEY_DOUBLE: u8 = 3;
75pub(super) const ANY_VALUE_KEY_INT64: u8 = 4;
76pub(super) const ANY_VALUE_KEY_BYTES: u8 = 5;
77pub(super) const ANY_VALUE_KEY_ARRAY: u8 = 6;
78pub(super) const ANY_VALUE_KEY_KEY_VALUE_LIST: u8 = 7;
79
80/// Number of msgpack items consumed per `[type, value]` pair in a typed `Array`.
81pub(super) const TYPED_VALUE_STRIDE: u32 = 2;
82
83/// Number of msgpack items consumed per `[key, type, value]` triplet in a typed attributes map.
84pub(super) const FLAT_ATTR_STRIDE: u32 = 3;
85
86/// Length in bytes of a V1 chunk's `trace_id` field (128-bit trace ID).
87pub(super) const TRACE_ID_LEN: u32 = 16;
88
89/// Streaming string intern table built up as the payload is decoded.
90///
91/// V1 strings are encoded inline the first time they appear (as msgpack `str`), and as a
92/// msgpack `uint` reference on every subsequent occurrence. ID 0 is reserved for the empty
93/// string and is pre-inserted on construction.
94pub(super) struct StringTable<T: DeserializableTraceData>
95where
96    T::Text: Clone,
97{
98    seen: Vec<T::Text>,
99}
100
101impl<T: DeserializableTraceData> StringTable<T>
102where
103    T::Text: Clone,
104{
105    pub(super) fn new() -> Self {
106        Self {
107            seen: vec![T::Text::default()],
108        }
109    }
110
111    /// Resolves a string reference by ID (encoded inline as msgpack `uint`).
112    fn resolve(&self, id: u64) -> Result<T::Text, DecodeError> {
113        usize::try_from(id)
114            .ok()
115            .and_then(|i| self.seen.get(i).cloned())
116            .ok_or_else(|| {
117                DecodeError::InvalidFormat(format!(
118                    "V1 string table reference out of range: id={id}, table_len={}",
119                    self.seen.len()
120                ))
121            })
122    }
123
124    /// Records a freshly-read inline string in the table.
125    fn record(&mut self, s: &T::Text) {
126        self.seen.push(s.clone());
127    }
128}
129
130/// Reads a string-or-reference value at the current buffer position.
131///
132/// Decides based on the next msgpack marker:
133/// - `str`/`fixstr` → read and intern, return the value
134/// - any unsigned int marker → resolve the table reference
135pub(super) fn read_interned_string<T: DeserializableTraceData>(
136    buf: &mut Buffer<T>,
137    table: &mut StringTable<T>,
138) -> Result<T::Text, DecodeError>
139where
140    T::Text: Clone,
141{
142    let slice: &[u8] = buf.as_slice();
143    let marker_byte = *slice.first().ok_or_else(|| {
144        DecodeError::InvalidFormat(
145            "Unexpected end of V1 buffer when reading interned string".to_owned(),
146        )
147    })?;
148
149    // msgpack markers:
150    //   fixstr           : 0xa0..=0xbf
151    //   str8/str16/str32 : 0xd9, 0xda, 0xdb
152    //   fixint (positive): 0x00..=0x7f
153    //   uint8/16/32/64   : 0xcc, 0xcd, 0xce, 0xcf
154    match marker_byte {
155        0xa0..=0xbf | 0xd9 | 0xda | 0xdb => {
156            let s = buf.read_string()?;
157            table.record(&s);
158            Ok(s)
159        }
160        0x00..=0x7f | 0xcc | 0xcd | 0xce | 0xcf => {
161            let id: u64 = decode::read_int(buf.as_mut_slice()).map_err(|_| {
162                DecodeError::InvalidFormat(
163                    "V1 interned string reference uint read failure".to_owned(),
164                )
165            })?;
166            table.resolve(id)
167        }
168        _ => Err(DecodeError::InvalidFormat(format!(
169            "Unexpected msgpack marker 0x{marker_byte:02x} for V1 interned string"
170        ))),
171    }
172}
173
174/// Decodes a V1 msgpack payload from owned bytes into a [`TracerPayloadBytes`].
175///
176/// # Returns
177///
178/// * `Ok((payload, payload_size))` — the decoded payload and the number of bytes consumed from the
179///   buffer.
180/// * `Err(DecodeError)` — if the payload is malformed.
181///
182/// # Errors
183///
184/// Returns an error for any malformed map / array length, unknown map key, missing required
185/// field, or any embedded msgpack read failure.
186pub fn from_bytes(
187    data: libdd_tinybytes::Bytes,
188) -> Result<(TracerPayloadBytes, usize), DecodeError> {
189    from_buffer(&mut Buffer::new(data))
190}
191
192/// Decodes a V1 msgpack payload from a borrowed slice into a [`TracerPayloadSlice`].
193/// The resulting payload borrows from the input buffer (same lifetime).
194pub fn from_slice(data: &[u8]) -> Result<(TracerPayloadSlice<'_>, usize), DecodeError> {
195    from_buffer(&mut Buffer::new(data))
196}
197
198/// Generic over the deserialization mode (owned `BytesData` or borrowed `SliceData`).
199pub fn from_buffer<T: DeserializableTraceData>(
200    data: &mut Buffer<T>,
201) -> Result<(TracerPayload<T>, usize), DecodeError>
202where
203    T::Text: Clone,
204{
205    let start_len = data.len();
206    let mut table = StringTable::<T>::new();
207    let payload = decode_payload(data, &mut table)?;
208    let consumed = start_len - data.len();
209    Ok((payload, consumed))
210}
211
212/// Consumes and discards the msgpack value at the current buffer position, regardless of its
213/// type. Used to skip unknown keys for forward compatibility: if the V1 format gains new fields,
214/// older decoders shouldn't reject the whole payload just because they don't recognize a key.
215///
216/// Any inline string encountered while skipping (at any nesting depth) is interned into `table`,
217/// same as a recognized field would: skipping a value must not desync later back-references to
218/// strings that happen to also appear inside it.
219pub(super) fn skip_unknown_value<T: DeserializableTraceData>(
220    buf: &mut Buffer<T>,
221    table: &mut StringTable<T>,
222) -> Result<(), DecodeError>
223where
224    T::Text: Clone,
225{
226    // Snapshot the buffer's owning handle *before* advancing past the skipped value: any string
227    // found inside it will be a substring of this exact allocation, so this is what
228    // `T::intern_skipped_str` must derive ownership from. Cloning is cheap (a refcount bump for
229    // `T::Bytes = Bytes`), unaffected by the lied `'static` lifetime `as_mut_slice` exposes.
230    let owner = buf.bytes().clone();
231    let value = rmpv::decode::read_value_ref(buf.as_mut_slice())
232        .map_err(|_| DecodeError::InvalidFormat("Failed to skip unknown V1 value".to_owned()))?;
233    record_strings_in_value_ref::<T>(&value, &owner, table);
234    Ok(())
235}
236
237/// Recursively walks a parsed [`rmpv::ValueRef`], interning every string it contains into
238/// `table`. Strings with invalid UTF-8 are ignored: they can never have been produced by
239/// [`read_interned_string`]'s own encoder-side counterpart, so they can't be the target of a
240/// later back-reference either.
241///
242/// `owner` must be a snapshot of the buffer taken before it was advanced past `value`: the
243/// strings inside `value` report a lied `'static` lifetime (see `Buffer::as_mut_slice`) but
244/// really borrow from `owner`'s memory.
245fn record_strings_in_value_ref<T: DeserializableTraceData>(
246    value: &rmpv::ValueRef<'static>,
247    owner: &T::Bytes,
248    table: &mut StringTable<T>,
249) where
250    T::Text: Clone,
251{
252    match value {
253        rmpv::ValueRef::String(s) => {
254            if let Some(s) = (*s).into_str() {
255                table.record(&T::intern_skipped_str(owner, s));
256            }
257        }
258        rmpv::ValueRef::Array(items) => {
259            for item in items {
260                record_strings_in_value_ref::<T>(item, owner, table);
261            }
262        }
263        rmpv::ValueRef::Map(entries) => {
264            for (key, val) in entries {
265                record_strings_in_value_ref::<T>(key, owner, table);
266                record_strings_in_value_ref::<T>(val, owner, table);
267            }
268        }
269        _ => {}
270    }
271}
272
273/// Decodes the top-level V1 payload map: tracer metadata fields + chunks array.
274fn decode_payload<T: DeserializableTraceData>(
275    buf: &mut Buffer<T>,
276    table: &mut StringTable<T>,
277) -> Result<TracerPayload<T>, DecodeError>
278where
279    T::Text: Clone,
280{
281    let map_len = decode::read_map_len(buf.as_mut_slice())
282        .map_err(|_| DecodeError::InvalidFormat("Unable to read V1 payload map len".to_owned()))?;
283
284    let mut payload = TracerPayload::<T>::default();
285    let mut saw_chunks = false;
286
287    for _ in 0..map_len {
288        let key = decode::read_int::<u8, _>(buf.as_mut_slice()).map_err(|_| {
289            DecodeError::InvalidFormat("V1 payload key (u8) read failure".to_owned())
290        })?;
291        match key {
292            trace_key::CHUNKS => {
293                payload.chunks = decode_chunks(buf, table)?;
294                saw_chunks = true;
295            }
296            trace_key::CONTAINER_ID => payload.container_id = read_interned_string(buf, table)?,
297            trace_key::LANGUAGE_NAME => payload.language_name = read_interned_string(buf, table)?,
298            trace_key::LANGUAGE_VERSION => {
299                payload.language_version = read_interned_string(buf, table)?
300            }
301            trace_key::TRACER_VERSION => payload.tracer_version = read_interned_string(buf, table)?,
302            trace_key::RUNTIME_ID => payload.runtime_id = read_interned_string(buf, table)?,
303            trace_key::ENV_REF => payload.env = read_interned_string(buf, table)?,
304            trace_key::HOSTNAME_REF => payload.hostname = read_interned_string(buf, table)?,
305            trace_key::APP_VERSION_REF => payload.app_version = read_interned_string(buf, table)?,
306            trace_key::ATTRIBUTES => {
307                payload.attributes = span::read_attributes_map(buf, table)?;
308            }
309            _unknown => skip_unknown_value(buf, table)?,
310        }
311    }
312
313    if !saw_chunks {
314        return Err(DecodeError::InvalidFormat(
315            "V1 payload is missing the chunks field".to_owned(),
316        ));
317    }
318
319    Ok(payload)
320}
321
322fn decode_chunks<T: DeserializableTraceData>(
323    buf: &mut Buffer<T>,
324    table: &mut StringTable<T>,
325) -> Result<Vec<TraceChunk<T>>, DecodeError>
326where
327    T::Text: Clone,
328{
329    let count = decode::read_array_len(buf.as_mut_slice())
330        .map_err(|_| DecodeError::InvalidFormat("V1 chunks array len read failure".to_owned()))?;
331    let mut chunks = Vec::with_capacity(buf.capped_capacity(count as usize));
332    for _ in 0..count {
333        chunks.push(decode_chunk(buf, table)?);
334    }
335    Ok(chunks)
336}
337
338fn decode_chunk<T: DeserializableTraceData>(
339    buf: &mut Buffer<T>,
340    table: &mut StringTable<T>,
341) -> Result<TraceChunk<T>, DecodeError>
342where
343    T::Text: Clone,
344{
345    let map_len = decode::read_map_len(buf.as_mut_slice())
346        .map_err(|_| DecodeError::InvalidFormat("V1 chunk map len read failure".to_owned()))?;
347    let mut chunk = TraceChunk::<T>::default();
348    let mut saw_trace_id = false;
349    let mut saw_spans = false;
350
351    for _ in 0..map_len {
352        let key = decode::read_int::<u8, _>(buf.as_mut_slice())
353            .map_err(|_| DecodeError::InvalidFormat("V1 chunk key (u8) read failure".to_owned()))?;
354        match key {
355            chunk_key::TRACE_ID => {
356                let len = decode::read_bin_len(buf.as_mut_slice()).map_err(|_| {
357                    DecodeError::InvalidFormat("V1 chunk trace_id bin len read failure".to_owned())
358                })?;
359                if len != TRACE_ID_LEN {
360                    return Err(DecodeError::InvalidFormat(format!(
361                        "V1 chunk trace_id must be {TRACE_ID_LEN} bytes, got {len}"
362                    )));
363                }
364                let bytes = buf
365                    .try_slice_and_advance(TRACE_ID_LEN as usize)
366                    .ok_or_else(|| {
367                        DecodeError::InvalidFormat("V1 chunk trace_id payload truncated".to_owned())
368                    })?;
369                let slice: &[u8] = bytes.borrow();
370                chunk.trace_id.copy_from_slice(slice);
371                saw_trace_id = true;
372            }
373            chunk_key::SPANS => {
374                let count = decode::read_array_len(buf.as_mut_slice()).map_err(|_| {
375                    DecodeError::InvalidFormat("V1 chunk spans array len read failure".to_owned())
376                })?;
377                let mut spans = Vec::with_capacity(buf.capped_capacity(count as usize));
378                for _ in 0..count {
379                    spans.push(span::decode_span(buf, table)?);
380                }
381                chunk.spans = spans;
382                saw_spans = true;
383            }
384            chunk_key::ORIGIN => chunk.origin = read_interned_string(buf, table)?,
385            chunk_key::PRIORITY => {
386                let v: i64 = decode::read_int(buf.as_mut_slice()).map_err(|_| {
387                    DecodeError::InvalidFormat("V1 chunk priority read failure".to_owned())
388                })?;
389                chunk.priority = Some(i32::try_from(v).map_err(|_| {
390                    DecodeError::InvalidFormat(format!("V1 chunk priority {v} exceeds i32 range"))
391                })?);
392            }
393            chunk_key::SAMPLING_MECHANISM => {
394                let v: u64 = decode::read_int(buf.as_mut_slice()).map_err(|_| {
395                    DecodeError::InvalidFormat(
396                        "V1 chunk sampling_mechanism read failure".to_owned(),
397                    )
398                })?;
399                chunk.sampling_mechanism = Some(u32::try_from(v).map_err(|_| {
400                    DecodeError::InvalidFormat(format!(
401                        "V1 chunk sampling_mechanism {v} exceeds u32::MAX"
402                    ))
403                })?);
404            }
405            chunk_key::ATTRIBUTES => {
406                chunk.attributes = span::read_attributes_map(buf, table)?;
407            }
408            chunk_key::DROPPED_TRACE => {
409                chunk.dropped_trace = decode::read_bool(buf.as_mut_slice()).map_err(|_| {
410                    DecodeError::InvalidFormat(
411                        "V1 chunk dropped_trace bool read failure".to_owned(),
412                    )
413                })?;
414            }
415            _unknown => skip_unknown_value(buf, table)?,
416        }
417    }
418
419    if !saw_trace_id {
420        return Err(DecodeError::InvalidFormat(
421            "V1 chunk is missing trace_id".to_owned(),
422        ));
423    }
424    if !saw_spans {
425        return Err(DecodeError::InvalidFormat(
426            "V1 chunk is missing spans array".to_owned(),
427        ));
428    }
429
430    Ok(chunk)
431}
432
433#[cfg(test)]
434mod tests {
435    use super::*;
436    use crate::msgpack_encoder::v1::to_vec_from_v1;
437    use crate::span::v1::{
438        AttributeValue, Span as V1Span, SpanBytes as V1SpanBytes, SpanKind, TraceChunkBytes,
439        TracerPayloadBytes,
440    };
441    use crate::span::vec_map::VecMap;
442    use bolero::check;
443    use libdd_tinybytes::{Bytes, BytesString};
444
445    fn bs(s: &str) -> BytesString {
446        BytesString::from_slice(s.as_bytes()).expect("test string must fit in BytesString")
447    }
448
449    fn sample_payload() -> TracerPayloadBytes {
450        let mut attrs = VecMap::<BytesString, AttributeValue<_>>::new();
451        attrs.insert(bs("http.method"), AttributeValue::String(bs("GET")));
452        attrs.insert(bs("http.status"), AttributeValue::Int(200));
453        attrs.insert(bs("is_root"), AttributeValue::Bool(true));
454        attrs.insert(bs("ratio"), AttributeValue::Float(0.75));
455        attrs.insert(
456            bs("ids"),
457            AttributeValue::List(vec![AttributeValue::Int(1), AttributeValue::Int(2)]),
458        );
459
460        let span = V1Span {
461            service: bs("svc"),
462            name: bs("GET /users"),
463            resource: bs("/users"),
464            r#type: bs("web"),
465            span_id: 42,
466            parent_id: 7,
467            start: 1_700_000_000_000,
468            duration: 1_500,
469            error: true,
470            span_kind: SpanKind::Server,
471            env: bs("prod"),
472            version: bs("1.2.3"),
473            component: bs("net/http"),
474            attributes: attrs,
475            ..Default::default()
476        };
477
478        let mut chunk_attrs = VecMap::<BytesString, AttributeValue<_>>::new();
479        chunk_attrs.insert(bs("_dd.p.dm"), AttributeValue::String(bs("-1")));
480
481        let chunk = TraceChunkBytes {
482            trace_id: [1u8; 16],
483            priority: Some(1),
484            origin: bs("synthetic"),
485            sampling_mechanism: Some(2),
486            dropped_trace: false,
487            attributes: chunk_attrs,
488            spans: vec![span],
489        };
490
491        TracerPayloadBytes {
492            language_name: bs("rust"),
493            language_version: bs("1.87"),
494            tracer_version: bs("9.9.9"),
495            runtime_id: bs("abcd-1234"),
496            env: bs("prod"),
497            hostname: bs("host-1"),
498            app_version: bs("1.2.3"),
499            chunks: vec![chunk],
500            ..Default::default()
501        }
502    }
503
504    #[test]
505    fn roundtrip_full_payload() {
506        let original = sample_payload();
507        let bytes = to_vec_from_v1(&original);
508        let payload_len = bytes.len();
509        let (decoded, consumed) =
510            from_bytes(Bytes::from(bytes)).expect("decoder should succeed on encoder output");
511
512        assert_eq!(consumed, payload_len, "decoder should consume all bytes");
513
514        // Tracer-level metadata
515        assert_eq!(decoded.language_name.as_str(), "rust");
516        assert_eq!(decoded.language_version.as_str(), "1.87");
517        assert_eq!(decoded.tracer_version.as_str(), "9.9.9");
518        assert_eq!(decoded.runtime_id.as_str(), "abcd-1234");
519        assert_eq!(decoded.env.as_str(), "prod");
520        assert_eq!(decoded.hostname.as_str(), "host-1");
521        assert_eq!(decoded.app_version.as_str(), "1.2.3");
522
523        // Chunk
524        assert_eq!(decoded.chunks.len(), 1);
525        let chunk = &decoded.chunks[0];
526        assert_eq!(chunk.trace_id, [1u8; 16]);
527        assert_eq!(chunk.priority, Some(1));
528        assert_eq!(chunk.sampling_mechanism, Some(2));
529        assert_eq!(chunk.origin.as_str(), "synthetic");
530        assert_eq!(chunk.attributes.len(), 1);
531
532        // Span
533        assert_eq!(chunk.spans.len(), 1);
534        let span = &chunk.spans[0];
535        assert_eq!(span.service.as_str(), "svc");
536        assert_eq!(span.name.as_str(), "GET /users");
537        assert_eq!(span.resource.as_str(), "/users");
538        assert_eq!(span.r#type.as_str(), "web");
539        assert_eq!(span.span_id, 42);
540        assert_eq!(span.parent_id, 7);
541        assert_eq!(span.start, 1_700_000_000_000);
542        assert_eq!(span.duration, 1_500);
543        assert!(span.error);
544        assert_eq!(span.span_kind, SpanKind::Server);
545        assert_eq!(span.env.as_str(), "prod");
546        assert_eq!(span.version.as_str(), "1.2.3");
547        assert_eq!(span.component.as_str(), "net/http");
548        assert_eq!(span.attributes.len(), 5);
549    }
550
551    #[test]
552    fn empty_payload_roundtrip() {
553        let original = TracerPayloadBytes::default();
554        let bytes = to_vec_from_v1(&original);
555        let (decoded, _) =
556            from_bytes(Bytes::from(bytes)).expect("decoder should succeed on empty payload");
557        assert!(decoded.chunks.is_empty());
558        assert!(decoded.language_name.as_str().is_empty());
559    }
560
561    #[test]
562    fn missing_chunks_field_is_rejected() {
563        // Manually encode a payload map with only one entry (env), no chunks field.
564        // `0x81` = fixmap len 1, key 0x07 (ENV_REF), value = inline str "x" (`0xa1 0x78`).
565        let bytes = vec![0x81, 0x07, 0xa1, 0x78];
566        let err = from_bytes(Bytes::from(bytes)).expect_err("missing chunks must error");
567        assert!(matches!(err, DecodeError::InvalidFormat(_)));
568    }
569
570    #[test]
571    fn truncated_trace_id_is_rejected_not_panicking() {
572        // Payload map with 1 entry: chunks -> [ chunk map with 1 entry: trace_id -> bin(16) ].
573        // The bin declares 16 bytes but only 4 are actually present, so the owned decoder's
574        // `try_slice_and_advance` must reject this instead of indexing out of bounds.
575        let bytes = vec![
576            0x81,
577            trace_key::CHUNKS,
578            0x91, // array len 1
579            0x81, // chunk fixmap len 1
580            chunk_key::TRACE_ID,
581            0xc4, // bin8 marker
582            0x10, // declared length: 16 bytes
583            0x01,
584            0x02,
585            0x03,
586            0x04, // only 4 bytes actually present
587        ];
588        let err = from_bytes(Bytes::from(bytes)).expect_err("truncated trace_id must error");
589        assert!(matches!(err, DecodeError::InvalidFormat(_)));
590    }
591
592    #[test]
593    fn string_interning_resolves_across_chunks() {
594        // Two chunks sharing the same service name. The decoded service strings must both
595        // be "shared" — verifying the streaming string table is preserved across chunks.
596        let span_a = V1Span {
597            service: bs("shared"),
598            name: bs("a"),
599            span_id: 1,
600            start: 1,
601            ..Default::default()
602        };
603        let span_b = V1Span {
604            service: bs("shared"),
605            name: bs("b"),
606            span_id: 2,
607            start: 1,
608            ..Default::default()
609        };
610        let payload = TracerPayloadBytes {
611            chunks: vec![
612                TraceChunkBytes {
613                    trace_id: [1u8; 16],
614                    spans: vec![span_a],
615                    ..Default::default()
616                },
617                TraceChunkBytes {
618                    trace_id: [2u8; 16],
619                    spans: vec![span_b],
620                    ..Default::default()
621                },
622            ],
623            ..Default::default()
624        };
625        let bytes = to_vec_from_v1(&payload);
626        let (decoded, _) =
627            from_bytes(Bytes::from(bytes)).expect("decoder should resolve interned strings");
628        assert_eq!(decoded.chunks[0].spans[0].service.as_str(), "shared");
629        assert_eq!(decoded.chunks[1].spans[0].service.as_str(), "shared");
630    }
631
632    #[test]
633    fn nested_keyvalue_attribute_roundtrip() {
634        let mut inner = VecMap::<BytesString, AttributeValue<_>>::new();
635        inner.insert(bs("k"), AttributeValue::String(bs("v")));
636        let mut attrs = VecMap::<BytesString, AttributeValue<_>>::new();
637        attrs.insert(bs("nested"), AttributeValue::KeyValue(inner));
638
639        let span = V1Span {
640            service: bs("svc"),
641            name: bs("op"),
642            span_id: 1,
643            start: 1,
644            attributes: attrs,
645            ..Default::default()
646        };
647        let payload = TracerPayloadBytes {
648            chunks: vec![TraceChunkBytes {
649                trace_id: [0u8; 16],
650                spans: vec![span],
651                ..Default::default()
652            }],
653            ..Default::default()
654        };
655        let bytes = to_vec_from_v1(&payload);
656        let (decoded, _) = from_bytes(Bytes::from(bytes)).expect("nested KeyValue roundtrip");
657
658        let decoded_attrs = &decoded.chunks[0].spans[0].attributes;
659        match decoded_attrs.get(&bs("nested")) {
660            Some(AttributeValue::KeyValue(map)) => {
661                assert_eq!(map.len(), 1);
662                match map.get(&bs("k")) {
663                    Some(AttributeValue::String(v)) => assert_eq!(v.as_str(), "v"),
664                    _ => panic!("inner value should be String"),
665                }
666            }
667            _ => panic!("attribute should decode as KeyValue"),
668        }
669    }
670
671    /// Fuzz test: bolero generates random strings + numbers for the V1 payload, the encoder
672    /// serialises it, and the decoder must accept its own output (no panic, no error). Mirrors
673    /// the v04 `fuzz_from_bytes` pattern. Bolero caps tuples at 12 fields — extra metadata is
674    /// either omitted or filled with deterministic defaults.
675    #[test]
676    #[cfg_attr(miri, ignore)]
677    fn fuzz_from_bytes() {
678        check!()
679            .with_type::<(
680                String, // language_name
681                String, // env (payload-level)
682                String, // service
683                String, // name
684                String, // resource
685                String, // span env
686                String, // attr_key
687                String, // attr_value
688                u64,    // span_id
689                u64,    // parent_id
690                u64,    // start
691                bool,   // error
692            )>()
693            .cloned()
694            .for_each(
695                |(
696                    lang,
697                    payload_env,
698                    service,
699                    name,
700                    resource,
701                    span_env,
702                    attr_key,
703                    attr_value,
704                    span_id,
705                    parent_id,
706                    start,
707                    error,
708                )| {
709                    let bs = |s: &str| BytesString::from_slice(s.as_ref()).unwrap();
710                    let mut attrs = VecMap::<BytesString, AttributeValue<_>>::new();
711                    attrs.insert(bs(&attr_key), AttributeValue::String(bs(&attr_value)));
712
713                    let span = V1SpanBytes {
714                        service: bs(&service),
715                        name: bs(&name),
716                        resource: bs(&resource),
717                        span_id,
718                        parent_id,
719                        start: start as i64,
720                        error,
721                        env: bs(&span_env),
722                        attributes: attrs,
723                        ..Default::default()
724                    };
725
726                    let payload = TracerPayloadBytes {
727                        language_name: bs(&lang),
728                        env: bs(&payload_env),
729                        chunks: vec![TraceChunkBytes {
730                            trace_id: [0xab; 16],
731                            spans: vec![span],
732                            ..Default::default()
733                        }],
734                        ..Default::default()
735                    };
736
737                    let encoded = to_vec_from_v1(&payload);
738                    let result = from_bytes(Bytes::from(encoded));
739                    assert!(
740                        result.is_ok(),
741                        "decoder rejected its own encoded output: {result:?}"
742                    );
743                },
744            );
745    }
746
747    // ---------------------------------------------------------------------------------------------
748    // Forward-compatibility: unknown map keys must be skipped for every V1 map type. This test
749    // hand-builds wire bytes (the encoder never emits unknown keys) with `rmp::encode`, injecting
750    // a future/unknown key at every nesting level (payload, chunk, span, span_link, span_event),
751    // and asserts the surrounding known fields still decode correctly.
752    // ---------------------------------------------------------------------------------------------
753
754    use rmp::encode::{self, ByteBuf};
755
756    /// Writes a `u8` msgpack map key.
757    fn wkey(buf: &mut ByteBuf, k: u8) {
758        encode::write_uint(buf, k as u64).unwrap();
759    }
760
761    /// Exercises unknown-key skipping at every V1 nesting level in a single payload:
762    /// - payload: unknown field 99 carries the first occurrence of "ghost" (must be harvested as
763    ///   table id 1, a scalar skip at the chunk level exercises the recursive skip, and a
764    ///   subsequent span field back-references "prod" by id to prove the table wasn't desynced).
765    /// - chunk: unknown field 77 carries a nested `[uint, str, map]` value (recursive skip), whose
766    ///   inline string "buried" must also be harvested (table id 3).
767    /// - span: unknown field 88 carries a scalar (f64) with no string to harvest.
768    /// - span_link / span_event: unknown fields 55 / 66 precede their known sibling field.
769    #[test]
770    fn unknown_keys_are_skipped_at_every_level() {
771        let mut span_link = ByteBuf::new();
772        encode::write_map_len(&mut span_link, 2).unwrap();
773        wkey(&mut span_link, 55); // unknown span_link key
774        encode::write_bool(&mut span_link, false).unwrap();
775        wkey(&mut span_link, span_link_key::SPAN_ID);
776        encode::write_uint(&mut span_link, 777).unwrap();
777
778        let mut span_event = ByteBuf::new();
779        encode::write_map_len(&mut span_event, 2).unwrap();
780        wkey(&mut span_event, 66); // unknown span_event key
781        encode::write_uint(&mut span_event, 999).unwrap();
782        wkey(&mut span_event, span_event_key::TIME);
783        encode::write_uint(&mut span_event, 123).unwrap();
784
785        let mut span = ByteBuf::new();
786        encode::write_map_len(&mut span, 6).unwrap();
787        wkey(&mut span, span_key::SPAN_ID);
788        encode::write_uint(&mut span, 42).unwrap();
789        wkey(&mut span, span_key::START);
790        encode::write_uint(&mut span, 100).unwrap();
791        wkey(&mut span, 88); // unknown span key: scalar, nothing to harvest
792        encode::write_f64(&mut span, 2.5).unwrap();
793        wkey(&mut span, span_key::SERVICE);
794        encode::write_uint(&mut span, 2).unwrap(); // back-reference to encoder id 2 ("prod")
795        wkey(&mut span, span_key::SPAN_LINKS);
796        encode::write_array_len(&mut span, 1).unwrap();
797        span.as_mut_vec().extend_from_slice(&span_link.into_vec());
798        wkey(&mut span, span_key::SPAN_EVENTS);
799        encode::write_array_len(&mut span, 1).unwrap();
800        span.as_mut_vec().extend_from_slice(&span_event.into_vec());
801
802        let mut chunk = ByteBuf::new();
803        encode::write_map_len(&mut chunk, 3).unwrap();
804        wkey(&mut chunk, 77); // unknown chunk key: nested value, recursive skip + string harvest
805        encode::write_array_len(&mut chunk, 3).unwrap();
806        encode::write_uint(&mut chunk, 1).unwrap();
807        encode::write_str(&mut chunk, "buried").unwrap(); // first occurrence -> table id 3
808        encode::write_map_len(&mut chunk, 1).unwrap();
809        encode::write_uint(&mut chunk, 5).unwrap();
810        encode::write_bool(&mut chunk, true).unwrap();
811        wkey(&mut chunk, chunk_key::TRACE_ID);
812        encode::write_bin(&mut chunk, &[9u8; 16]).unwrap();
813        wkey(&mut chunk, chunk_key::SPANS);
814        encode::write_array_len(&mut chunk, 1).unwrap();
815        chunk.as_mut_vec().extend_from_slice(&span.into_vec());
816
817        let mut buf = ByteBuf::new();
818        encode::write_map_len(&mut buf, 3).unwrap();
819        wkey(&mut buf, 99); // unknown payload key: first occurrence "ghost" -> table id 1
820        encode::write_str(&mut buf, "ghost").unwrap();
821        wkey(&mut buf, trace_key::ENV_REF);
822        encode::write_str(&mut buf, "prod").unwrap(); // first occurrence -> table id 2
823        wkey(&mut buf, trace_key::CHUNKS);
824        encode::write_array_len(&mut buf, 1).unwrap();
825        buf.as_mut_vec().extend_from_slice(&chunk.into_vec());
826        let buf = buf.into_vec();
827
828        let (decoded, consumed) =
829            from_bytes(Bytes::from(buf.clone())).expect("unknown keys must be skipped");
830        assert_eq!(
831            consumed,
832            buf.len(),
833            "decoder must consume every skipped value"
834        );
835        assert_eq!(decoded.env.as_str(), "prod");
836        let chunk = &decoded.chunks[0];
837        assert_eq!(chunk.trace_id, [9u8; 16]);
838        let span = &chunk.spans[0];
839        assert_eq!(span.span_id, 42);
840        assert_eq!(span.start, 100);
841        assert_eq!(
842            span.service.as_str(),
843            "prod",
844            "back-reference must still resolve correctly: harvesting \"ghost\" and \"buried\" \
845             while skipping unknown fields must not desync the string table"
846        );
847        assert_eq!(span.span_links[0].span_id, 777);
848        assert_eq!(span.span_events[0].time_unix_nano, 123);
849    }
850}
libdd_trace_utils/msgpack_decoder/v1/mod.rs

libdd_trace_utils/msgpack_decoder/v1/
mod.rs
