Skip to main content

moqtap_trace/
event.rs

1use ciborium::Value;
2
3use crate::error::MoqTraceError;
4use crate::header::{as_i64, as_u64, normalised, store_entries, unrecognised, DetailLevel};
5
6/// Direction of a message or stream relative to the recording endpoint.
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub enum Direction {
9    /// Sent (outgoing). Wire value: `0`.
10    Send,
11    /// Received (incoming). Wire value: `1`.
12    Receive,
13}
14
15impl Direction {
16    fn from_cbor(v: &Value) -> Result<Self, MoqTraceError> {
17        match as_u64(v) {
18            Some(0) => Ok(Direction::Send),
19            Some(1) => Ok(Direction::Receive),
20            _ => Err(MoqTraceError::InvalidEvent("invalid direction value".into())),
21        }
22    }
23}
24
25/// Data stream type.
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub enum StreamType {
28    /// Subgroup stream. Wire value: `0`.
29    Subgroup,
30    /// Datagram. Wire value: `1`.
31    Datagram,
32    /// Fetch stream. Wire value: `2`.
33    Fetch,
34}
35
36impl StreamType {
37    fn from_cbor(v: &Value) -> Result<Self, MoqTraceError> {
38        match as_u64(v) {
39            Some(0) => Ok(StreamType::Subgroup),
40            Some(1) => Ok(StreamType::Datagram),
41            Some(2) => Ok(StreamType::Fetch),
42            _ => Err(MoqTraceError::InvalidEvent("invalid stream type value".into())),
43        }
44    }
45}
46
47/// What sort of failure an [`Error`](EventData::Error) event records.
48///
49/// An open vocabulary. SPEC.md's Event 6 section names three kinds and says
50/// others may be added without a format version bump, so a spelling this crate
51/// does not know is kept verbatim in [`Other`](ErrorKind::Other) rather than
52/// refused. That is the treatment [`Perspective`](crate::header::Perspective)
53/// gets, and it is the rule SPEC.md points at for this key.
54///
55/// An enum and not a bare `String` because these three are the recorder's own
56/// classification of what it just saw, chosen at the point the event is built:
57/// a misspelling is a value no reader can group by, and the compiler is the
58/// only thing that catches it before the trace is written. The wire spellings
59/// live in [`as_str`](ErrorKind::as_str) alone, so a reader and a writer cannot
60/// disagree about them.
61#[derive(Debug, Clone, PartialEq, Eq)]
62#[non_exhaustive]
63pub enum ErrorKind {
64    /// The peer violated the protocol.
65    Protocol,
66    /// The QUIC or WebTransport layer failed.
67    Transport,
68    /// Bytes that would not parse as any message this recorder knows.
69    Decode,
70    /// A kind this version of the crate does not know, kept verbatim.
71    Other(String),
72}
73
74impl ErrorKind {
75    /// The wire spelling of this kind.
76    pub fn as_str(&self) -> &str {
77        match self {
78            ErrorKind::Protocol => "protocol",
79            ErrorKind::Transport => "transport",
80            ErrorKind::Decode => "decode",
81            ErrorKind::Other(s) => s,
82        }
83    }
84
85    fn parse(s: &str) -> Self {
86        match s {
87            "protocol" => ErrorKind::Protocol,
88            "transport" => ErrorKind::Transport,
89            "decode" => ErrorKind::Decode,
90            other => ErrorKind::Other(other.to_string()),
91        }
92    }
93}
94
95/// Which end of a relay a peer sits on.
96///
97/// This is the direction the *connection* was made in, not the direction
98/// subscriptions flow. A downstream peer may publish and an upstream peer may
99/// subscribe; for subscription causality see
100/// [`EventData::SubscriptionDerivation`].
101#[derive(Debug, Clone, PartialEq, Eq, Hash)]
102#[non_exhaustive]
103pub enum Side {
104    /// The peer connected to this relay.
105    Downstream,
106    /// This relay connected to the peer — an origin or an upstream relay.
107    Upstream,
108    /// A side this version of the crate does not know, kept verbatim.
109    Other(String),
110}
111
112impl Side {
113    /// The wire spelling of this side.
114    pub fn as_str(&self) -> &str {
115        match self {
116            Side::Downstream => "downstream",
117            Side::Upstream => "upstream",
118            Side::Other(s) => s,
119        }
120    }
121
122    fn parse(s: &str) -> Self {
123        match s {
124            "downstream" => Side::Downstream,
125            "upstream" => Side::Upstream,
126            other => Side::Other(other.to_string()),
127        }
128    }
129}
130
131/// Role a peer is acting in at connection time.
132#[derive(Debug, Clone, PartialEq, Eq)]
133#[non_exhaustive]
134pub enum PeerRole {
135    /// Peer is acting as a publisher.
136    Publisher,
137    /// Peer is acting as a subscriber.
138    Subscriber,
139    /// Peer is acting as both.
140    Both,
141    /// A role this version of the crate does not know, kept verbatim.
142    Other(String),
143}
144
145impl PeerRole {
146    /// The wire spelling of this role.
147    pub fn as_str(&self) -> &str {
148        match self {
149            PeerRole::Publisher => "publisher",
150            PeerRole::Subscriber => "subscriber",
151            PeerRole::Both => "both",
152            PeerRole::Other(s) => s,
153        }
154    }
155
156    fn parse(s: &str) -> Self {
157        match s {
158            "publisher" => PeerRole::Publisher,
159            "subscriber" => PeerRole::Subscriber,
160            "both" => PeerRole::Both,
161            other => PeerRole::Other(other.to_string()),
162        }
163    }
164}
165
166/// How an upstream subscription came to serve a downstream one.
167#[derive(Debug, Clone, PartialEq, Eq)]
168#[non_exhaustive]
169pub enum DerivationKind {
170    /// A new upstream subscription was created to satisfy downstream subs.
171    Created,
172    /// An existing upstream subscription now also serves an additional
173    /// downstream sub — subscription fan-in.
174    Shared,
175    /// A kind this version of the crate does not know, kept verbatim.
176    Other(String),
177}
178
179impl DerivationKind {
180    /// The wire spelling of this kind.
181    pub fn as_str(&self) -> &str {
182        match self {
183            DerivationKind::Created => "created",
184            DerivationKind::Shared => "shared",
185            DerivationKind::Other(s) => s,
186        }
187    }
188
189    fn parse(s: &str) -> Self {
190        match s {
191            "created" => DerivationKind::Created,
192            "shared" => DerivationKind::Shared,
193            other => DerivationKind::Other(other.to_string()),
194        }
195    }
196}
197
198/// A `(peer, request id)` reference naming one subscription on one peer.
199///
200/// Request IDs are only unique within a peer's session, so neither half
201/// identifies a subscription on its own.
202#[derive(Debug, Clone, PartialEq, Eq)]
203pub struct SubscriptionRef {
204    /// Source-local peer identifier.
205    pub peer: String,
206    /// Request ID of the subscription on that peer.
207    pub request_id: u64,
208}
209
210impl SubscriptionRef {
211    /// A reference to `request_id` on `peer`.
212    pub fn new(peer: impl Into<String>, request_id: u64) -> Self {
213        SubscriptionRef { peer: peer.into(), request_id }
214    }
215
216    fn to_value(&self) -> Value {
217        Value::Array(vec![Value::Text(self.peer.clone()), Value::Integer(self.request_id.into())])
218    }
219
220    fn from_value(v: &Value) -> Result<Self, MoqTraceError> {
221        let arr = match v {
222            Value::Array(a) => a,
223            _ => {
224                return Err(MoqTraceError::InvalidEvent(
225                    "subscription ref is not a CBOR array".into(),
226                ))
227            }
228        };
229        if arr.len() != 2 {
230            return Err(MoqTraceError::InvalidEvent(
231                "subscription ref must be [peer, request_id]".into(),
232            ));
233        }
234        let peer = arr[0]
235            .as_text()
236            .ok_or_else(|| MoqTraceError::InvalidEvent("subscription ref peer is not text".into()))?
237            .to_string();
238        let request_id = as_u64(&arr[1]).ok_or_else(|| {
239            MoqTraceError::InvalidEvent("subscription ref request_id is not uint".into())
240        })?;
241        Ok(SubscriptionRef { peer, request_id })
242    }
243}
244
245/// Length of a trace ID, in bytes. Fixed by the format.
246pub const TRACE_ID_LEN: usize = 16;
247
248/// The most bytes a recorder may put in an [`Error`](EventData::Error) event's
249/// `"raw"`. Fixed by the format — SPEC.md, Event 6, under the heading about
250/// the cap and `"rawlen"`.
251///
252/// The cap binds the party building an event out of bytes it has just
253/// observed, and nothing else. It is deliberately **not** enforced when an
254/// event is serialized, and this crate does not enforce it there: a serializer
255/// cannot tell a freshly recorded event from one that arrived by being read,
256/// so a cap applied at that point either shortens evidence on a rewrite or
257/// refuses a file the reader was required to accept — whichever it does, it
258/// does to the wrong events. A `"raw"` longer than this reads back at its full
259/// length and is written back at its full length.
260///
261/// [`EventData::error_observed`] is where the cap belongs and where this crate
262/// applies it. A recorder assembling the variant by hand applies it here.
263///
264/// One obligation stays with the recorder and cannot live in a constructor at
265/// all: SPEC.md allows `"raw"` once per flow — per stream where the error names
266/// one, per peer where it does not. Later errors on that flow are still
267/// recorded and simply carry no bytes. That is state across events, so it
268/// belongs to whatever is holding the flow.
269pub const ERROR_RAW_CAP: usize = 4096;
270
271/// Event type discriminants, matching the specification's `"e"` values.
272const EVENT_CONTROL_MESSAGE: u64 = 0;
273const EVENT_STREAM_OPENED: u64 = 1;
274const EVENT_STREAM_CLOSED: u64 = 2;
275const EVENT_OBJECT_HEADER: u64 = 3;
276const EVENT_OBJECT_PAYLOAD: u64 = 4;
277const EVENT_STATE_CHANGE: u64 = 5;
278const EVENT_ERROR: u64 = 6;
279const EVENT_ANNOTATION: u64 = 7;
280const EVENT_PEER_CONNECTED: u64 = 8;
281const EVENT_PEER_DISCONNECTED: u64 = 9;
282const EVENT_SUBSCRIPTION_DERIVATION: u64 = 10;
283
284/// A single event in a `.moqtrace` file.
285#[derive(Debug, Clone, PartialEq)]
286pub struct TraceEvent {
287    /// Monotonically increasing sequence number (0-based). Segment-local in a
288    /// segmented trace, so ordering across segments is `(segment, seq)`.
289    pub seq: u64,
290    /// Timestamp in microseconds since the containing segment's start time.
291    pub timestamp: i64,
292    /// Which peer this event pertains to.
293    ///
294    /// Required when the trace's perspective is
295    /// [`RelayTap`](crate::header::Perspective::RelayTap), where a single
296    /// trace covers many concurrent sessions; omitted otherwise, since a
297    /// single-session trace has only one peer to speak of. The identifier is
298    /// **source-local**: the same string in two traces from different sources
299    /// does not name the same peer.
300    pub peer: Option<String>,
301    /// Event-specific data.
302    pub data: EventData,
303    /// Keys on this event that this version of the crate could not use, kept
304    /// verbatim.
305    ///
306    /// Optional keys may be added to an existing event type without a format
307    /// version bump, so "unknown keys MUST be ignored" is a rule about
308    /// *reading past* them. It is not a licence to drop them: a tool that
309    /// reads a trace and writes it back — a redaction pass, a filter, a
310    /// re-segmentation — would otherwise emit a valid file that looks like it
311    /// never carried them, and one tool's ignorance would become permanent for
312    /// every reader downstream of it.
313    ///
314    /// A key this crate *does* know lands here too when its value is not of a
315    /// type that key can hold — `"ta": "hello"` on an event 1, say. SPEC.md
316    /// treats such a key as unrecognised: the value is ignored for meaning,
317    /// the field that would have held it reads `None`, and the entry is
318    /// written back unchanged. Knowing more about a key must not mean
319    /// preserving it less.
320    ///
321    /// [`EventData::Unknown`] already does this for an event type the crate
322    /// cannot name. This is the same guarantee one level down, for a key on a
323    /// type it can.
324    ///
325    /// Unchanged binds the value and not its encoding, exactly as it does for
326    /// [`TraceHeader::extra`](crate::header::TraceHeader::extra): on the way
327    /// out an integral float in a stored value is written as a CBOR integer
328    /// and a byte string under RFC 8746's tag 64 as major type 2, at any
329    /// depth. SPEC.md's two encoding rules are about every byte a writer
330    /// emits rather than only the keys it understood, and the JavaScript
331    /// implementation's decoder folds both shapes away before its own code
332    /// runs, so it could not emit either however hard it tried. Nothing a
333    /// comparison of the two values can see changes, with the single
334    /// exception SPEC.md names: `-0.0` written as `0` loses its sign.
335    ///
336    /// A CBOR map may not carry one key twice, and this list can: it is an
337    /// ordered list of pairs, not a map. So on the way out an entry naming a
338    /// key the event writes from a field is dropped, and of two entries
339    /// sharing a key the first is written — as it is for a map nested inside
340    /// a stored value, which is a map this crate emits too.
341    ///
342    /// The same two passes govern the other opaque values an event carries:
343    /// a control message's `"msg"`, an annotation's `"data"` and an
344    /// [`EventData::Unknown`]'s fields.
345    ///
346    /// Empty for every event this crate constructs itself.
347    pub extra: Vec<(Value, Value)>,
348}
349
350/// Event-specific payload, discriminated by type.
351#[derive(Debug, Clone, PartialEq)]
352#[non_exhaustive]
353pub enum EventData {
354    /// A control-stream message was sent or received (event type 0).
355    ControlMessage {
356        /// Send or receive.
357        direction: Direction,
358        /// Wire message type ID (e.g. `0x03` for SUBSCRIBE).
359        message_type: u64,
360        /// Decoded message fields, and **not necessarily a map**: match on it
361        /// rather than assuming, or use [`TraceEvent::request_id`].
362        ///
363        /// A *recorder* MUST write a CBOR map keyed in snake_case here, and an
364        /// empty map when it decoded nothing. A *reader* has to take what
365        /// earlier writers produced, and every `capture-*` case in the
366        /// conformance corpus holds a text rendering of the message instead.
367        /// Such a value is handed back verbatim — unaddressable by key, but
368        /// not a reason to reject an event the format forbids dropping — and
369        /// is written back unchanged, because replacing it would destroy the
370        /// only record of a message nobody will see again.
371        ///
372        /// Unchanged binds what the value says and not how it is encoded.
373        /// This is a value the reader never looked at, so the two encoding
374        /// rules reach it on the way out like any other:
375        /// [`TraceEvent::extra`] has the whole of it.
376        message: Value,
377        /// QUIC stream the message travelled on, when the recorder knows it.
378        ///
379        /// `None` means unknown, which a reader must not confuse with stream
380        /// 0. Recorders should fill it in: from draft-17 each request has its
381        /// own bidirectional stream and responses carry no request ID, so the
382        /// stream is the only thing pairing a response with its request.
383        stream_id: Option<u64>,
384        /// Raw wire bytes (only at `full` detail level).
385        raw: Option<Vec<u8>>,
386    },
387    /// A stream was opened, or a datagram arrived (event type 1).
388    ///
389    /// The four optional identifiers are what the recorder read from the
390    /// stream's header, and [`header_raw`](EventData::StreamOpened::header_raw)
391    /// is the header itself when the recorder held it. Without them a
392    /// `headers` trace cannot say which track a stream belonged to.
393    ///
394    /// Each is `None` when the recorder did not know it — and also when the
395    /// file carried the key with a value of the wrong type, in which case the
396    /// entry is kept verbatim in [`TraceEvent::extra`] instead of being read
397    /// here. A reader must not refuse a stream for lacking one.
398    StreamOpened {
399        /// The QUIC stream ID, or, where the recorder sees none, one it
400        /// assigned below 2^62. A datagram's is its own: at least 2^62 and unique
401        /// within the trace, and older traces put every datagram on 0. Tell a
402        /// datagram by `stream_type`, never by this value. SPEC.md
403        /// "`"sid"` — the stream ID".
404        stream_id: u64,
405        /// Outgoing or incoming.
406        direction: Direction,
407        /// Stream type.
408        stream_type: StreamType,
409        /// Track alias the stream carries. Meaningful on any stream type.
410        track_alias: Option<u64>,
411        /// Subgroup ID, on a [`StreamType::Subgroup`] stream.
412        subgroup_id: Option<u64>,
413        /// Fetch request ID, on a [`StreamType::Fetch`] stream.
414        ///
415        /// A recorder must write it there. It is the only correlation between
416        /// the stream and the FETCH that asked for it, and unlike a subgroup
417        /// stream a fetch stream carries no track alias to identify it by
418        /// instead — so without it a fetch stream in a `headers` trace names
419        /// nothing at all.
420        fetch_request_id: Option<u64>,
421        /// Group ID, on a [`StreamType::Datagram`] stream.
422        ///
423        /// Scoped to datagrams because on a subgroup stream every object
424        /// carries the group of the stream by construction, so a copy here
425        /// would be a second field with no independent source. Where it does
426        /// appear alongside an [`ObjectHeader`](EventData::ObjectHeader) for
427        /// the same stream, the object header is authoritative: this copy
428        /// serves a reader that has not yet seen an object, and a
429        /// disagreement between the two is not corruption.
430        group_id: Option<u64>,
431        /// The header's wire bytes (`"hdr"`, `headers` level and up): from the
432        /// stream type through the last header field, or a datagram's bytes
433        /// before its payload. No transport framing.
434        ///
435        /// Where these and the fields above disagree, the bytes are the
436        /// evidence: the fields are what the recorder read from them.
437        header_raw: Option<Vec<u8>>,
438    },
439    /// A QUIC stream was closed (event type 2).
440    StreamClosed {
441        /// Stream ID, as on [`StreamOpened`](EventData::StreamOpened::stream_id).
442        stream_id: u64,
443        /// Error code (0 = clean close).
444        error_code: u64,
445    },
446    /// An object header was parsed from a data stream (event type 3).
447    ObjectHeader {
448        /// Stream ID this object arrived on, as on
449        /// [`StreamOpened`](EventData::StreamOpened::stream_id). A datagram's
450        /// object carries its datagram's ID.
451        stream_id: u64,
452        /// Group ID.
453        group: u64,
454        /// Object ID.
455        object: u64,
456        /// Publisher priority.
457        publisher_priority: u64,
458        /// Object status (0=normal, 1=end-of-group, etc.).
459        object_status: u64,
460    },
461    /// Object payload bytes were received or sent (event type 4).
462    ObjectPayload {
463        /// Stream ID, as on [`ObjectHeader`](EventData::ObjectHeader::stream_id).
464        stream_id: u64,
465        /// Group ID.
466        group: u64,
467        /// Object ID.
468        object: u64,
469        /// Payload size in bytes.
470        size: u64,
471        /// Payload bytes (only at `headers+data` or `full` level).
472        payload: Option<Vec<u8>>,
473    },
474    /// Session FSM phase transition (event type 5).
475    StateChange {
476        /// Previous session phase.
477        from: String,
478        /// New session phase.
479        to: String,
480    },
481    /// Protocol or transport error (event type 6).
482    ///
483    /// The four optional fields are what makes the event evidence rather than
484    /// an assertion. A peer that sends something malformed is one of the few
485    /// things a shared trace is uniquely good for — the recording party can
486    /// see it and the sending party cannot — and until these existed the only
487    /// field in the format able to hold bytes was a control message's `"raw"`,
488    /// so a recorder wanting to keep the offending bytes had to record the
489    /// violation as a decodable message in order to have somewhere to put
490    /// them.
491    ///
492    /// Each is `None` when the recorder did not write it — and also when the
493    /// file carried the key with a value the field cannot hold, in which case
494    /// the entry stays verbatim in [`TraceEvent::extra`] instead of being read
495    /// here.
496    ///
497    /// [`error_observed`](EventData::error_observed) builds one of these from
498    /// bytes a recorder has just seen, which is where the cap on `raw` and the
499    /// detail levels the two byte-bearing fields sit at are applied.
500    Error {
501        /// Error code.
502        error_code: u64,
503        /// Human-readable reason.
504        reason: String,
505        /// QUIC stream the error was observed on, when there was one and the
506        /// recorder knows it.
507        ///
508        /// Optional on the same terms as a control message's `"sid"`: `None`
509        /// means there was no stream or none is known, which a reader must not
510        /// confuse with stream 0.
511        stream_id: Option<u64>,
512        /// What sort of failure this was.
513        kind: Option<ErrorKind>,
514        /// Byte length of the input the recorder held for this error, before
515        /// any truncation.
516        ///
517        /// Recorded from `headers+sizes` upwards, one level below the bytes
518        /// themselves: how large a malformed message was is often enough on
519        /// its own to tell a truncated message from a mistyped one, and it
520        /// carries no content. It is still a size, and this format gates sizes
521        /// deliberately, so it does not reach a `control`-level trace.
522        ///
523        /// Where both are present, a value larger than `raw`'s length is how a
524        /// reader learns the capture is partial and by how much. Where this is
525        /// absent, a `raw` of exactly [`ERROR_RAW_CAP`] bytes is the one length
526        /// the cap makes ambiguous and must be taken as possibly truncated.
527        raw_len: Option<u64>,
528        /// The offending bytes, at `full` detail only.
529        ///
530        /// Payload-bearing, and gated a level above the event's own
531        /// `control`+: an error naming a *data* stream has subgroup framing
532        /// and object payload behind it, so inheriting the event's level would
533        /// have put media into traces whose declared level excludes payloads
534        /// outright.
535        ///
536        /// A recorder caps this at [`ERROR_RAW_CAP`] bytes. A reader does not:
537        /// a longer one read from a file is neither shortened nor refused.
538        raw: Option<Vec<u8>>,
539    },
540    /// User-defined annotation (event type 7).
541    Annotation {
542        /// User-defined label.
543        label: String,
544        /// User-defined data (any CBOR type).
545        ///
546        /// Opaque: nothing here reads it, and it is written back saying what
547        /// it said. Its encoding is the writer's, though — see
548        /// [`TraceEvent::extra`].
549        data: Value,
550    },
551    /// A new peer session was established (event type 8).
552    ///
553    /// The identifier for the peer lives on [`TraceEvent::peer`]; every later
554    /// event for that peer repeats it.
555    PeerConnected {
556        /// Peer-reported endpoint URI or remote address, best-effort.
557        endpoint: Option<String>,
558        /// Transport type (`"webtransport"`, `"raw-quic"`, …).
559        transport: Option<String>,
560        /// Role at connection time, if known then.
561        role: Option<PeerRole>,
562        /// Which end of the relay the peer connected from.
563        side: Option<Side>,
564    },
565    /// A peer session ended (event type 9).
566    PeerDisconnected {
567        /// Error or close code (0 = clean close).
568        error_code: u64,
569        /// Human-readable reason.
570        reason: Option<String>,
571    },
572    /// An upstream subscription was created or extended in causal response to
573    /// downstream ones (event type 10).
574    ///
575    /// This is the primitive multi-hop correlation is built from: a collector
576    /// reconstructs an end-to-end tree from these links plus the trace IDs
577    /// propagated along them.
578    ///
579    /// The four timestamps share the `timestamp` field's timebase — the
580    /// emitting source's own clock. Differences between them are therefore
581    /// meaningful with no cross-hop clock agreement, which is the point: they
582    /// measure how long *this* hop took. Never subtract one source's
583    /// timestamp from another's.
584    ///
585    /// A source emits the event as soon as the downstream SUBSCRIBE arrives,
586    /// carrying whichever timestamps it has, and may emit it again for the
587    /// same pair as the rest arrive. A consumer treats the later event as an
588    /// update to the earlier one, not as a second derivation.
589    SubscriptionDerivation {
590        /// The upstream subscription.
591        upstream: SubscriptionRef,
592        /// Every downstream subscription currently served by `upstream`.
593        downstream: Vec<SubscriptionRef>,
594        /// Whether the upstream sub was created here or was already running.
595        kind: DerivationKind,
596        /// Trace ID propagated along the subscription chain, carried as raw
597        /// bytes at every layer so two implementations produce identical
598        /// values for the same chain.
599        trace_id: Option<[u8; TRACE_ID_LEN]>,
600        /// Track namespace the subscription targets, one entry per field.
601        namespace: Option<Vec<Vec<u8>>>,
602        /// Track name the subscription targets.
603        track_name: Option<Vec<u8>>,
604        /// When the downstream SUBSCRIBE was received.
605        t_downstream_received: Option<i64>,
606        /// When the upstream SUBSCRIBE was transmitted. Absent for terminal
607        /// subscriptions, where this source is the content origin.
608        t_upstream_sent: Option<i64>,
609        /// When SUBSCRIBE_OK was received from upstream. Absent for terminal
610        /// or still-in-flight subscriptions.
611        t_upstream_ok_received: Option<i64>,
612        /// When SUBSCRIBE_OK was transmitted downstream. Absent while the
613        /// subscription is still in flight.
614        t_downstream_ok_sent: Option<i64>,
615    },
616    /// An event whose type this version of the crate does not know.
617    ///
618    /// New event types may be added without a format version bump, so a
619    /// reader that rejected them would turn every future addition into a
620    /// breaking change. The fields are kept verbatim, which means an unknown
621    /// event survives a read-modify-write round trip intact — intact in what
622    /// it says, since the encoding written back is this crate's own. See
623    /// [`TraceEvent::extra`].
624    Unknown {
625        /// The event type discriminant that was read.
626        event_type: u64,
627        /// Every key and value from the event map other than `"n"`, `"t"`
628        /// and `"e"` — and other than `"p"`, when a peer was read from it. A
629        /// `"p"` that is not text is kept here like any other value this crate
630        /// could not use.
631        fields: Vec<(Value, Value)>,
632    },
633}
634
635/// Where `detail` sits in the levels' ordering, or `None` for a level this
636/// crate cannot place.
637///
638/// The levels are a chain — each records everything the one before it does —
639/// so one rank answers every question about which of two a recorder is at.
640/// A level this crate has never heard of has no place in the chain, which is
641/// why this is an `Option` rather than a number with a default: a default
642/// would guess, and the two ways of guessing wrong are not symmetric.
643fn detail_rank(detail: &DetailLevel) -> Option<u8> {
644    match detail {
645        DetailLevel::Control => Some(0),
646        DetailLevel::Headers => Some(1),
647        DetailLevel::HeadersSizes => Some(2),
648        DetailLevel::HeadersData => Some(3),
649        DetailLevel::Full => Some(4),
650        DetailLevel::Other(_) => None,
651    }
652}
653
654/// Whether a recorder at `detail` records everything a recorder at `floor`
655/// does.
656///
657/// `false` when either level is one this crate cannot place. A future level
658/// might sit above `floor` or below it, and answering `true` on a guess would
659/// put payload bytes into a trace whose declared level excludes them — a loss
660/// of one diagnostic against a leak that cannot be taken back.
661fn detail_reaches(detail: &DetailLevel, floor: &DetailLevel) -> bool {
662    match (detail_rank(detail), detail_rank(floor)) {
663        (Some(level), Some(floor)) => level >= floor,
664        _ => false,
665    }
666}
667
668impl EventData {
669    /// An [`Error`](EventData::Error) built from the bytes a recorder has just
670    /// observed, for a trace recorded at `detail`.
671    ///
672    /// This is where the cap on `"raw"` belongs and the only place this crate
673    /// applies it: the party constructing an event out of traffic it just saw
674    /// is the one SPEC.md addresses, and it is the only party that can tell a
675    /// fresh event from one that arrived by being read. Serializing does not
676    /// cap, and reading does not refuse — see [`ERROR_RAW_CAP`].
677    ///
678    /// `observed` is the whole input the recorder held, uncapped. What comes
679    /// back depends on `detail`, because the two byte-bearing fields sit at
680    /// different levels and that is deliberate:
681    ///
682    /// - `raw_len` is the full length of `observed`, from `headers+sizes`
683    ///   upwards. It is a size, and sizes are gated; it is not gated *with*
684    ///   the bytes, so it is available in every trace where the bytes must not
685    ///   appear, which is the point of having it.
686    /// - `raw` is the first [`ERROR_RAW_CAP`] bytes of `observed`, at `full`
687    ///   only. Below that level nothing is copied.
688    ///
689    /// A level this crate cannot place yields neither.
690    ///
691    /// Where both come back, comparing them is how a reader learns the capture
692    /// was truncated: `raw_len` is the length before the cap bit, not after.
693    pub fn error_observed(
694        error_code: u64,
695        reason: impl Into<String>,
696        kind: Option<ErrorKind>,
697        stream_id: Option<u64>,
698        observed: &[u8],
699        detail: &DetailLevel,
700    ) -> Self {
701        let raw_len = detail_reaches(detail, &DetailLevel::HeadersSizes)
702            .then(|| u64::try_from(observed.len()).unwrap_or(u64::MAX));
703        let raw = detail_reaches(detail, &DetailLevel::Full)
704            .then(|| observed[..observed.len().min(ERROR_RAW_CAP)].to_vec());
705        EventData::Error { error_code, reason: reason.into(), stream_id, kind, raw_len, raw }
706    }
707}
708
709impl TraceEvent {
710    /// An event with no peer identifier — the single-session case.
711    pub fn new(seq: u64, timestamp: i64, data: EventData) -> Self {
712        TraceEvent { seq, timestamp, peer: None, data, extra: Vec::new() }
713    }
714
715    /// An event attributed to `peer` — the relay-tap case.
716    pub fn for_peer(seq: u64, timestamp: i64, peer: impl Into<String>, data: EventData) -> Self {
717        TraceEvent { seq, timestamp, peer: Some(peer.into()), data, extra: Vec::new() }
718    }
719
720    /// Attach unrecognised keys, for a caller reconstructing an event it did
721    /// not decode itself.
722    ///
723    /// Keys that collide with ones the event writes from its own fields are
724    /// dropped on serialization rather than written twice, since a CBOR map
725    /// with a repeated key is malformed and the event's own value is the one
726    /// the reader would have produced. A key the event's type merely *defines*
727    /// does not collide: an optional field holding `None` writes nothing, so
728    /// the entry here is the only copy of that key and is written.
729    #[must_use]
730    pub fn with_extra(mut self, extra: Vec<(Value, Value)>) -> Self {
731        self.extra = extra;
732        self
733    }
734
735    /// The event type discriminant this event serializes as.
736    pub fn event_type(&self) -> u64 {
737        match &self.data {
738            EventData::ControlMessage { .. } => EVENT_CONTROL_MESSAGE,
739            EventData::StreamOpened { .. } => EVENT_STREAM_OPENED,
740            EventData::StreamClosed { .. } => EVENT_STREAM_CLOSED,
741            EventData::ObjectHeader { .. } => EVENT_OBJECT_HEADER,
742            EventData::ObjectPayload { .. } => EVENT_OBJECT_PAYLOAD,
743            EventData::StateChange { .. } => EVENT_STATE_CHANGE,
744            EventData::Error { .. } => EVENT_ERROR,
745            EventData::Annotation { .. } => EVENT_ANNOTATION,
746            EventData::PeerConnected { .. } => EVENT_PEER_CONNECTED,
747            EventData::PeerDisconnected { .. } => EVENT_PEER_DISCONNECTED,
748            EventData::SubscriptionDerivation { .. } => EVENT_SUBSCRIPTION_DERIVATION,
749            EventData::Unknown { event_type, .. } => *event_type,
750        }
751    }
752
753    /// Extract the `request_id` from a control message's decoded `"msg"`
754    /// field, if present.
755    ///
756    /// Returns `None` for non-control-message events, for a `"msg"` that is
757    /// not a map, and for a map that names the field something else. Drafts 07
758    /// through 10 call it `subscribe_id`, and this does not answer for them:
759    /// the two names sit on different messages with different meanings, and a
760    /// reader that wants either can ask the map itself.
761    ///
762    /// The key is `request_id`, in the snake_case the drafts use, because that
763    /// is what every writer of these files produces. It read `requestId` until
764    /// draft-20, and matched nothing — not this crate's own corpus, and not a
765    /// trace written by any other implementation.
766    pub fn request_id(&self) -> Option<u64> {
767        if let EventData::ControlMessage { message: Value::Map(ref pairs), .. } = self.data {
768            for (k, v) in pairs {
769                if k.as_text() == Some("request_id") {
770                    return v.as_integer().and_then(|i| u64::try_from(i).ok());
771                }
772            }
773        }
774        None
775    }
776
777    /// Return the message type for control message events.
778    pub fn message_type(&self) -> Option<u64> {
779        if let EventData::ControlMessage { message_type, .. } = self.data {
780            Some(message_type)
781        } else {
782            None
783        }
784    }
785
786    /// Return the direction for events that have one.
787    pub fn direction(&self) -> Option<Direction> {
788        match &self.data {
789            EventData::ControlMessage { direction, .. }
790            | EventData::StreamOpened { direction, .. } => Some(*direction),
791            _ => None,
792        }
793    }
794}
795
796// ── CBOR conversion ────────────────────────────────────────
797
798/// Wrapper that serializes as a CBOR byte string (major type 2). Without this
799/// wrapper, serializing `&[u8]` via generic `Serialize` produces a CBOR array
800/// of u8.
801struct ByteStr<'a>(&'a [u8]);
802
803impl serde::Serialize for ByteStr<'_> {
804    #[inline]
805    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
806        s.serialize_bytes(self.0)
807    }
808}
809
810impl Direction {
811    #[inline]
812    fn to_u64(self) -> u64 {
813        match self {
814            Direction::Send => 0,
815            Direction::Receive => 1,
816        }
817    }
818}
819
820impl StreamType {
821    #[inline]
822    fn to_u64(self) -> u64 {
823        match self {
824            StreamType::Subgroup => 0,
825            StreamType::Datagram => 1,
826            StreamType::Fetch => 2,
827        }
828    }
829}
830
831/// Build a CBOR [`Value`] from a [`TraceEvent`] by running it through the
832/// crate's `Serialize` impl. Convenience for tests and inspection — the
833/// hot write path in [`MoqTraceWriter`](crate::writer::MoqTraceWriter) uses
834/// `Serialize` directly and never materializes a `Value`.
835impl From<&TraceEvent> for Value {
836    fn from(event: &TraceEvent) -> Self {
837        Value::serialized(event).expect("TraceEvent serialization is infallible")
838    }
839}
840
841impl serde::Serialize for TraceEvent {
842    fn serialize<S: serde::Serializer>(&self, ser: S) -> Result<S::Ok, S::Error> {
843        use serde::ser::SerializeMap;
844
845        // Every value an event carries out of a file rather than out of a
846        // typed field goes out through `normalised`: an unknown event type's
847        // fields and the store here, a control message's `"msg"` and an
848        // annotation's `"data"` below. The two encoding rules bind the bytes a
849        // writer emits and so bind every value it emits, the ones it never
850        // looked at included — SPEC.md, Interoperability. A typed field needs
851        // none of this and gets none: `"raw"`, `"pl"`, `"traceId"`, `"tn"` and
852        // every field of `"ns"` are `Vec<u8>` by the time they reach here and
853        // go out as major type 2 by construction, and every integer field goes
854        // out as a CBOR integer for the same reason.
855        //
856        // Filtered against the common fields on the way out, the same way the
857        // store below is filtered against every field the event writes: an
858        // unknown event type writes `"n"`, `"t"`, `"e"` and `"p"` from the
859        // common fields, so an entry here naming one of those is dropped
860        // rather than putting that key in the map twice. A caller can put one
861        // in this list by hand, and a file that repeated a common key leaves
862        // one here too — that is where the entry `seq` did not take now lives.
863        let unknown_fields = match &self.data {
864            EventData::Unknown { fields, .. } => {
865                store_entries(fields, |key| writes_common_key(self.peer.as_deref(), key))
866            }
867            _ => Vec::new(),
868        };
869
870        // A key this event writes from one of its own fields is written from
871        // there, so an `extra` entry naming it is dropped rather than written
872        // twice: a CBOR map with a duplicate key is malformed, and the field
873        // is what a reader produced. A key the type merely *defines* is not
874        // one of those — an optional field holding `None` writes nothing, and
875        // the entry here is then a value the decode could not use, which has
876        // to go back out.
877        //
878        // The same pass the header's stores go through, for the same three
879        // reasons: the filter above, the encoding rules, and no key twice.
880        let extra = store_entries(&self.extra, |key| {
881            writes_common_key(self.peer.as_deref(), key) || writes_variant_key(&self.data, key)
882        });
883
884        // Pre-count entries so the CBOR map is encoded with a definite length.
885        let variant_entries = match &self.data {
886            EventData::ControlMessage { stream_id, raw, .. } => {
887                3 + usize::from(stream_id.is_some()) + usize::from(raw.is_some())
888            }
889            EventData::StreamOpened {
890                track_alias,
891                subgroup_id,
892                fetch_request_id,
893                group_id,
894                header_raw,
895                ..
896            } => {
897                3 + usize::from(track_alias.is_some())
898                    + usize::from(subgroup_id.is_some())
899                    + usize::from(fetch_request_id.is_some())
900                    + usize::from(group_id.is_some())
901                    + usize::from(header_raw.is_some())
902            }
903            EventData::StreamClosed { .. } => 2,
904            EventData::ObjectHeader { .. } => 5,
905            EventData::ObjectPayload { payload, .. } => 4 + usize::from(payload.is_some()),
906            EventData::StateChange { .. } => 2,
907            EventData::Error { stream_id, kind, raw_len, raw, .. } => {
908                2 + usize::from(stream_id.is_some())
909                    + usize::from(kind.is_some())
910                    + usize::from(raw_len.is_some())
911                    + usize::from(raw.is_some())
912            }
913            EventData::Annotation { .. } => 2,
914            EventData::PeerConnected { endpoint, transport, role, side } => {
915                usize::from(endpoint.is_some())
916                    + usize::from(transport.is_some())
917                    + usize::from(role.is_some())
918                    + usize::from(side.is_some())
919            }
920            EventData::PeerDisconnected { reason, .. } => 1 + usize::from(reason.is_some()),
921            EventData::SubscriptionDerivation {
922                trace_id,
923                namespace,
924                track_name,
925                t_downstream_received,
926                t_upstream_sent,
927                t_upstream_ok_received,
928                t_downstream_ok_sent,
929                ..
930            } => {
931                3 + usize::from(trace_id.is_some())
932                    + usize::from(namespace.is_some())
933                    + usize::from(track_name.is_some())
934                    + usize::from(t_downstream_received.is_some())
935                    + usize::from(t_upstream_sent.is_some())
936                    + usize::from(t_upstream_ok_received.is_some())
937                    + usize::from(t_downstream_ok_sent.is_some())
938            }
939            EventData::Unknown { .. } => unknown_fields.len(),
940        };
941
942        let entries = 3 /* n, t, e */
943            + usize::from(self.peer.is_some())
944            + variant_entries
945            + extra.len();
946
947        let mut map = ser.serialize_map(Some(entries))?;
948        map.serialize_entry("n", &self.seq)?;
949        map.serialize_entry("t", &self.timestamp)?;
950        if let Some(ref peer) = self.peer {
951            map.serialize_entry("p", peer)?;
952        }
953
954        match &self.data {
955            EventData::ControlMessage { direction, message_type, message, stream_id, raw } => {
956                map.serialize_entry("e", &EVENT_CONTROL_MESSAGE)?;
957                map.serialize_entry("d", &direction.to_u64())?;
958                map.serialize_entry("mt", message_type)?;
959                map.serialize_entry("msg", &normalised(message))?;
960                if let Some(sid) = stream_id {
961                    map.serialize_entry("sid", sid)?;
962                }
963                if let Some(raw) = raw {
964                    map.serialize_entry("raw", &ByteStr(raw))?;
965                }
966            }
967            EventData::StreamOpened {
968                stream_id,
969                direction,
970                stream_type,
971                track_alias,
972                subgroup_id,
973                fetch_request_id,
974                group_id,
975                header_raw,
976            } => {
977                map.serialize_entry("e", &EVENT_STREAM_OPENED)?;
978                map.serialize_entry("sid", stream_id)?;
979                map.serialize_entry("d", &direction.to_u64())?;
980                map.serialize_entry("st", &stream_type.to_u64())?;
981                if let Some(ta) = track_alias {
982                    map.serialize_entry("ta", ta)?;
983                }
984                if let Some(sg) = subgroup_id {
985                    map.serialize_entry("sg", sg)?;
986                }
987                if let Some(fri) = fetch_request_id {
988                    map.serialize_entry("fri", fri)?;
989                }
990                if let Some(g) = group_id {
991                    map.serialize_entry("g", g)?;
992                }
993                if let Some(hdr) = header_raw {
994                    map.serialize_entry("hdr", &ByteStr(hdr))?;
995                }
996            }
997            EventData::StreamClosed { stream_id, error_code } => {
998                map.serialize_entry("e", &EVENT_STREAM_CLOSED)?;
999                map.serialize_entry("sid", stream_id)?;
1000                map.serialize_entry("ec", error_code)?;
1001            }
1002            EventData::ObjectHeader {
1003                stream_id,
1004                group,
1005                object,
1006                publisher_priority,
1007                object_status,
1008            } => {
1009                map.serialize_entry("e", &EVENT_OBJECT_HEADER)?;
1010                map.serialize_entry("sid", stream_id)?;
1011                map.serialize_entry("g", group)?;
1012                map.serialize_entry("o", object)?;
1013                map.serialize_entry("pp", publisher_priority)?;
1014                map.serialize_entry("os", object_status)?;
1015            }
1016            EventData::ObjectPayload { stream_id, group, object, size, payload } => {
1017                map.serialize_entry("e", &EVENT_OBJECT_PAYLOAD)?;
1018                map.serialize_entry("sid", stream_id)?;
1019                map.serialize_entry("g", group)?;
1020                map.serialize_entry("o", object)?;
1021                map.serialize_entry("sz", size)?;
1022                if let Some(pl) = payload {
1023                    map.serialize_entry("pl", &ByteStr(pl))?;
1024                }
1025            }
1026            EventData::StateChange { from, to } => {
1027                map.serialize_entry("e", &EVENT_STATE_CHANGE)?;
1028                map.serialize_entry("from", from)?;
1029                map.serialize_entry("to", to)?;
1030            }
1031            EventData::Error { error_code, reason, stream_id, kind, raw_len, raw } => {
1032                map.serialize_entry("e", &EVENT_ERROR)?;
1033                map.serialize_entry("ec", error_code)?;
1034                map.serialize_entry("reason", reason)?;
1035                if let Some(sid) = stream_id {
1036                    map.serialize_entry("sid", sid)?;
1037                }
1038                if let Some(kind) = kind {
1039                    map.serialize_entry("ek", kind.as_str())?;
1040                }
1041                if let Some(raw_len) = raw_len {
1042                    map.serialize_entry("rawlen", raw_len)?;
1043                }
1044                // At whatever length it arrived. The cap is the recorder's,
1045                // and re-applying it here would shorten evidence read out of
1046                // somebody else's file — see [`ERROR_RAW_CAP`].
1047                if let Some(raw) = raw {
1048                    map.serialize_entry("raw", &ByteStr(raw))?;
1049                }
1050            }
1051            EventData::Annotation { label, data } => {
1052                map.serialize_entry("e", &EVENT_ANNOTATION)?;
1053                map.serialize_entry("label", label)?;
1054                map.serialize_entry("data", &normalised(data))?;
1055            }
1056            EventData::PeerConnected { endpoint, transport, role, side } => {
1057                map.serialize_entry("e", &EVENT_PEER_CONNECTED)?;
1058                if let Some(endpoint) = endpoint {
1059                    map.serialize_entry("endpoint", endpoint)?;
1060                }
1061                if let Some(transport) = transport {
1062                    map.serialize_entry("transport", transport)?;
1063                }
1064                if let Some(role) = role {
1065                    map.serialize_entry("role", role.as_str())?;
1066                }
1067                if let Some(side) = side {
1068                    map.serialize_entry("side", side.as_str())?;
1069                }
1070            }
1071            EventData::PeerDisconnected { error_code, reason } => {
1072                map.serialize_entry("e", &EVENT_PEER_DISCONNECTED)?;
1073                map.serialize_entry("ec", error_code)?;
1074                if let Some(reason) = reason {
1075                    map.serialize_entry("reason", reason)?;
1076                }
1077            }
1078            EventData::SubscriptionDerivation {
1079                upstream,
1080                downstream,
1081                kind,
1082                trace_id,
1083                namespace,
1084                track_name,
1085                t_downstream_received,
1086                t_upstream_sent,
1087                t_upstream_ok_received,
1088                t_downstream_ok_sent,
1089            } => {
1090                map.serialize_entry("e", &EVENT_SUBSCRIPTION_DERIVATION)?;
1091                map.serialize_entry("u", &upstream.to_value())?;
1092                let downstream_refs: Vec<Value> =
1093                    downstream.iter().map(SubscriptionRef::to_value).collect();
1094                map.serialize_entry("d", &Value::Array(downstream_refs))?;
1095                map.serialize_entry("kind", kind.as_str())?;
1096                if let Some(trace_id) = trace_id {
1097                    map.serialize_entry("traceId", &ByteStr(trace_id))?;
1098                }
1099                if let Some(namespace) = namespace {
1100                    let fields: Vec<Value> =
1101                        namespace.iter().map(|f| Value::Bytes(f.clone())).collect();
1102                    map.serialize_entry("ns", &Value::Array(fields))?;
1103                }
1104                if let Some(track_name) = track_name {
1105                    map.serialize_entry("tn", &ByteStr(track_name))?;
1106                }
1107                if let Some(t) = t_downstream_received {
1108                    map.serialize_entry("tdr", t)?;
1109                }
1110                if let Some(t) = t_upstream_sent {
1111                    map.serialize_entry("tus", t)?;
1112                }
1113                if let Some(t) = t_upstream_ok_received {
1114                    map.serialize_entry("tuo", t)?;
1115                }
1116                if let Some(t) = t_downstream_ok_sent {
1117                    map.serialize_entry("tdo", t)?;
1118                }
1119            }
1120            EventData::Unknown { event_type, .. } => {
1121                map.serialize_entry("e", event_type)?;
1122                for (k, v) in &unknown_fields {
1123                    map.serialize_entry(k, v)?;
1124                }
1125            }
1126        }
1127
1128        // Last, so the event's own keys keep the positions a reader expects
1129        // and the file stays diffable against one written without them.
1130        for (k, v) in &extra {
1131            map.serialize_entry(k, v)?;
1132        }
1133
1134        map.end()
1135    }
1136}
1137
1138// ── decoding helpers ───────────────────────────────────────
1139
1140/// Helper to extract a u64 from a CBOR map by key.
1141///
1142/// Accepts the float form an encoder may use for a value past 32 bits — see
1143/// [`as_u64`](crate::header::as_u64).
1144fn get_uint(pairs: &[(Value, Value)], key: &str) -> Option<u64> {
1145    pairs.iter().find_map(|(k, v)| if k.as_text() == Some(key) { as_u64(v) } else { None })
1146}
1147
1148/// Helper to extract an i64 from a CBOR map by key. See [`get_uint`].
1149fn get_int(pairs: &[(Value, Value)], key: &str) -> Option<i64> {
1150    pairs.iter().find_map(|(k, v)| if k.as_text() == Some(key) { as_i64(v) } else { None })
1151}
1152
1153/// Helper to extract a text string from a CBOR map by key.
1154fn get_text(pairs: &[(Value, Value)], key: &str) -> Option<String> {
1155    pairs.iter().find_map(|(k, v)| {
1156        if k.as_text() == Some(key) {
1157            v.as_text().map(|s| s.to_string())
1158        } else {
1159            None
1160        }
1161    })
1162}
1163
1164/// Helper to extract a value from a CBOR map by key.
1165fn get_value(pairs: &[(Value, Value)], key: &str) -> Option<Value> {
1166    pairs.iter().find_map(|(k, v)| if k.as_text() == Some(key) { Some(v.clone()) } else { None })
1167}
1168
1169/// Tag 64 marks a byte string as an array of unsigned 8-bit integers.
1170const TAG_UINT8_ARRAY: u64 = 64;
1171
1172/// Read a byte string, unwrapping the typed-array tag some encoders add.
1173///
1174/// A byte string may arrive plain or wrapped in tag 64, and one of the two
1175/// implementations of this format wraps every one it writes. Reading only the
1176/// plain form meant every payload-bearing field that encoder produced — raw
1177/// wire bytes, object payloads, track names, trace ids — read as absent, and
1178/// a payload-bearing field that reads as absent is indistinguishable from one
1179/// the recorder chose not to capture.
1180fn as_bytes(value: &Value) -> Option<Vec<u8>> {
1181    match value {
1182        Value::Bytes(b) => Some(b.clone()),
1183        Value::Tag(TAG_UINT8_ARRAY, inner) => match inner.as_ref() {
1184            Value::Bytes(b) => Some(b.clone()),
1185            _ => None,
1186        },
1187        _ => None,
1188    }
1189}
1190
1191/// Helper to extract byte string from a CBOR map by key.
1192fn get_bytes(pairs: &[(Value, Value)], key: &str) -> Option<Vec<u8>> {
1193    pairs.iter().find_map(|(k, v)| if k.as_text() == Some(key) { as_bytes(v) } else { None })
1194}
1195
1196/// Read a control message's decoded `"msg"` field.
1197///
1198/// An absent key reads as the empty map a writer should have written, not as
1199/// an error. The format requires a writer with nothing decoded — an
1200/// unparseable message type, or a recorder not decoding bodies at all — to
1201/// emit `{}` rather than omit the key, but requires a reader to be more
1202/// tolerant still, and this is why: event 0 is one of the types sampling MUST
1203/// NOT drop, so refusing it over a missing `"msg"` discards exactly the events
1204/// the format promises to keep.
1205///
1206/// How much is lost depends on how the caller drives the reader, and the worse
1207/// case is the idiomatic one. [`MoqTraceReader::read_next`] returns the error,
1208/// so `collect::<Result<Vec<_>, _>>()` — what this crate's own tests use —
1209/// stops at the first offending event and yields none of the ones after it. A
1210/// caller that skips errors and continues loses only the offending events;
1211/// `moqtap trace` does that, and prints each one, so the loss was at least
1212/// visible there.
1213///
1214/// A value that is present is kept whatever its type. Recordings predating
1215/// the rule carry a text rendering of the message here — every `capture-*`
1216/// case in the conformance corpus is such a file — and they stay readable,
1217/// with the field simply not addressable by key.
1218fn get_message(pairs: &[(Value, Value)]) -> Value {
1219    get_value(pairs, "msg").unwrap_or_else(|| Value::Map(Vec::new()))
1220}
1221
1222fn require_uint(pairs: &[(Value, Value)], key: &str) -> Result<u64, MoqTraceError> {
1223    get_uint(pairs, key).ok_or_else(|| MoqTraceError::InvalidEvent(format!("missing '{key}'")))
1224}
1225
1226fn require_int(pairs: &[(Value, Value)], key: &str) -> Result<i64, MoqTraceError> {
1227    get_int(pairs, key).ok_or_else(|| MoqTraceError::InvalidEvent(format!("missing '{key}'")))
1228}
1229
1230fn require_text(pairs: &[(Value, Value)], key: &str) -> Result<String, MoqTraceError> {
1231    get_text(pairs, key).ok_or_else(|| MoqTraceError::InvalidEvent(format!("missing '{key}'")))
1232}
1233
1234fn require_value(pairs: &[(Value, Value)], key: &str) -> Result<Value, MoqTraceError> {
1235    get_value(pairs, key).ok_or_else(|| MoqTraceError::InvalidEvent(format!("missing '{key}'")))
1236}
1237
1238fn require_direction(pairs: &[(Value, Value)], key: &str) -> Result<Direction, MoqTraceError> {
1239    let v = require_value(pairs, key)?;
1240    Direction::from_cbor(&v)
1241}
1242
1243/// Read a trace ID, refusing any length but the one the format fixes.
1244///
1245/// A wrong-length value is malformed rather than something to pad or
1246/// truncate: the identifier exists so that two implementations independently
1247/// derive byte-identical values for one subscription chain, and a reader that
1248/// quietly reshapes it breaks exactly the property it is there for.
1249fn get_trace_id(pairs: &[(Value, Value)]) -> Result<Option<[u8; TRACE_ID_LEN]>, MoqTraceError> {
1250    let Some(bytes) = get_bytes(pairs, "traceId") else {
1251        return Ok(None);
1252    };
1253    <[u8; TRACE_ID_LEN]>::try_from(bytes.as_slice()).map(Some).map_err(|_| {
1254        MoqTraceError::InvalidEvent(format!(
1255            "'traceId' is {} bytes, must be {TRACE_ID_LEN}",
1256            bytes.len()
1257        ))
1258    })
1259}
1260
1261/// Read a track namespace: an array of byte strings, tolerating text entries
1262/// from encoders that lose the distinction.
1263///
1264/// Answers `None` for anything else — a `"ns"` that is not an array, or an
1265/// array holding something that is neither bytes nor text. `"ns"` is optional,
1266/// and the rule for an optional key whose value this reader cannot use is that
1267/// the key is kept as an unrecognised one rather than that the event dies; the
1268/// caller then still has the derivation's upstream and downstream
1269/// subscriptions, which are what make it a derivation at all.
1270///
1271/// An error here would cost more than the field. `read_next` propagates it,
1272/// so `collect::<Result<Vec<_>, _>>()` — the documented idiom — would stop at
1273/// that event and yield none of the ones after it: one namespace an encoder
1274/// wrote oddly would take the rest of the recording with it.
1275fn get_namespace(pairs: &[(Value, Value)]) -> Option<Vec<Vec<u8>>> {
1276    let Some(Value::Array(items)) = get_value(pairs, "ns") else {
1277        return None;
1278    };
1279    let mut namespace = Vec::with_capacity(items.len());
1280    for item in items {
1281        let field = as_bytes(&item).or_else(|| item.as_text().map(|t| t.as_bytes().to_vec()))?;
1282        namespace.push(field);
1283    }
1284    Some(namespace)
1285}
1286
1287/// The three keys every event carries whatever its type. All are required, so
1288/// a value the reader cannot use is a malformed event rather than something to
1289/// keep: there would be no event left to hang the kept key on.
1290const REQUIRED_COMMON_KEYS: [&str; 3] = ["n", "t", "e"];
1291
1292/// Whether the common event fields account for `key` — the required three
1293/// always, and `"p"` only when a peer was actually read from it.
1294///
1295/// `"p"` is optional, and optional means it can also be *unusable*: a `"p"`
1296/// that is not text leaves [`TraceEvent::peer`] `None`, and the entry then
1297/// belongs in [`TraceEvent::extra`] like any other value the reader could not
1298/// use. Excluding the key unconditionally would delete it instead.
1299fn writes_common_key(peer: Option<&str>, key: &str) -> bool {
1300    REQUIRED_COMMON_KEYS.contains(&key) || (key == "p" && peer.is_some())
1301}
1302
1303/// Whether `key` is written from one of `data`'s own fields — equivalently,
1304/// on an event that was decoded, whether the decode used it. The common keys
1305/// are not this function's business; [`writes_common_key`] answers for those.
1306///
1307/// Deliberately not "does this event type define `key`". The two part company
1308/// on a defined key whose value the decode could not use: SPEC.md treats such
1309/// a key as unrecognised, so its field stays `None` and the entry goes to
1310/// [`TraceEvent::extra`], from where the serializer writes it back unchanged.
1311/// Asking about the type's whole vocabulary instead keeps the key out of
1312/// `extra` while no field holds it either, and merely reading the file deletes
1313/// the value — which is how adding `"ta"`, `"sg"`, `"fri"` and `"g"` to event
1314/// 1 made this reader preserve *less* than while it had never heard of them.
1315///
1316/// The match is exhaustive on purpose: a new variant does not compile until
1317/// its keys are listed, and both ways of getting a list wrong are silent. A
1318/// key left out is written twice, once from the field and once from `extra`,
1319/// and a CBOR map with a duplicate key is malformed; a key wrongly listed is
1320/// dropped from every rewrite.
1321fn writes_variant_key(data: &EventData, key: &str) -> bool {
1322    /// `always` are the keys the variant writes unconditionally; `optional`
1323    /// pairs each of the rest with whether its field holds a value.
1324    fn among(key: &str, always: &[&str], optional: &[(&str, bool)]) -> bool {
1325        always.contains(&key) || optional.iter().any(|&(k, written)| written && k == key)
1326    }
1327
1328    match data {
1329        EventData::ControlMessage { stream_id, raw, .. } => {
1330            among(key, &["d", "mt", "msg"], &[("sid", stream_id.is_some()), ("raw", raw.is_some())])
1331        }
1332        EventData::StreamOpened {
1333            track_alias,
1334            subgroup_id,
1335            fetch_request_id,
1336            group_id,
1337            header_raw,
1338            ..
1339        } => among(
1340            key,
1341            &["sid", "d", "st"],
1342            &[
1343                ("ta", track_alias.is_some()),
1344                ("sg", subgroup_id.is_some()),
1345                ("fri", fetch_request_id.is_some()),
1346                ("g", group_id.is_some()),
1347                ("hdr", header_raw.is_some()),
1348            ],
1349        ),
1350        EventData::StreamClosed { .. } => among(key, &["sid", "ec"], &[]),
1351        EventData::ObjectHeader { .. } => among(key, &["sid", "g", "o", "pp", "os"], &[]),
1352        EventData::ObjectPayload { payload, .. } => {
1353            among(key, &["sid", "g", "o", "sz"], &[("pl", payload.is_some())])
1354        }
1355        EventData::StateChange { .. } => among(key, &["from", "to"], &[]),
1356        EventData::Error { stream_id, kind, raw_len, raw, .. } => among(
1357            key,
1358            &["ec", "reason"],
1359            &[
1360                ("sid", stream_id.is_some()),
1361                ("ek", kind.is_some()),
1362                ("rawlen", raw_len.is_some()),
1363                ("raw", raw.is_some()),
1364            ],
1365        ),
1366        EventData::Annotation { .. } => among(key, &["label", "data"], &[]),
1367        EventData::PeerConnected { endpoint, transport, role, side } => among(
1368            key,
1369            &[],
1370            &[
1371                ("endpoint", endpoint.is_some()),
1372                ("transport", transport.is_some()),
1373                ("role", role.is_some()),
1374                ("side", side.is_some()),
1375            ],
1376        ),
1377        EventData::PeerDisconnected { reason, .. } => {
1378            among(key, &["ec"], &[("reason", reason.is_some())])
1379        }
1380        EventData::SubscriptionDerivation {
1381            trace_id,
1382            namespace,
1383            track_name,
1384            t_downstream_received,
1385            t_upstream_sent,
1386            t_upstream_ok_received,
1387            t_downstream_ok_sent,
1388            ..
1389        } => among(
1390            key,
1391            &["u", "d", "kind"],
1392            &[
1393                ("traceId", trace_id.is_some()),
1394                ("ns", namespace.is_some()),
1395                ("tn", track_name.is_some()),
1396                ("tdr", t_downstream_received.is_some()),
1397                ("tus", t_upstream_sent.is_some()),
1398                ("tuo", t_upstream_ok_received.is_some()),
1399                ("tdo", t_downstream_ok_sent.is_some()),
1400            ],
1401        ),
1402        // An unknown event type writes every key it read straight back out of
1403        // `fields`, so those are the ones an `extra` entry would duplicate.
1404        // Decoding one collects nothing into `extra` — see the `Unknown` arm
1405        // below — but a caller may have attached some by hand.
1406        EventData::Unknown { fields, .. } => fields.iter().any(|(k, _)| k.as_text() == Some(key)),
1407    }
1408}
1409
1410/// Every entry on `pairs` that the event decoded from it does not write back
1411/// out of a field of its own. See [`unrecognised`], which the header uses for
1412/// the same reason.
1413///
1414/// One difference from the header's use of it is worth naming, because it
1415/// decides *which* entry of a duplicate pair survives. A getter here searches
1416/// for the first entry it can **use**, where the header's lookup returns the
1417/// first entry for a key whatever it holds. So on a file repeating a key with
1418/// two different types — `"sid": "x"` and then `"sid": 9` — the field takes
1419/// the second and the walk drops the first. Without this walk both are lost,
1420/// and SPEC.md leaves readers free to disagree over which of a
1421/// duplicate pair wins, so nothing may depend on which one it is.
1422fn unrecognised_keys(
1423    pairs: &[(Value, Value)],
1424    peer: Option<&str>,
1425    data: &EventData,
1426) -> Vec<(Value, Value)> {
1427    unrecognised(pairs, |key| writes_common_key(peer, key) || writes_variant_key(data, key))
1428}
1429
1430impl TryFrom<Value> for TraceEvent {
1431    type Error = MoqTraceError;
1432
1433    fn try_from(value: Value) -> Result<Self, MoqTraceError> {
1434        let pairs = match value {
1435            Value::Map(pairs) => pairs,
1436            _ => return Err(MoqTraceError::InvalidEvent("event is not a CBOR map".into())),
1437        };
1438
1439        let seq = require_uint(&pairs, "n")?;
1440        let timestamp = require_int(&pairs, "t")?;
1441        let peer = get_text(&pairs, "p");
1442        let event_type = require_uint(&pairs, "e")?;
1443
1444        let data = match event_type {
1445            EVENT_CONTROL_MESSAGE => EventData::ControlMessage {
1446                direction: require_direction(&pairs, "d")?,
1447                message_type: require_uint(&pairs, "mt")?,
1448                message: get_message(&pairs),
1449                stream_id: get_uint(&pairs, "sid"),
1450                raw: get_bytes(&pairs, "raw"),
1451            },
1452            EVENT_STREAM_OPENED => {
1453                let st_val = require_value(&pairs, "st")?;
1454                EventData::StreamOpened {
1455                    stream_id: require_uint(&pairs, "sid")?,
1456                    direction: require_direction(&pairs, "d")?,
1457                    stream_type: StreamType::from_cbor(&st_val)?,
1458                    // Read whatever is usable, including a key outside the
1459                    // stream type it is scoped to. A writer must not produce
1460                    // one, but "ignore" is the same read-past-and-keep rule
1461                    // that governs `extra`: dropping it here would make this
1462                    // reader's opinion permanent for every reader downstream.
1463                    // A value that is not an unsigned integer is not usable —
1464                    // the field stays `None` and the key goes to `extra`,
1465                    // which is that same rule one step over. Neither reading
1466                    // past a key nor failing to understand its value is a
1467                    // licence to delete it.
1468                    track_alias: get_uint(&pairs, "ta"),
1469                    subgroup_id: get_uint(&pairs, "sg"),
1470                    fetch_request_id: get_uint(&pairs, "fri"),
1471                    group_id: get_uint(&pairs, "g"),
1472                    header_raw: get_bytes(&pairs, "hdr"),
1473                }
1474            }
1475            EVENT_STREAM_CLOSED => EventData::StreamClosed {
1476                stream_id: require_uint(&pairs, "sid")?,
1477                error_code: require_uint(&pairs, "ec")?,
1478            },
1479            EVENT_OBJECT_HEADER => EventData::ObjectHeader {
1480                stream_id: require_uint(&pairs, "sid")?,
1481                group: require_uint(&pairs, "g")?,
1482                object: require_uint(&pairs, "o")?,
1483                publisher_priority: require_uint(&pairs, "pp")?,
1484                object_status: require_uint(&pairs, "os")?,
1485            },
1486            EVENT_OBJECT_PAYLOAD => EventData::ObjectPayload {
1487                stream_id: require_uint(&pairs, "sid")?,
1488                group: require_uint(&pairs, "g")?,
1489                object: require_uint(&pairs, "o")?,
1490                size: require_uint(&pairs, "sz")?,
1491                payload: get_bytes(&pairs, "pl"),
1492            },
1493            EVENT_STATE_CHANGE => EventData::StateChange {
1494                from: require_text(&pairs, "from")?,
1495                to: require_text(&pairs, "to")?,
1496            },
1497            EVENT_ERROR => EventData::Error {
1498                error_code: require_uint(&pairs, "ec")?,
1499                reason: require_text(&pairs, "reason")?,
1500                // All four optional, and every recording made before they
1501                // existed carries none of them. A `"raw"` longer than the cap
1502                // is read at the length the file gave it: the cap is addressed
1503                // to a recorder building an event from what it observed, and
1504                // re-truncating here would destroy evidence to make somebody
1505                // else's file conform to a rule it was never handed. Report
1506                // the non-conformance if it is worth reporting; do not repair
1507                // it. SPEC.md, Event 6.
1508                stream_id: get_uint(&pairs, "sid"),
1509                kind: get_text(&pairs, "ek").as_deref().map(ErrorKind::parse),
1510                raw_len: get_uint(&pairs, "rawlen"),
1511                raw: get_bytes(&pairs, "raw"),
1512            },
1513            EVENT_ANNOTATION => EventData::Annotation {
1514                label: require_text(&pairs, "label")?,
1515                data: require_value(&pairs, "data")?,
1516            },
1517            EVENT_PEER_CONNECTED => EventData::PeerConnected {
1518                endpoint: get_text(&pairs, "endpoint"),
1519                transport: get_text(&pairs, "transport"),
1520                role: get_text(&pairs, "role").as_deref().map(PeerRole::parse),
1521                side: get_text(&pairs, "side").as_deref().map(Side::parse),
1522            },
1523            EVENT_PEER_DISCONNECTED => EventData::PeerDisconnected {
1524                error_code: require_uint(&pairs, "ec")?,
1525                reason: get_text(&pairs, "reason"),
1526            },
1527            EVENT_SUBSCRIPTION_DERIVATION => {
1528                let upstream = SubscriptionRef::from_value(&require_value(&pairs, "u")?)?;
1529                let downstream = match require_value(&pairs, "d")? {
1530                    Value::Array(items) => items
1531                        .iter()
1532                        .map(SubscriptionRef::from_value)
1533                        .collect::<Result<Vec<_>, _>>()?,
1534                    _ => {
1535                        return Err(MoqTraceError::InvalidEvent(
1536                            "'d' (downstream subs) is not a CBOR array".into(),
1537                        ))
1538                    }
1539                };
1540                EventData::SubscriptionDerivation {
1541                    upstream,
1542                    downstream,
1543                    kind: DerivationKind::parse(&require_text(&pairs, "kind")?),
1544                    trace_id: get_trace_id(&pairs)?,
1545                    namespace: get_namespace(&pairs),
1546                    track_name: get_bytes(&pairs, "tn")
1547                        .or_else(|| get_text(&pairs, "tn").map(String::into_bytes)),
1548                    t_downstream_received: get_int(&pairs, "tdr"),
1549                    t_upstream_sent: get_int(&pairs, "tus"),
1550                    t_upstream_ok_received: get_int(&pairs, "tuo"),
1551                    t_downstream_ok_sent: get_int(&pairs, "tdo"),
1552                }
1553            }
1554            other => EventData::Unknown {
1555                event_type: other,
1556                // The same walk `unrecognised_keys` does, with only the common
1557                // fields to ask about: an unknown event type writes every
1558                // other key straight back out of here. It had the same defect
1559                // for the same reason — a repeated `"n"` was filtered out by
1560                // name, so the entry `seq` did not take reached neither — and
1561                // this variant collects nothing into `extra`, which makes
1562                // `fields` the only place such an entry can land.
1563                fields: unrecognised(&pairs, |key| writes_common_key(peer.as_deref(), key)),
1564            },
1565        };
1566
1567        let extra = match &data {
1568            // `fields` already holds every key on this event the common
1569            // fields did not consume, a `"p"` this crate could not use among
1570            // them. Collecting them into `extra` as well would write each one
1571            // twice and produce a CBOR map with duplicate keys.
1572            EventData::Unknown { .. } => Vec::new(),
1573            other => unrecognised_keys(&pairs, peer.as_deref(), other),
1574        };
1575        Ok(TraceEvent { seq, timestamp, peer, data, extra })
1576    }
1577}