Skip to main content

moqtap_codec/draft13/
data_stream.rs

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