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