Skip to main content

moqtap_codec/draft12/
data_stream.rs

1//! Draft-12 data stream header encoding and decoding.
2//!
3//! Changes from draft-11:
4//! - Subgroup stream type IDs shift from 0x08-0x0D to 0x10-0x15
5//! - Datagram types (separate namespace): 0x00-0x05
6//! - Fetch type: same as draft-11 (0x05)
7
8use super::types::ObjectStatus;
9use crate::error::CodecError;
10use crate::types::read_bytes;
11use crate::varint::VarInt;
12use bytes::{Buf, BufMut};
13
14/// Stream type IDs for draft-12 data streams.
15///
16/// # Draft-12 contradicts itself about where the subgroup types sit
17///
18/// Three places in this draft name the SUBGROUP_HEADER Type range, and they do
19/// not agree. Two of them carry draft-11's answer:
20///
21///   - Section 9, Table 10, the table of unidirectional stream types, whose
22///     SUBGROUP_HEADER row reads 0x08-0x0D.
23///   - Section 9.4.2, Figure 35, the header's own layout:
24///     `Type (i) = 0x8..0xD`.
25///
26/// The third does not. Section 9.4.2 says "There are 12 defined Type values for
27/// SUBGROUP_HEADER" and Table 13, immediately under that figure, lists them:
28/// 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x18, 0x19, 0x1A, 0x1B, 0x1C and 0x1D.
29///
30/// **This enum implements Table 13**, so a stream opening with 0x08 through
31/// 0x0D is [`CodecError::UnknownStreamType`] and closes the session.
32///
33/// Table 13 is the surviving half, for the same reason and by the same
34/// mechanism as the datagram-status contradiction documented on
35/// [`DatagramType`] — a code-point update that missed a spot:
36///
37///   - The range 0x08-0x0D holds six values, and this draft defines twelve
38///     types. Draft-11 Section 9.4.2 Table 11 lists exactly six, at 0x08
39///     through 0x0D, and the twelve here are those six crossed with the End Of
40///     Group bit draft-12 added — its own change log says so: "Use bits in
41///     SUBGROUP_HEADER and DATAGRAM* types to compress subgroup ID and
42///     extensions". So 0x08-0x0D and `0x8..0xD` are draft-11's range left
43///     behind, and they cannot hold what this draft defines.
44///   - Table 13 is the only one of the three that says what each value *means*.
45///     The other two give a range and nothing else, so following either would
46///     leave every framing decision — whether a Subgroup ID field is on the
47///     wire, whether objects carry extensions, whether the stream ends the
48///     group — with nothing to read it from.
49///   - Draft-14 keeps Table 13 unchanged and corrects the other two to match:
50///     its Table 10 reads "0x10-0x1D" and its Figure reads
51///     `Type (i) = 0x10..0x1D`. That is the disagreement being resolved in
52///     favour of Table 13 by the working group, one draft later.
53///
54/// Draft-13 repeats all three halves unchanged, and carries the same note.
55///
56/// The cost of being wrong is asymmetric and points the same way. Accepting
57/// 0x08-0x0D as well would mean parsing a stream under framing no table
58/// assigns it, which is the failure this codebase refuses elsewhere — an
59/// out-of-range Type aliased onto a valid one produces objects with plausible,
60/// wrong contents. Refusing them closes a session with a peer that followed the
61/// stale half of its own draft, which is visible, reportable, and what
62/// draft-14 says the peer should not have done.
63#[derive(Debug, Clone, Copy, PartialEq, Eq)]
64#[repr(u64)]
65pub enum StreamType {
66    /// Fetch response stream (0x05).
67    Fetch = 0x05,
68    /// Subgroup: subgroup_id=0, no extensions (0x10).
69    SubgroupZero = 0x10,
70    /// Subgroup: subgroup_id=0, with extensions (0x11).
71    SubgroupZeroExt = 0x11,
72    /// Subgroup: subgroup_id=first object ID, no extensions (0x12).
73    SubgroupFirstObj = 0x12,
74    /// Subgroup: subgroup_id=first object ID, with extensions (0x13).
75    SubgroupFirstObjExt = 0x13,
76    /// Subgroup: explicit subgroup_id, no extensions (0x14).
77    SubgroupExplicit = 0x14,
78    /// Subgroup: explicit subgroup_id, with extensions (0x15).
79    SubgroupExplicitExt = 0x15,
80    /// Subgroup: subgroup_id=0, contains end of group, no extensions (0x18).
81    SubgroupZeroEog = 0x18,
82    /// Subgroup: subgroup_id=0, contains end of group, with extensions (0x19).
83    SubgroupZeroEogExt = 0x19,
84    /// Subgroup: subgroup_id=first object ID, contains end of group, no extensions (0x1A).
85    SubgroupFirstObjEog = 0x1A,
86    /// Subgroup: subgroup_id=first object ID, contains end of group, with extensions (0x1B).
87    SubgroupFirstObjEogExt = 0x1B,
88    /// Subgroup: explicit subgroup_id, contains end of group, no extensions (0x1C).
89    SubgroupExplicitEog = 0x1C,
90    /// Subgroup: explicit subgroup_id, contains end of group, with extensions (0x1D).
91    SubgroupExplicitEogExt = 0x1D,
92}
93
94/// Hold an object to the rule that a non-existent object carries no extensions.
95///
96/// Section 9.2.1.2: "Any Object may have extension headers except those with
97/// Object Status 'Object Does Not Exist'. If an endpoint receives a non-existent
98/// Object containing extension headers it MUST close the session with a Protocol
99/// Violation."
100///
101/// The sentence is about a receiver, and it reaches all three carriers that can
102/// announce a status: an object on a subgroup stream, an object on a fetch
103/// stream, and a status datagram. A plain datagram has no status field, so it
104/// is the only carrier that cannot break the rule.
105///
106/// Reported under [`CodecError::ExtensionsOnNonExistentObject`], which is this
107/// rule and nothing else. [`CodecError::InvalidField`] is too coarse for it:
108/// shared with a dozen unrelated malformations the draft does not answer with a
109/// close, it leaves a caller unable to act on the sentence above.
110fn check_extensions_against_status(
111    status: ObjectStatus,
112    extensions: &[u8],
113) -> Result<(), CodecError> {
114    if status == ObjectStatus::ObjectDoesNotExist && !extensions.is_empty() {
115        return Err(CodecError::ExtensionsOnNonExistentObject(extensions.len()));
116    }
117    Ok(())
118}
119
120impl StreamType {
121    pub fn from_id(id: u64) -> Option<Self> {
122        match id {
123            0x05 => Some(StreamType::Fetch),
124            0x10 => Some(StreamType::SubgroupZero),
125            0x11 => Some(StreamType::SubgroupZeroExt),
126            0x12 => Some(StreamType::SubgroupFirstObj),
127            0x13 => Some(StreamType::SubgroupFirstObjExt),
128            0x14 => Some(StreamType::SubgroupExplicit),
129            0x15 => Some(StreamType::SubgroupExplicitExt),
130            0x18 => Some(StreamType::SubgroupZeroEog),
131            0x19 => Some(StreamType::SubgroupZeroEogExt),
132            0x1A => Some(StreamType::SubgroupFirstObjEog),
133            0x1B => Some(StreamType::SubgroupFirstObjEogExt),
134            0x1C => Some(StreamType::SubgroupExplicitEog),
135            0x1D => Some(StreamType::SubgroupExplicitEogExt),
136            _ => None,
137        }
138    }
139
140    pub fn is_subgroup(&self) -> bool {
141        matches!(
142            self,
143            StreamType::SubgroupZero
144                | StreamType::SubgroupZeroExt
145                | StreamType::SubgroupFirstObj
146                | StreamType::SubgroupFirstObjExt
147                | StreamType::SubgroupExplicit
148                | StreamType::SubgroupExplicitExt
149                | StreamType::SubgroupZeroEog
150                | StreamType::SubgroupZeroEogExt
151                | StreamType::SubgroupFirstObjEog
152                | StreamType::SubgroupFirstObjEogExt
153                | StreamType::SubgroupExplicitEog
154                | StreamType::SubgroupExplicitEogExt
155        )
156    }
157
158    pub fn has_extensions(&self) -> bool {
159        matches!(
160            self,
161            StreamType::SubgroupZeroExt
162                | StreamType::SubgroupFirstObjExt
163                | StreamType::SubgroupExplicitExt
164                | StreamType::SubgroupZeroEogExt
165                | StreamType::SubgroupFirstObjEogExt
166                | StreamType::SubgroupExplicitEogExt
167        )
168    }
169
170    /// True if this subgroup stream type indicates the stream contains the end of its group.
171    pub fn contains_end_of_group(&self) -> bool {
172        matches!(
173            self,
174            StreamType::SubgroupZeroEog
175                | StreamType::SubgroupZeroEogExt
176                | StreamType::SubgroupFirstObjEog
177                | StreamType::SubgroupFirstObjEogExt
178                | StreamType::SubgroupExplicitEog
179                | StreamType::SubgroupExplicitEogExt
180        )
181    }
182
183    /// True if this subgroup stream type puts an explicit Subgroup ID on the
184    /// wire.
185    ///
186    /// The Subgroup ID Field Present column of the SUBGROUP_HEADER type table
187    /// in Section 9.4.2. The other two columns of that row say what the
188    /// Subgroup ID *is* where the field is absent — zero, or the first
189    /// Object's ID — so this is only about the field, never about the value.
190    pub fn writes_subgroup_id(&self) -> bool {
191        matches!(
192            self,
193            StreamType::SubgroupExplicit
194                | StreamType::SubgroupExplicitExt
195                | StreamType::SubgroupExplicitEog
196                | StreamType::SubgroupExplicitEogExt
197        )
198    }
199}
200
201/// Datagram wire types (separate namespace from QUIC stream types).
202///
203/// The two namespaces overlap in draft-12 and cannot share one enum: 0x05 is
204/// FETCH_HEADER among stream types and OBJECT_DATAGRAM_STATUS with extensions
205/// among datagram types.
206///
207/// Draft-12 contradicts itself about where the status types sit. Its table of
208/// datagram types gives OBJECT_DATAGRAM 0x00 through 0x03 and
209/// OBJECT_DATAGRAM_STATUS 0x04 through 0x05, and the four OBJECT_DATAGRAM
210/// values are spelled out one by one as the End Of Group bit crossed with the
211/// Extensions bit — the End Of Group bit being what this draft added. But the
212/// sentence under the OBJECT_DATAGRAM_STATUS figure still reads "the set of
213/// values from 0x02 to 0x03", which is draft-11's range from before the bit
214/// existed and cannot be squared with the table above it. Draft-13 repeats both
215/// halves unchanged; draft-14 keeps the table's answer and records the sentence
216/// as a missed code-point update. The table is therefore the surviving half and
217/// the values below follow it.
218#[derive(Debug, Clone, Copy, PartialEq, Eq)]
219#[repr(u64)]
220pub enum DatagramType {
221    /// Object datagram, no extensions (0x00).
222    Datagram = 0x00,
223    /// Object datagram, with extensions (0x01).
224    DatagramExt = 0x01,
225    /// Object datagram carrying the end of its group, no extensions (0x02).
226    DatagramEog = 0x02,
227    /// Object datagram carrying the end of its group, with extensions (0x03).
228    DatagramEogExt = 0x03,
229    /// Object datagram status, no extensions (0x04).
230    DatagramStatus = 0x04,
231    /// Object datagram status, with extensions (0x05).
232    DatagramStatusExt = 0x05,
233}
234
235impl DatagramType {
236    pub fn from_id(id: u64) -> Option<Self> {
237        match id {
238            0x00 => Some(DatagramType::Datagram),
239            0x01 => Some(DatagramType::DatagramExt),
240            0x02 => Some(DatagramType::DatagramEog),
241            0x03 => Some(DatagramType::DatagramEogExt),
242            0x04 => Some(DatagramType::DatagramStatus),
243            0x05 => Some(DatagramType::DatagramStatusExt),
244            _ => None,
245        }
246    }
247
248    pub fn has_extensions(&self) -> bool {
249        matches!(
250            self,
251            DatagramType::DatagramExt
252                | DatagramType::DatagramEogExt
253                | DatagramType::DatagramStatusExt
254        )
255    }
256
257    pub fn is_status(&self) -> bool {
258        matches!(self, DatagramType::DatagramStatus | DatagramType::DatagramStatusExt)
259    }
260
261    pub fn is_end_of_group(&self) -> bool {
262        matches!(self, DatagramType::DatagramEog | DatagramType::DatagramEogExt)
263    }
264}
265
266/// Which failure a leading unidirectional stream type that is not the one a
267/// reader wants is.
268///
269/// Section 9: "An endpoint that receives an unknown stream or datagram type
270/// MUST close the session." One sentence, two tables, and on this draft the two
271/// tables collide: 0x05 is FETCH_HEADER in the stream table and
272/// OBJECT_DATAGRAM_STATUS with extensions in the datagram table. Which table
273/// was consulted is therefore part of the answer, not a detail, and it is why
274/// [`CodecError::UnknownStreamType`] and [`CodecError::UnknownDatagramType`]
275/// are separate variants rather than one.
276///
277/// The stream table assigns 0x05 and the range 0x10 to 0x1D. Everything outside
278/// them is unknown at the head of a stream, and the session ends.
279fn stream_type_error(raw: u64) -> CodecError {
280    if StreamType::from_id(raw).is_some() {
281        CodecError::InvalidField
282    } else {
283        CodecError::UnknownStreamType(raw)
284    }
285}
286
287/// Which failure a leading datagram type that is not one a reader wants is.
288///
289/// The datagram half of the sentence quoted on `stream_type_error`, read
290/// against the other table: 0x00 to 0x05 are what it assigns, and everything
291/// else arriving as a datagram is unknown.
292///
293/// Answered from [`DatagramType`] alone, never from [`StreamType`]. A value in
294/// both tables means one thing as a datagram and another as a stream, and
295/// consulting the wrong one turns an assigned datagram type into an unknown one
296/// or the reverse.
297fn datagram_type_error(raw: u64) -> CodecError {
298    if DatagramType::from_id(raw).is_some() {
299        CodecError::InvalidField
300    } else {
301        CodecError::UnknownDatagramType(raw)
302    }
303}
304
305// ── Extension helpers ─────────────────────────────────────────
306
307fn read_extension_bytes(buf: &mut impl Buf, byte_len: u64) -> Result<Vec<u8>, CodecError> {
308    read_bytes(buf, byte_len as usize)
309}
310
311// ============================================================
312// Subgroup stream header
313// ============================================================
314
315#[derive(Debug, Clone, PartialEq, Eq)]
316pub struct SubgroupHeader {
317    pub stream_type: StreamType,
318    pub track_alias: VarInt,
319    pub group_id: VarInt,
320    pub subgroup_id: VarInt,
321    pub publisher_priority: u8,
322}
323
324impl SubgroupHeader {
325    /// Encode a subgroup stream header including its leading stream-type
326    /// field, so the bytes form the start of a data stream a peer can read.
327    ///
328    /// [`Self::encode`] writes the body alone, which is what a caller wants
329    /// once the stream is already open and what a caller must not use for its
330    /// first write. It is also the half that cannot stand on its own here,
331    /// because the stream type is what says whether a Subgroup ID follows it
332    /// and whether the objects on the stream carry extension headers.
333    pub fn encode_stream(&self, buf: &mut impl BufMut) {
334        VarInt::from_usize(self.stream_type as usize).encode(buf);
335        self.encode(buf);
336    }
337
338    /// Encode the header body, without its leading stream-type field.
339    ///
340    /// Driven by the stream type, and silent about a `subgroup_id` it decides
341    /// not to write: on a type whose Subgroup ID Field Present column reads No
342    /// the field is dropped, and the peer reads the subgroup the *type* names -
343    /// zero, or the first Object's ID - rather than the one in hand. Nothing is
344    /// malformed about the result, which is what makes it worth refusing rather
345    /// than tolerating. [`Self::encode_checked`] refuses it.
346    pub fn encode(&self, buf: &mut impl BufMut) {
347        self.track_alias.encode(buf);
348        self.group_id.encode(buf);
349        if self.stream_type.writes_subgroup_id() {
350            self.subgroup_id.encode(buf);
351        }
352        buf.put_u8(self.publisher_priority);
353    }
354
355    /// Encode the header body, refusing a Subgroup ID this stream type has
356    /// nowhere to put.
357    ///
358    /// [`Self::decode_with_type`] leaves the field at zero for every type that
359    /// does not carry it, so a decoded header always passes: the refusal is for
360    /// a header assembled by hand, where a caller set an ID the type will
361    /// discard.
362    ///
363    /// A zero is accepted under any type. It is what the decoder produces, and
364    /// on a Subgroup ID Value column reading `0` it is also the truth, so
365    /// refusing it would refuse the ordinary case to catch nothing.
366    ///
367    /// # Errors
368    ///
369    /// [`CodecError::InvalidField`] if a non-zero Subgroup ID sits under a
370    /// stream type that writes no Subgroup ID field.
371    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
372        if !self.stream_type.writes_subgroup_id() && self.subgroup_id.into_inner() != 0 {
373            return Err(CodecError::InvalidField);
374        }
375        self.encode(buf);
376        Ok(())
377    }
378
379    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
380        Self::decode_with_type(StreamType::SubgroupExplicit, buf)
381    }
382
383    pub fn decode_with_type(
384        stream_type: StreamType,
385        buf: &mut impl Buf,
386    ) -> Result<Self, CodecError> {
387        let track_alias = VarInt::decode(buf)?;
388        let group_id = VarInt::decode(buf)?;
389        let subgroup_id = match stream_type {
390            StreamType::SubgroupZero
391            | StreamType::SubgroupZeroExt
392            | StreamType::SubgroupZeroEog
393            | StreamType::SubgroupZeroEogExt => VarInt::from_usize(0),
394            StreamType::SubgroupExplicit
395            | StreamType::SubgroupExplicitExt
396            | StreamType::SubgroupExplicitEog
397            | StreamType::SubgroupExplicitEogExt => VarInt::decode(buf)?,
398            StreamType::SubgroupFirstObj
399            | StreamType::SubgroupFirstObjExt
400            | StreamType::SubgroupFirstObjEog
401            | StreamType::SubgroupFirstObjEogExt => VarInt::from_usize(0),
402            _ => return Err(CodecError::InvalidField),
403        };
404        if buf.remaining() < 1 {
405            return Err(CodecError::UnexpectedEnd);
406        }
407        let publisher_priority = buf.get_u8();
408        Ok(Self { stream_type, track_alias, group_id, subgroup_id, publisher_priority })
409    }
410
411    /// Decode a subgroup header from the start of a data stream, consuming
412    /// the leading stream type varint and using it to select the variant.
413    ///
414    /// Errors with [`CodecError::UnknownStreamType`] when the stream table does
415    /// not assign the leading type, which this draft answers with a close, and
416    /// with [`CodecError::InvalidField`] when it does assign it but not to a
417    /// subgroup. `stream_type_error` draws that line.
418    pub fn decode_stream(buf: &mut impl Buf) -> Result<Self, CodecError> {
419        let raw = VarInt::decode(buf)?.into_inner();
420        let stream_type = StreamType::from_id(raw).ok_or_else(|| stream_type_error(raw))?;
421        if !stream_type.is_subgroup() {
422            return Err(stream_type_error(raw));
423        }
424        Self::decode_with_type(stream_type, buf)
425    }
426}
427
428// ============================================================
429// Object header within subgroup
430// ============================================================
431
432#[derive(Debug, Clone, PartialEq, Eq)]
433pub struct ObjectHeader {
434    pub object_id: VarInt,
435    pub extension_headers_length: VarInt,
436    pub extensions: Vec<u8>,
437    pub payload_length: VarInt,
438    pub object_status: ObjectStatus,
439}
440
441impl ObjectHeader {
442    /// Encode the object header in the framing that carries no extension
443    /// block.
444    ///
445    /// Lossy, and lossy in a way the caller cannot see: an object holding
446    /// extension headers is written without them and without a word. Which
447    /// framing is correct is not a property of the object at all - Section
448    /// 9.4.2 gives the stream's type an Extensions Present column, and every
449    /// object on the stream follows it - so this entry point can only guess,
450    /// and it guesses "absent". Prefer [`Self::encode_with_extensions`], which
451    /// is told, or [`Self::encode_checked`], which refuses what it would
452    /// otherwise drop.
453    pub fn encode(&self, buf: &mut impl BufMut) {
454        self.encode_with_extensions(false, buf);
455    }
456
457    /// Encode the header, refusing a status the framing cannot carry.
458    ///
459    /// Section 9.4.2 puts the Object Status field on the wire only when the
460    /// Object Payload Length is zero, and Section 9.2.1.1 says "Any object
461    /// with a status code other than zero MUST have an empty payload". A
462    /// non-zero status paired with a non-zero payload length therefore has no
463    /// encoding at all: [`Self::encode`] drops the status and the peer reads an
464    /// ordinary object, which is a different object from the one the caller
465    /// described. This refuses instead.
466    ///
467    /// The datagram types on this draft already refuse the same pairing. These
468    /// two did not, and they are the ones a publisher writes on every stream.
469    ///
470    /// Extension headers are refused here rather than dropped, for a reason
471    /// the status rule does not share: this entry point writes the framing
472    /// that has no Extension Headers Length field, so the bytes have nowhere
473    /// to go. Writing them anyway is not an option and losing them silently
474    /// puts a stream on the wire that no reader can follow - a reader on an
475    /// extensions-bearing stream takes the Object Payload Length as the
476    /// extension length and every object after it is misread. A caller that
477    /// knows the stream's framing wants
478    /// [`Self::encode_checked_with_extensions`].
479    ///
480    /// # Errors
481    ///
482    /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
483    /// a non-zero Object Payload Length, or if the object carries extension
484    /// headers this framing cannot write.
485    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
486        self.encode_checked_with_extensions(false, buf)
487    }
488
489    /// Encode the object header into a stream whose type has already settled
490    /// whether objects carry an extension block, refusing what that framing
491    /// cannot express.
492    ///
493    /// `has_extensions` is the stream's answer, not the object's: Section
494    /// 9.4.2 fixes it for the whole stream from the SUBGROUP_HEADER type, so
495    /// an object with no extensions on a stream that carries them still writes
496    /// a length of zero, and that is the one direction not refused here. The
497    /// other direction has no encoding, so it is refused.
498    ///
499    /// # Errors
500    ///
501    /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
502    /// a non-zero Object Payload Length, or if `has_extensions` is `false`
503    /// while the object carries extension headers.
504    pub fn encode_checked_with_extensions(
505        &self,
506        has_extensions: bool,
507        buf: &mut impl BufMut,
508    ) -> Result<(), CodecError> {
509        if self.payload_length.into_inner() != 0 && self.object_status as usize != 0 {
510            return Err(CodecError::InvalidField);
511        }
512        if !has_extensions && !self.extensions.is_empty() {
513            return Err(CodecError::InvalidField);
514        }
515        self.encode_with_extensions(has_extensions, buf);
516        Ok(())
517    }
518
519    /// Encode the object header, writing the extension block only when the
520    /// stream's type says objects carry one.
521    ///
522    /// Infallible, and so unable to say that a `false` here discards the
523    /// extension headers the object holds. [`Self::encode_checked_with_extensions`]
524    /// is the same write with that refusal in front of it.
525    pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
526        self.object_id.encode(buf);
527        if has_extensions {
528            VarInt::from_usize(self.extensions.len()).encode(buf);
529            buf.put_slice(&self.extensions);
530        }
531        self.payload_length.encode(buf);
532        if self.payload_length.into_inner() == 0 {
533            VarInt::from_usize(self.object_status as usize).encode(buf);
534        }
535    }
536
537    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
538        Self::decode_with_extensions(false, buf)
539    }
540
541    pub fn decode_with_extensions(
542        has_extensions: bool,
543        buf: &mut impl Buf,
544    ) -> Result<Self, CodecError> {
545        let object_id = VarInt::decode(buf)?;
546        let (extension_headers_length, extensions) = if has_extensions {
547            let ehl = VarInt::decode(buf)?;
548            let ext = read_extension_bytes(buf, ehl.into_inner())?;
549            (ehl, ext)
550        } else {
551            (VarInt::from_usize(0), Vec::new())
552        };
553        let payload_length = VarInt::decode(buf)?;
554        let object_status = if payload_length.into_inner() == 0 {
555            let sv = VarInt::decode(buf)?.into_inner();
556            ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?
557        } else {
558            ObjectStatus::Normal
559        };
560        check_extensions_against_status(object_status, &extensions)?;
561        Ok(Self { object_id, extension_headers_length, extensions, payload_length, object_status })
562    }
563}
564
565// ============================================================
566// Datagram (types 0x00 through 0x03)
567// ============================================================
568
569#[derive(Debug, Clone, PartialEq, Eq)]
570pub struct DatagramHeader {
571    pub track_alias: VarInt,
572    pub group_id: VarInt,
573    pub object_id: VarInt,
574    pub publisher_priority: u8,
575    pub extension_headers_length: VarInt,
576    pub extensions: Vec<u8>,
577    /// Whether this object is the last one in its group.
578    ///
579    /// Draft-12 added this flag and put it in the datagram type rather than in
580    /// the header body, so it is not written or read by the methods below;
581    /// [`DatagramType::is_end_of_group`] is where it lives on the wire.
582    pub end_of_group: bool,
583}
584
585impl DatagramHeader {
586    /// Encode the datagram header in the framing that carries no extension
587    /// block.
588    ///
589    /// Lossy in the same way the subgroup object header is: an extension block
590    /// this value holds is dropped, because the framing being written has no
591    /// field for it. The type byte decides which framing is right, and this
592    /// entry point does not write the type byte, so it cannot consult it.
593    /// [`Datagram::encode`] does both together and never disagrees with itself;
594    /// this is the piece for a caller that has already written the type.
595    pub fn encode(&self, buf: &mut impl BufMut) {
596        self.encode_with_extensions(false, buf);
597    }
598
599    pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
600        self.track_alias.encode(buf);
601        self.group_id.encode(buf);
602        self.object_id.encode(buf);
603        buf.put_u8(self.publisher_priority);
604        if has_extensions {
605            VarInt::from_usize(self.extensions.len()).encode(buf);
606            buf.put_slice(&self.extensions);
607        }
608    }
609
610    /// Encode the datagram header, refusing what this framing cannot carry.
611    ///
612    /// No status is ever refused, and that is a fact about this draft rather
613    /// than a check left out. This is the OBJECT_DATAGRAM of Section 9.3.1,
614    /// whose layout carries no Object Status field at all; a datagram that
615    /// states a status is the separate OBJECT_DATAGRAM_STATUS message, modelled
616    /// here as [`DatagramStatusHeader`]. So there is no status for
617    /// [`Self::encode`] to drop, and nothing for Section 9.2.1.1's "Any object
618    /// with a status code other than zero MUST have an empty payload" to rule
619    /// on: an object framed this way has status zero by construction.
620    ///
621    /// The extension block is a different matter. [`Self::encode`] writes the
622    /// framing without one, so a block this value holds has nowhere to go, and
623    /// dropping it silently is what puts a datagram on the wire describing
624    /// something other than what the caller built. That is refused here.
625    ///
626    /// The fallible signature is also what lets one entry point span every
627    /// draft. `dispatch::AnyDatagramHeader::encode` calls this on all thirteen,
628    /// and the drafts whose payload-bearing datagram *does* carry a status field
629    /// need somewhere to say no.
630    ///
631    /// # Errors
632    ///
633    /// [`CodecError::InvalidField`] if the value carries extension headers,
634    /// which this framing has no field for.
635    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
636        if !self.extensions.is_empty() {
637            return Err(CodecError::InvalidField);
638        }
639        self.encode(buf);
640        Ok(())
641    }
642
643    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
644        Self::decode_with_extensions(false, buf)
645    }
646
647    pub fn decode_with_extensions(
648        has_extensions: bool,
649        buf: &mut impl Buf,
650    ) -> Result<Self, CodecError> {
651        let track_alias = VarInt::decode(buf)?;
652        let group_id = VarInt::decode(buf)?;
653        let object_id = VarInt::decode(buf)?;
654        if buf.remaining() < 1 {
655            return Err(CodecError::UnexpectedEnd);
656        }
657        let publisher_priority = buf.get_u8();
658        let (extension_headers_length, extensions) = if has_extensions {
659            let ehl = VarInt::decode(buf)?;
660            // A datagram whose type says extensions are present must actually carry
661            // some: receiving one with an Extension Headers Length of 0 closes the
662            // session. The opposite holds on a subgroup stream, where the type byte is
663            // fixed for the whole stream and an object with no extensions has no other
664            // way to say so, which is why this check belongs to the datagram readers
665            // alone.
666            if ehl.into_inner() == 0 {
667                return Err(CodecError::InvalidField);
668            }
669            let ext = read_extension_bytes(buf, ehl.into_inner())?;
670            (ehl, ext)
671        } else {
672            (VarInt::from_usize(0), Vec::new())
673        };
674        Ok(Self {
675            track_alias,
676            group_id,
677            object_id,
678            publisher_priority,
679            extension_headers_length,
680            extensions,
681            end_of_group: false,
682        })
683    }
684}
685
686// ============================================================
687// Datagram Status (types 0x04, 0x05)
688// ============================================================
689
690#[derive(Debug, Clone, PartialEq, Eq)]
691pub struct DatagramStatusHeader {
692    pub track_alias: VarInt,
693    pub group_id: VarInt,
694    pub object_id: VarInt,
695    pub publisher_priority: u8,
696    pub extension_headers_length: VarInt,
697    pub extensions: Vec<u8>,
698    pub object_status: ObjectStatus,
699}
700
701impl DatagramStatusHeader {
702    pub fn encode(&self, buf: &mut impl BufMut) {
703        self.encode_with_extensions(false, buf);
704    }
705
706    /// Encode the status datagram header, refusing an extension block this
707    /// framing cannot carry.
708    ///
709    /// The same one-sided rule the payload-bearing header obeys, and the same
710    /// reason for it: [`Self::encode`] writes the framing without a block, so
711    /// bytes held here would be dropped rather than written.
712    ///
713    /// # Errors
714    ///
715    /// [`CodecError::InvalidField`] if the value carries extension headers.
716    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
717        if !self.extensions.is_empty() {
718            return Err(CodecError::InvalidField);
719        }
720        self.encode(buf);
721        Ok(())
722    }
723
724    pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
725        self.track_alias.encode(buf);
726        self.group_id.encode(buf);
727        self.object_id.encode(buf);
728        buf.put_u8(self.publisher_priority);
729        if has_extensions {
730            VarInt::from_usize(self.extensions.len()).encode(buf);
731            buf.put_slice(&self.extensions);
732        }
733        VarInt::from_usize(self.object_status as usize).encode(buf);
734    }
735
736    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
737        Self::decode_with_extensions(false, buf)
738    }
739
740    pub fn decode_with_extensions(
741        has_extensions: bool,
742        buf: &mut impl Buf,
743    ) -> Result<Self, CodecError> {
744        let track_alias = VarInt::decode(buf)?;
745        let group_id = VarInt::decode(buf)?;
746        let object_id = VarInt::decode(buf)?;
747        if buf.remaining() < 1 {
748            return Err(CodecError::UnexpectedEnd);
749        }
750        let publisher_priority = buf.get_u8();
751        let (extension_headers_length, extensions) = if has_extensions {
752            let ehl = VarInt::decode(buf)?;
753            // A datagram whose type says extensions are present must actually carry
754            // some: receiving one with an Extension Headers Length of 0 closes the
755            // session. The opposite holds on a subgroup stream, where the type byte is
756            // fixed for the whole stream and an object with no extensions has no other
757            // way to say so, which is why this check belongs to the datagram readers
758            // alone.
759            if ehl.into_inner() == 0 {
760                return Err(CodecError::InvalidField);
761            }
762            let ext = read_extension_bytes(buf, ehl.into_inner())?;
763            (ehl, ext)
764        } else {
765            (VarInt::from_usize(0), Vec::new())
766        };
767        let sv = VarInt::decode(buf)?.into_inner();
768        let object_status = ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?;
769        check_extensions_against_status(object_status, &extensions)?;
770        Ok(Self {
771            track_alias,
772            group_id,
773            object_id,
774            publisher_priority,
775            extension_headers_length,
776            extensions,
777            object_status,
778        })
779    }
780}
781
782// ============================================================
783// Datagram framing
784// ============================================================
785
786/// One datagram, of whichever shape its type field names.
787///
788/// A MoQT datagram opens with a variable-length integer naming its type, and
789/// that integer is what says which of the layouts above follows it, and whether an extension block sits inside it.
790/// Neither [`DatagramHeader`] nor [`DatagramStatusHeader`] reads or writes it, so neither can be handed
791/// the first byte of a datagram a peer sent, and neither produces bytes a peer
792/// can read. This is the entry point that does both.
793///
794/// The payload of a payload-bearing datagram runs to the end of the QUIC
795/// datagram, so it is not part of this value: [`Self::decode`] stops at the end
796/// of the header and leaves the payload in the buffer, and a caller appends the
797/// payload after [`Self::encode`].
798#[derive(Debug, Clone, PartialEq, Eq)]
799pub enum Datagram {
800    /// An object carrying a payload.
801    Payload(DatagramHeader),
802    /// An object stating a status, with no payload.
803    Status(DatagramStatusHeader),
804}
805
806impl Datagram {
807    /// Whether this datagram states an Object Status instead of carrying a
808    /// payload.
809    pub fn is_status(&self) -> bool {
810        matches!(self, Self::Status(_))
811    }
812
813    /// The type field this value writes.
814    ///
815    /// The extensions bit is taken from the extension bytes themselves rather
816    /// than from the declared length beside them, which is what keeps the type
817    /// and the body from contradicting each other: a datagram whose type
818    /// announces extensions and then declares a length of 0 closes the session
819    /// on receipt, and one that announces none has nowhere to put them. The end
820    /// of group bit has no home in the body at all, so it comes from the header
821    /// flag and goes nowhere else.
822    pub fn datagram_type(&self) -> DatagramType {
823        match self {
824            Self::Payload(header) => match (header.end_of_group, header.extensions.is_empty()) {
825                (false, true) => DatagramType::Datagram,
826                (false, false) => DatagramType::DatagramExt,
827                (true, true) => DatagramType::DatagramEog,
828                (true, false) => DatagramType::DatagramEogExt,
829            },
830            Self::Status(header) => {
831                if header.extensions.is_empty() {
832                    DatagramType::DatagramStatus
833                } else {
834                    DatagramType::DatagramStatusExt
835                }
836            }
837        }
838    }
839
840    /// Decode a datagram from its first byte, type field included.
841    ///
842    /// Errors with [`CodecError::UnknownDatagramType`] when the datagram table
843    /// does not assign the leading type, which this draft answers with a close.
844    /// `datagram_type_error` settles it against that table alone — 0x05 is
845    /// assigned in both tables here and means different things in each.
846    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
847        let raw = VarInt::decode(buf)?.into_inner();
848        let datagram_type = DatagramType::from_id(raw).ok_or_else(|| datagram_type_error(raw))?;
849        let has_extensions = datagram_type.has_extensions();
850        if datagram_type.is_status() {
851            return Ok(Self::Status(DatagramStatusHeader::decode_with_extensions(
852                has_extensions,
853                buf,
854            )?));
855        }
856        let mut header = DatagramHeader::decode_with_extensions(has_extensions, buf)?;
857        header.end_of_group = datagram_type.is_end_of_group();
858        Ok(Self::Payload(header))
859    }
860
861    /// Encode the datagram, type field included.
862    pub fn encode(&self, buf: &mut impl BufMut) {
863        let datagram_type = self.datagram_type();
864        let has_extensions = datagram_type.has_extensions();
865        VarInt::from_usize(datagram_type as usize).encode(buf);
866        match self {
867            Self::Payload(header) => header.encode_with_extensions(has_extensions, buf),
868            Self::Status(header) => header.encode_with_extensions(has_extensions, buf),
869        }
870    }
871
872    /// Encode the datagram, refusing a header the framing it names cannot
873    /// carry.
874    ///
875    /// The body is built before anything reaches `buf`, so a refused datagram
876    /// leaves `buf` untouched rather than a type field with no body under it.
877    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
878        let mut body = Vec::with_capacity(64);
879        let datagram_type = self.datagram_type();
880        let has_extensions = datagram_type.has_extensions();
881        match self {
882            Self::Payload(header) => header.encode_with_extensions(has_extensions, &mut body),
883            Self::Status(header) => {
884                check_extensions_against_status(header.object_status, &header.extensions)?;
885                header.encode_with_extensions(has_extensions, &mut body);
886            }
887        }
888        VarInt::from_usize(datagram_type as usize).encode(buf);
889        buf.put_slice(&body);
890        Ok(())
891    }
892}
893
894// ============================================================
895// Fetch stream (type 0x05)
896// ============================================================
897
898#[derive(Debug, Clone, PartialEq, Eq)]
899pub struct FetchHeader {
900    pub request_id: VarInt,
901}
902
903#[derive(Debug, Clone, PartialEq, Eq)]
904pub struct FetchObjectHeader {
905    pub group_id: VarInt,
906    pub subgroup_id: VarInt,
907    pub object_id: VarInt,
908    pub publisher_priority: u8,
909    pub extension_headers_length: VarInt,
910    pub extensions: Vec<u8>,
911    pub payload_length: VarInt,
912    pub object_status: ObjectStatus,
913}
914
915impl FetchHeader {
916    /// Encode a fetch stream header including its leading stream-type field,
917    /// so the bytes form the start of a data stream a peer can read.
918    ///
919    /// [`Self::encode`] writes the body alone, which is what a caller wants
920    /// once the stream is already open and what a caller must not use for its
921    /// first write. The read side has had [`Self::decode_stream`] all along,
922    /// so without this the codec could not round-trip its own fetch stream
923    /// through its own reader.
924    pub fn encode_stream(&self, buf: &mut impl BufMut) {
925        VarInt::from_usize(StreamType::Fetch as usize).encode(buf);
926        self.encode(buf);
927    }
928
929    pub fn encode(&self, buf: &mut impl BufMut) {
930        self.request_id.encode(buf);
931    }
932
933    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
934        let request_id = VarInt::decode(buf)?;
935        Ok(Self { request_id })
936    }
937
938    /// Decode a fetch header from the start of a data stream, consuming the
939    /// leading stream type varint.
940    ///
941    /// Errors with [`CodecError::UnknownStreamType`] when the stream table does
942    /// not assign the leading type, which this draft answers with a close, and
943    /// with [`CodecError::InvalidField`] when it is assigned but is not
944    /// [`StreamType::Fetch`]. `stream_type_error` draws that line.
945    pub fn decode_stream(buf: &mut impl Buf) -> Result<Self, CodecError> {
946        let stream_type = VarInt::decode(buf)?.into_inner();
947        if stream_type != StreamType::Fetch as u64 {
948            return Err(stream_type_error(stream_type));
949        }
950        Self::decode(buf)
951    }
952}
953
954impl FetchObjectHeader {
955    pub fn encode(&self, buf: &mut impl BufMut) {
956        self.group_id.encode(buf);
957        self.subgroup_id.encode(buf);
958        self.object_id.encode(buf);
959        buf.put_u8(self.publisher_priority);
960        VarInt::from_usize(self.extensions.len()).encode(buf);
961        buf.put_slice(&self.extensions);
962        self.payload_length.encode(buf);
963        if self.payload_length.into_inner() == 0 {
964            VarInt::from_usize(self.object_status as usize).encode(buf);
965        }
966    }
967
968    /// Encode the header, refusing a status the framing cannot carry.
969    ///
970    /// Section 9.4.4 puts the Object Status field on the wire only when the
971    /// Object Payload Length is zero, and Section 9.2.1.1 says "Any object
972    /// with a status code other than zero MUST have an empty payload". A
973    /// non-zero status paired with a non-zero payload length therefore has no
974    /// encoding at all: [`Self::encode`] drops the status and the peer reads an
975    /// ordinary object, which is a different object from the one the caller
976    /// described. This refuses instead.
977    ///
978    /// The datagram types on this draft already refuse the same pairing. These
979    /// two did not, and they are the ones a publisher writes on every stream.
980    ///
981    /// # Errors
982    ///
983    /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
984    /// a non-zero Object Payload Length.
985    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
986        if self.payload_length.into_inner() != 0 && self.object_status as usize != 0 {
987            return Err(CodecError::InvalidField);
988        }
989        self.encode(buf);
990        Ok(())
991    }
992
993    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
994        let group_id = VarInt::decode(buf)?;
995        let subgroup_id = VarInt::decode(buf)?;
996        let object_id = VarInt::decode(buf)?;
997        if buf.remaining() < 1 {
998            return Err(CodecError::UnexpectedEnd);
999        }
1000        let publisher_priority = buf.get_u8();
1001        let extension_headers_length = VarInt::decode(buf)?;
1002        let extensions = read_extension_bytes(buf, extension_headers_length.into_inner())?;
1003        let payload_length = VarInt::decode(buf)?;
1004        let object_status = if payload_length.into_inner() == 0 {
1005            let sv = VarInt::decode(buf)?.into_inner();
1006            ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?
1007        } else {
1008            ObjectStatus::Normal
1009        };
1010        check_extensions_against_status(object_status, &extensions)?;
1011        Ok(Self {
1012            group_id,
1013            subgroup_id,
1014            object_id,
1015            publisher_priority,
1016            extension_headers_length,
1017            extensions,
1018            payload_length,
1019            object_status,
1020        })
1021    }
1022}
1023
1024#[cfg(test)]
1025mod tests {
1026    use super::*;
1027
1028    /// A payload datagram for track 1, group 0, object 7, priority 128, with
1029    /// no extension block — the shape [`DatagramHeader::encode`] writes.
1030    fn payload_datagram() -> DatagramHeader {
1031        DatagramHeader {
1032            track_alias: VarInt::from_usize(1),
1033            group_id: VarInt::from_usize(0),
1034            object_id: VarInt::from_usize(7),
1035            publisher_priority: 128,
1036            extension_headers_length: VarInt::from_usize(0),
1037            extensions: Vec::new(),
1038            end_of_group: false,
1039        }
1040    }
1041
1042    /// The same datagram on the message that does carry a status.
1043    fn status_datagram(status: ObjectStatus) -> DatagramStatusHeader {
1044        DatagramStatusHeader {
1045            track_alias: VarInt::from_usize(1),
1046            group_id: VarInt::from_usize(0),
1047            object_id: VarInt::from_usize(7),
1048            publisher_priority: 128,
1049            extension_headers_length: VarInt::from_usize(0),
1050            extensions: Vec::new(),
1051            object_status: status,
1052        }
1053    }
1054
1055    /// Draft-12's payload datagram has no status for the encoder to drop, and
1056    /// so nothing for the fallible encode to refuse.
1057    ///
1058    /// [`DatagramHeader`] is the OBJECT_DATAGRAM of Section 9.3.1, whose layout
1059    /// carries no Object Status field at all. A datagram that states a status
1060    /// is the separate OBJECT_DATAGRAM_STATUS message of Section 9.3.2, modelled
1061    /// here as [`DatagramStatusHeader`]. The refusal drafts 07, 08 and 14-19
1062    /// need on this path therefore has nothing to bite on, and this gate holds
1063    /// [`DatagramHeader::encode_checked`] to writing exactly what
1064    /// [`DatagramHeader::encode`] writes and never refusing — the alternative
1065    /// being a codec that answers `Err` for a datagram every draft-12 publisher
1066    /// is entitled to send.
1067    ///
1068    /// The second half is what makes the first half safe rather than merely
1069    /// permissive. Every status draft-12 assigns travels intact on the message
1070    /// that can express one, so leaving the payload datagram unchecked loses
1071    /// nothing; if a status could ride this header, the check being skipped
1072    /// here would be the check drafts 07 and 18 need.
1073    ///
1074    /// # What this catches, observed by making each change and running it
1075    ///
1076    /// Making `encode_checked` refuse unconditionally, as an over-eager copy of
1077    /// the drafts that do need a check would:
1078    ///
1079    /// ```text
1080    /// draft-12's payload datagram has no status to refuse: InvalidField
1081    /// ```
1082    ///
1083    /// Making `encode_checked` return `Ok(())` without writing anything:
1084    ///
1085    /// ```text
1086    /// assertion `left == right` failed: the fallible encode must write exactly what `encode` writes
1087    ///   left: []
1088    ///  right: [1, 0, 7, 128]
1089    /// ```
1090    ///
1091    /// Making `DatagramStatusHeader::encode_with_extensions` write a constant
1092    /// `ObjectStatus::Normal` instead of the header's own status, so the
1093    /// message that is supposed to carry a status stops doing so:
1094    ///
1095    /// ```text
1096    /// assertion `left == right` failed: ObjectDoesNotExist must survive on the message that carries a status
1097    ///   left: Normal
1098    ///  right: ObjectDoesNotExist
1099    /// ```
1100    #[test]
1101    fn a_payload_datagram_has_no_status_to_refuse() {
1102        let header = payload_datagram();
1103
1104        let mut checked = Vec::new();
1105        header.encode_checked(&mut checked).unwrap_or_else(|e| {
1106            panic!("draft-12's payload datagram has no status to refuse: {e:?}")
1107        });
1108
1109        let mut plain = Vec::new();
1110        header.encode(&mut plain);
1111        assert_eq!(checked, plain, "the fallible encode must write exactly what `encode` writes");
1112
1113        let decoded = DatagramHeader::decode(&mut &checked[..])
1114            .expect("the bytes encode_checked wrote must parse back");
1115        assert_eq!(decoded, header, "the payload datagram did not survive its own encoding");
1116
1117        for &status in ObjectStatus::ALL {
1118            let mut buf = Vec::new();
1119            status_datagram(status).encode(&mut buf);
1120            let decoded = DatagramStatusHeader::decode(&mut &buf[..])
1121                .unwrap_or_else(|e| panic!("the status datagram for {status:?} must parse: {e:?}"));
1122            assert_eq!(
1123                decoded.object_status, status,
1124                "{status:?} must survive on the message that carries a status"
1125            );
1126        }
1127    }
1128}