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.
215pub(super) fn skip_unknown_value<T: DeserializableTraceData>(
216    buf: &mut Buffer<T>,
217) -> Result<(), DecodeError> {
218    rmpv::decode::read_value(buf.as_mut_slice())
219        .map_err(|_| DecodeError::InvalidFormat("Failed to skip unknown V1 value".to_owned()))?;
220    Ok(())
221}
222
223/// Decodes the top-level V1 payload map: tracer metadata fields + chunks array.
224fn decode_payload<T: DeserializableTraceData>(
225    buf: &mut Buffer<T>,
226    table: &mut StringTable<T>,
227) -> Result<TracerPayload<T>, DecodeError>
228where
229    T::Text: Clone,
230{
231    let map_len = decode::read_map_len(buf.as_mut_slice())
232        .map_err(|_| DecodeError::InvalidFormat("Unable to read V1 payload map len".to_owned()))?;
233
234    let mut payload = TracerPayload::<T>::default();
235    let mut saw_chunks = false;
236
237    for _ in 0..map_len {
238        let key = decode::read_int::<u8, _>(buf.as_mut_slice()).map_err(|_| {
239            DecodeError::InvalidFormat("V1 payload key (u8) read failure".to_owned())
240        })?;
241        match key {
242            trace_key::CHUNKS => {
243                payload.chunks = decode_chunks(buf, table)?;
244                saw_chunks = true;
245            }
246            trace_key::CONTAINER_ID => payload.container_id = read_interned_string(buf, table)?,
247            trace_key::LANGUAGE_NAME => payload.language_name = read_interned_string(buf, table)?,
248            trace_key::LANGUAGE_VERSION => {
249                payload.language_version = read_interned_string(buf, table)?
250            }
251            trace_key::TRACER_VERSION => payload.tracer_version = read_interned_string(buf, table)?,
252            trace_key::RUNTIME_ID => payload.runtime_id = read_interned_string(buf, table)?,
253            trace_key::ENV_REF => payload.env = read_interned_string(buf, table)?,
254            trace_key::HOSTNAME_REF => payload.hostname = read_interned_string(buf, table)?,
255            trace_key::APP_VERSION_REF => payload.app_version = read_interned_string(buf, table)?,
256            trace_key::ATTRIBUTES => {
257                payload.attributes = span::read_attributes_map(buf, table)?;
258            }
259            _unknown => skip_unknown_value(buf)?,
260        }
261    }
262
263    if !saw_chunks {
264        return Err(DecodeError::InvalidFormat(
265            "V1 payload is missing the chunks field".to_owned(),
266        ));
267    }
268
269    Ok(payload)
270}
271
272fn decode_chunks<T: DeserializableTraceData>(
273    buf: &mut Buffer<T>,
274    table: &mut StringTable<T>,
275) -> Result<Vec<TraceChunk<T>>, DecodeError>
276where
277    T::Text: Clone,
278{
279    let count = decode::read_array_len(buf.as_mut_slice())
280        .map_err(|_| DecodeError::InvalidFormat("V1 chunks array len read failure".to_owned()))?;
281    let mut chunks = Vec::with_capacity(buf.capped_capacity(count as usize));
282    for _ in 0..count {
283        chunks.push(decode_chunk(buf, table)?);
284    }
285    Ok(chunks)
286}
287
288fn decode_chunk<T: DeserializableTraceData>(
289    buf: &mut Buffer<T>,
290    table: &mut StringTable<T>,
291) -> Result<TraceChunk<T>, DecodeError>
292where
293    T::Text: Clone,
294{
295    let map_len = decode::read_map_len(buf.as_mut_slice())
296        .map_err(|_| DecodeError::InvalidFormat("V1 chunk map len read failure".to_owned()))?;
297    let mut chunk = TraceChunk::<T>::default();
298    let mut saw_trace_id = false;
299    let mut saw_spans = false;
300
301    for _ in 0..map_len {
302        let key = decode::read_int::<u8, _>(buf.as_mut_slice())
303            .map_err(|_| DecodeError::InvalidFormat("V1 chunk key (u8) read failure".to_owned()))?;
304        match key {
305            chunk_key::TRACE_ID => {
306                let len = decode::read_bin_len(buf.as_mut_slice()).map_err(|_| {
307                    DecodeError::InvalidFormat("V1 chunk trace_id bin len read failure".to_owned())
308                })?;
309                if len != TRACE_ID_LEN {
310                    return Err(DecodeError::InvalidFormat(format!(
311                        "V1 chunk trace_id must be {TRACE_ID_LEN} bytes, got {len}"
312                    )));
313                }
314                let bytes = buf
315                    .try_slice_and_advance(TRACE_ID_LEN as usize)
316                    .ok_or_else(|| {
317                        DecodeError::InvalidFormat("V1 chunk trace_id payload truncated".to_owned())
318                    })?;
319                let slice: &[u8] = bytes.borrow();
320                chunk.trace_id.copy_from_slice(slice);
321                saw_trace_id = true;
322            }
323            chunk_key::SPANS => {
324                let count = decode::read_array_len(buf.as_mut_slice()).map_err(|_| {
325                    DecodeError::InvalidFormat("V1 chunk spans array len read failure".to_owned())
326                })?;
327                let mut spans = Vec::with_capacity(buf.capped_capacity(count as usize));
328                for _ in 0..count {
329                    spans.push(span::decode_span(buf, table)?);
330                }
331                chunk.spans = spans;
332                saw_spans = true;
333            }
334            chunk_key::ORIGIN => chunk.origin = read_interned_string(buf, table)?,
335            chunk_key::PRIORITY => {
336                let v: i64 = decode::read_int(buf.as_mut_slice()).map_err(|_| {
337                    DecodeError::InvalidFormat("V1 chunk priority read failure".to_owned())
338                })?;
339                chunk.priority = Some(i32::try_from(v).map_err(|_| {
340                    DecodeError::InvalidFormat(format!("V1 chunk priority {v} exceeds i32 range"))
341                })?);
342            }
343            chunk_key::SAMPLING_MECHANISM => {
344                let v: u64 = decode::read_int(buf.as_mut_slice()).map_err(|_| {
345                    DecodeError::InvalidFormat(
346                        "V1 chunk sampling_mechanism read failure".to_owned(),
347                    )
348                })?;
349                chunk.sampling_mechanism = Some(u32::try_from(v).map_err(|_| {
350                    DecodeError::InvalidFormat(format!(
351                        "V1 chunk sampling_mechanism {v} exceeds u32::MAX"
352                    ))
353                })?);
354            }
355            chunk_key::ATTRIBUTES => {
356                chunk.attributes = span::read_attributes_map(buf, table)?;
357            }
358            chunk_key::DROPPED_TRACE => {
359                chunk.dropped_trace = decode::read_bool(buf.as_mut_slice()).map_err(|_| {
360                    DecodeError::InvalidFormat(
361                        "V1 chunk dropped_trace bool read failure".to_owned(),
362                    )
363                })?;
364            }
365            _unknown => skip_unknown_value(buf)?,
366        }
367    }
368
369    if !saw_trace_id {
370        return Err(DecodeError::InvalidFormat(
371            "V1 chunk is missing trace_id".to_owned(),
372        ));
373    }
374    if !saw_spans {
375        return Err(DecodeError::InvalidFormat(
376            "V1 chunk is missing spans array".to_owned(),
377        ));
378    }
379
380    Ok(chunk)
381}
382
383#[cfg(test)]
384mod tests {
385    use super::*;
386    use crate::msgpack_encoder::v1::to_vec_from_v1;
387    use crate::span::v1::{
388        AttributeValue, Span as V1Span, SpanBytes as V1SpanBytes, SpanKind, TraceChunkBytes,
389        TracerPayloadBytes,
390    };
391    use crate::span::vec_map::VecMap;
392    use bolero::check;
393    use libdd_tinybytes::{Bytes, BytesString};
394
395    fn bs(s: &str) -> BytesString {
396        BytesString::from_slice(s.as_bytes()).expect("test string must fit in BytesString")
397    }
398
399    fn sample_payload() -> TracerPayloadBytes {
400        let mut attrs = VecMap::<BytesString, AttributeValue<_>>::new();
401        attrs.insert(bs("http.method"), AttributeValue::String(bs("GET")));
402        attrs.insert(bs("http.status"), AttributeValue::Int(200));
403        attrs.insert(bs("is_root"), AttributeValue::Bool(true));
404        attrs.insert(bs("ratio"), AttributeValue::Float(0.75));
405        attrs.insert(
406            bs("ids"),
407            AttributeValue::List(vec![AttributeValue::Int(1), AttributeValue::Int(2)]),
408        );
409
410        let span = V1Span {
411            service: bs("svc"),
412            name: bs("GET /users"),
413            resource: bs("/users"),
414            r#type: bs("web"),
415            span_id: 42,
416            parent_id: 7,
417            start: 1_700_000_000_000,
418            duration: 1_500,
419            error: true,
420            span_kind: SpanKind::Server,
421            env: bs("prod"),
422            version: bs("1.2.3"),
423            component: bs("net/http"),
424            attributes: attrs,
425            ..Default::default()
426        };
427
428        let mut chunk_attrs = VecMap::<BytesString, AttributeValue<_>>::new();
429        chunk_attrs.insert(bs("_dd.p.dm"), AttributeValue::String(bs("-1")));
430
431        let chunk = TraceChunkBytes {
432            trace_id: [1u8; 16],
433            priority: Some(1),
434            origin: bs("synthetic"),
435            sampling_mechanism: Some(2),
436            dropped_trace: false,
437            attributes: chunk_attrs,
438            spans: vec![span],
439        };
440
441        TracerPayloadBytes {
442            language_name: bs("rust"),
443            language_version: bs("1.87"),
444            tracer_version: bs("9.9.9"),
445            runtime_id: bs("abcd-1234"),
446            env: bs("prod"),
447            hostname: bs("host-1"),
448            app_version: bs("1.2.3"),
449            chunks: vec![chunk],
450            ..Default::default()
451        }
452    }
453
454    #[test]
455    fn roundtrip_full_payload() {
456        let original = sample_payload();
457        let bytes = to_vec_from_v1(&original);
458        let payload_len = bytes.len();
459        let (decoded, consumed) =
460            from_bytes(Bytes::from(bytes)).expect("decoder should succeed on encoder output");
461
462        assert_eq!(consumed, payload_len, "decoder should consume all bytes");
463
464        // Tracer-level metadata
465        assert_eq!(decoded.language_name.as_str(), "rust");
466        assert_eq!(decoded.language_version.as_str(), "1.87");
467        assert_eq!(decoded.tracer_version.as_str(), "9.9.9");
468        assert_eq!(decoded.runtime_id.as_str(), "abcd-1234");
469        assert_eq!(decoded.env.as_str(), "prod");
470        assert_eq!(decoded.hostname.as_str(), "host-1");
471        assert_eq!(decoded.app_version.as_str(), "1.2.3");
472
473        // Chunk
474        assert_eq!(decoded.chunks.len(), 1);
475        let chunk = &decoded.chunks[0];
476        assert_eq!(chunk.trace_id, [1u8; 16]);
477        assert_eq!(chunk.priority, Some(1));
478        assert_eq!(chunk.sampling_mechanism, Some(2));
479        assert_eq!(chunk.origin.as_str(), "synthetic");
480        assert_eq!(chunk.attributes.len(), 1);
481
482        // Span
483        assert_eq!(chunk.spans.len(), 1);
484        let span = &chunk.spans[0];
485        assert_eq!(span.service.as_str(), "svc");
486        assert_eq!(span.name.as_str(), "GET /users");
487        assert_eq!(span.resource.as_str(), "/users");
488        assert_eq!(span.r#type.as_str(), "web");
489        assert_eq!(span.span_id, 42);
490        assert_eq!(span.parent_id, 7);
491        assert_eq!(span.start, 1_700_000_000_000);
492        assert_eq!(span.duration, 1_500);
493        assert!(span.error);
494        assert_eq!(span.span_kind, SpanKind::Server);
495        assert_eq!(span.env.as_str(), "prod");
496        assert_eq!(span.version.as_str(), "1.2.3");
497        assert_eq!(span.component.as_str(), "net/http");
498        assert_eq!(span.attributes.len(), 5);
499    }
500
501    #[test]
502    fn empty_payload_roundtrip() {
503        let original = TracerPayloadBytes::default();
504        let bytes = to_vec_from_v1(&original);
505        let (decoded, _) =
506            from_bytes(Bytes::from(bytes)).expect("decoder should succeed on empty payload");
507        assert!(decoded.chunks.is_empty());
508        assert!(decoded.language_name.as_str().is_empty());
509    }
510
511    #[test]
512    fn missing_chunks_field_is_rejected() {
513        // Manually encode a payload map with only one entry (env), no chunks field.
514        // `0x81` = fixmap len 1, key 0x07 (ENV_REF), value = inline str "x" (`0xa1 0x78`).
515        let bytes = vec![0x81, 0x07, 0xa1, 0x78];
516        let err = from_bytes(Bytes::from(bytes)).expect_err("missing chunks must error");
517        assert!(matches!(err, DecodeError::InvalidFormat(_)));
518    }
519
520    #[test]
521    fn truncated_trace_id_is_rejected_not_panicking() {
522        // Payload map with 1 entry: chunks -> [ chunk map with 1 entry: trace_id -> bin(16) ].
523        // The bin declares 16 bytes but only 4 are actually present, so the owned decoder's
524        // `try_slice_and_advance` must reject this instead of indexing out of bounds.
525        let bytes = vec![
526            0x81,
527            trace_key::CHUNKS,
528            0x91, // array len 1
529            0x81, // chunk fixmap len 1
530            chunk_key::TRACE_ID,
531            0xc4, // bin8 marker
532            0x10, // declared length: 16 bytes
533            0x01,
534            0x02,
535            0x03,
536            0x04, // only 4 bytes actually present
537        ];
538        let err = from_bytes(Bytes::from(bytes)).expect_err("truncated trace_id must error");
539        assert!(matches!(err, DecodeError::InvalidFormat(_)));
540    }
541
542    #[test]
543    fn string_interning_resolves_across_chunks() {
544        // Two chunks sharing the same service name. The decoded service strings must both
545        // be "shared" — verifying the streaming string table is preserved across chunks.
546        let span_a = V1Span {
547            service: bs("shared"),
548            name: bs("a"),
549            span_id: 1,
550            start: 1,
551            ..Default::default()
552        };
553        let span_b = V1Span {
554            service: bs("shared"),
555            name: bs("b"),
556            span_id: 2,
557            start: 1,
558            ..Default::default()
559        };
560        let payload = TracerPayloadBytes {
561            chunks: vec![
562                TraceChunkBytes {
563                    trace_id: [1u8; 16],
564                    spans: vec![span_a],
565                    ..Default::default()
566                },
567                TraceChunkBytes {
568                    trace_id: [2u8; 16],
569                    spans: vec![span_b],
570                    ..Default::default()
571                },
572            ],
573            ..Default::default()
574        };
575        let bytes = to_vec_from_v1(&payload);
576        let (decoded, _) =
577            from_bytes(Bytes::from(bytes)).expect("decoder should resolve interned strings");
578        assert_eq!(decoded.chunks[0].spans[0].service.as_str(), "shared");
579        assert_eq!(decoded.chunks[1].spans[0].service.as_str(), "shared");
580    }
581
582    #[test]
583    fn nested_keyvalue_attribute_roundtrip() {
584        let mut inner = VecMap::<BytesString, AttributeValue<_>>::new();
585        inner.insert(bs("k"), AttributeValue::String(bs("v")));
586        let mut attrs = VecMap::<BytesString, AttributeValue<_>>::new();
587        attrs.insert(bs("nested"), AttributeValue::KeyValue(inner));
588
589        let span = V1Span {
590            service: bs("svc"),
591            name: bs("op"),
592            span_id: 1,
593            start: 1,
594            attributes: attrs,
595            ..Default::default()
596        };
597        let payload = TracerPayloadBytes {
598            chunks: vec![TraceChunkBytes {
599                trace_id: [0u8; 16],
600                spans: vec![span],
601                ..Default::default()
602            }],
603            ..Default::default()
604        };
605        let bytes = to_vec_from_v1(&payload);
606        let (decoded, _) = from_bytes(Bytes::from(bytes)).expect("nested KeyValue roundtrip");
607
608        let decoded_attrs = &decoded.chunks[0].spans[0].attributes;
609        match decoded_attrs.get(&bs("nested")) {
610            Some(AttributeValue::KeyValue(map)) => {
611                assert_eq!(map.len(), 1);
612                match map.get(&bs("k")) {
613                    Some(AttributeValue::String(v)) => assert_eq!(v.as_str(), "v"),
614                    _ => panic!("inner value should be String"),
615                }
616            }
617            _ => panic!("attribute should decode as KeyValue"),
618        }
619    }
620
621    /// Fuzz test: bolero generates random strings + numbers for the V1 payload, the encoder
622    /// serialises it, and the decoder must accept its own output (no panic, no error). Mirrors
623    /// the v04 `fuzz_from_bytes` pattern. Bolero caps tuples at 12 fields — extra metadata is
624    /// either omitted or filled with deterministic defaults.
625    #[test]
626    #[cfg_attr(miri, ignore)]
627    fn fuzz_from_bytes() {
628        check!()
629            .with_type::<(
630                String, // language_name
631                String, // env (payload-level)
632                String, // service
633                String, // name
634                String, // resource
635                String, // span env
636                String, // attr_key
637                String, // attr_value
638                u64,    // span_id
639                u64,    // parent_id
640                u64,    // start
641                bool,   // error
642            )>()
643            .cloned()
644            .for_each(
645                |(
646                    lang,
647                    payload_env,
648                    service,
649                    name,
650                    resource,
651                    span_env,
652                    attr_key,
653                    attr_value,
654                    span_id,
655                    parent_id,
656                    start,
657                    error,
658                )| {
659                    let bs = |s: &str| BytesString::from_slice(s.as_ref()).unwrap();
660                    let mut attrs = VecMap::<BytesString, AttributeValue<_>>::new();
661                    attrs.insert(bs(&attr_key), AttributeValue::String(bs(&attr_value)));
662
663                    let span = V1SpanBytes {
664                        service: bs(&service),
665                        name: bs(&name),
666                        resource: bs(&resource),
667                        span_id,
668                        parent_id,
669                        start: start as i64,
670                        error,
671                        env: bs(&span_env),
672                        attributes: attrs,
673                        ..Default::default()
674                    };
675
676                    let payload = TracerPayloadBytes {
677                        language_name: bs(&lang),
678                        env: bs(&payload_env),
679                        chunks: vec![TraceChunkBytes {
680                            trace_id: [0xab; 16],
681                            spans: vec![span],
682                            ..Default::default()
683                        }],
684                        ..Default::default()
685                    };
686
687                    let encoded = to_vec_from_v1(&payload);
688                    let result = from_bytes(Bytes::from(encoded));
689                    assert!(
690                        result.is_ok(),
691                        "decoder rejected its own encoded output: {result:?}"
692                    );
693                },
694            );
695    }
696}
libdd_trace_utils/msgpack_decoder/v1/mod.rs

libdd_trace_utils/msgpack_decoder/v1/
mod.rs
