Skip to main content

moqtap_codec/draft19/
message.rs

1//! Draft-19 control message encoding and decoding.
2//!
3//! Key differences from draft-18:
4//! - `Request ID` field removed from GOAWAY entirely; the control-stream and
5//!   request-stream forms are now identical.
6//! - New Range Filter parameters (length-prefixed): SUBGROUP_FILTER (0x25),
7//!   OBJECTID_FILTER (0x26), PRIORITY_FILTER (0x27), OBJECT_PROPERTY_FILTER
8//!   (0x28) and TRACK_PROPERTY_FILTER (0x29).
9//! - New Setup Options MAX_FILTER_RANGES (0x06) and MAX_REQUEST_UPDATES (0x08);
10//!   both are even KVP types carrying a varint value.
11//! - GROUP_ORDER (0x22) moves from PUBLISH_OK to SUBSCRIBE_TRACKS (the wire
12//!   encoding of the parameter is unchanged).
13//! - PUBLISH_BLOCKED renamed to PUBLISH_SKIPPED (still type 0x0F; wire
14//!   identical).
15//! - SUBSCRIPTION_FILTER renamed to LOCATION_FILTER (still parameter 0x21).
16//! - REQUEST_ERROR adds CONFLICTING_FILTERS (0x35) and INVALID_FILTER (0x36);
17//!   DUPLICATE_SUBSCRIPTION (0x19) is removed.
18//! - The framing field after Message Length is named Message Body (draft-19
19//!   Section 10, Figure 3); earlier drafts called it Message Payload. The
20//!   change is editorial, so the bytes are unchanged, but this module uses the
21//!   new name.
22
23use crate::auth_token::{AuthorizationToken, AUTH_TOKEN_PARAMETER};
24use crate::error::MAX_FULL_TRACK_NAME_LENGTH;
25pub use crate::error::{
26    CodecError, MAX_GOAWAY_URI_LENGTH, MAX_MESSAGE_LENGTH, MAX_NAMESPACE_TUPLE_SIZE,
27    MAX_REASON_PHRASE_LENGTH,
28};
29use crate::kvp::{KeyValuePair, KvpError, KvpValue, MAX_KVP_VALUE_LEN};
30use crate::subscription_filter::{SubscriptionFilter, SUBSCRIPTION_FILTER_PARAMETER};
31use crate::types::check_location_range;
32use crate::types::*;
33use crate::varint::{Moqt18 as Wire, VarInt};
34use bytes::{Buf, BufMut};
35
36// ============================================================
37// Parameter encoding helpers for draft-19
38// ============================================================
39
40/// How a parameter value is encoded on the wire.
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42enum ParamEncoding {
43    /// Bare varint.
44    Varint,
45    /// Single byte (uint8).
46    Uint8,
47    /// Two consecutive varints (group, object).
48    Location,
49    /// Length-prefixed bytes.
50    LengthPrefixed,
51    /// A Track Namespace as defined in draft-19 Section 2.4.1: a varint field
52    /// count followed by that many length-prefixed fields.
53    ///
54    /// Not one of the four value encodings draft-19 Section 10.2 lists. A
55    /// parameter definition is free to name an encoding from elsewhere in the
56    /// document, and TRACK_NAMESPACE_PREFIX does exactly that; the field count
57    /// is the only length the wire carries.
58    TrackNamespaceValue,
59}
60
61fn param_encoding(key: u64) -> Option<ParamEncoding> {
62    match key {
63        // 0x02 = OBJECT_DELIVERY_TIMEOUT (DELIVERY_TIMEOUT on drafts 11-17)
64        // 0x04 = RENDEZVOUS_TIMEOUT (draft-19 Section 10.2.6). Not
65        //        MAX_CACHE_DURATION: that is Property Type 0x04 in the
66        //        separate Properties registry (Section 15.8), a different
67        //        namespace that happens to reuse the number.
68        // 0x06 = SUBGROUP_DELIVERY_TIMEOUT (new in draft-18)
69        // 0x08 = EXPIRES
70        // 0x0A = FILL_TIMEOUT (new in draft-18, FETCH only)
71        // 0x32 = NEW_GROUP_REQUEST
72        0x02 | 0x04 | 0x06 | 0x08 | 0x0A | 0x32 => Some(ParamEncoding::Varint),
73        // 0x10 = FORWARD, 0x20 = SUBSCRIBER_PRIORITY, 0x22 = GROUP_ORDER
74        0x10 | 0x20 | 0x22 => Some(ParamEncoding::Uint8),
75        // 0x09 = LARGEST_OBJECT. Draft-19 Section 10.2.16: "The LARGEST_OBJECT
76        //        parameter (Parameter Type 0x9) is a Location." A Location is
77        //        two consecutive varints (Section 10.2), with no length ahead
78        //        of them.
79        0x09 => Some(ParamEncoding::Location),
80        // 0x34 = TRACK_NAMESPACE_PREFIX. Section 10.2.19: it "uses the Track
81        //        Namespace encoding described in Section 2.4.1".
82        0x34 => Some(ParamEncoding::TrackNamespaceValue),
83        // 0x03 = AUTHORIZATION_TOKEN
84        // 0x21 = LOCATION_FILTER (SUBSCRIPTION_FILTER on drafts 15-18)
85        // 0x25 = SUBGROUP_FILTER, 0x26 = OBJECTID_FILTER, 0x27 = PRIORITY_FILTER,
86        // 0x28 = OBJECT_PROPERTY_FILTER, 0x29 = TRACK_PROPERTY_FILTER
87        //        (Range Filters, new in draft-19)
88        0x03 | 0x21 | 0x25 | 0x26 | 0x27 | 0x28 | 0x29 => Some(ParamEncoding::LengthPrefixed),
89        _ => None,
90    }
91}
92
93/// Whether `value` is inside the range draft-19 allows for a uint8-valued
94/// parameter.
95///
96/// Two of the three uint8 parameters restrict their range and say the receiver
97/// MUST close the session with PROTOCOL_VIOLATION on anything outside it:
98/// GROUP_ORDER allows only Ascending (0x1) and Descending (0x2) (Section
99/// 10.2.8), and FORWARD allows only 0 and 1 (Section 10.2.17).
100/// SUBSCRIBER_PRIORITY (Section 10.2.7) uses the whole 0-255 range, so it has
101/// no entry here.
102///
103/// Range-checking on decode is what makes the values usable: an application
104/// that tests `group_order == 2` for descending would otherwise treat 7 as
105/// neither ascending nor descending and carry on.
106fn uint8_value_in_range(key: u64, value: u8) -> bool {
107    match key {
108        // FORWARD (0x10)
109        0x10 => value <= 1,
110        // GROUP_ORDER (0x22)
111        0x22 => value == 1 || value == 2,
112        _ => true,
113    }
114}
115
116const AUTHORIZATION_TOKEN: u64 = 0x03;
117
118/// Whether a message may carry `key` more than once.
119///
120/// Section 10.2 states the default: "Senders MUST NOT repeat the same Parameter
121/// Type in a message unless the parameter definition explicitly allows multiple
122/// instances of that type to be sent in a single message." Two definitions do.
123///
124/// * `AUTHORIZATION_TOKEN` (0x03), Section 10.2.2: "The AUTHORIZATION TOKEN
125///   parameter MAY be repeated within a message as long as the combination of
126///   Token Type and Token Value are unique after resolving any aliases."
127/// * The five Range Filters (0x25 through 0x29), Section 5.1.3: "The Track
128///   Property filter parameter MAY appear multiple times in a SUBSCRIBE_TRACKS
129///   message or REQUEST_UPDATE for it. All other filter parameters MAY appear
130///   multiple times in a FETCH, SUBSCRIBE, SUBSCRIBE_TRACKS, PUBLISH_OK, or
131///   REQUEST_UPDATE (on a subscription, from the subscriber only) message."
132///
133/// # A zero `Type Delta` is "the same type again", not an error
134///
135/// The two rules interact, and **the draft does not say how**. Section 10.2 also
136/// requires that "Parameters MUST be serialized in ascending order by Type", so
137/// a second instance of a repeatable type produces a `Type Delta` of 0 — well
138/// formed only if the decoder reads a zero delta as a repeat rather than as a
139/// malformation. That reading is the one taken here; the alternative makes
140/// Section 5.1.3's permission unusable, because there is no other encoding for
141/// a second filter of the same type.
142///
143/// # Why the permission is not decoration
144///
145/// "All filter parameters with the same SetID value are combined using logical
146/// 'AND' operations, then all the resulting sets are combined using logical
147/// 'OR' operations." One filter parameter carries one SetID, so a subscriber
148/// asking for two alternatives that each constrain the same field — Subgroup 1
149/// to 3 at low priority, or Subgroup 10 to 12 at high — has to send SUBGROUP_
150/// FILTER twice, once per set. Refusing the repeat does not narrow what a peer
151/// can express; it collapses the set lattice to one filter per type and turns
152/// conforming traffic into a session close.
153///
154/// What the filters may **not** do is repeat the same (Parameter Type, SetID,
155/// Property Type) triple, and Section 5.1.3 answers that with a REQUEST_ERROR
156/// carrying INVALID_FILTER rather than with a session close. A reply an endpoint
157/// sends is not a frame a decoder refuses, so nothing here enforces it — see
158/// [`crate::range_filter`] for the reader an endpoint uses to decide.
159fn parameter_may_repeat(key: u64) -> bool {
160    matches!(key, AUTHORIZATION_TOKEN | 0x25..=0x29)
161}
162
163/// Add a delta to the previous delta-encoded key.
164///
165/// Draft-19 Section 1.4.3: "The previous Type value plus the Delta Type MUST NOT
166/// be greater than 2^64 - 1. If a Delta Type is received that would be too
167/// large, the Session MUST be closed with a PROTOCOL_VIOLATION." MoQT varints
168/// span the whole 64-bit range, so a peer can drive the sum past the end: a
169/// debug build panicked on the addition and a release build wrapped the key and
170/// reported the parameter under a type its sender never wrote.
171fn add_delta(prev_key: u64, delta: u64) -> Result<u64, CodecError> {
172    prev_key.checked_add(delta).ok_or(CodecError::KeyDeltaOverflow(prev_key, delta))
173}
174
175/// Hold a namespace-plus-name pair to the Full Track Name cap.
176///
177/// Draft-19 Section 2.4.1: "The maximum total length of a Full Track Name is
178/// 4,096 bytes. The length of a Full Track Name is computed as the sum of the
179/// Track Namespace Field Length fields and the Track Name Length field... If an
180/// endpoint receives a Track Namespace or a Full Track Name exceeding 4,096
181/// bytes, it MUST close the session with a PROTOCOL_VIOLATION."
182///
183/// The namespace half of that sentence is enforced inside the namespace decoder,
184/// which is the only place that sees a namespace with no name beside it. This is
185/// the other half, and it has to live where the two are decoded together: a
186/// namespace at 4,000 bytes and a name at 500 are each legal alone.
187fn check_full_track_name(namespace: &TrackNamespace, track_name: &[u8]) -> Result<(), CodecError> {
188    let total = namespace.field_bytes_len().saturating_add(track_name.len());
189    if total > MAX_FULL_TRACK_NAME_LENGTH {
190        return Err(CodecError::TrackNameTooLong);
191    }
192    Ok(())
193}
194
195/// Hold every AUTHORIZATION TOKEN parameter to the Token structure it names.
196///
197/// Section 10.2.2: "If the Token structure cannot be decoded, the receiver
198/// MUST close the Session with KEY_VALUE_FORMATTING_ERROR." That is the answer
199/// Section 1.4.3 gives for any Type whose value does not match the
200/// serialization that Type defines; the Token is the one structure this draft
201/// spells out, and the only parameter value in it that is more than opaque
202/// bytes.
203///
204/// Both namespaces carry the type on this draft, and both reach here.
205///
206/// A type this draft cannot name is left alone. The rule is conditional on the
207/// receiver understanding the Type, and an extension's parameter carries bytes
208/// no rule here describes.
209fn check_authorization_tokens(parameters: &[KeyValuePair]) -> Result<(), CodecError> {
210    for parameter in parameters {
211        let key = parameter.key.into_inner();
212        if key != AUTH_TOKEN_PARAMETER {
213            continue;
214        }
215        match &parameter.value {
216            KvpValue::Bytes(value) => {
217                AuthorizationToken::decode_moqt::<Wire>(key, value)?;
218            }
219            // Unreachable from the decoder, which picks the shape from the
220            // type and finds this one length-prefixed. A caller that built the
221            // pair in memory can still get here, and it is the same rule: the
222            // value is not the serialization the type defines.
223            KvpValue::Varint(_) => {
224                return Err(CodecError::KeyValueFormatting {
225                    key,
226                    detail: "its value is a bare varint where the type defines a Token structure",
227                });
228            }
229        }
230    }
231    Ok(())
232}
233
234/// Hold every LOCATION_FILTER parameter to the filter structure it names.
235///
236/// Section 5.1.2: "An endpoint that receives a filter type other than the above
237/// MUST close the session with PROTOCOL_VIOLATION." Section 10.2.9 defines the
238/// parameter, which this draft renamed from SUBSCRIPTION_FILTER to
239/// LOCATION_FILTER when it added the Range Filters beside it. The number, 0x21,
240/// and the structure are the ones draft-18 had.
241///
242/// Drafts 15 and 16 stated the length rule of this parameter directly, at
243/// draft-16 Section 9.2.2.5 — "If the length of the Subscription Filter does
244/// not match the parameter length, the publisher MUST close the session with
245/// PROTOCOL_VIOLATION." Draft-17 dropped that sentence, and what answers the
246/// same malformation here is the general rule of Section 1.4.3, which names
247/// KEY_VALUE_FORMATTING_ERROR. Same malformation, different code, and the
248/// session table is where the two part.
249///
250/// The End Group is a delta, and this draft states what happens when resolving
251/// it leaves the number space: "the last Group ID to be delivered
252/// will be the Group ID in Start Location plus the End Group Delta. If the
253/// resulting Group ID would be greater than 2^64 - 1, the endpoint MUST close
254/// the session with a PROTOCOL_VIOLATION." That is why the sum is taken here and
255/// not left to the caller — draft-17, which introduced the delta and states no
256/// such sentence, does not take it.
257///
258/// The filter is otherwise decoded and discarded. What is kept is the refusal —
259/// the value stays on the parameter as the bytes that arrived, so a caller reads
260/// it through [`SubscriptionFilter::decode_moqt`] when it wants the filter
261/// rather than the frame.
262fn check_subscription_filters(parameters: &[KeyValuePair]) -> Result<(), CodecError> {
263    for parameter in parameters {
264        if parameter.key.into_inner() != SUBSCRIPTION_FILTER_PARAMETER {
265            continue;
266        }
267        match &parameter.value {
268            KvpValue::Bytes(value) => {
269                SubscriptionFilter::decode_moqt::<Wire>(value)?.last_group()?;
270            }
271            // Unreachable from the decoder, which picks the shape from the type
272            // and finds this one length-prefixed. A caller that built the pair
273            // in memory can still get here, and it is the same rule.
274            KvpValue::Varint(_) => {
275                return Err(CodecError::SubscriptionFilterMalformed {
276                    detail: "its value is a bare varint where the type defines a filter",
277                });
278            }
279        }
280    }
281    Ok(())
282}
283
284/// Decode a count-prefixed list of parameters with delta-encoded types.
285fn decode_parameters(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
286    let count = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
287    let mut params = crate::types::reserve_bounded(count, buf);
288    let mut prev_key: u64 = 0;
289
290    for i in 0..count {
291        let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
292        let abs_key = add_delta(prev_key, delta)?;
293        // Types ascend, so a repeat is always a zero delta against the
294        // parameter before it. Draft-19 Section 10.2: "Receivers SHOULD check
295        // that there are no unexpected duplicate parameters and close the
296        // session with PROTOCOL_VIOLATION if found." Downstream code that scans
297        // the list for a key takes whichever copy it meets first, so two
298        // implementations reading one frame can pick opposite values.
299        //
300        // "Unexpected" is what `parameter_may_repeat` reads: a zero delta on a
301        // type whose own definition permits repeats is the second instance,
302        // which is the only encoding such an instance has.
303        if i > 0 && delta == 0 && !parameter_may_repeat(abs_key) {
304            return Err(CodecError::DuplicateParameter(abs_key));
305        }
306        prev_key = abs_key;
307
308        // Section 10.2: "All Message Parameters MUST be defined in the
309        // negotiated version of MOQT or negotiated via Setup Options. An
310        // endpoint that receives an unknown Message Parameter MUST close the
311        // session with PROTOCOL_VIOLATION. Because the receiver has to
312        // understand every Message Parameter, there is no need for a mechanism
313        // to skip unknown parameters." Because unknown parameters
314        // cannot be skipped, the block is bounded by a parameter count rather
315        // than a length.
316        //
317        // The table this consults is the registry's, so a type it cannot name
318        // is one this draft does not define. Reporting it as an ordinary
319        // malformation, which is what it did before, left the rule enforced
320        // against the frame and invisible to the session.
321        let encoding =
322            param_encoding(abs_key).ok_or(CodecError::UnknownMessageParameter(abs_key))?;
323
324        let value = match encoding {
325            ParamEncoding::Varint => {
326                let v = VarInt::decode_moqt::<Wire>(buf)?;
327                KvpValue::Varint(v)
328            }
329            ParamEncoding::Uint8 => {
330                if buf.remaining() < 1 {
331                    return Err(CodecError::UnexpectedEnd);
332                }
333                let byte = buf.get_u8();
334                if !uint8_value_in_range(abs_key, byte) {
335                    return Err(CodecError::ParameterValueOutOfRange {
336                        key: abs_key,
337                        value: byte as u64,
338                    });
339                }
340                KvpValue::Varint(VarInt::from_u64_moqt(byte as u64))
341            }
342            ParamEncoding::Location => {
343                let group = VarInt::decode_moqt::<Wire>(buf)?;
344                let object = VarInt::decode_moqt::<Wire>(buf)?;
345                let mut encoded = Vec::new();
346                group.encode_moqt::<Wire>(&mut encoded);
347                object.encode_moqt::<Wire>(&mut encoded);
348                KvpValue::Bytes(encoded)
349            }
350            ParamEncoding::LengthPrefixed => {
351                let len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
352                let data = read_bytes(buf, len)?;
353                KvpValue::Bytes(data)
354            }
355            ParamEncoding::TrackNamespaceValue => {
356                // A prefix of zero fields is legal: Section 2.4.1 puts a Track
357                // Namespace at "between 0 and 32 Track Namespace Fields", and
358                // an empty prefix matches every namespace.
359                let ns = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
360                let mut encoded = Vec::new();
361                ns.encode_moqt::<Wire>(&mut encoded);
362                KvpValue::Bytes(encoded)
363            }
364        };
365
366        params.push(KeyValuePair { key: VarInt::from_u64_moqt(abs_key), value });
367    }
368    check_authorization_tokens(&params)?;
369    check_subscription_filters(&params)?;
370    Ok(params)
371}
372
373/// Whether `bytes` is exactly the wire form of a Location — two consecutive
374/// varints and nothing after them.
375///
376/// `decode_parameters` builds this value by reading two varints and
377/// re-serialising them, so every value it produces satisfies this. A value
378/// built in memory need not, and the encode arm writes these bytes verbatim
379/// because a Location carries no length of its own. Without this check a
380/// caller could hand over one varint, or three, and the codec would put a
381/// frame on the wire that its own decoder answers with an error.
382fn is_location_value(bytes: &[u8]) -> bool {
383    let mut buf = bytes;
384    VarInt::decode_moqt::<Wire>(&mut buf).is_ok()
385        && VarInt::decode_moqt::<Wire>(&mut buf).is_ok()
386        && !buf.has_remaining()
387}
388
389/// Whether `bytes` is exactly the wire form of a Track Namespace, with
390/// nothing after it. The same reasoning as [`is_location_value`]: the value
391/// goes out verbatim, so it has to be something this draft can read back.
392fn is_track_namespace_value(bytes: &[u8]) -> bool {
393    let mut buf = bytes;
394    TrackNamespace::decode_allow_empty_moqt::<Wire>(&mut buf).is_ok() && !buf.has_remaining()
395}
396
397/// Encode a count-prefixed list of parameters with delta-encoded types.
398///
399/// Errors with [`CodecError::InvalidField`] on a uint8-valued parameter whose
400/// value [`decode_parameters`] would refuse, so the two directions accept the
401/// same set of frames.
402///
403/// The check is not a mirror added for tidiness. A uint8 parameter's value is
404/// written as one octet, and a value that does not fit one is otherwise
405/// truncated to its low byte: GROUP_ORDER 258 becomes the byte 0x02, which is
406/// Descending — a well-formed frame carrying a value the caller never asked
407/// for, and one no receiver could tell from a genuine Descending. Refusing is
408/// the only outcome that does not silently rewrite the message.
409///
410/// The two structure rules are here for a plainer reason. A value under a type
411/// that defines a structure and is not that structure — a Token, a filter — is
412/// one the receiver must close the session over, so writing it is not a way to
413/// send it; the sender's first sign of trouble would be the session going.
414fn encode_parameters(params: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
415    check_authorization_tokens(params)?;
416    check_subscription_filters(params)?;
417    VarInt::from_usize(params.len()).encode_moqt::<Wire>(buf);
418    let mut prev_key: u64 = 0;
419
420    for (i, p) in params.iter().enumerate() {
421        let abs_key = p.key.into_inner();
422        // The delta is a difference, so a descending pair wraps the subtraction
423        // into a nine-byte delta the peer resolves to an unrelated key, and a
424        // repeated type is a frame `decode_parameters` refuses. Both are
425        // refused here so the two directions accept the same set of frames.
426        let delta = abs_key
427            .checked_sub(prev_key)
428            .ok_or(CodecError::ParametersOutOfOrder(prev_key, abs_key))?;
429        if i > 0 && delta == 0 && !parameter_may_repeat(abs_key) {
430            return Err(CodecError::DuplicateParameter(abs_key));
431        }
432        prev_key = abs_key;
433        VarInt::from_u64_moqt(delta).encode_moqt::<Wire>(buf);
434
435        // The same maximum the decoder below applies, and the same one this
436        // draft's Setup Option encoder has always applied: "The maximum length
437        // of a value is 2^16-1 bytes. If an endpoint receives a length larger
438        // than the maximum, it MUST close the session with a PROTOCOL_VIOLATION."
439        // A value past it is one the peer must end the session over, so writing
440        // it is not a way to send it.
441        //
442        // Hoisted above the shape table rather than repeated inside it: a
443        // Location is bytes as well, and one past the maximum is not a Location.
444        if let KvpValue::Bytes(b) = &p.value {
445            if b.len() > MAX_KVP_VALUE_LEN {
446                return Err(KvpError::ValueTooLong(b.len()).into());
447            }
448        }
449
450        let encoding = param_encoding(abs_key);
451        match (&p.value, encoding) {
452            (KvpValue::Varint(v), Some(ParamEncoding::Varint)) => {
453                v.encode_moqt::<Wire>(buf);
454            }
455            (KvpValue::Varint(v), Some(ParamEncoding::Uint8)) => {
456                let raw = v.into_inner();
457                let byte = u8::try_from(raw).map_err(|_| CodecError::InvalidField)?;
458                if !uint8_value_in_range(abs_key, byte) {
459                    return Err(CodecError::ParameterValueOutOfRange {
460                        key: abs_key,
461                        value: byte as u64,
462                    });
463                }
464                buf.put_u8(byte);
465            }
466            // Both values are already stored in their own wire form — two
467            // varints for a Location, a field count and its fields for a Track
468            // Namespace — so they go out as they are. Adding a length here is
469            // the bug these arms exist to avoid.
470            (KvpValue::Bytes(b), Some(ParamEncoding::Location)) => {
471                if !is_location_value(b) {
472                    return Err(CodecError::InvalidField);
473                }
474                buf.put_slice(b);
475            }
476            (KvpValue::Bytes(b), Some(ParamEncoding::TrackNamespaceValue)) => {
477                if !is_track_namespace_value(b) {
478                    return Err(CodecError::InvalidField);
479                }
480                buf.put_slice(b);
481            }
482            (KvpValue::Bytes(b), Some(ParamEncoding::LengthPrefixed)) => {
483                VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
484                buf.put_slice(b);
485            }
486            _ => {
487                // Fallback: encode as KVP even/odd
488                match &p.value {
489                    KvpValue::Varint(v) => v.encode_moqt::<Wire>(buf),
490                    KvpValue::Bytes(b) => {
491                        VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
492                        buf.put_slice(b);
493                    }
494                }
495            }
496        }
497    }
498    Ok(())
499}
500
501/// Decode delta-encoded KVPs with even/odd convention (for setup options
502/// and track properties). Read until buffer is exhausted.
503fn decode_kvp_delta(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
504    let mut pairs = Vec::new();
505    let mut prev_key: u64 = 0;
506
507    while buf.has_remaining() {
508        let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
509        let abs_key = add_delta(prev_key, delta)?;
510        prev_key = abs_key;
511
512        let value = if abs_key.is_multiple_of(2) {
513            let v = VarInt::decode_moqt::<Wire>(buf)?;
514            KvpValue::Varint(v)
515        } else {
516            let len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
517            // Draft-19 Section 1.4.3: "The maximum length of a value is 2^16-1
518            // bytes. If an endpoint receives a length larger than the maximum,
519            // it MUST close the session with a PROTOCOL_VIOLATION." The
520            // standalone `KeyValuePair::decode` already enforces this; stating
521            // it here too means the two readers of the same wire shape answer
522            // the same way, rather than this one leaning on the caller having
523            // clipped the buffer to a control message first.
524            if len > MAX_KVP_VALUE_LEN {
525                return Err(KvpError::ValueTooLong(len).into());
526            }
527            let data = read_bytes(buf, len)?;
528            KvpValue::Bytes(data)
529        };
530
531        pairs.push(KeyValuePair { key: VarInt::from_u64_moqt(abs_key), value });
532    }
533    Ok(pairs)
534}
535
536/// Encode delta-encoded KVPs with even/odd convention.
537///
538/// Refuses a list that is not in ascending order by type, for the same reason
539/// [`encode_parameters`] does: the delta is a difference, and a descending pair
540/// wraps it into a nine-byte delta the peer resolves to an unrelated key.
541fn encode_kvp_delta(pairs: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
542    let mut prev_key: u64 = 0;
543    for p in pairs {
544        let abs_key = p.key.into_inner();
545        let delta = abs_key
546            .checked_sub(prev_key)
547            .ok_or(CodecError::ParametersOutOfOrder(prev_key, abs_key))?;
548        prev_key = abs_key;
549        VarInt::from_u64_moqt(delta).encode_moqt::<Wire>(buf);
550        match &p.value {
551            KvpValue::Varint(v) => v.encode_moqt::<Wire>(buf),
552            KvpValue::Bytes(b) => {
553                if b.len() > MAX_KVP_VALUE_LEN {
554                    return Err(KvpError::ValueTooLong(b.len()).into());
555                }
556                VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
557                buf.put_slice(b);
558            }
559        }
560    }
561    Ok(())
562}
563
564/// Immutable Properties, Property Type 0xB.
565///
566/// Section 12.7: Immutable Properties are "a Track or Object Property that
567/// contains a sequence of Key-Value-Pairs (see Figure 2) that are themselves
568/// Track or Object Properties, respectively". The Type is odd, so its value is
569/// length-prefixed bytes, and those bytes are another delta-typed run starting
570/// from 0.
571const IMMUTABLE_PROPERTIES: u64 = 0x0B;
572
573/// Whether `value` is inside the range draft-19 allows for a Track Property
574/// type that restricts one.
575///
576/// Two types do, and each answers anything outside its range with a session
577/// close. DEFAULT_PUBLISHER_GROUP_ORDER (0x22), Section 12.5: "The allowed
578/// values are Ascending (0x1) or Descending (0x2). If an endpoint receives a
579/// value outside this range, it MUST close the session with
580/// PROTOCOL_VIOLATION." DYNAMIC_GROUPS (0x30), Section 12.6: "The allowed
581/// values are 0 or 1... If an endpoint receives a value larger than 1, it MUST
582/// close the session with PROTOCOL_VIOLATION."
583///
584/// Both are Track Properties, so the list they arrive in is the one carried by
585/// a control message rather than the properties on an object.
586///
587/// DEFAULT_PUBLISHER_PRIORITY (0x0E) is not here. Section 12.4 says
588/// "Priorities above 255 are invalid" and stops, where the two above name a
589/// consequence in the next clause. A range stated without one is not a close.
590///
591/// The numbers belong to the Property registry and not the Message Parameter
592/// one. Type 0x22 is GROUP_ORDER as a parameter and
593/// DEFAULT_PUBLISHER_GROUP_ORDER as a property, and the two happen to permit the
594/// same pair of values while meaning different things — one subscriber's
595/// preference against a property of the track. Reading either table for the
596/// other's types would be right by accident here and wrong at the next entry.
597fn track_property_value_in_range(key: u64, value: u64) -> bool {
598    match key {
599        // DEFAULT_PUBLISHER_GROUP_ORDER (0x22)
600        0x22 => value == 1 || value == 2,
601        // DYNAMIC_GROUPS (0x30)
602        0x30 => value <= 1,
603        _ => true,
604    }
605}
606
607/// Refuse a Track Property whose value falls outside the range its type allows,
608/// wherever in the list it is carried.
609///
610/// # Inside Immutable Properties as well as beside them
611///
612/// The list is walked one level down through Immutable Properties, whose
613/// contents Section 12.7 defines as properties themselves. The draft asks for
614/// this in as many words: "When looking for the value of a property, processors
615/// MUST search both the mutable properties and the contents of Immutable
616/// Properties." A check applied only to the outer list is one a peer opts out of
617/// by moving a pair inside the block, and the block is where an Original
618/// Publisher puts what a relay must not rewrite — which is where a track's group
619/// order and dynamic-group support belong.
620///
621/// Bytes under 0xB that do not parse as a Key-Value-Pair run are left alone
622/// rather than refused. Section 12.7 says relays "MAY decode and view the
623/// Properties in the Key-Value-Pairs", which is a permission and not a
624/// requirement, so a block this codec cannot read is carried to the caller
625/// intact instead of ending the session.
626fn check_track_property_values(properties: &[KeyValuePair]) -> Result<(), CodecError> {
627    for property in properties {
628        let key = property.key.into_inner();
629        match &property.value {
630            KvpValue::Varint(value) => {
631                let value = value.into_inner();
632                if !track_property_value_in_range(key, value) {
633                    return Err(CodecError::TrackPropertyValueOutOfRange { key, value });
634                }
635            }
636            KvpValue::Bytes(bytes) if key == IMMUTABLE_PROPERTIES => {
637                let mut inner = &bytes[..];
638                // A block that is not a Key-Value-Pair run is skipped rather
639                // than refused. See the note above: reading inside it is a
640                // permission, so one that cannot be read is carried.
641                //
642                // Skipped means this block and only this block. The rule the
643                // draft states here is about the block whose pairs will not
644                // parse, and says nothing about its neighbours; ending the
645                // whole walk would let a peer keep an out-of-range property
646                // from being looked at by putting an unparseable block in
647                // front of it.
648                if let Ok(nested) = decode_kvp_delta(&mut inner) {
649                    check_track_property_values(&nested)?;
650                }
651            }
652            KvpValue::Bytes(_) => {}
653        }
654    }
655    Ok(())
656}
657
658/// Decode the Track Properties that fill the tail of a control message.
659///
660/// [`decode_kvp_delta`] with the Property registry's value rules applied. The
661/// two are separate because that function also reads Setup Options, which are a
662/// third namespace numbering its entries independently of this one.
663fn decode_track_properties(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
664    let properties = decode_kvp_delta(buf)?;
665    check_track_property_values(&properties)?;
666    Ok(properties)
667}
668
669/// Encode a control message's Track Properties.
670///
671/// Held to the same value ranges as the decoder. A value this codec refuses to
672/// read is one it must not write: the peer that receives it is required to close
673/// the session, so the sender's first sign of trouble would be the session
674/// going.
675fn encode_track_properties(
676    properties: &[KeyValuePair],
677    buf: &mut impl BufMut,
678) -> Result<(), CodecError> {
679    check_track_property_values(properties)?;
680    encode_kvp_delta(properties, buf)
681}
682
683/// The Setup Option types this draft defines.
684///
685/// Section 10.3.1 assigns PATH, AUTHORIZATION TOKEN, MAX_AUTH_TOKEN_CACHE_SIZE, AUTHORITY,
686/// MAX_FILTER_RANGES, MOQT_IMPLEMENTATION and MAX_REQUEST_UPDATES.
687///
688/// The list exists for one rule and one direction. Section 10.3: "Receivers
689/// MUST allow duplicates of unknown Setup Options." A receiver may therefore
690/// refuse a repeat only of a type it can name, and an option outside this list
691/// is one an extension defined and this codec has no business closing a session
692/// over. Nothing else reads it - unknown options are still decoded and carried,
693/// as "Receivers MUST ignore unrecognized Setup Options" requires.
694const KNOWN_SETUP_OPTIONS: &[u64] = &[0x01, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08];
695
696/// The one Setup Option whose definition allows more than one instance.
697///
698/// Section 10.3.1.4: "The AUTHORIZATION TOKEN Setup Option (Option Type 0x03)
699/// is functionally equivalent to the AUTHORIZATION TOKEN message parameter...
700/// The endpoint can specify one or more tokens in SETUP that the peer can use to
701/// authorize MOQT session establishment." That is the "unless the option
702/// definition explicitly allows multiple instances" carve-out, and it is the
703/// only one on this draft.
704const REPEATABLE_SETUP_OPTION: u64 = 0x03;
705
706/// Decode the Setup Options of a SETUP message.
707///
708/// Section 10.3: "Senders MUST NOT repeat the same Option Type in a message
709/// unless the option definition explicitly allows multiple instances. Receivers
710/// MUST allow duplicates of unknown Setup Options."
711///
712/// The second sentence is why this is not the mirror of
713/// [`encode_setup_options`]: a repeat of a type this draft names is refused, and
714/// a repeat of any other type is carried. Types ascend and are delta-encoded, so
715/// a repeat is always a zero delta against the option before it.
716fn decode_setup_options(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
717    let options = decode_kvp_delta(buf)?;
718    for (i, option) in options.iter().enumerate() {
719        let key = option.key.into_inner();
720        if key == REPEATABLE_SETUP_OPTION || !KNOWN_SETUP_OPTIONS.contains(&key) {
721            continue;
722        }
723        if options[..i].iter().any(|earlier| earlier.key == option.key) {
724            return Err(CodecError::DuplicateParameter(key));
725        }
726    }
727    check_authorization_tokens(&options)?;
728    Ok(options)
729}
730
731/// Encode the Setup Options of a SETUP message.
732///
733/// The sender's half of the same sentence, and it is the wider half: "Senders
734/// MUST NOT repeat the same Option Type in a message" names no exception for
735/// types the sender does not recognise, so every repeat is refused here except
736/// the one the draft allows. A caller holding an option this codec has never
737/// heard of still may not send it twice.
738///
739/// The token is in this namespace as well, and is held to its structure here for
740/// the reason [`encode_parameters`] gives.
741fn encode_setup_options(options: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
742    check_authorization_tokens(options)?;
743    for (i, option) in options.iter().enumerate() {
744        if option.key.into_inner() == REPEATABLE_SETUP_OPTION {
745            continue;
746        }
747        if options[..i].iter().any(|earlier| earlier.key == option.key) {
748            return Err(CodecError::DuplicateParameter(option.key.into_inner()));
749        }
750    }
751    encode_kvp_delta(options, buf)
752}
753
754// ============================================================
755// Message Types
756// ============================================================
757
758#[derive(Debug, Clone, Copy, PartialEq, Eq)]
759#[repr(u64)]
760pub enum MessageType {
761    RequestUpdate = 0x02,
762    Subscribe = 0x03,
763    SubscribeOk = 0x04,
764    RequestError = 0x05,
765    PublishNamespace = 0x06,
766    /// REQUEST_OK (0x07). PUBLISH_OK is now an alias of this type.
767    RequestOk = 0x07,
768    Namespace = 0x08,
769    PublishDone = 0x0B,
770    TrackStatus = 0x0D,
771    NamespaceDone = 0x0E,
772    PublishSkipped = 0x0F,
773    GoAway = 0x10,
774    Fetch = 0x16,
775    FetchOk = 0x18,
776    Publish = 0x1D,
777    /// SUBSCRIBE_NAMESPACE (renumbered to 0x50 in draft-18).
778    SubscribeNamespace = 0x50,
779    /// SUBSCRIBE_TRACKS (new message in draft-18).
780    SubscribeTracks = 0x51,
781    Setup = 0x2F00,
782}
783
784impl MessageType {
785    pub fn from_id(id: u64) -> Option<Self> {
786        match id {
787            0x02 => Some(MessageType::RequestUpdate),
788            0x03 => Some(MessageType::Subscribe),
789            0x04 => Some(MessageType::SubscribeOk),
790            0x05 => Some(MessageType::RequestError),
791            0x06 => Some(MessageType::PublishNamespace),
792            0x07 => Some(MessageType::RequestOk),
793            0x08 => Some(MessageType::Namespace),
794            0x0B => Some(MessageType::PublishDone),
795            0x0D => Some(MessageType::TrackStatus),
796            0x0E => Some(MessageType::NamespaceDone),
797            0x0F => Some(MessageType::PublishSkipped),
798            0x10 => Some(MessageType::GoAway),
799            0x16 => Some(MessageType::Fetch),
800            0x18 => Some(MessageType::FetchOk),
801            0x1D => Some(MessageType::Publish),
802            0x50 => Some(MessageType::SubscribeNamespace),
803            0x51 => Some(MessageType::SubscribeTracks),
804            0x2F00 => Some(MessageType::Setup),
805            _ => None,
806        }
807    }
808
809    pub fn id(&self) -> u64 {
810        *self as u64
811    }
812
813    /// This type's name in the shared vector corpus: the `message_type` its
814    /// draft's `codec/messages/*.json` files carry, in `snake_case`.
815    pub fn name(&self) -> &'static str {
816        match self {
817            MessageType::RequestUpdate => "request_update",
818            MessageType::Subscribe => "subscribe",
819            MessageType::SubscribeOk => "subscribe_ok",
820            MessageType::RequestError => "request_error",
821            MessageType::PublishNamespace => "publish_namespace",
822            MessageType::RequestOk => "request_ok",
823            MessageType::Namespace => "namespace",
824            MessageType::PublishDone => "publish_done",
825            MessageType::TrackStatus => "track_status",
826            MessageType::NamespaceDone => "namespace_done",
827            MessageType::PublishSkipped => "publish_skipped",
828            MessageType::GoAway => "goaway",
829            MessageType::Fetch => "fetch",
830            MessageType::FetchOk => "fetch_ok",
831            MessageType::Publish => "publish",
832            MessageType::SubscribeNamespace => "subscribe_namespace",
833            MessageType::SubscribeTracks => "subscribe_tracks",
834            MessageType::Setup => "setup",
835        }
836    }
837}
838
839// ============================================================
840// Session Lifecycle Messages
841// ============================================================
842
843/// Unified SETUP (0x2F00).
844#[derive(Debug, Clone, PartialEq, Eq)]
845pub struct Setup {
846    pub options: Vec<KeyValuePair>,
847}
848
849/// GOAWAY (0x10). Draft-19's GOAWAY has no Request ID field, so the
850/// control-stream and request-stream forms are identical on the wire.
851/// Draft-18 is the one draft that carries the field, and only when the
852/// message is sent on the control stream.
853#[derive(Debug, Clone, PartialEq, Eq)]
854pub struct GoAway {
855    pub new_session_uri: Vec<u8>,
856    pub timeout: VarInt,
857}
858
859// ============================================================
860// Consolidated Response Messages
861// ============================================================
862
863/// REQUEST_OK (0x07). Used as a generic OK response and as the alias for
864/// PUBLISH_OK / REQUEST_UPDATE_OK / TRACK_STATUS_OK / SUBSCRIBE_NAMESPACE_OK
865/// / PUBLISH_NAMESPACE_OK.
866///
867/// `track_properties` is only populated for TRACK_STATUS_OK; for every
868/// other shape it MUST be empty (length implicit from the message length).
869#[derive(Debug, Clone, PartialEq, Eq)]
870pub struct RequestOk {
871    pub parameters: Vec<KeyValuePair>,
872    pub track_properties: Vec<KeyValuePair>,
873}
874
875/// Optional Redirect structure carried in REQUEST_ERROR with code 0x34.
876#[derive(Debug, Clone, PartialEq, Eq)]
877pub struct Redirect {
878    pub connect_uri: Vec<u8>,
879    pub track_namespace: TrackNamespace,
880    pub track_name: Vec<u8>,
881}
882
883/// REQUEST_ERROR (0x05). Adds an optional Redirect structure when
884/// `error_code` is REDIRECT (0x34).
885#[derive(Debug, Clone, PartialEq, Eq)]
886pub struct RequestError {
887    pub error_code: VarInt,
888    pub retry_interval: VarInt,
889    pub reason_phrase: Vec<u8>,
890    pub redirect: Option<Redirect>,
891}
892
893/// REQUEST_ERROR error codes with dedicated meaning.
894///
895/// Note: DUPLICATE_SUBSCRIPTION (0x19) is removed in draft-19, as multiple
896/// concurrent subscriptions per Track are now allowed.
897pub mod request_error_codes {
898    /// A Mandatory Track Property the receiver does not understand.
899    pub const UNSUPPORTED_EXTENSION: u64 = 0x33;
900    /// Response carries a [`super::Redirect`] structure.
901    pub const REDIRECT: u64 = 0x34;
902    /// New in draft-19: SUBSCRIBE_TRACKS filter parameters conflict among too
903    /// many subscribers to aggregate the subscription upstream.
904    pub const CONFLICTING_FILTERS: u64 = 0x35;
905    /// New in draft-19: a Range Filter parameter is invalid or exceeds
906    /// MAX_FILTER_RANGES.
907    pub const INVALID_FILTER: u64 = 0x36;
908}
909
910// ============================================================
911// Subscribe Messages
912// ============================================================
913
914#[derive(Debug, Clone, PartialEq, Eq)]
915pub struct Subscribe {
916    pub request_id: VarInt,
917    pub track_namespace: TrackNamespace,
918    pub track_name: Vec<u8>,
919    pub parameters: Vec<KeyValuePair>,
920}
921
922/// SUBSCRIBE_OK (0x04).
923#[derive(Debug, Clone, PartialEq, Eq)]
924pub struct SubscribeOk {
925    pub track_alias: VarInt,
926    pub parameters: Vec<KeyValuePair>,
927    pub track_properties: Vec<KeyValuePair>,
928}
929
930#[derive(Debug, Clone, PartialEq, Eq)]
931pub struct RequestUpdate {
932    pub request_id: VarInt,
933    pub parameters: Vec<KeyValuePair>,
934}
935
936// ============================================================
937// Publish Messages
938// ============================================================
939
940#[derive(Debug, Clone, PartialEq, Eq)]
941pub struct Publish {
942    pub request_id: VarInt,
943    pub track_namespace: TrackNamespace,
944    pub track_name: Vec<u8>,
945    pub track_alias: VarInt,
946    pub parameters: Vec<KeyValuePair>,
947    pub track_properties: Vec<KeyValuePair>,
948}
949
950/// PUBLISH_DONE (0x0B). Status codes 0x5/0x6 are swapped vs draft-17.
951#[derive(Debug, Clone, PartialEq, Eq)]
952pub struct PublishDone {
953    pub status_code: VarInt,
954    pub stream_count: VarInt,
955    pub reason_phrase: Vec<u8>,
956}
957
958/// Numeric values for the [`PublishDone::status_code`] field.
959pub mod publish_done_codes {
960    /// Draft-18: TOO_FAR_BEHIND is 0x05 (was 0x06 in draft-17).
961    pub const TOO_FAR_BEHIND: u64 = 0x05;
962    /// Draft-18: EXPIRED is 0x06 (was 0x05 in draft-17).
963    pub const EXPIRED: u64 = 0x06;
964}
965
966// ============================================================
967// Publish Namespace Messages
968// ============================================================
969
970#[derive(Debug, Clone, PartialEq, Eq)]
971pub struct PublishNamespace {
972    pub request_id: VarInt,
973    pub track_namespace: TrackNamespace,
974    pub parameters: Vec<KeyValuePair>,
975}
976
977// ============================================================
978// Namespace Messages
979// ============================================================
980
981#[derive(Debug, Clone, PartialEq, Eq)]
982pub struct Namespace {
983    pub namespace_suffix: TrackNamespace,
984}
985
986#[derive(Debug, Clone, PartialEq, Eq)]
987pub struct NamespaceDone {
988    pub namespace_suffix: TrackNamespace,
989}
990
991// ============================================================
992// Subscribe Namespace / Tracks Messages
993// ============================================================
994
995/// SUBSCRIBE_NAMESPACE (0x50). Subscribes to NAMESPACE / NAMESPACE_DONE
996/// advertisements for namespaces matching `namespace_prefix`. The
997/// `subscribe_options` byte from draft-17 is removed; namespace subscriptions
998/// only produce NAMESPACE / NAMESPACE_DONE.
999#[derive(Debug, Clone, PartialEq, Eq)]
1000pub struct SubscribeNamespace {
1001    pub request_id: VarInt,
1002    pub namespace_prefix: TrackNamespace,
1003    pub parameters: Vec<KeyValuePair>,
1004}
1005
1006/// SUBSCRIBE_TRACKS (0x51, new in draft-18). Subscribes to PUBLISH messages
1007/// for tracks whose namespace matches `namespace_prefix`. Carries the FORWARD
1008/// parameter (Section 10.2.17), which on drafts 15 through 17 may appear on
1009/// SUBSCRIBE_NAMESPACE instead.
1010#[derive(Debug, Clone, PartialEq, Eq)]
1011pub struct SubscribeTracks {
1012    pub request_id: VarInt,
1013    pub namespace_prefix: TrackNamespace,
1014    pub parameters: Vec<KeyValuePair>,
1015}
1016
1017// ============================================================
1018// Track Status Messages
1019// ============================================================
1020
1021#[derive(Debug, Clone, PartialEq, Eq)]
1022pub struct TrackStatus {
1023    pub request_id: VarInt,
1024    pub track_namespace: TrackNamespace,
1025    pub track_name: Vec<u8>,
1026    pub parameters: Vec<KeyValuePair>,
1027}
1028
1029// ============================================================
1030// Fetch Messages
1031// ============================================================
1032
1033#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1034#[repr(u64)]
1035pub enum FetchType {
1036    Standalone = 1,
1037    RelativeJoining = 2,
1038    AbsoluteJoining = 3,
1039}
1040
1041impl FetchType {
1042    pub fn from_u64(v: u64) -> Option<Self> {
1043        match v {
1044            1 => Some(FetchType::Standalone),
1045            2 => Some(FetchType::RelativeJoining),
1046            3 => Some(FetchType::AbsoluteJoining),
1047            _ => None,
1048        }
1049    }
1050}
1051
1052#[derive(Debug, Clone, PartialEq, Eq)]
1053pub struct Fetch {
1054    pub request_id: VarInt,
1055    pub fetch_type: FetchType,
1056    pub fetch_payload: FetchPayload,
1057    pub parameters: Vec<KeyValuePair>,
1058}
1059
1060#[derive(Debug, Clone, PartialEq, Eq)]
1061pub enum FetchPayload {
1062    Standalone {
1063        track_namespace: TrackNamespace,
1064        track_name: Vec<u8>,
1065        start_group: VarInt,
1066        start_object: VarInt,
1067        end_group: VarInt,
1068        end_object: VarInt,
1069    },
1070    Joining {
1071        joining_request_id: VarInt,
1072        joining_start: VarInt,
1073    },
1074}
1075
1076/// FETCH_OK (0x18). `end_of_track` is uint8.
1077#[derive(Debug, Clone, PartialEq, Eq)]
1078pub struct FetchOk {
1079    pub end_of_track: u8,
1080    pub end_group: VarInt,
1081    pub end_object: VarInt,
1082    pub parameters: Vec<KeyValuePair>,
1083    pub track_properties: Vec<KeyValuePair>,
1084}
1085
1086// ============================================================
1087// Publish Skipped
1088// ============================================================
1089
1090/// PUBLISH_SKIPPED (0x0F). Drafts 17 and 18 name the same codepoint
1091/// PUBLISH_BLOCKED, with the same two fields.
1092#[derive(Debug, Clone, PartialEq, Eq)]
1093pub struct PublishSkipped {
1094    pub namespace_suffix: TrackNamespace,
1095    pub track_name: Vec<u8>,
1096}
1097
1098// ============================================================
1099// Unified Message Enum
1100// ============================================================
1101
1102#[derive(Debug, Clone, PartialEq, Eq)]
1103pub enum ControlMessage {
1104    Setup(Setup),
1105    GoAway(GoAway),
1106    RequestOk(RequestOk),
1107    RequestError(RequestError),
1108    Subscribe(Subscribe),
1109    SubscribeOk(SubscribeOk),
1110    RequestUpdate(RequestUpdate),
1111    Publish(Publish),
1112    PublishDone(PublishDone),
1113    PublishNamespace(PublishNamespace),
1114    Namespace(Namespace),
1115    NamespaceDone(NamespaceDone),
1116    SubscribeNamespace(SubscribeNamespace),
1117    SubscribeTracks(SubscribeTracks),
1118    TrackStatus(TrackStatus),
1119    Fetch(Fetch),
1120    FetchOk(FetchOk),
1121    PublishSkipped(PublishSkipped),
1122}
1123
1124/// Refuse a FETCH whose range ends before it starts.
1125///
1126/// Section 10.12.3: "Fetch specifies an inclusive range of Objects starting at
1127/// Start Location and ending at End Location. End Location MUST specify the
1128/// same or a larger Location than Start Location for Standalone and Absolute Joining Fetches." A Joining Fetch names
1129/// no explicit range - it is computed from the subscription it joins - so only
1130/// a standalone range is checked here.
1131///
1132/// SUBSCRIBE is not checked here, and needs no check: this draft's
1133/// AbsoluteRange filter carries an End Group Delta measured from the start
1134/// location rather than an absolute End Group, so an end before the start
1135/// has no encoding.
1136///
1137/// Applied on both sides. A range that ends before it starts selects nothing,
1138/// and the peer's only recourse is an error response or a session close, so
1139/// writing one is not a way to ask for anything.
1140fn check_ranges(message: &ControlMessage) -> Result<(), CodecError> {
1141    match message {
1142        ControlMessage::Fetch(m) => match &m.fetch_payload {
1143            FetchPayload::Standalone {
1144                start_group, start_object, end_group, end_object, ..
1145            } => check_location_range(
1146                start_group.into_inner(),
1147                start_object.into_inner(),
1148                end_group.into_inner(),
1149                end_object.into_inner(),
1150            ),
1151            FetchPayload::Joining { .. } => Ok(()),
1152        },
1153        _ => Ok(()),
1154    }
1155}
1156
1157/// Refuse a message whose discriminator disagrees with the fields beside it.
1158///
1159/// Two draft-19 messages carry a field that says which of the following fields
1160/// are on the wire: FETCH's Fetch Type, and REQUEST_ERROR's Error Code, whose
1161/// REDIRECT value (0x34) is what puts the Redirect structure on the wire. This
1162/// codec holds the alternatives in an enum and an `Option`, so a value can say
1163/// one thing in its discriminator and another in its body, and the two sides of
1164/// the codec resolve that differently — the encoder writes whatever the body
1165/// holds, and the decoder reads whatever the discriminator announces.
1166///
1167/// The result is a message that does not survive its own round trip:
1168///
1169/// - A FETCH whose type says Standalone and whose body is a joining pair
1170///   encodes to a joining request id and a joining start where a Track
1171///   Namespace and a Track Name belong, and comes back as a Standalone fetch of
1172///   a track named after two integers — or, more often, as an error, which at
1173///   least is honest. The two joining types share one body shape, so the check
1174///   is between Standalone and everything else rather than one arm per type.
1175/// - A REQUEST_ERROR with code REDIRECT and no Redirect body encodes to a
1176///   message that ends where the decoder expects a Connect URI length, so the
1177///   peer reads the redirect out of whatever follows or runs off the end. The
1178///   mirror case is quieter and no better: a Redirect body under any other
1179///   error code is written out and then skipped by a decoder that was never
1180///   told to look for it, so the sender believes it redirected a peer that
1181///   never saw a redirect.
1182///
1183/// Refusing at the encoder keeps the two readings from ever diverging on the
1184/// wire.
1185fn check_discriminators(message: &ControlMessage) -> Result<(), CodecError> {
1186    match message {
1187        ControlMessage::Fetch(m) => {
1188            let body_is_standalone = matches!(m.fetch_payload, FetchPayload::Standalone { .. });
1189            if body_is_standalone != (m.fetch_type == FetchType::Standalone) {
1190                return Err(CodecError::InvalidField);
1191            }
1192        }
1193        ControlMessage::RequestError(m) => {
1194            let code_is_redirect = m.error_code.into_inner() == request_error_codes::REDIRECT;
1195            if code_is_redirect != m.redirect.is_some() {
1196                return Err(CodecError::InvalidField);
1197            }
1198        }
1199        _ => {}
1200    }
1201    Ok(())
1202}
1203
1204/// Whether draft-19 lets Message Parameter `key` appear in `message`.
1205///
1206/// Section 10.2.1: "Each Message Parameter definition indicates the message
1207/// types in which it can appear. If it appears in some other type of message,
1208/// the receiving endpoint MUST close the connection with a PROTOCOL_VIOLATION."
1209/// One arm per entry in the Message Parameters registry (Section 15.7),
1210/// carrying the message types that entry's own definition names.
1211///
1212/// Three things about this draft the arms below fold in:
1213///
1214/// * Six of the names are one wire type. Section 10.5: "This document uses the
1215///   shorthand PUBLISH_OK, REQUEST_UPDATE_OK, TRACK_STATUS_OK,
1216///   SUBSCRIBE_NAMESPACE_OK, and PUBLISH_NAMESPACE_OK to refer to a REQUEST_OK
1217///   sent in response to the corresponding request type", and the same section
1218///   sends a REQUEST_OK in answer to SUBSCRIBE_TRACKS as well. Which one a
1219///   given REQUEST_OK is depends on the request its Request ID answers, which
1220///   is session state and not in the frame, so each of those names widens the
1221///   same arm and a REQUEST_OK is held to their union.
1222/// * The five Range Filters state their scope in Section 5.1.3 rather than in
1223///   their own subsections: the Track Property filter "MAY appear multiple
1224///   times in a SUBSCRIBE_TRACKS message or REQUEST_UPDATE for it", and "all
1225///   other filter parameters MAY appear multiple times in a FETCH, SUBSCRIBE,
1226///   SUBSCRIBE_TRACKS, PUBLISH_OK, or REQUEST_UPDATE" message. A parameter
1227///   definition is free to state its scope elsewhere, and these do.
1228/// * SUBSCRIBE_TRACKS inherits SUBSCRIBE's whole set. Section 10.19.1: "Any
1229///   Parameter that can be specified on a Subscription (ie: in SUBSCRIBE) is
1230///   valid in SUBSCRIBE_TRACKS, unless otherwise specified." Draft-18 has no
1231///   such sentence, which is why its SUBSCRIBE_TRACKS admits two types and
1232///   this one admits fourteen.
1233///
1234/// FETCH_OK has no arm in the table below, and that is the draft's doing rather
1235/// than an omission here: Section 10.13 gives it a Parameters field and no
1236/// parameter definition names it, so every type this draft defines is "some
1237/// other type of message" there.
1238///
1239/// The table decides scope only. A type this draft does not define has no scope
1240/// to be outside of and is answered by [`CodecError::UnknownMessageParameter`],
1241/// which is why the final arm carries rather than refuses.
1242fn parameter_in_scope(key: u64, message: MessageType) -> bool {
1243    use MessageType as M;
1244    // Section 10.19.1 makes SUBSCRIBE_TRACKS a superset of SUBSCRIBE, so every
1245    // arm admitting one admits the other. The arms spell both out rather than
1246    // wrapping the call, so each still reads against its own sentence.
1247    match key {
1248        // Section 10.2.4 OBJECT_DELIVERY_TIMEOUT: "It MAY appear in a
1249        // PUBLISH_OK, SUBSCRIBE, or REQUEST_UPDATE message."
1250        0x02 => {
1251            matches!(message, M::RequestOk | M::Subscribe | M::RequestUpdate | M::SubscribeTracks)
1252        }
1253        // Section 10.2.2 AUTHORIZATION TOKEN: "It MAY appear in a PUBLISH,
1254        // SUBSCRIBE, REQUEST_UPDATE, SUBSCRIBE_NAMESPACE, SUBSCRIBE_TRACKS,
1255        // PUBLISH_NAMESPACE, TRACK_STATUS or FETCH message."
1256        0x03 => matches!(
1257            message,
1258            M::Publish
1259                | M::Subscribe
1260                | M::RequestUpdate
1261                | M::SubscribeNamespace
1262                | M::SubscribeTracks
1263                | M::PublishNamespace
1264                | M::TrackStatus
1265                | M::Fetch
1266        ),
1267        // Section 10.2.6 RENDEZVOUS TIMEOUT: it "MAY appear in a SUBSCRIBE
1268        // message".
1269        0x04 => matches!(message, M::Subscribe | M::SubscribeTracks),
1270        // Section 10.2.3 SUBGROUP_DELIVERY_TIMEOUT: "It MAY appear in a
1271        // PUBLISH_OK, SUBSCRIBE, or REQUEST_UPDATE message."
1272        0x06 => {
1273            matches!(message, M::RequestOk | M::Subscribe | M::RequestUpdate | M::SubscribeTracks)
1274        }
1275        // Section 10.2.15 EXPIRES: "It MAY appear in SUBSCRIBE_OK, PUBLISH,
1276        // PUBLISH_OK, SUBSCRIBE_NAMESPACE_OK, SUBSCRIBE_TRACKS_OK,
1277        // PUBLISH_NAMESPACE_OK, or REQUEST_UPDATE_OK." Five of those seven are
1278        // a REQUEST_OK.
1279        0x08 => matches!(message, M::SubscribeOk | M::Publish | M::RequestOk),
1280        // Section 10.2.16 LARGEST OBJECT: "It MAY appear in SUBSCRIBE_OK,
1281        // PUBLISH, REQUEST_UPDATE_OK, or TRACK_STATUS_OK."
1282        0x09 => matches!(message, M::SubscribeOk | M::Publish | M::RequestOk),
1283        // Section 10.2.5 FILL TIMEOUT: it "MAY appear in a FETCH message".
1284        0x0A => matches!(message, M::Fetch),
1285        // Section 10.2.17 FORWARD: "It MAY appear in SUBSCRIBE, REQUEST_UPDATE
1286        // (for a subscription), PUBLISH, PUBLISH_OK and SUBSCRIBE_TRACKS."
1287        0x10 => matches!(
1288            message,
1289            M::Subscribe | M::RequestUpdate | M::Publish | M::RequestOk | M::SubscribeTracks
1290        ),
1291        // Section 10.2.7 SUBSCRIBER PRIORITY: "It MAY appear in a SUBSCRIBE,
1292        // FETCH, REQUEST_UPDATE (for a subscription or FETCH), or PUBLISH_OK
1293        // message."
1294        0x20 => matches!(
1295            message,
1296            M::Subscribe | M::Fetch | M::RequestUpdate | M::RequestOk | M::SubscribeTracks
1297        ),
1298        // Section 10.2.9 LOCATION FILTER: "It MAY appear in a SUBSCRIBE,
1299        // PUBLISH_OK or REQUEST_UPDATE (for a subscription) message."
1300        0x21 => {
1301            matches!(message, M::Subscribe | M::RequestOk | M::RequestUpdate | M::SubscribeTracks)
1302        }
1303        // Section 10.2.8 GROUP ORDER: "It MAY appear in a SUBSCRIBE,
1304        // SUBSCRIBE_TRACKS, or FETCH."
1305        0x22 => matches!(message, M::Subscribe | M::SubscribeTracks | M::Fetch),
1306        // Section 5.1.3: "All other filter parameters MAY appear multiple times
1307        // in a FETCH, SUBSCRIBE, SUBSCRIBE_TRACKS, PUBLISH_OK, or
1308        // REQUEST_UPDATE (on a subscription, from the subscriber only)
1309        // message." SUBGROUP_FILTER (Section 10.2.10), OBJECTID_FILTER
1310        // (10.2.11), PRIORITY_FILTER (10.2.12) and OBJECT_PROPERTY_FILTER
1311        // (10.2.13) are those four.
1312        0x25..=0x28 => matches!(
1313            message,
1314            M::Fetch | M::Subscribe | M::SubscribeTracks | M::RequestOk | M::RequestUpdate
1315        ),
1316        // Section 5.1.3, of TRACK_PROPERTY_FILTER (Section 10.2.14) alone: it
1317        // "MAY appear multiple times in a SUBSCRIBE_TRACKS message or
1318        // REQUEST_UPDATE for it". It selects tracks rather than objects, which
1319        // is why it is the one filter a SUBSCRIBE may not carry.
1320        0x29 => matches!(message, M::SubscribeTracks | M::RequestUpdate),
1321        // Section 10.2.18 NEW GROUP REQUEST: "It MAY appear in PUBLISH_OK,
1322        // SUBSCRIBE or REQUEST_UPDATE for a subscription."
1323        0x32 => {
1324            matches!(message, M::RequestOk | M::Subscribe | M::RequestUpdate | M::SubscribeTracks)
1325        }
1326        // Section 10.2.19 TRACK_NAMESPACE_PREFIX: "It MAY appear in
1327        // REQUEST_UPDATE for a SUBSCRIBE_NAMESPACE or SUBSCRIBE_TRACKS
1328        // request." The two named there are the request being updated, not two
1329        // more places the parameter may be written.
1330        0x34 => matches!(message, M::RequestUpdate),
1331        _ => true,
1332    }
1333}
1334
1335/// Refuse a message carrying a Message Parameter its own definition does not
1336/// place there.
1337///
1338/// Section 10.2.1 answers this with a close, which the drafts below do not.
1339/// Draft-16 Section 9.2.2, and drafts 07 through 15 under the older name
1340/// Version Specific Parameters, end the same sentence "it MUST be ignored" —
1341/// so this check belongs to drafts 17, 18 and 19 and to no draft before them.
1342///
1343/// Applied on both sides. A parameter outside its scope is one the peer must
1344/// close the session over, so writing one is a way to end a session rather than
1345/// a way to ask for anything.
1346fn check_parameter_scope(message: &ControlMessage) -> Result<(), CodecError> {
1347    let parameters = match message {
1348        ControlMessage::RequestOk(m) => &m.parameters,
1349        ControlMessage::Subscribe(m) => &m.parameters,
1350        ControlMessage::SubscribeOk(m) => &m.parameters,
1351        ControlMessage::RequestUpdate(m) => &m.parameters,
1352        ControlMessage::Publish(m) => &m.parameters,
1353        ControlMessage::PublishNamespace(m) => &m.parameters,
1354        ControlMessage::SubscribeNamespace(m) => &m.parameters,
1355        ControlMessage::SubscribeTracks(m) => &m.parameters,
1356        ControlMessage::TrackStatus(m) => &m.parameters,
1357        ControlMessage::Fetch(m) => &m.parameters,
1358        ControlMessage::FetchOk(m) => &m.parameters,
1359        // No Message Parameters field. SETUP is named here rather than left to
1360        // a wildcard because the draft says why it can never have one: Section
1361        // 10.2.1 notes that "since Setup Options use a separate namespace, it
1362        // is impossible for Message Parameters to appear in Setup messages",
1363        // and this codec keeps the two namespaces in separate fields.
1364        ControlMessage::Setup(_)
1365        | ControlMessage::GoAway(_)
1366        | ControlMessage::RequestError(_)
1367        | ControlMessage::PublishDone(_)
1368        | ControlMessage::Namespace(_)
1369        | ControlMessage::NamespaceDone(_)
1370        | ControlMessage::PublishSkipped(_) => return Ok(()),
1371    };
1372
1373    let message_type = message.message_type();
1374    for parameter in parameters {
1375        let key = parameter.key.into_inner();
1376        if !parameter_in_scope(key, message_type) {
1377            return Err(CodecError::ParameterOutOfScope { key, message_type: message_type.id() });
1378        }
1379    }
1380    Ok(())
1381}
1382
1383impl ControlMessage {
1384    pub fn encode(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1385        check_discriminators(self)?;
1386        check_ranges(self)?;
1387        check_parameter_scope(self)?;
1388        let mut body = Vec::with_capacity(256);
1389        self.encode_body(&mut body)?;
1390
1391        if body.len() > MAX_MESSAGE_LENGTH {
1392            return Err(CodecError::MessageTooLong(body.len()));
1393        }
1394
1395        let msg_type = self.message_type();
1396        VarInt::from_usize(msg_type.id() as usize).encode_moqt::<Wire>(buf);
1397        // Draft-19: 16-bit length (big-endian)
1398        buf.put_u16(body.len() as u16);
1399        buf.put_slice(&body);
1400        Ok(())
1401    }
1402
1403    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1404        let type_id = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
1405        let msg_type =
1406            MessageType::from_id(type_id).ok_or(CodecError::UnknownMessageType(type_id))?;
1407        // Draft-19: 16-bit length (big-endian)
1408        if buf.remaining() < 2 {
1409            return Err(CodecError::UnexpectedEnd);
1410        }
1411        let body_len = buf.get_u16() as usize;
1412        if buf.remaining() < body_len {
1413            return Err(CodecError::UnexpectedEnd);
1414        }
1415        let body_bytes = buf.copy_to_bytes(body_len);
1416        let mut body = &body_bytes[..];
1417        let msg = match Self::decode_body(msg_type, &mut body) {
1418            Ok(msg) => msg,
1419            // The fields wanted more bytes than the Length allowed. This buffer
1420            // is already bounded by that Length, so running out inside it cannot
1421            // mean the message is still arriving - which is what the same error
1422            // means everywhere else, and why a reader loops on it rather than
1423            // closing. Here there is nothing left to arrive.
1424            Err(
1425                CodecError::UnexpectedEnd
1426                | CodecError::Kvp(crate::kvp::KvpError::UnexpectedEnd)
1427                | CodecError::Kvp(crate::kvp::KvpError::VarInt(
1428                    crate::varint::VarIntError::UnexpectedEnd,
1429                ))
1430                | CodecError::VarInt(crate::varint::VarIntError::UnexpectedEnd),
1431            ) => {
1432                return Err(CodecError::ControlMessageLengthMismatch {
1433                    declared: body_len,
1434                    detail: "its fields ran past the end",
1435                });
1436            }
1437            Err(e) => return Err(e),
1438        };
1439        check_ranges(&msg)?;
1440        check_parameter_scope(&msg)?;
1441        // Draft-19 Section 10: "If the length does not match the length of the
1442        // Message Body, the receiver MUST close the session with a
1443        // PROTOCOL_VIOLATION." A body parser that stops short leaves bytes
1444        // here; without this the surplus is discarded and a truncated or
1445        // mis-framed field looks like a well-formed message.
1446        if body.has_remaining() {
1447            return Err(CodecError::ControlMessageLengthMismatch {
1448                declared: body_len,
1449                detail: "its fields left bytes unread",
1450            });
1451        }
1452        Ok(msg)
1453    }
1454
1455    fn encode_body(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1456        match self {
1457            ControlMessage::Setup(m) => {
1458                encode_setup_options(&m.options, buf)?;
1459            }
1460            ControlMessage::GoAway(m) => {
1461                if m.new_session_uri.len() > MAX_GOAWAY_URI_LENGTH {
1462                    return Err(CodecError::GoAwayUriTooLong);
1463                }
1464                VarInt::from_usize(m.new_session_uri.len()).encode_moqt::<Wire>(buf);
1465                buf.put_slice(&m.new_session_uri);
1466                m.timeout.encode_moqt::<Wire>(buf);
1467            }
1468            ControlMessage::RequestOk(m) => {
1469                encode_parameters(&m.parameters, buf)?;
1470                encode_track_properties(&m.track_properties, buf)?;
1471            }
1472            ControlMessage::RequestError(m) => {
1473                if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1474                    return Err(CodecError::ReasonPhraseTooLong);
1475                }
1476                m.error_code.encode_moqt::<Wire>(buf);
1477                m.retry_interval.encode_moqt::<Wire>(buf);
1478                VarInt::from_usize(m.reason_phrase.len()).encode_moqt::<Wire>(buf);
1479                buf.put_slice(&m.reason_phrase);
1480                if let Some(r) = &m.redirect {
1481                    r.track_namespace.validate_moqt()?;
1482                    check_full_track_name(&r.track_namespace, &r.track_name)?;
1483                    VarInt::from_usize(r.connect_uri.len()).encode_moqt::<Wire>(buf);
1484                    buf.put_slice(&r.connect_uri);
1485                    r.track_namespace.encode_moqt::<Wire>(buf);
1486                    VarInt::from_usize(r.track_name.len()).encode_moqt::<Wire>(buf);
1487                    buf.put_slice(&r.track_name);
1488                }
1489            }
1490            ControlMessage::Subscribe(m) => {
1491                m.track_namespace.validate_moqt()?;
1492                check_full_track_name(&m.track_namespace, &m.track_name)?;
1493                m.request_id.encode_moqt::<Wire>(buf);
1494                m.track_namespace.encode_moqt::<Wire>(buf);
1495                VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
1496                buf.put_slice(&m.track_name);
1497                encode_parameters(&m.parameters, buf)?;
1498            }
1499            ControlMessage::SubscribeOk(m) => {
1500                m.track_alias.encode_moqt::<Wire>(buf);
1501                encode_parameters(&m.parameters, buf)?;
1502                encode_track_properties(&m.track_properties, buf)?;
1503            }
1504            ControlMessage::RequestUpdate(m) => {
1505                m.request_id.encode_moqt::<Wire>(buf);
1506                encode_parameters(&m.parameters, buf)?;
1507            }
1508            ControlMessage::Publish(m) => {
1509                m.track_namespace.validate_moqt()?;
1510                check_full_track_name(&m.track_namespace, &m.track_name)?;
1511                m.request_id.encode_moqt::<Wire>(buf);
1512                m.track_namespace.encode_moqt::<Wire>(buf);
1513                VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
1514                buf.put_slice(&m.track_name);
1515                m.track_alias.encode_moqt::<Wire>(buf);
1516                encode_parameters(&m.parameters, buf)?;
1517                encode_track_properties(&m.track_properties, buf)?;
1518            }
1519            ControlMessage::PublishDone(m) => {
1520                if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1521                    return Err(CodecError::ReasonPhraseTooLong);
1522                }
1523                m.status_code.encode_moqt::<Wire>(buf);
1524                m.stream_count.encode_moqt::<Wire>(buf);
1525                VarInt::from_usize(m.reason_phrase.len()).encode_moqt::<Wire>(buf);
1526                buf.put_slice(&m.reason_phrase);
1527            }
1528            ControlMessage::PublishNamespace(m) => {
1529                m.track_namespace.validate_moqt()?;
1530                m.request_id.encode_moqt::<Wire>(buf);
1531                m.track_namespace.encode_moqt::<Wire>(buf);
1532                encode_parameters(&m.parameters, buf)?;
1533            }
1534            ControlMessage::Namespace(m) => {
1535                m.namespace_suffix.validate_moqt()?;
1536                m.namespace_suffix.encode_moqt::<Wire>(buf);
1537            }
1538            ControlMessage::NamespaceDone(m) => {
1539                m.namespace_suffix.validate_moqt()?;
1540                m.namespace_suffix.encode_moqt::<Wire>(buf);
1541            }
1542            ControlMessage::SubscribeNamespace(m) => {
1543                m.namespace_prefix.validate_moqt()?;
1544                m.request_id.encode_moqt::<Wire>(buf);
1545                m.namespace_prefix.encode_moqt::<Wire>(buf);
1546                encode_parameters(&m.parameters, buf)?;
1547            }
1548            ControlMessage::SubscribeTracks(m) => {
1549                m.namespace_prefix.validate_moqt()?;
1550                m.request_id.encode_moqt::<Wire>(buf);
1551                m.namespace_prefix.encode_moqt::<Wire>(buf);
1552                encode_parameters(&m.parameters, buf)?;
1553            }
1554            ControlMessage::TrackStatus(m) => {
1555                m.track_namespace.validate_moqt()?;
1556                check_full_track_name(&m.track_namespace, &m.track_name)?;
1557                m.request_id.encode_moqt::<Wire>(buf);
1558                m.track_namespace.encode_moqt::<Wire>(buf);
1559                VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
1560                buf.put_slice(&m.track_name);
1561                encode_parameters(&m.parameters, buf)?;
1562            }
1563            ControlMessage::Fetch(m) => {
1564                m.request_id.encode_moqt::<Wire>(buf);
1565                VarInt::from_usize(m.fetch_type as usize).encode_moqt::<Wire>(buf);
1566                match &m.fetch_payload {
1567                    FetchPayload::Standalone {
1568                        track_namespace,
1569                        track_name,
1570                        start_group,
1571                        start_object,
1572                        end_group,
1573                        end_object,
1574                    } => {
1575                        track_namespace.validate_moqt()?;
1576                        check_full_track_name(track_namespace, track_name)?;
1577                        track_namespace.encode_moqt::<Wire>(buf);
1578                        VarInt::from_usize(track_name.len()).encode_moqt::<Wire>(buf);
1579                        buf.put_slice(track_name);
1580                        start_group.encode_moqt::<Wire>(buf);
1581                        start_object.encode_moqt::<Wire>(buf);
1582                        end_group.encode_moqt::<Wire>(buf);
1583                        end_object.encode_moqt::<Wire>(buf);
1584                    }
1585                    FetchPayload::Joining { joining_request_id, joining_start } => {
1586                        joining_request_id.encode_moqt::<Wire>(buf);
1587                        joining_start.encode_moqt::<Wire>(buf);
1588                    }
1589                }
1590                encode_parameters(&m.parameters, buf)?;
1591            }
1592            ControlMessage::FetchOk(m) => {
1593                buf.put_u8(m.end_of_track);
1594                m.end_group.encode_moqt::<Wire>(buf);
1595                m.end_object.encode_moqt::<Wire>(buf);
1596                encode_parameters(&m.parameters, buf)?;
1597                encode_track_properties(&m.track_properties, buf)?;
1598            }
1599            ControlMessage::PublishSkipped(m) => {
1600                m.namespace_suffix.validate_moqt()?;
1601                check_full_track_name(&m.namespace_suffix, &m.track_name)?;
1602                m.namespace_suffix.encode_moqt::<Wire>(buf);
1603                VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
1604                buf.put_slice(&m.track_name);
1605            }
1606        }
1607        Ok(())
1608    }
1609
1610    fn decode_body(msg_type: MessageType, buf: &mut impl Buf) -> Result<Self, CodecError> {
1611        match msg_type {
1612            MessageType::Setup => {
1613                let options = decode_setup_options(buf)?;
1614                Ok(ControlMessage::Setup(Setup { options }))
1615            }
1616            MessageType::GoAway => {
1617                let uri_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1618                // Draft-19 Section 10.4: an endpoint that receives a New
1619                // Session URI Length above the maximum MUST close the session
1620                // with a PROTOCOL_VIOLATION. Checked here as well as on encode
1621                // so an oversized URI never reaches the application.
1622                if uri_len > MAX_GOAWAY_URI_LENGTH {
1623                    return Err(CodecError::GoAwayUriTooLong);
1624                }
1625                let uri = read_bytes(buf, uri_len)?;
1626                let timeout = VarInt::decode_moqt::<Wire>(buf)?;
1627                Ok(ControlMessage::GoAway(GoAway { new_session_uri: uri, timeout }))
1628            }
1629            MessageType::RequestOk => {
1630                let parameters = decode_parameters(buf)?;
1631                let track_properties = decode_track_properties(buf)?;
1632                Ok(ControlMessage::RequestOk(RequestOk { parameters, track_properties }))
1633            }
1634            MessageType::RequestError => {
1635                let error_code = VarInt::decode_moqt::<Wire>(buf)?;
1636                let retry_interval = VarInt::decode_moqt::<Wire>(buf)?;
1637                let reason_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1638                // Draft-19 Section 1.4.4: a received reason phrase length above
1639                // the maximum MUST close the session with a PROTOCOL_VIOLATION.
1640                if reason_len > MAX_REASON_PHRASE_LENGTH {
1641                    return Err(CodecError::ReasonPhraseTooLong);
1642                }
1643                let reason_phrase = read_bytes(buf, reason_len)?;
1644                let redirect = if error_code.into_inner() == request_error_codes::REDIRECT {
1645                    let uri_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1646                    let connect_uri = read_bytes(buf, uri_len)?;
1647                    let track_namespace = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1648                    let name_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1649                    let track_name = read_bytes(buf, name_len)?;
1650                    check_full_track_name(&track_namespace, &track_name)?;
1651                    Some(Redirect { connect_uri, track_namespace, track_name })
1652                } else {
1653                    None
1654                };
1655                Ok(ControlMessage::RequestError(RequestError {
1656                    error_code,
1657                    retry_interval,
1658                    reason_phrase,
1659                    redirect,
1660                }))
1661            }
1662            MessageType::Subscribe => {
1663                let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1664                let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
1665                let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1666                let track_name = read_bytes(buf, tn_len)?;
1667                check_full_track_name(&track_namespace, &track_name)?;
1668                let parameters = decode_parameters(buf)?;
1669                Ok(ControlMessage::Subscribe(Subscribe {
1670                    request_id,
1671                    track_namespace,
1672                    track_name,
1673                    parameters,
1674                }))
1675            }
1676            MessageType::SubscribeOk => {
1677                let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
1678                let parameters = decode_parameters(buf)?;
1679                let track_properties = decode_track_properties(buf)?;
1680                Ok(ControlMessage::SubscribeOk(SubscribeOk {
1681                    track_alias,
1682                    parameters,
1683                    track_properties,
1684                }))
1685            }
1686            MessageType::RequestUpdate => {
1687                let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1688                let parameters = decode_parameters(buf)?;
1689                Ok(ControlMessage::RequestUpdate(RequestUpdate { request_id, parameters }))
1690            }
1691            MessageType::Publish => {
1692                let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1693                let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
1694                let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1695                let track_name = read_bytes(buf, tn_len)?;
1696                let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
1697                check_full_track_name(&track_namespace, &track_name)?;
1698                let parameters = decode_parameters(buf)?;
1699                let track_properties = decode_track_properties(buf)?;
1700                Ok(ControlMessage::Publish(Publish {
1701                    request_id,
1702                    track_namespace,
1703                    track_name,
1704                    track_alias,
1705                    parameters,
1706                    track_properties,
1707                }))
1708            }
1709            MessageType::PublishDone => {
1710                let status_code = VarInt::decode_moqt::<Wire>(buf)?;
1711                let stream_count = VarInt::decode_moqt::<Wire>(buf)?;
1712                let reason_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1713                // Draft-19 Section 1.4.4, same bound as REQUEST_ERROR above.
1714                if reason_len > MAX_REASON_PHRASE_LENGTH {
1715                    return Err(CodecError::ReasonPhraseTooLong);
1716                }
1717                let reason_phrase = read_bytes(buf, reason_len)?;
1718                Ok(ControlMessage::PublishDone(PublishDone {
1719                    status_code,
1720                    stream_count,
1721                    reason_phrase,
1722                }))
1723            }
1724            MessageType::PublishNamespace => {
1725                let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1726                let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
1727                let parameters = decode_parameters(buf)?;
1728                Ok(ControlMessage::PublishNamespace(PublishNamespace {
1729                    request_id,
1730                    track_namespace,
1731                    parameters,
1732                }))
1733            }
1734            MessageType::Namespace => {
1735                let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1736                Ok(ControlMessage::Namespace(Namespace { namespace_suffix }))
1737            }
1738            MessageType::NamespaceDone => {
1739                let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1740                Ok(ControlMessage::NamespaceDone(NamespaceDone { namespace_suffix }))
1741            }
1742            MessageType::SubscribeNamespace => {
1743                let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1744                let namespace_prefix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1745                let parameters = decode_parameters(buf)?;
1746                Ok(ControlMessage::SubscribeNamespace(SubscribeNamespace {
1747                    request_id,
1748                    namespace_prefix,
1749                    parameters,
1750                }))
1751            }
1752            MessageType::SubscribeTracks => {
1753                let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1754                let namespace_prefix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1755                let parameters = decode_parameters(buf)?;
1756                Ok(ControlMessage::SubscribeTracks(SubscribeTracks {
1757                    request_id,
1758                    namespace_prefix,
1759                    parameters,
1760                }))
1761            }
1762            MessageType::TrackStatus => {
1763                let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1764                let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
1765                let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1766                let track_name = read_bytes(buf, tn_len)?;
1767                check_full_track_name(&track_namespace, &track_name)?;
1768                let parameters = decode_parameters(buf)?;
1769                Ok(ControlMessage::TrackStatus(TrackStatus {
1770                    request_id,
1771                    track_namespace,
1772                    track_name,
1773                    parameters,
1774                }))
1775            }
1776            MessageType::Fetch => {
1777                let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1778                let fetch_type_val = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
1779                let fetch_type = FetchType::from_u64(fetch_type_val)
1780                    .ok_or(CodecError::InvalidFetchType(fetch_type_val))?;
1781                let fetch_payload = match fetch_type {
1782                    FetchType::Standalone => {
1783                        let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
1784                        let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1785                        let track_name = read_bytes(buf, tn_len)?;
1786                        let start_group = VarInt::decode_moqt::<Wire>(buf)?;
1787                        let start_object = VarInt::decode_moqt::<Wire>(buf)?;
1788                        let end_group = VarInt::decode_moqt::<Wire>(buf)?;
1789                        let end_object = VarInt::decode_moqt::<Wire>(buf)?;
1790                        check_full_track_name(&track_namespace, &track_name)?;
1791                        FetchPayload::Standalone {
1792                            track_namespace,
1793                            track_name,
1794                            start_group,
1795                            start_object,
1796                            end_group,
1797                            end_object,
1798                        }
1799                    }
1800                    FetchType::RelativeJoining | FetchType::AbsoluteJoining => {
1801                        let joining_request_id = VarInt::decode_moqt::<Wire>(buf)?;
1802                        let joining_start = VarInt::decode_moqt::<Wire>(buf)?;
1803                        FetchPayload::Joining { joining_request_id, joining_start }
1804                    }
1805                };
1806                let parameters = decode_parameters(buf)?;
1807                Ok(ControlMessage::Fetch(Fetch {
1808                    request_id,
1809                    fetch_type,
1810                    fetch_payload,
1811                    parameters,
1812                }))
1813            }
1814            MessageType::FetchOk => {
1815                if buf.remaining() < 1 {
1816                    return Err(CodecError::UnexpectedEnd);
1817                }
1818                let end_of_track = buf.get_u8();
1819                let end_group = VarInt::decode_moqt::<Wire>(buf)?;
1820                let end_object = VarInt::decode_moqt::<Wire>(buf)?;
1821                let parameters = decode_parameters(buf)?;
1822                let track_properties = decode_track_properties(buf)?;
1823                Ok(ControlMessage::FetchOk(FetchOk {
1824                    end_of_track,
1825                    end_group,
1826                    end_object,
1827                    parameters,
1828                    track_properties,
1829                }))
1830            }
1831            MessageType::PublishSkipped => {
1832                let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1833                let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1834                let track_name = read_bytes(buf, tn_len)?;
1835                check_full_track_name(&namespace_suffix, &track_name)?;
1836                Ok(ControlMessage::PublishSkipped(PublishSkipped { namespace_suffix, track_name }))
1837            }
1838        }
1839    }
1840
1841    pub fn message_type(&self) -> MessageType {
1842        match self {
1843            ControlMessage::Setup(_) => MessageType::Setup,
1844            ControlMessage::GoAway(_) => MessageType::GoAway,
1845            ControlMessage::RequestOk(_) => MessageType::RequestOk,
1846            ControlMessage::RequestError(_) => MessageType::RequestError,
1847            ControlMessage::Subscribe(_) => MessageType::Subscribe,
1848            ControlMessage::SubscribeOk(_) => MessageType::SubscribeOk,
1849            ControlMessage::RequestUpdate(_) => MessageType::RequestUpdate,
1850            ControlMessage::Publish(_) => MessageType::Publish,
1851            ControlMessage::PublishDone(_) => MessageType::PublishDone,
1852            ControlMessage::PublishNamespace(_) => MessageType::PublishNamespace,
1853            ControlMessage::Namespace(_) => MessageType::Namespace,
1854            ControlMessage::NamespaceDone(_) => MessageType::NamespaceDone,
1855            ControlMessage::SubscribeNamespace(_) => MessageType::SubscribeNamespace,
1856            ControlMessage::SubscribeTracks(_) => MessageType::SubscribeTracks,
1857            ControlMessage::TrackStatus(_) => MessageType::TrackStatus,
1858            ControlMessage::Fetch(_) => MessageType::Fetch,
1859            ControlMessage::FetchOk(_) => MessageType::FetchOk,
1860            ControlMessage::PublishSkipped(_) => MessageType::PublishSkipped,
1861        }
1862    }
1863}
1864
1865#[cfg(test)]
1866mod tests {
1867    use super::*;
1868
1869    /// Frame `body` as a draft-19 control message of `type_id`, declaring
1870    /// `declared_len` rather than the body's real length. Used to build the
1871    /// mismatched frame the length rule is about.
1872    fn frame_with_declared_len(type_id: u64, declared_len: u16, body: &[u8]) -> Vec<u8> {
1873        let mut out = Vec::new();
1874        VarInt::from_u64_moqt(type_id).encode_moqt::<Wire>(&mut out);
1875        out.put_u16(declared_len);
1876        out.put_slice(body);
1877        out
1878    }
1879
1880    fn frame(type_id: u64, body: &[u8]) -> Vec<u8> {
1881        frame_with_declared_len(type_id, body.len() as u16, body)
1882    }
1883
1884    /// A SUBSCRIBE body: request id 1, namespace ("a"), track name "b", and
1885    /// `params` already encoded.
1886    fn subscribe_body(params: &[u8]) -> Vec<u8> {
1887        let mut body = vec![0x01, 0x01, 0x01, b'a', 0x01, b'b'];
1888        body.extend_from_slice(params);
1889        body
1890    }
1891
1892    /// Draft-19 Section 10: "If the length does not match the length of the
1893    /// Message Body, the receiver MUST close the session with a
1894    /// PROTOCOL_VIOLATION."
1895    ///
1896    /// Without the trailing-byte check in `decode` this SUBSCRIBE parses and
1897    /// the two surplus bytes vanish:
1898    ///
1899    /// ```text
1900    /// assertion `left == right` failed
1901    ///   left: Ok(Subscribe(Subscribe { request_id: VarInt(1), track_namespace:
1902    ///         TrackNamespace([[97]]), track_name: [98], parameters: [] }))
1903    ///  right: Err(InvalidField)
1904    /// ```
1905    #[test]
1906    fn a_message_body_shorter_than_the_declared_length_is_refused() {
1907        let body = subscribe_body(&[0x00]);
1908        let mut junked = body.clone();
1909        junked.extend_from_slice(&[0xff, 0xff]);
1910        let bytes = frame_with_declared_len(0x03, (body.len() + 2) as u16, &junked);
1911
1912        let mut buf = &bytes[..];
1913        assert_eq!(
1914            ControlMessage::decode(&mut buf),
1915            Err(CodecError::ControlMessageLengthMismatch {
1916                declared: (body.len() + 2),
1917                detail: "its fields left bytes unread",
1918            })
1919        );
1920
1921        // The same body with an honest length still decodes, so the guard
1922        // rejects the mismatch and not the message.
1923        let honest = frame(0x03, &body);
1924        let mut buf = &honest[..];
1925        assert!(ControlMessage::decode(&mut buf).is_ok());
1926    }
1927
1928    /// Draft-19 Section 1.4.4: "The reason phrase length has a maximum value of
1929    /// 1024 bytes. If an endpoint receives a length exceeding the maximum, it
1930    /// MUST close the session with a PROTOCOL_VIOLATION".
1931    ///
1932    /// Without the decode-side bound the 2000-byte phrase is handed to the
1933    /// application:
1934    ///
1935    /// ```text
1936    /// assertion `left == right` failed
1937    ///   left: Ok(RequestError(RequestError { error_code: VarInt(1),
1938    ///         retry_interval: VarInt(0), reason_phrase: [120, 120, ...],
1939    ///         redirect: None }))
1940    ///  right: Err(ReasonPhraseTooLong)
1941    /// ```
1942    ///
1943    /// (The 2000 repeated bytes of the phrase are elided from that transcript.)
1944    #[test]
1945    fn an_over_long_reason_phrase_is_refused_on_decode() {
1946        for (type_id, prefix) in [(0x05u64, vec![0x01, 0x00]), (0x0B, vec![0x01, 0x00])] {
1947            let mut body = prefix;
1948            let over = MAX_REASON_PHRASE_LENGTH + 976;
1949            VarInt::from_usize(over).encode_moqt::<Wire>(&mut body);
1950            body.extend(std::iter::repeat_n(b'x', over));
1951            let bytes = frame(type_id, &body);
1952
1953            let mut buf = &bytes[..];
1954            assert_eq!(
1955                ControlMessage::decode(&mut buf),
1956                Err(CodecError::ReasonPhraseTooLong),
1957                "message type 0x{type_id:x}"
1958            );
1959        }
1960    }
1961
1962    /// Draft-19 Section 10.4: "The maximum length of the New Session URI is
1963    /// 8,192 bytes. If an endpoint receives a length exceeding the maximum, it
1964    /// MUST close the session with a PROTOCOL_VIOLATION."
1965    ///
1966    /// Without the decode-side bound the oversized URI reaches the application
1967    /// and a migrating endpoint follows it:
1968    ///
1969    /// ```text
1970    /// assertion `left == right` failed
1971    ///   left: Ok(GoAway(GoAway { new_session_uri: [117, 117, ...],
1972    ///         timeout: VarInt(0) }))
1973    ///  right: Err(GoAwayUriTooLong)
1974    /// ```
1975    ///
1976    /// (The 9000 repeated bytes of the URI are elided from that transcript.)
1977    #[test]
1978    fn an_over_long_goaway_uri_is_refused_on_decode() {
1979        let over = MAX_GOAWAY_URI_LENGTH + 808;
1980        let mut body = Vec::new();
1981        VarInt::from_usize(over).encode_moqt::<Wire>(&mut body);
1982        body.extend(std::iter::repeat_n(b'u', over));
1983        body.push(0x00); // timeout
1984        let bytes = frame(0x10, &body);
1985
1986        let mut buf = &bytes[..];
1987        assert_eq!(ControlMessage::decode(&mut buf), Err(CodecError::GoAwayUriTooLong));
1988    }
1989
1990    /// Draft-19 Section 10.2.8 (GROUP_ORDER): "The allowed values are Ascending
1991    /// (0x1) or Descending (0x2). If an endpoint receives a value outside this
1992    /// range, it MUST close the session with PROTOCOL_VIOLATION." Section
1993    /// 10.2.17 says the same of FORWARD with the values 0 and 1.
1994    ///
1995    /// Without `uint8_value_in_range` the out-of-range byte is handed up as an
1996    /// ordinary parameter:
1997    ///
1998    /// ```text
1999    /// assertion `left == right` failed
2000    ///   left: Ok(Subscribe(Subscribe { request_id: VarInt(1), track_namespace:
2001    ///         TrackNamespace([[97]]), track_name: [98], parameters:
2002    ///         [KeyValuePair { key: VarInt(34), value: Varint(VarInt(7)) }] }))
2003    ///  right: Err(InvalidField)
2004    /// ```
2005    #[test]
2006    fn a_uint8_parameter_outside_its_range_is_refused() {
2007        // key, rejected value, accepted value
2008        let cases = [(0x22u8, 7u8, 2u8), (0x10, 9, 1)];
2009        for (key, bad, good) in cases {
2010            let bytes = frame(0x03, &subscribe_body(&[0x01, key, bad]));
2011            let mut buf = &bytes[..];
2012            assert_eq!(
2013                ControlMessage::decode(&mut buf),
2014                Err(CodecError::ParameterValueOutOfRange { key: key as u64, value: bad as u64 }),
2015                "parameter 0x{key:x} value {bad}"
2016            );
2017
2018            let bytes = frame(0x03, &subscribe_body(&[0x01, key, good]));
2019            let mut buf = &bytes[..];
2020            assert!(
2021                ControlMessage::decode(&mut buf).is_ok(),
2022                "parameter 0x{key:x} value {good} should still decode"
2023            );
2024        }
2025    }
2026
2027    /// SUBSCRIBER_PRIORITY (0x20) is a uint8 with no restricted range, so it
2028    /// must keep accepting the whole 0-255 span. This is the negative half of
2029    /// the range check: a table that over-reached would fail here.
2030    #[test]
2031    fn subscriber_priority_still_accepts_the_whole_byte_range() {
2032        for value in [0u8, 1, 2, 128, 255] {
2033            let bytes = frame(0x03, &subscribe_body(&[0x01, 0x20, value]));
2034            let mut buf = &bytes[..];
2035            assert!(ControlMessage::decode(&mut buf).is_ok(), "priority {value}");
2036        }
2037    }
2038
2039    fn param(key: u64, value: &[u8]) -> KeyValuePair {
2040        KeyValuePair { key: VarInt::from_u64_moqt(key), value: KvpValue::Bytes(value.to_vec()) }
2041    }
2042
2043    /// Draft-19 Section 10.2.16: "The LARGEST_OBJECT parameter (Parameter Type
2044    /// 0x9) is a Location." Section 10.2 defines Location as "Two consecutive
2045    /// varints (Group, Object)" — the value carries no length of its own.
2046    ///
2047    /// The frame below is built from the draft rather than from this encoder:
2048    /// REQUEST_OK, four body bytes, one parameter, type delta `0x09`, then the
2049    /// two varints `0x0a` and `0x03` for Location (10, 3). A length-prefixed
2050    /// spelling would need a fifth byte.
2051    ///
2052    /// With `0x09` back in the Length-prefixed arm, the decoder reads the
2053    /// group varint `0x0a` as a value length of 10 and runs off the end of a
2054    /// four-byte body. Both this test and
2055    /// [`a_location_does_not_eat_the_block_that_follows_it`] fail with:
2056    ///
2057    /// ```text
2058    /// spec-correct frame must decode: UnexpectedEnd
2059    /// ```
2060    #[test]
2061    fn largest_object_is_two_bare_varints() {
2062        let body = [0x01, 0x09, 0x0a, 0x03];
2063        let bytes = frame(0x07, &body);
2064
2065        let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
2066        let ControlMessage::RequestOk(ok) = &msg else {
2067            panic!("expected REQUEST_OK, got {msg:?}")
2068        };
2069        assert_eq!(ok.parameters, vec![param(0x09, &[0x0a, 0x03])]);
2070        assert!(ok.track_properties.is_empty(), "the four body bytes are all parameter");
2071
2072        let mut out = Vec::new();
2073        msg.encode(&mut out).expect("re-encode");
2074        assert_eq!(out, bytes, "the value must go back out as the two bare varints it came in as");
2075    }
2076
2077    /// A Location value the encoder was handed but the decoder could not read
2078    /// back is refused on the way out, not written.
2079    ///
2080    /// LARGEST_OBJECT carries no length of its own — that is the whole point
2081    /// of the encoding — so `encode_parameters` writes its bytes verbatim. A
2082    /// value built in memory rather than decoded is under no obligation to be
2083    /// two varints, and without this check the codec answers `Ok(())` and puts
2084    /// a frame on the wire that `ControlMessage::decode` then refuses. One
2085    /// varint short and one varint long are the two ways to get it wrong.
2086    ///
2087    /// # What it catches
2088    ///
2089    /// Dropping the `is_location_value` guard from this draft's encode arm,
2090    /// run:
2091    ///
2092    /// ```text
2093    /// panicked at crates\moqtap-codec\src\draft19\message.rs:
2094    /// LARGEST_OBJECT of one varint must not encode: the decoder cannot read it back
2095    ///
2096    /// test result: FAILED. 1 passed; 1 failed; 0 ignored; 0 measured; 108 filtered out
2097    /// ```
2098    ///
2099    /// The sibling draft kept its guard and kept passing, which is what shows
2100    /// the check is per-draft and not inherited from somewhere shared.
2101    #[test]
2102    fn a_location_value_that_is_not_two_varints_is_refused_on_encode() {
2103        for (label, value) in
2104            [("one varint", vec![0x0a]), ("three varints", vec![0x0a, 0x03, 0x05])]
2105        {
2106            let msg = ControlMessage::RequestOk(RequestOk {
2107                parameters: vec![param(0x09, &value)],
2108                track_properties: Vec::new(),
2109            });
2110            let mut out = Vec::new();
2111            assert!(
2112                msg.encode(&mut out).is_err(),
2113                "LARGEST_OBJECT of {label} must not encode: the decoder cannot read it back"
2114            );
2115        }
2116
2117        // The well-formed value still goes out, so the check refuses the
2118        // malformed case and not the encoding itself.
2119        let msg = ControlMessage::RequestOk(RequestOk {
2120            parameters: vec![param(0x09, &[0x0a, 0x03])],
2121            track_properties: Vec::new(),
2122        });
2123        let mut out = Vec::new();
2124        msg.encode(&mut out).expect("a Location of exactly two varints must still encode");
2125        ControlMessage::decode(&mut &out[..]).expect("and must decode back");
2126    }
2127
2128    /// The same Location read through a message that carries other fields
2129    /// after it, so a stray length byte cannot hide in a trailing block.
2130    ///
2131    /// SUBSCRIBE_OK is track alias `0x05`, then the parameters, then the track
2132    /// properties. With LARGEST_OBJECT (10, 3) and one property
2133    /// (OBJECT_DELIVERY_TIMEOUT, type `0x02`, 5000ms as the two-byte varint
2134    /// `0x93 0x88`), the body is `05 01 09 0a 03 02 93 88`.
2135    #[test]
2136    fn a_location_does_not_eat_the_block_that_follows_it() {
2137        let body = [0x05, 0x01, 0x09, 0x0a, 0x03, 0x02, 0x93, 0x88];
2138        let bytes = frame(0x04, &body);
2139
2140        let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
2141        let ControlMessage::SubscribeOk(ok) = &msg else {
2142            panic!("expected SUBSCRIBE_OK, got {msg:?}")
2143        };
2144        assert_eq!(ok.parameters, vec![param(0x09, &[0x0a, 0x03])]);
2145        assert_eq!(
2146            ok.track_properties,
2147            vec![KeyValuePair {
2148                key: VarInt::from_u64_moqt(0x02),
2149                value: KvpValue::Varint(VarInt::from_u64_moqt(5000)),
2150            }]
2151        );
2152
2153        let mut out = Vec::new();
2154        msg.encode(&mut out).expect("re-encode");
2155        assert_eq!(out, bytes);
2156    }
2157
2158    /// Draft-19 Section 10.2.19: the TRACK_NAMESPACE_PREFIX parameter
2159    /// (Parameter Type 0x34) "uses the Track Namespace encoding described in
2160    /// Section 2.4.1" — a varint field count followed by that many
2161    /// length-prefixed fields, and nothing in front of it. That encoding is not
2162    /// one of the four Section 10.2 lists, so it cannot be assumed to be
2163    /// Length-prefixed by default.
2164    ///
2165    /// The frame below is built from Section 2.4.1: REQUEST_UPDATE for request
2166    /// `7`, one parameter, type delta `0x34`, then the namespace ("live",
2167    /// "sports") as `02 04 "live" 06 "sports"`. Sixteen body bytes; a
2168    /// length-prefixed spelling would need a seventeenth for the outer length.
2169    ///
2170    /// With `0x34` back in the Length-prefixed arm the field count `0x02` is
2171    /// read as an outer length of two bytes, leaving eleven bytes of namespace
2172    /// unread. Draft-19's Section 10 body-length check turns that into a
2173    /// refusal rather than a truncated value:
2174    ///
2175    /// ```text
2176    /// spec-correct frame must decode: InvalidField
2177    /// ```
2178    ///
2179    /// That check is not a safety net here. Where the surplus lands inside the
2180    /// declared body — as in
2181    /// [`an_empty_track_namespace_prefix_is_one_zero_byte`], whose namespace is
2182    /// one byte long — the misread is silent, and that test fails instead with:
2183    ///
2184    /// ```text
2185    /// assertion `left == right` failed
2186    ///   left: [KeyValuePair { key: VarInt(52), value: Bytes([]) }]
2187    ///  right: [KeyValuePair { key: VarInt(52), value: Bytes([0]) }]
2188    /// ```
2189    #[test]
2190    fn track_namespace_prefix_is_a_bare_track_namespace() {
2191        let namespace: Vec<u8> = [&[0x02, 0x04][..], b"live", &[0x06][..], b"sports"].concat();
2192        assert_eq!(namespace.len(), 13);
2193
2194        let body: Vec<u8> = [&[0x07, 0x01, 0x34][..], &namespace].concat();
2195        assert_eq!(body.len(), 16);
2196        let bytes = frame(0x02, &body);
2197
2198        let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
2199        let ControlMessage::RequestUpdate(update) = &msg else {
2200            panic!("expected REQUEST_UPDATE, got {msg:?}")
2201        };
2202        assert_eq!(update.request_id.into_inner(), 7);
2203        assert_eq!(update.parameters, vec![param(0x34, &namespace)]);
2204
2205        let mut out = Vec::new();
2206        msg.encode(&mut out).expect("re-encode");
2207        assert_eq!(out, bytes, "no outer length may appear in front of the Track Namespace");
2208    }
2209
2210    /// An empty prefix is a legal Track Namespace: Section 2.4.1 puts one at
2211    /// "between 0 and 32 Track Namespace Fields". On the wire that is the
2212    /// single byte `0x00`, and it must not be confused with a length-prefixed
2213    /// value of zero bytes.
2214    #[test]
2215    fn an_empty_track_namespace_prefix_is_one_zero_byte() {
2216        let body = [0x07, 0x01, 0x34, 0x00];
2217        let bytes = frame(0x02, &body);
2218
2219        let msg = ControlMessage::decode(&mut &bytes[..]).expect("empty prefix must decode");
2220        let ControlMessage::RequestUpdate(update) = &msg else {
2221            panic!("expected REQUEST_UPDATE, got {msg:?}")
2222        };
2223        assert_eq!(update.parameters, vec![param(0x34, &[0x00])]);
2224
2225        let mut out = Vec::new();
2226        msg.encode(&mut out).expect("re-encode");
2227        assert_eq!(out, bytes);
2228    }
2229}