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}