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::object_properties::{parse_object_properties, ObjectProperty, PropertyValue};
11use crate::types::read_bytes;
12use crate::varint::VarInt;
13use crate::version::DraftVersion;
14use bytes::{Buf, BufMut};
15
16/// Stream type IDs for draft-12 data streams.
17///
18/// # Draft-12 contradicts itself about where the subgroup types sit
19///
20/// Three places in this draft name the SUBGROUP_HEADER Type range, and they do
21/// not agree. Two of them carry draft-11's answer:
22///
23///   - Section 9, Table 10, the table of unidirectional stream types, whose
24///     SUBGROUP_HEADER row reads 0x08-0x0D.
25///   - Section 9.4.2, Figure 35, the header's own layout:
26///     `Type (i) = 0x8..0xD`.
27///
28/// The third does not. Section 9.4.2 says "There are 12 defined Type values for
29/// SUBGROUP_HEADER" and Table 13, immediately under that figure, lists them:
30/// 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x18, 0x19, 0x1A, 0x1B, 0x1C and 0x1D.
31///
32/// **This enum implements Table 13**, so a stream opening with 0x08 through
33/// 0x0D is [`CodecError::UnknownStreamType`] and closes the session.
34///
35/// Table 13 is the surviving half: a code-point update that missed a spot.
36/// The datagram-status disagreement recorded on [`DatagramType`] has the same
37/// shape and is settled the same way there.
38///
39///   - The range 0x08-0x0D holds six values, and this draft defines twelve
40///     types. Draft-11 Section 9.4.2 Table 11 lists exactly six, at 0x08
41///     through 0x0D, and the twelve here are those six crossed with the End Of
42///     Group bit draft-12 added — its own change log says so: "Use bits in
43///     SUBGROUP_HEADER and DATAGRAM* types to compress subgroup ID and
44///     extensions". So 0x08-0x0D and `0x8..0xD` are draft-11's range left
45///     behind, and they cannot hold what this draft defines.
46///   - Table 13 is the only one of the three that says what each value *means*.
47///     The other two give a range and nothing else, so following either would
48///     leave every framing decision — whether a Subgroup ID field is on the
49///     wire, whether objects carry extensions, whether the stream ends the
50///     group — with nothing to read it from.
51///   - Draft-14 keeps Table 13 unchanged and corrects the other two to match:
52///     its Table 10 reads "0x10-0x1D" and its Figure reads
53///     `Type (i) = 0x10..0x1D`. That is the disagreement being resolved in
54///     favour of Table 13 by the working group, one draft later.
55///
56/// Draft-13 repeats all three halves unchanged, and carries the same note.
57///
58/// The cost of being wrong is asymmetric and points the same way. Accepting
59/// 0x08-0x0D as well would mean parsing a stream under framing no table
60/// assigns it, which is the failure this codebase refuses elsewhere — an
61/// out-of-range Type aliased onto a valid one produces objects with plausible,
62/// wrong contents. Refusing them closes a session with a peer that followed the
63/// stale half of its own draft, which is visible, reportable, and what
64/// draft-14 says the peer should not have done.
65#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66#[repr(u64)]
67pub enum StreamType {
68    /// Fetch response stream (0x05).
69    Fetch = 0x05,
70    /// Subgroup: subgroup_id=0, no extensions (0x10).
71    SubgroupZero = 0x10,
72    /// Subgroup: subgroup_id=0, with extensions (0x11).
73    SubgroupZeroExt = 0x11,
74    /// Subgroup: subgroup_id=first object ID, no extensions (0x12).
75    SubgroupFirstObj = 0x12,
76    /// Subgroup: subgroup_id=first object ID, with extensions (0x13).
77    SubgroupFirstObjExt = 0x13,
78    /// Subgroup: explicit subgroup_id, no extensions (0x14).
79    SubgroupExplicit = 0x14,
80    /// Subgroup: explicit subgroup_id, with extensions (0x15).
81    SubgroupExplicitExt = 0x15,
82    /// Subgroup: subgroup_id=0, contains end of group, no extensions (0x18).
83    SubgroupZeroEog = 0x18,
84    /// Subgroup: subgroup_id=0, contains end of group, with extensions (0x19).
85    SubgroupZeroEogExt = 0x19,
86    /// Subgroup: subgroup_id=first object ID, contains end of group, no extensions (0x1A).
87    SubgroupFirstObjEog = 0x1A,
88    /// Subgroup: subgroup_id=first object ID, contains end of group, with extensions (0x1B).
89    SubgroupFirstObjEogExt = 0x1B,
90    /// Subgroup: explicit subgroup_id, contains end of group, no extensions (0x1C).
91    SubgroupExplicitEog = 0x1C,
92    /// Subgroup: explicit subgroup_id, contains end of group, with extensions (0x1D).
93    SubgroupExplicitEogExt = 0x1D,
94}
95
96/// Hold an object to the rule that a non-existent object carries no extensions.
97///
98/// Section 9.2.1.2: "Any Object may have extension headers except those with
99/// Object Status 'Object Does Not Exist'. If an endpoint receives a non-existent
100/// Object containing extension headers it MUST close the session with a Protocol
101/// Violation."
102///
103/// The sentence is about a receiver, and it reaches all three carriers that can
104/// announce a status: an object on a subgroup stream, an object on a fetch
105/// stream, and a status datagram. A plain datagram has no status field, so it
106/// is the only carrier that cannot break the rule.
107///
108/// Reported under [`CodecError::ExtensionsOnNonExistentObject`], which is this
109/// rule and nothing else. [`CodecError::InvalidField`] is too coarse for it:
110/// shared with a dozen unrelated malformations the draft does not answer with a
111/// close, it leaves a caller unable to act on the sentence above.
112fn check_extensions_against_status(
113    status: ObjectStatus,
114    extensions: &[u8],
115) -> Result<(), CodecError> {
116    if status == ObjectStatus::ObjectDoesNotExist && !extensions.is_empty() {
117        return Err(CodecError::ExtensionsOnNonExistentObject(extensions.len()));
118    }
119    Ok(())
120}
121
122impl StreamType {
123    pub fn from_id(id: u64) -> Option<Self> {
124        match id {
125            0x05 => Some(StreamType::Fetch),
126            0x10 => Some(StreamType::SubgroupZero),
127            0x11 => Some(StreamType::SubgroupZeroExt),
128            0x12 => Some(StreamType::SubgroupFirstObj),
129            0x13 => Some(StreamType::SubgroupFirstObjExt),
130            0x14 => Some(StreamType::SubgroupExplicit),
131            0x15 => Some(StreamType::SubgroupExplicitExt),
132            0x18 => Some(StreamType::SubgroupZeroEog),
133            0x19 => Some(StreamType::SubgroupZeroEogExt),
134            0x1A => Some(StreamType::SubgroupFirstObjEog),
135            0x1B => Some(StreamType::SubgroupFirstObjEogExt),
136            0x1C => Some(StreamType::SubgroupExplicitEog),
137            0x1D => Some(StreamType::SubgroupExplicitEogExt),
138            _ => None,
139        }
140    }
141
142    pub fn is_subgroup(&self) -> bool {
143        matches!(
144            self,
145            StreamType::SubgroupZero
146                | StreamType::SubgroupZeroExt
147                | StreamType::SubgroupFirstObj
148                | StreamType::SubgroupFirstObjExt
149                | StreamType::SubgroupExplicit
150                | StreamType::SubgroupExplicitExt
151                | StreamType::SubgroupZeroEog
152                | StreamType::SubgroupZeroEogExt
153                | StreamType::SubgroupFirstObjEog
154                | StreamType::SubgroupFirstObjEogExt
155                | StreamType::SubgroupExplicitEog
156                | StreamType::SubgroupExplicitEogExt
157        )
158    }
159
160    pub fn has_extensions(&self) -> bool {
161        matches!(
162            self,
163            StreamType::SubgroupZeroExt
164                | StreamType::SubgroupFirstObjExt
165                | StreamType::SubgroupExplicitExt
166                | StreamType::SubgroupZeroEogExt
167                | StreamType::SubgroupFirstObjEogExt
168                | StreamType::SubgroupExplicitEogExt
169        )
170    }
171
172    /// True if this subgroup stream type indicates the stream contains the end of its group.
173    pub fn contains_end_of_group(&self) -> bool {
174        matches!(
175            self,
176            StreamType::SubgroupZeroEog
177                | StreamType::SubgroupZeroEogExt
178                | StreamType::SubgroupFirstObjEog
179                | StreamType::SubgroupFirstObjEogExt
180                | StreamType::SubgroupExplicitEog
181                | StreamType::SubgroupExplicitEogExt
182        )
183    }
184
185    /// True if this subgroup stream type puts an explicit Subgroup ID on the
186    /// wire.
187    ///
188    /// The Subgroup ID Field Present column of the SUBGROUP_HEADER type table
189    /// in Section 9.4.2. The other two columns of that row say what the
190    /// Subgroup ID *is* where the field is absent — zero, or the first
191    /// Object's ID — so this is only about the field, never about the value.
192    pub fn writes_subgroup_id(&self) -> bool {
193        matches!(
194            self,
195            StreamType::SubgroupExplicit
196                | StreamType::SubgroupExplicitExt
197                | StreamType::SubgroupExplicitEog
198                | StreamType::SubgroupExplicitEogExt
199        )
200    }
201}
202
203/// Datagram wire types (separate namespace from QUIC stream types).
204///
205/// The two namespaces overlap in draft-12 and cannot share one enum: 0x05 is
206/// FETCH_HEADER among stream types and OBJECT_DATAGRAM_STATUS with extensions
207/// among datagram types.
208///
209/// The values follow the table in draft-12 Section 9: 0x00-0x03 are
210/// OBJECT_DATAGRAM and 0x04-0x05 are OBJECT_DATAGRAM_STATUS. The two close rules
211/// for an empty extension block still carry draft-11's code points, and are
212/// read through the table:
213///
214///   - draft-12 Section 9.3.2: "The Type field takes the form 0b0000001X (or
215///     the set of values from 0x02 to 0x03)" and "If an endpoint receives a
216///     datagram with Type 0x03 and Extension Headers Length is 0, it MUST close
217///     the session with Protocol Violation." draft-11 Section 9 assigns
218///     0x02-0x03 to OBJECT_DATAGRAM_STATUS. draft-14 Appendix A.1 lists
219///     "Corrected missed code point updates to Object Datagram Status (#1082)",
220///     and that change rewrites the sentence to "Type 0x05". It names 0x05.
221///   - draft-12 Section 9.3.1: "If an endpoint receives a datagram with Type
222///     0x01 and Extension Headers Length is 0, it MUST close the session with
223///     Protocol Violation." 0x01 is the only OBJECT_DATAGRAM Type with
224///     extensions in draft-11 Section 9.2. draft-12 Section 9.3.1 Table 12
225///     adds 0x03, End Of Group with Extensions Present "Yes", and the sentence
226///     was not updated. draft-14 Section 10.3.1 states the rule for every such
227///     Type: "If an endpoint receives a datagram with Extensions Present as
228///     "Yes" and a Extension Headers Length of 0, it MUST close the session
229///     with PROTOCOL_VIOLATION." It names 0x01 and 0x03.
230///
231/// So 0x01, 0x03 and 0x05 carry extensions, and each is refused with an
232/// Extension Headers Length of 0. 0x00, 0x02 and 0x04 carry no Extension
233/// Headers Length. The client's `EmptyDatagramExtensions` closes the session
234/// over the same three Types.
235#[derive(Debug, Clone, Copy, PartialEq, Eq)]
236#[repr(u64)]
237pub enum DatagramType {
238    /// Object datagram, no extensions (0x00).
239    Datagram = 0x00,
240    /// Object datagram, with extensions (0x01).
241    DatagramExt = 0x01,
242    /// Object datagram carrying the end of its group, no extensions (0x02).
243    DatagramEog = 0x02,
244    /// Object datagram carrying the end of its group, with extensions (0x03).
245    DatagramEogExt = 0x03,
246    /// Object datagram status, no extensions (0x04).
247    DatagramStatus = 0x04,
248    /// Object datagram status, with extensions (0x05).
249    DatagramStatusExt = 0x05,
250}
251
252impl DatagramType {
253    pub fn from_id(id: u64) -> Option<Self> {
254        match id {
255            0x00 => Some(DatagramType::Datagram),
256            0x01 => Some(DatagramType::DatagramExt),
257            0x02 => Some(DatagramType::DatagramEog),
258            0x03 => Some(DatagramType::DatagramEogExt),
259            0x04 => Some(DatagramType::DatagramStatus),
260            0x05 => Some(DatagramType::DatagramStatusExt),
261            _ => None,
262        }
263    }
264
265    pub fn has_extensions(&self) -> bool {
266        matches!(
267            self,
268            DatagramType::DatagramExt
269                | DatagramType::DatagramEogExt
270                | DatagramType::DatagramStatusExt
271        )
272    }
273
274    pub fn is_status(&self) -> bool {
275        matches!(self, DatagramType::DatagramStatus | DatagramType::DatagramStatusExt)
276    }
277
278    pub fn is_end_of_group(&self) -> bool {
279        matches!(self, DatagramType::DatagramEog | DatagramType::DatagramEogExt)
280    }
281}
282
283/// Which failure a leading unidirectional stream type that is not the one a
284/// reader wants is.
285///
286/// Section 9: "An endpoint that receives an unknown stream or datagram type
287/// MUST close the session." One sentence, two tables, and on this draft the two
288/// tables collide: 0x05 is FETCH_HEADER in the stream table and
289/// OBJECT_DATAGRAM_STATUS with extensions in the datagram table. Which table
290/// was consulted is therefore part of the answer, not a detail, and it is why
291/// [`CodecError::UnknownStreamType`] and [`CodecError::UnknownDatagramType`]
292/// are separate variants rather than one.
293///
294/// The stream table assigns 0x05 and the range 0x10 to 0x1D. Everything outside
295/// them is unknown at the head of a stream, and the session ends.
296fn stream_type_error(raw: u64) -> CodecError {
297    if StreamType::from_id(raw).is_some() {
298        CodecError::InvalidField
299    } else {
300        CodecError::UnknownStreamType(raw)
301    }
302}
303
304/// Which failure a leading datagram type that is not one a reader wants is.
305///
306/// The datagram half of the sentence quoted on `stream_type_error`, read
307/// against the other table: 0x00 to 0x05 are what it assigns, and everything
308/// else arriving as a datagram is unknown.
309///
310/// Answered from [`DatagramType`] alone, never from [`StreamType`]. A value in
311/// both tables means one thing as a datagram and another as a stream, and
312/// consulting the wrong one turns an assigned datagram type into an unknown one
313/// or the reverse.
314fn datagram_type_error(raw: u64) -> CodecError {
315    if DatagramType::from_id(raw).is_some() {
316        CodecError::InvalidField
317    } else {
318        CodecError::UnknownDatagramType(raw)
319    }
320}
321
322// ── Extension helpers ─────────────────────────────────────────
323
324fn read_extension_bytes(buf: &mut impl Buf, byte_len: u64) -> Result<Vec<u8>, CodecError> {
325    read_bytes(buf, byte_len as usize)
326}
327
328// ============================================================
329// Subgroup stream header
330// ============================================================
331
332#[derive(Debug, Clone, PartialEq, Eq)]
333pub struct SubgroupHeader {
334    pub stream_type: StreamType,
335    pub track_alias: VarInt,
336    pub group_id: VarInt,
337    pub subgroup_id: VarInt,
338    pub publisher_priority: u8,
339}
340
341impl SubgroupHeader {
342    /// Encode a subgroup stream header including its leading stream-type
343    /// field, so the bytes form the start of a data stream a peer can read.
344    ///
345    /// [`Self::encode`] writes the body alone, which is what a caller wants
346    /// once the stream is already open and what a caller must not use for its
347    /// first write. It is also the half that cannot stand on its own here,
348    /// because the stream type is what says whether a Subgroup ID follows it
349    /// and whether the objects on the stream carry extension headers.
350    pub fn encode_stream(&self, buf: &mut impl BufMut) {
351        VarInt::from_usize(self.stream_type as usize).encode(buf);
352        self.encode(buf);
353    }
354
355    /// Encode the header body, without its leading stream-type field.
356    ///
357    /// Driven by the stream type, and silent about a `subgroup_id` it decides
358    /// not to write: on a type whose Subgroup ID Field Present column reads No
359    /// the field is dropped, and the peer reads the subgroup the *type* names -
360    /// zero, or the first Object's ID - rather than the one in hand. Nothing is
361    /// malformed about the result, which is what makes it worth refusing rather
362    /// than tolerating. [`Self::encode_checked`] refuses it.
363    pub fn encode(&self, buf: &mut impl BufMut) {
364        self.track_alias.encode(buf);
365        self.group_id.encode(buf);
366        if self.stream_type.writes_subgroup_id() {
367            self.subgroup_id.encode(buf);
368        }
369        buf.put_u8(self.publisher_priority);
370    }
371
372    /// Encode the header body, refusing a Subgroup ID this stream type has
373    /// nowhere to put.
374    ///
375    /// [`Self::decode_with_type`] leaves the field at zero for every type that
376    /// does not carry it, so a decoded header always passes: the refusal is for
377    /// a header assembled by hand, where a caller set an ID the type will
378    /// discard.
379    ///
380    /// A zero is accepted under any type. It is what the decoder produces, and
381    /// on a Subgroup ID Value column reading `0` it is also the truth, so
382    /// refusing it would refuse the ordinary case to catch nothing.
383    ///
384    /// # Errors
385    ///
386    /// [`CodecError::InvalidField`] if a non-zero Subgroup ID sits under a
387    /// stream type that writes no Subgroup ID field.
388    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
389        if !self.stream_type.writes_subgroup_id() && self.subgroup_id.into_inner() != 0 {
390            return Err(CodecError::InvalidField);
391        }
392        self.encode(buf);
393        Ok(())
394    }
395
396    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
397        Self::decode_with_type(StreamType::SubgroupExplicit, buf)
398    }
399
400    pub fn decode_with_type(
401        stream_type: StreamType,
402        buf: &mut impl Buf,
403    ) -> Result<Self, CodecError> {
404        let track_alias = VarInt::decode(buf)?;
405        let group_id = VarInt::decode(buf)?;
406        let subgroup_id = match stream_type {
407            StreamType::SubgroupZero
408            | StreamType::SubgroupZeroExt
409            | StreamType::SubgroupZeroEog
410            | StreamType::SubgroupZeroEogExt => VarInt::from_usize(0),
411            StreamType::SubgroupExplicit
412            | StreamType::SubgroupExplicitExt
413            | StreamType::SubgroupExplicitEog
414            | StreamType::SubgroupExplicitEogExt => VarInt::decode(buf)?,
415            StreamType::SubgroupFirstObj
416            | StreamType::SubgroupFirstObjExt
417            | StreamType::SubgroupFirstObjEog
418            | StreamType::SubgroupFirstObjEogExt => VarInt::from_usize(0),
419            _ => return Err(CodecError::InvalidField),
420        };
421        if buf.remaining() < 1 {
422            return Err(CodecError::UnexpectedEnd);
423        }
424        let publisher_priority = buf.get_u8();
425        Ok(Self { stream_type, track_alias, group_id, subgroup_id, publisher_priority })
426    }
427
428    /// Decode a subgroup header from the start of a data stream, consuming
429    /// the leading stream type varint and using it to select the variant.
430    ///
431    /// Errors with [`CodecError::UnknownStreamType`] when the stream table does
432    /// not assign the leading type, which this draft answers with a close, and
433    /// with [`CodecError::InvalidField`] when it does assign it but not to a
434    /// subgroup. `stream_type_error` draws that line.
435    pub fn decode_stream(buf: &mut impl Buf) -> Result<Self, CodecError> {
436        let raw = VarInt::decode(buf)?.into_inner();
437        let stream_type = StreamType::from_id(raw).ok_or_else(|| stream_type_error(raw))?;
438        if !stream_type.is_subgroup() {
439            return Err(stream_type_error(raw));
440        }
441        Self::decode_with_type(stream_type, buf)
442    }
443}
444
445// ============================================================
446// Object header within subgroup
447// ============================================================
448
449#[derive(Debug, Clone, PartialEq, Eq)]
450pub struct ObjectHeader {
451    pub object_id: VarInt,
452    pub extension_headers_length: VarInt,
453    pub extensions: Vec<u8>,
454    pub payload_length: VarInt,
455    pub object_status: ObjectStatus,
456}
457
458impl ObjectHeader {
459    /// Encode the object header in the framing that carries no extension
460    /// block.
461    ///
462    /// Lossy, and lossy in a way the caller cannot see: an object holding
463    /// extension headers is written without them and without a word. Which
464    /// framing is correct is not a property of the object at all - Section
465    /// 9.4.2 gives the stream's type an Extensions Present column, and every
466    /// object on the stream follows it - so this entry point can only guess,
467    /// and it guesses "absent". Prefer [`Self::encode_with_extensions`], which
468    /// is told, or [`Self::encode_checked`], which refuses what it would
469    /// otherwise drop.
470    pub fn encode(&self, buf: &mut impl BufMut) {
471        self.encode_with_extensions(false, buf);
472    }
473
474    /// Encode the header, refusing a status the framing cannot carry.
475    ///
476    /// Section 9.4.2 puts the Object Status field on the wire only when the
477    /// Object Payload Length is zero, and Section 9.2.1.1 says "Any object
478    /// with a status code other than zero MUST have an empty payload". A
479    /// non-zero status paired with a non-zero payload length therefore has no
480    /// encoding at all: [`Self::encode`] drops the status and the peer reads an
481    /// ordinary object, which is a different object from the one the caller
482    /// described. This refuses instead.
483    ///
484    /// The datagram types on this draft already refuse the same pairing. These
485    /// two did not, and they are the ones a publisher writes on every stream.
486    ///
487    /// Extension headers are refused here rather than dropped, for a reason
488    /// the status rule does not share: this entry point writes the framing
489    /// that has no Extension Headers Length field, so the bytes have nowhere
490    /// to go. Writing them anyway is not an option and losing them silently
491    /// puts a stream on the wire that no reader can follow - a reader on an
492    /// extensions-bearing stream takes the Object Payload Length as the
493    /// extension length and every object after it is misread. A caller that
494    /// knows the stream's framing wants
495    /// [`Self::encode_checked_with_extensions`].
496    ///
497    /// # Errors
498    ///
499    /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
500    /// a non-zero Object Payload Length, or if the object carries extension
501    /// headers this framing cannot write.
502    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
503        self.encode_checked_with_extensions(false, buf)
504    }
505
506    /// Encode the object header into a stream whose type has already settled
507    /// whether objects carry an extension block, refusing what that framing
508    /// cannot express.
509    ///
510    /// `has_extensions` is the stream's answer, not the object's: Section
511    /// 9.4.2 fixes it for the whole stream from the SUBGROUP_HEADER type, so
512    /// an object with no extensions on a stream that carries them still writes
513    /// a length of zero, and that is the one direction not refused here. The
514    /// other direction has no encoding, so it is refused.
515    ///
516    /// # Errors
517    ///
518    /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
519    /// a non-zero Object Payload Length, or if `has_extensions` is `false`
520    /// while the object carries extension headers.
521    pub fn encode_checked_with_extensions(
522        &self,
523        has_extensions: bool,
524        buf: &mut impl BufMut,
525    ) -> Result<(), CodecError> {
526        if self.payload_length.into_inner() != 0 && self.object_status as usize != 0 {
527            return Err(CodecError::InvalidField);
528        }
529        if !has_extensions && !self.extensions.is_empty() {
530            return Err(CodecError::InvalidField);
531        }
532        self.encode_with_extensions(has_extensions, buf);
533        Ok(())
534    }
535
536    /// Encode the object header, writing the extension block only when the
537    /// stream's type says objects carry one.
538    ///
539    /// Infallible, and so unable to say that a `false` here discards the
540    /// extension headers the object holds. [`Self::encode_checked_with_extensions`]
541    /// is the same write with that refusal in front of it.
542    pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
543        self.object_id.encode(buf);
544        if has_extensions {
545            VarInt::from_usize(self.extensions.len()).encode(buf);
546            buf.put_slice(&self.extensions);
547        }
548        self.payload_length.encode(buf);
549        if self.payload_length.into_inner() == 0 {
550            VarInt::from_usize(self.object_status as usize).encode(buf);
551        }
552    }
553
554    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
555        Self::decode_with_extensions(false, buf)
556    }
557
558    pub fn decode_with_extensions(
559        has_extensions: bool,
560        buf: &mut impl Buf,
561    ) -> Result<Self, CodecError> {
562        let object_id = VarInt::decode(buf)?;
563        let (extension_headers_length, extensions) = if has_extensions {
564            let ehl = VarInt::decode(buf)?;
565            let ext = read_extension_bytes(buf, ehl.into_inner())?;
566            (ehl, ext)
567        } else {
568            (VarInt::from_usize(0), Vec::new())
569        };
570        let payload_length = VarInt::decode(buf)?;
571        let object_status = if payload_length.into_inner() == 0 {
572            let sv = VarInt::decode(buf)?.into_inner();
573            ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?
574        } else {
575            ObjectStatus::Normal
576        };
577        check_extensions_against_status(object_status, &extensions)?;
578        Ok(Self { object_id, extension_headers_length, extensions, payload_length, object_status })
579    }
580}
581
582// ============================================================
583// Prior Group ID Gap
584// ============================================================
585
586/// Prior Group ID Gap, Extension Header Type 0x40 (draft-12 Section 10.1).
587pub const PRIOR_GROUP_ID_GAP: u64 = 0x40;
588
589/// The Prior Group ID Gap values one Object carries.
590///
591/// draft-12 Section 10.1: "A track with a Group that contains more than one
592/// Object with different values for Prior Group ID Gap or has a Prior Group ID
593/// Gap larger than the Group ID is considered malformed." Malformed Tracks are
594/// draft-12 Section 2.5. This draft does not limit an Object to one instance,
595/// so every value is kept. The least is held to the Group ID, for the reason
596/// [`SubgroupGaps`] gives.
597#[derive(Debug, Clone, Default, PartialEq, Eq)]
598pub struct PriorGaps {
599    /// Each Prior Group ID Gap, in wire order.
600    pub group: Vec<u64>,
601}
602
603impl PriorGaps {
604    /// Read the gaps from an Object's parsed extension headers, refusing them
605    /// with [`CodecError::MalformedTrack`] when every one is larger than
606    /// `group_id`.
607    pub fn from_properties(
608        properties: &[ObjectProperty<'_>],
609        group_id: u64,
610    ) -> Result<Self, CodecError> {
611        let group: Vec<u64> = properties
612            .iter()
613            .filter_map(|property| match (property.property_type, &property.value) {
614                (PRIOR_GROUP_ID_GAP, PropertyValue::Varint(value)) => Some(*value),
615                _ => None,
616            })
617            .collect();
618        if let Some(gap) = group.iter().copied().min().filter(|&gap| gap > group_id) {
619            return Err(CodecError::MalformedTrack {
620                detail: format!("Prior Group ID Gap {gap} is larger than Group ID {group_id}"),
621            });
622        }
623        Ok(Self { group })
624    }
625
626    /// [`Self::from_properties`] on an extension block, refusing a block that
627    /// does not parse as [`crate::object_properties::parse_object_properties`]
628    /// does.
629    pub fn from_extensions(extensions: &[u8], group_id: u64) -> Result<Self, CodecError> {
630        let properties = parse_object_properties(DraftVersion::Draft12, extensions)?;
631        Self::from_properties(&properties, group_id)
632    }
633}
634
635/// The Prior Group ID Gap values the Objects of one subgroup stream carried.
636///
637/// A subgroup stream carries one Group, so its Objects are where "a Group that
638/// contains more than one Object with different values for Prior Group ID Gap"
639/// (draft-12 Section 10.1) shows on one stream.
640///
641/// This draft does not say whether an Object may carry 0x40 more than once, or
642/// which value is that Object's when it does. draft-12 Section 9.2.1.2 says
643/// only "Object Extension Headers are serialized as Key-Value-Pairs", and the
644/// no-repeat rule of draft-12 Section 8.2, "Senders MUST NOT repeat the same
645/// parameter type in a message", covers message Parameters. draft-12 Section 12
646/// leaves it open: "List which headers can be repeated in the table." So the
647/// Objects carry different values only when no value is carried by every one of
648/// them: then, whichever of its values is taken as each Object's, two of them
649/// differ. While a value is common to all, taking it as each Object's makes
650/// them equal, and they are not refused. The values one Object carries are not
651/// compared with each other, since the condition needs "more than one Object".
652///
653/// "has a Prior Group ID Gap larger than the Group ID" (draft-12 Section 10.1)
654/// also depends on which value is the Object's. An Object carrying a value
655/// larger than the Group ID and one that is not has a gap larger than the Group
656/// ID under one reading and not under another, so it is refused only when every
657/// value it carries is larger.
658#[derive(Debug, Clone, PartialEq, Eq)]
659pub struct SubgroupGaps {
660    group_id: u64,
661    /// The values every earlier Object that carried a gap carried, or `None`
662    /// before the first such Object.
663    common: Option<Vec<u64>>,
664}
665
666impl SubgroupGaps {
667    /// Track the Objects of the stream `header` opens.
668    pub fn new(header: &SubgroupHeader) -> Self {
669        Self { group_id: header.group_id.into_inner(), common: None }
670    }
671
672    /// Check the next Object on this stream, refusing with
673    /// [`CodecError::MalformedTrack`] a gap larger than the Group ID, or
674    /// values that leave no value common to every Object on the stream that
675    /// carried a gap. Records the Object only when it passes.
676    pub fn check(&mut self, object: &ObjectHeader) -> Result<PriorGaps, CodecError> {
677        let gaps = PriorGaps::from_extensions(&object.extensions, self.group_id)?;
678        if gaps.group.is_empty() {
679            return Ok(gaps);
680        }
681        let common: Vec<u64> = match &self.common {
682            Some(common) => common.iter().copied().filter(|v| gaps.group.contains(v)).collect(),
683            None => {
684                let mut values = gaps.group.clone();
685                values.sort_unstable();
686                values.dedup();
687                values
688            }
689        };
690        if common.is_empty() {
691            return Err(CodecError::MalformedTrack {
692                detail: format!(
693                    "the Objects in Group {} carry no Prior Group ID Gap value in common",
694                    self.group_id
695                ),
696            });
697        }
698        self.common = Some(common);
699        Ok(gaps)
700    }
701}
702
703/// The Prior Group ID Gap conditions of draft-12 Section 10.1 that one
704/// fetch stream shows.
705///
706/// [`PriorGaps::from_extensions`] refuses a gap larger than the Group ID. Across
707/// the stream this refuses Objects of one Group with no value in common, read
708/// as [`SubgroupGaps`] reads them, and "an Object with a Group ID within a
709/// previously communicated gap" for the Group before on the stream. An Object
710/// whose values differ has communicated at least the least of them, whichever
711/// is that Object's. draft-12 Section 8.16: "Within each group,
712/// objects are sent in Object ID order; subgroup ID is not used for ordering."
713///
714/// [`FetchObjectHeader::decode`] reads one Object and applies none of this: a
715/// reader of a fetch stream checks each Object with one `FetchGaps`.
716#[derive(Debug, Clone, PartialEq, Eq)]
717// Boxed so that the fetch readers holding one stay small.
718pub struct FetchGaps(Box<crate::prior_gaps::FetchGapTracker>);
719
720impl Default for FetchGaps {
721    fn default() -> Self {
722        Self::new()
723    }
724}
725
726impl FetchGaps {
727    /// Track the Objects of one fetch stream.
728    pub fn new() -> Self {
729        Self(Box::new(crate::prior_gaps::FetchGapTracker::new(
730            crate::prior_gaps::GapRules::Draft11To13,
731        )))
732    }
733
734    /// Check the next Object on the stream, from its Group ID, Object ID and
735    /// extension block, refusing a gap condition with
736    /// [`CodecError::MalformedTrack`] and a block that does not parse as
737    /// [`parse_object_properties`] does. Records the Object only when it
738    /// passes.
739    pub fn check(
740        &mut self,
741        group_id: u64,
742        object_id: u64,
743        extensions: &[u8],
744    ) -> Result<PriorGaps, CodecError> {
745        let gaps = PriorGaps::from_extensions(extensions, group_id)?;
746        self.0.check(group_id, object_id, &gaps.group, None)?;
747        Ok(gaps)
748    }
749}
750
751// ============================================================
752// Datagram (types 0x00 through 0x03)
753// ============================================================
754
755#[derive(Debug, Clone, PartialEq, Eq)]
756pub struct DatagramHeader {
757    pub track_alias: VarInt,
758    pub group_id: VarInt,
759    pub object_id: VarInt,
760    pub publisher_priority: u8,
761    pub extension_headers_length: VarInt,
762    pub extensions: Vec<u8>,
763    /// Whether this object is the last one in its group.
764    ///
765    /// Draft-12 added this flag and put it in the datagram type rather than in
766    /// the header body, so it is not written or read by the methods below;
767    /// [`DatagramType::is_end_of_group`] is where it lives on the wire.
768    pub end_of_group: bool,
769}
770
771impl DatagramHeader {
772    /// Encode the datagram header in the framing that carries no extension
773    /// block.
774    ///
775    /// Lossy in the same way the subgroup object header is: an extension block
776    /// this value holds is dropped, because the framing being written has no
777    /// field for it. The type byte decides which framing is right, and this
778    /// entry point does not write the type byte, so it cannot consult it.
779    /// [`Datagram::encode`] does both together and never disagrees with itself;
780    /// this is the piece for a caller that has already written the type.
781    pub fn encode(&self, buf: &mut impl BufMut) {
782        self.encode_with_extensions(false, buf);
783    }
784
785    pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
786        self.track_alias.encode(buf);
787        self.group_id.encode(buf);
788        self.object_id.encode(buf);
789        buf.put_u8(self.publisher_priority);
790        if has_extensions {
791            VarInt::from_usize(self.extensions.len()).encode(buf);
792            buf.put_slice(&self.extensions);
793        }
794    }
795
796    /// Encode the datagram header, refusing what this framing cannot carry.
797    ///
798    /// No status is ever refused, and that is a fact about this draft rather
799    /// than a check left out. This is the OBJECT_DATAGRAM of Section 9.3.1,
800    /// whose layout carries no Object Status field at all; a datagram that
801    /// states a status is the separate OBJECT_DATAGRAM_STATUS message, modelled
802    /// here as [`DatagramStatusHeader`]. So there is no status for
803    /// [`Self::encode`] to drop, and nothing for Section 9.2.1.1's "Any object
804    /// with a status code other than zero MUST have an empty payload" to rule
805    /// on: an object framed this way has status zero by construction.
806    ///
807    /// The extension block is a different matter. [`Self::encode`] writes the
808    /// framing without one, so a block this value holds has nowhere to go, and
809    /// dropping it silently is what puts a datagram on the wire describing
810    /// something other than what the caller built. That is refused here.
811    ///
812    /// The fallible signature is also what lets one entry point span every
813    /// draft. `dispatch::AnyDatagramHeader::encode` calls this on all thirteen,
814    /// and the drafts whose payload-bearing datagram *does* carry a status field
815    /// need somewhere to say no.
816    ///
817    /// # Errors
818    ///
819    /// [`CodecError::InvalidField`] if the value carries extension headers,
820    /// which this framing has no field for.
821    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
822        if !self.extensions.is_empty() {
823            return Err(CodecError::InvalidField);
824        }
825        self.encode(buf);
826        Ok(())
827    }
828
829    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
830        Self::decode_with_extensions(false, buf)
831    }
832
833    pub fn decode_with_extensions(
834        has_extensions: bool,
835        buf: &mut impl Buf,
836    ) -> Result<Self, CodecError> {
837        let track_alias = VarInt::decode(buf)?;
838        let group_id = VarInt::decode(buf)?;
839        let object_id = VarInt::decode(buf)?;
840        if buf.remaining() < 1 {
841            return Err(CodecError::UnexpectedEnd);
842        }
843        let publisher_priority = buf.get_u8();
844        let (extension_headers_length, extensions) = if has_extensions {
845            let ehl = VarInt::decode(buf)?;
846            // A datagram whose type says extensions are present must actually carry
847            // some: receiving one with an Extension Headers Length of 0 closes the
848            // session. The opposite holds on a subgroup stream, where the type byte is
849            // fixed for the whole stream and an object with no extensions has no other
850            // way to say so, which is why this check belongs to the datagram readers
851            // alone.
852            if ehl.into_inner() == 0 {
853                return Err(CodecError::InvalidField);
854            }
855            let ext = read_extension_bytes(buf, ehl.into_inner())?;
856            (ehl, ext)
857        } else {
858            (VarInt::from_usize(0), Vec::new())
859        };
860        Ok(Self {
861            track_alias,
862            group_id,
863            object_id,
864            publisher_priority,
865            extension_headers_length,
866            extensions,
867            end_of_group: false,
868        })
869    }
870}
871
872// ============================================================
873// Datagram Status (types 0x04, 0x05)
874// ============================================================
875
876#[derive(Debug, Clone, PartialEq, Eq)]
877pub struct DatagramStatusHeader {
878    pub track_alias: VarInt,
879    pub group_id: VarInt,
880    pub object_id: VarInt,
881    pub publisher_priority: u8,
882    pub extension_headers_length: VarInt,
883    pub extensions: Vec<u8>,
884    pub object_status: ObjectStatus,
885}
886
887impl DatagramStatusHeader {
888    pub fn encode(&self, buf: &mut impl BufMut) {
889        self.encode_with_extensions(false, buf);
890    }
891
892    /// Encode the status datagram header, refusing an extension block this
893    /// framing cannot carry.
894    ///
895    /// The same one-sided rule the payload-bearing header obeys, and the same
896    /// reason for it: [`Self::encode`] writes the framing without a block, so
897    /// bytes held here would be dropped rather than written.
898    ///
899    /// # Errors
900    ///
901    /// [`CodecError::InvalidField`] if the value carries extension headers.
902    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
903        if !self.extensions.is_empty() {
904            return Err(CodecError::InvalidField);
905        }
906        self.encode(buf);
907        Ok(())
908    }
909
910    pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
911        self.track_alias.encode(buf);
912        self.group_id.encode(buf);
913        self.object_id.encode(buf);
914        buf.put_u8(self.publisher_priority);
915        if has_extensions {
916            VarInt::from_usize(self.extensions.len()).encode(buf);
917            buf.put_slice(&self.extensions);
918        }
919        VarInt::from_usize(self.object_status as usize).encode(buf);
920    }
921
922    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
923        Self::decode_with_extensions(false, buf)
924    }
925
926    pub fn decode_with_extensions(
927        has_extensions: bool,
928        buf: &mut impl Buf,
929    ) -> Result<Self, CodecError> {
930        let track_alias = VarInt::decode(buf)?;
931        let group_id = VarInt::decode(buf)?;
932        let object_id = VarInt::decode(buf)?;
933        if buf.remaining() < 1 {
934            return Err(CodecError::UnexpectedEnd);
935        }
936        let publisher_priority = buf.get_u8();
937        let (extension_headers_length, extensions) = if has_extensions {
938            let ehl = VarInt::decode(buf)?;
939            // A datagram whose type says extensions are present must actually carry
940            // some: receiving one with an Extension Headers Length of 0 closes the
941            // session. The opposite holds on a subgroup stream, where the type byte is
942            // fixed for the whole stream and an object with no extensions has no other
943            // way to say so, which is why this check belongs to the datagram readers
944            // alone.
945            if ehl.into_inner() == 0 {
946                return Err(CodecError::InvalidField);
947            }
948            let ext = read_extension_bytes(buf, ehl.into_inner())?;
949            (ehl, ext)
950        } else {
951            (VarInt::from_usize(0), Vec::new())
952        };
953        let sv = VarInt::decode(buf)?.into_inner();
954        let object_status = ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?;
955        check_extensions_against_status(object_status, &extensions)?;
956        Ok(Self {
957            track_alias,
958            group_id,
959            object_id,
960            publisher_priority,
961            extension_headers_length,
962            extensions,
963            object_status,
964        })
965    }
966}
967
968// ============================================================
969// Datagram framing
970// ============================================================
971
972/// One datagram, of whichever shape its type field names.
973///
974/// A MoQT datagram opens with a variable-length integer naming its type, and
975/// that integer is what says which of the layouts above follows it, and whether an extension block sits inside it.
976/// Neither [`DatagramHeader`] nor [`DatagramStatusHeader`] reads or writes it, so neither can be handed
977/// the first byte of a datagram a peer sent, and neither produces bytes a peer
978/// can read. This is the entry point that does both.
979///
980/// The payload of a payload-bearing datagram runs to the end of the QUIC
981/// datagram, so it is not part of this value: [`Self::decode`] stops at the end
982/// of the header and leaves the payload in the buffer, and a caller appends the
983/// payload after [`Self::encode`].
984#[derive(Debug, Clone, PartialEq, Eq)]
985pub enum Datagram {
986    /// An object carrying a payload.
987    Payload(DatagramHeader),
988    /// An object stating a status, with no payload.
989    Status(DatagramStatusHeader),
990}
991
992impl Datagram {
993    /// Whether this datagram states an Object Status instead of carrying a
994    /// payload.
995    pub fn is_status(&self) -> bool {
996        matches!(self, Self::Status(_))
997    }
998
999    /// The type field this value writes.
1000    ///
1001    /// The extensions bit is taken from the extension bytes themselves rather
1002    /// than from the declared length beside them, which is what keeps the type
1003    /// and the body from contradicting each other: a datagram whose type
1004    /// announces extensions and then declares a length of 0 closes the session
1005    /// on receipt, and one that announces none has nowhere to put them. The end
1006    /// of group bit has no home in the body at all, so it comes from the header
1007    /// flag and goes nowhere else.
1008    pub fn datagram_type(&self) -> DatagramType {
1009        match self {
1010            Self::Payload(header) => match (header.end_of_group, header.extensions.is_empty()) {
1011                (false, true) => DatagramType::Datagram,
1012                (false, false) => DatagramType::DatagramExt,
1013                (true, true) => DatagramType::DatagramEog,
1014                (true, false) => DatagramType::DatagramEogExt,
1015            },
1016            Self::Status(header) => {
1017                if header.extensions.is_empty() {
1018                    DatagramType::DatagramStatus
1019                } else {
1020                    DatagramType::DatagramStatusExt
1021                }
1022            }
1023        }
1024    }
1025
1026    /// Decode a datagram from its first byte, type field included.
1027    ///
1028    /// Errors with [`CodecError::UnknownDatagramType`] when the datagram table
1029    /// does not assign the leading type, which this draft answers with a close.
1030    /// `datagram_type_error` settles it against that table alone — 0x05 is
1031    /// assigned in both tables here and means different things in each.
1032    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1033        let raw = VarInt::decode(buf)?.into_inner();
1034        let datagram_type = DatagramType::from_id(raw).ok_or_else(|| datagram_type_error(raw))?;
1035        let has_extensions = datagram_type.has_extensions();
1036        if datagram_type.is_status() {
1037            return Ok(Self::Status(DatagramStatusHeader::decode_with_extensions(
1038                has_extensions,
1039                buf,
1040            )?));
1041        }
1042        let mut header = DatagramHeader::decode_with_extensions(has_extensions, buf)?;
1043        header.end_of_group = datagram_type.is_end_of_group();
1044        Ok(Self::Payload(header))
1045    }
1046
1047    /// Decode one whole datagram: [`Self::decode`], then the payload that runs
1048    /// to the end of `buf`, which must hold exactly one transport datagram.
1049    ///
1050    /// Unlike [`Self::decode`], this refuses an extension block that does not
1051    /// parse, as [`crate::object_properties::parse_object_properties`] does,
1052    /// and, with [`CodecError::MalformedTrack`], a Prior Group ID Gap larger
1053    /// than the datagram's Group ID, through [`PriorGaps::from_extensions`].
1054    /// draft-12 Section 9.3 carries the Group ID and the extension headers in
1055    /// one datagram, so this half of the draft-12 Section 10.1 rule shows in it.
1056    ///
1057    /// It also refuses bytes after a status datagram's Object Status, with
1058    /// [`CodecError::PayloadNotPermitted`]. draft-12 Section 9.3.2: "An
1059    /// OBJECT_DATAGRAM_STATUS is similar to OBJECT_DATAGRAM except it conveys
1060    /// an Object Status and has no payload."
1061    pub fn decode_object(buf: &mut impl Buf) -> Result<(Self, Vec<u8>), CodecError> {
1062        let datagram = Self::decode(buf)?;
1063        let (group_id, extensions) = match &datagram {
1064            Self::Payload(header) => (header.group_id.into_inner(), &header.extensions),
1065            Self::Status(header) => (header.group_id.into_inner(), &header.extensions),
1066        };
1067        PriorGaps::from_extensions(extensions, group_id)?;
1068        let payload = read_bytes(buf, buf.remaining())?;
1069        if let Self::Status(header) = &datagram {
1070            if !payload.is_empty() {
1071                return Err(CodecError::PayloadNotPermitted {
1072                    status: header.object_status.as_u64(),
1073                    len: payload.len(),
1074                    detail: "its type states a status in place of a payload",
1075                });
1076            }
1077        }
1078        Ok((datagram, payload))
1079    }
1080
1081    /// Encode the datagram, type field included.
1082    pub fn encode(&self, buf: &mut impl BufMut) {
1083        let datagram_type = self.datagram_type();
1084        let has_extensions = datagram_type.has_extensions();
1085        VarInt::from_usize(datagram_type as usize).encode(buf);
1086        match self {
1087            Self::Payload(header) => header.encode_with_extensions(has_extensions, buf),
1088            Self::Status(header) => header.encode_with_extensions(has_extensions, buf),
1089        }
1090    }
1091
1092    /// Encode the datagram, refusing a header the framing it names cannot
1093    /// carry.
1094    ///
1095    /// The body is built before anything reaches `buf`, so a refused datagram
1096    /// leaves `buf` untouched rather than a type field with no body under it.
1097    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1098        let mut body = Vec::with_capacity(64);
1099        let datagram_type = self.datagram_type();
1100        let has_extensions = datagram_type.has_extensions();
1101        match self {
1102            Self::Payload(header) => header.encode_with_extensions(has_extensions, &mut body),
1103            Self::Status(header) => {
1104                check_extensions_against_status(header.object_status, &header.extensions)?;
1105                header.encode_with_extensions(has_extensions, &mut body);
1106            }
1107        }
1108        VarInt::from_usize(datagram_type as usize).encode(buf);
1109        buf.put_slice(&body);
1110        Ok(())
1111    }
1112}
1113
1114// ============================================================
1115// Fetch stream (type 0x05)
1116// ============================================================
1117
1118#[derive(Debug, Clone, PartialEq, Eq)]
1119pub struct FetchHeader {
1120    pub request_id: VarInt,
1121}
1122
1123#[derive(Debug, Clone, PartialEq, Eq)]
1124pub struct FetchObjectHeader {
1125    pub group_id: VarInt,
1126    pub subgroup_id: VarInt,
1127    pub object_id: VarInt,
1128    pub publisher_priority: u8,
1129    pub extension_headers_length: VarInt,
1130    pub extensions: Vec<u8>,
1131    pub payload_length: VarInt,
1132    pub object_status: ObjectStatus,
1133}
1134
1135impl FetchHeader {
1136    /// Encode a fetch stream header including its leading stream-type field,
1137    /// so the bytes form the start of a data stream a peer can read.
1138    ///
1139    /// [`Self::encode`] writes the body alone, which is what a caller wants
1140    /// once the stream is already open and what a caller must not use for its
1141    /// first write. The read side has had [`Self::decode_stream`] all along,
1142    /// so without this the codec could not round-trip its own fetch stream
1143    /// through its own reader.
1144    pub fn encode_stream(&self, buf: &mut impl BufMut) {
1145        VarInt::from_usize(StreamType::Fetch as usize).encode(buf);
1146        self.encode(buf);
1147    }
1148
1149    pub fn encode(&self, buf: &mut impl BufMut) {
1150        self.request_id.encode(buf);
1151    }
1152
1153    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1154        let request_id = VarInt::decode(buf)?;
1155        Ok(Self { request_id })
1156    }
1157
1158    /// Decode a fetch header from the start of a data stream, consuming the
1159    /// leading stream type varint.
1160    ///
1161    /// Errors with [`CodecError::UnknownStreamType`] when the stream table does
1162    /// not assign the leading type, which this draft answers with a close, and
1163    /// with [`CodecError::InvalidField`] when it is assigned but is not
1164    /// [`StreamType::Fetch`]. `stream_type_error` draws that line.
1165    pub fn decode_stream(buf: &mut impl Buf) -> Result<Self, CodecError> {
1166        let stream_type = VarInt::decode(buf)?.into_inner();
1167        if stream_type != StreamType::Fetch as u64 {
1168            return Err(stream_type_error(stream_type));
1169        }
1170        Self::decode(buf)
1171    }
1172}
1173
1174impl FetchObjectHeader {
1175    pub fn encode(&self, buf: &mut impl BufMut) {
1176        self.group_id.encode(buf);
1177        self.subgroup_id.encode(buf);
1178        self.object_id.encode(buf);
1179        buf.put_u8(self.publisher_priority);
1180        VarInt::from_usize(self.extensions.len()).encode(buf);
1181        buf.put_slice(&self.extensions);
1182        self.payload_length.encode(buf);
1183        if self.payload_length.into_inner() == 0 {
1184            VarInt::from_usize(self.object_status as usize).encode(buf);
1185        }
1186    }
1187
1188    /// Encode the header, refusing a status the framing cannot carry.
1189    ///
1190    /// Section 9.4.4 puts the Object Status field on the wire only when the
1191    /// Object Payload Length is zero, and Section 9.2.1.1 says "Any object
1192    /// with a status code other than zero MUST have an empty payload". A
1193    /// non-zero status paired with a non-zero payload length therefore has no
1194    /// encoding at all: [`Self::encode`] drops the status and the peer reads an
1195    /// ordinary object, which is a different object from the one the caller
1196    /// described. This refuses instead.
1197    ///
1198    /// The datagram types on this draft already refuse the same pairing. These
1199    /// two did not, and they are the ones a publisher writes on every stream.
1200    ///
1201    /// # Errors
1202    ///
1203    /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
1204    /// a non-zero Object Payload Length.
1205    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1206        if self.payload_length.into_inner() != 0 && self.object_status as usize != 0 {
1207            return Err(CodecError::InvalidField);
1208        }
1209        self.encode(buf);
1210        Ok(())
1211    }
1212
1213    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1214        let group_id = VarInt::decode(buf)?;
1215        let subgroup_id = VarInt::decode(buf)?;
1216        let object_id = VarInt::decode(buf)?;
1217        if buf.remaining() < 1 {
1218            return Err(CodecError::UnexpectedEnd);
1219        }
1220        let publisher_priority = buf.get_u8();
1221        let extension_headers_length = VarInt::decode(buf)?;
1222        let extensions = read_extension_bytes(buf, extension_headers_length.into_inner())?;
1223        let payload_length = VarInt::decode(buf)?;
1224        let object_status = if payload_length.into_inner() == 0 {
1225            let sv = VarInt::decode(buf)?.into_inner();
1226            ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?
1227        } else {
1228            ObjectStatus::Normal
1229        };
1230        check_extensions_against_status(object_status, &extensions)?;
1231        Ok(Self {
1232            group_id,
1233            subgroup_id,
1234            object_id,
1235            publisher_priority,
1236            extension_headers_length,
1237            extensions,
1238            payload_length,
1239            object_status,
1240        })
1241    }
1242}
1243
1244#[cfg(test)]
1245mod tests {
1246    use super::*;
1247
1248    /// A payload datagram for track 1, group 0, object 7, priority 128, with
1249    /// no extension block — the shape [`DatagramHeader::encode`] writes.
1250    fn payload_datagram() -> DatagramHeader {
1251        DatagramHeader {
1252            track_alias: VarInt::from_usize(1),
1253            group_id: VarInt::from_usize(0),
1254            object_id: VarInt::from_usize(7),
1255            publisher_priority: 128,
1256            extension_headers_length: VarInt::from_usize(0),
1257            extensions: Vec::new(),
1258            end_of_group: false,
1259        }
1260    }
1261
1262    /// The same datagram on the message that does carry a status.
1263    fn status_datagram(status: ObjectStatus) -> DatagramStatusHeader {
1264        DatagramStatusHeader {
1265            track_alias: VarInt::from_usize(1),
1266            group_id: VarInt::from_usize(0),
1267            object_id: VarInt::from_usize(7),
1268            publisher_priority: 128,
1269            extension_headers_length: VarInt::from_usize(0),
1270            extensions: Vec::new(),
1271            object_status: status,
1272        }
1273    }
1274
1275    /// Draft-12's payload datagram has no status for the encoder to drop, and
1276    /// so nothing for the fallible encode to refuse.
1277    ///
1278    /// [`DatagramHeader`] is the OBJECT_DATAGRAM of Section 9.3.1, whose layout
1279    /// carries no Object Status field at all. A datagram that states a status
1280    /// is the separate OBJECT_DATAGRAM_STATUS message of Section 9.3.2, modelled
1281    /// here as [`DatagramStatusHeader`]. The refusal drafts 07, 08 and 14-19
1282    /// need on this path therefore has nothing to bite on, and this gate holds
1283    /// [`DatagramHeader::encode_checked`] to writing exactly what
1284    /// [`DatagramHeader::encode`] writes and never refusing — the alternative
1285    /// being a codec that answers `Err` for a datagram every draft-12 publisher
1286    /// is entitled to send.
1287    ///
1288    /// The second half is what makes the first half safe rather than merely
1289    /// permissive. Every status draft-12 assigns travels intact on the message
1290    /// that can express one, so leaving the payload datagram unchecked loses
1291    /// nothing; if a status could ride this header, the check being skipped
1292    /// here would be the check drafts 07 and 18 need.
1293    #[test]
1294    fn a_payload_datagram_has_no_status_to_refuse() {
1295        let header = payload_datagram();
1296
1297        let mut checked = Vec::new();
1298        header.encode_checked(&mut checked).unwrap_or_else(|e| {
1299            panic!("draft-12's payload datagram has no status to refuse: {e:?}")
1300        });
1301
1302        let mut plain = Vec::new();
1303        header.encode(&mut plain);
1304        assert_eq!(checked, plain, "the fallible encode must write exactly what `encode` writes");
1305
1306        let decoded = DatagramHeader::decode(&mut &checked[..])
1307            .expect("the bytes encode_checked wrote must parse back");
1308        assert_eq!(decoded, header, "the payload datagram did not survive its own encoding");
1309
1310        for &status in ObjectStatus::ALL {
1311            let mut buf = Vec::new();
1312            status_datagram(status).encode(&mut buf);
1313            let decoded = DatagramStatusHeader::decode(&mut &buf[..])
1314                .unwrap_or_else(|e| panic!("the status datagram for {status:?} must parse: {e:?}"));
1315            assert_eq!(
1316                decoded.object_status, status,
1317                "{status:?} must survive on the message that carries a status"
1318            );
1319        }
1320    }
1321}
1322
1323#[cfg(test)]
1324mod prior_gap_tests {
1325    use super::*;
1326
1327    fn header(group: usize) -> SubgroupHeader {
1328        SubgroupHeader {
1329            stream_type: StreamType::SubgroupZeroExt,
1330            track_alias: VarInt::from_usize(1),
1331            group_id: VarInt::from_usize(group),
1332            subgroup_id: VarInt::from_usize(0),
1333            publisher_priority: 0x80,
1334        }
1335    }
1336
1337    fn gaps_block(gaps: &[usize]) -> Vec<u8> {
1338        let mut block = Vec::new();
1339        for &gap in gaps {
1340            VarInt::from_usize(PRIOR_GROUP_ID_GAP as usize).encode(&mut block);
1341            VarInt::from_usize(gap).encode(&mut block);
1342        }
1343        block
1344    }
1345
1346    fn object(id: usize, gaps: &[usize]) -> ObjectHeader {
1347        let extensions = gaps_block(gaps);
1348        ObjectHeader {
1349            object_id: VarInt::from_usize(id),
1350            extension_headers_length: VarInt::from_usize(extensions.len()),
1351            extensions,
1352            payload_length: VarInt::from_usize(1),
1353            object_status: ObjectStatus::Normal,
1354        }
1355    }
1356
1357    fn is_malformed_track<T>(result: Result<T, CodecError>) -> bool {
1358        matches!(result, Err(CodecError::MalformedTrack { .. }))
1359    }
1360
1361    #[test]
1362    fn gap_larger_than_the_group_id_is_refused() {
1363        assert!(is_malformed_track(SubgroupGaps::new(&header(1)).check(&object(0, &[2]))));
1364        let gaps = SubgroupGaps::new(&header(2)).check(&object(0, &[2])).unwrap();
1365        assert_eq!(gaps.group, vec![2]);
1366    }
1367
1368    #[test]
1369    fn objects_with_different_gaps_are_refused() {
1370        let mut gaps = SubgroupGaps::new(&header(5));
1371        gaps.check(&object(0, &[2])).unwrap();
1372        gaps.check(&object(1, &[])).unwrap();
1373        gaps.check(&object(2, &[2])).unwrap();
1374        assert!(is_malformed_track(gaps.check(&object(3, &[3]))));
1375    }
1376
1377    #[test]
1378    fn values_within_one_object_are_not_compared() {
1379        let mut gaps = SubgroupGaps::new(&header(5));
1380        assert_eq!(gaps.check(&object(0, &[2, 3])).unwrap().group, vec![2, 3]);
1381        assert_eq!(gaps.check(&object(1, &[3, 2, 3])).unwrap().group, vec![3, 2, 3]);
1382    }
1383
1384    #[test]
1385    fn objects_with_a_common_value_are_accepted() {
1386        let mut gaps = SubgroupGaps::new(&header(5));
1387        gaps.check(&object(0, &[2, 3])).unwrap();
1388        gaps.check(&object(1, &[3, 4])).unwrap();
1389        gaps.check(&object(2, &[])).unwrap();
1390        gaps.check(&object(3, &[3])).unwrap();
1391    }
1392
1393    #[test]
1394    fn objects_that_share_no_value_are_refused() {
1395        let mut gaps = SubgroupGaps::new(&header(5));
1396        gaps.check(&object(0, &[2, 3])).unwrap();
1397        assert!(is_malformed_track(gaps.check(&object(1, &[4, 5]))));
1398        gaps.check(&object(2, &[3, 4])).unwrap();
1399        assert!(is_malformed_track(gaps.check(&object(3, &[4]))));
1400    }
1401
1402    #[test]
1403    fn objects_with_no_value_common_to_all_are_refused() {
1404        let mut gaps = SubgroupGaps::new(&header(5));
1405        gaps.check(&object(0, &[1, 2])).unwrap();
1406        gaps.check(&object(1, &[2, 3])).unwrap();
1407        assert!(is_malformed_track(gaps.check(&object(2, &[1, 3]))));
1408    }
1409
1410    #[test]
1411    fn a_gap_larger_than_the_group_id_under_every_reading_is_refused() {
1412        assert!(is_malformed_track(SubgroupGaps::new(&header(5)).check(&object(0, &[7, 6]))));
1413        SubgroupGaps::new(&header(5)).check(&object(0, &[2, 6])).unwrap();
1414    }
1415
1416    #[test]
1417    fn a_refused_object_is_not_recorded() {
1418        let mut gaps = SubgroupGaps::new(&header(5));
1419        assert!(is_malformed_track(gaps.check(&object(0, &[6]))));
1420        gaps.check(&object(1, &[3])).unwrap();
1421    }
1422
1423    fn datagram(group: usize, gaps: &[usize]) -> Vec<u8> {
1424        let extensions = gaps_block(gaps);
1425        let mut bytes = Vec::new();
1426        Datagram::Payload(DatagramHeader {
1427            track_alias: VarInt::from_usize(1),
1428            group_id: VarInt::from_usize(group),
1429            object_id: VarInt::from_usize(0),
1430            publisher_priority: 0x80,
1431            extension_headers_length: VarInt::from_usize(extensions.len()),
1432            extensions,
1433            end_of_group: false,
1434        })
1435        .encode(&mut bytes);
1436        bytes.push(0xAA);
1437        bytes
1438    }
1439
1440    #[test]
1441    fn decode_object_refuses_a_gap_larger_than_the_datagram_group_id() {
1442        let wire = datagram(1, &[2]);
1443        assert!(is_malformed_track(Datagram::decode_object(&mut &wire[..])));
1444        assert!(Datagram::decode(&mut &wire[..]).is_ok());
1445        let wire = datagram(2, &[2]);
1446        let (decoded, payload) = Datagram::decode_object(&mut &wire[..]).unwrap();
1447        assert_eq!(decoded.datagram_type(), DatagramType::DatagramExt);
1448        assert_eq!(payload, vec![0xAA]);
1449    }
1450
1451    #[test]
1452    fn decode_object_refuses_a_payload_after_a_status_datagram() {
1453        let mut wire = Vec::new();
1454        Datagram::Status(DatagramStatusHeader {
1455            track_alias: VarInt::from_usize(1),
1456            group_id: VarInt::from_usize(0),
1457            object_id: VarInt::from_usize(0),
1458            publisher_priority: 0x80,
1459            extension_headers_length: VarInt::from_usize(0),
1460            extensions: Vec::new(),
1461            object_status: ObjectStatus::Normal,
1462        })
1463        .encode(&mut wire);
1464        let (_, payload) = Datagram::decode_object(&mut &wire[..]).unwrap();
1465        assert!(payload.is_empty());
1466        wire.push(0xAA);
1467        assert!(matches!(
1468            Datagram::decode_object(&mut &wire[..]),
1469            Err(CodecError::PayloadNotPermitted { len: 1, .. })
1470        ));
1471    }
1472}