Skip to main content

moqtap_codec/draft15/
data_stream.rs

1//! Draft-15 data stream header encoding and decoding.
2//!
3//! Draft-15 data streams differ significantly from draft-14:
4//! - Subgroup stream types encode flags in the type byte. Section 10.4.2
5//!   Table 6 assigns twenty-four: `0x10`-`0x15`, `0x18`-`0x1D`, `0x30`-`0x35`
6//!   and `0x38`-`0x3D`.
7//! - Priority is absent when `type & 0x20`, and the object then inherits the
8//!   priority the subscription established
9//! - `type & 0x06` decides how the Subgroup ID is carried: it is zero, it is
10//!   the first object's ID, or it is a field on the wire. Table 6 states that
11//!   as two columns — Subgroup ID Field Present, and Subgroup ID Value — not
12//!   as a mode field, and the fourth combination is simply not a row in it
13//! - `type & 0x08` marks a stream whose last object ends its group
14//! - Extensions flag (`type & 0x01`) affects per-object parsing
15//! - Fetch objects use serialization_flags for delta encoding
16//! - Object IDs in subgroups use delta encoding (first=absolute, subsequent=delta+1)
17
18use super::types::ObjectStatus;
19use crate::error::CodecError;
20use crate::object_properties::{ObjectProperty, PropertyValue};
21use crate::varint::VarInt;
22use bytes::{Buf, BufMut};
23
24/// Advance `buf` past `len` bytes without copying them.
25fn skip(buf: &mut impl Buf, len: u64) -> Result<(), CodecError> {
26    let len = usize::try_from(len).map_err(|_| CodecError::UnexpectedEnd)?;
27    if buf.remaining() < len {
28        return Err(CodecError::UnexpectedEnd);
29    }
30    buf.advance(len);
31    Ok(())
32}
33
34/// Turn a wire Object Status code into the status draft-15 gives it, refusing
35/// any code the draft does not assign.
36///
37/// Draft-15 Section 10.2.1.1 lists the codes an object may carry and says any
38/// other value SHOULD be treated as a protocol error and the session
39/// terminated with a PROTOCOL_VIOLATION. Every place this module reads a
40/// status runs the wire code through here.
41///
42/// [`SubgroupObject::object_status`] and [`DatagramHeader::object_status`]
43/// then store the [`ObjectStatus`] this returns rather than the raw code, so
44/// the refusal is not something a future decode site can forget: those fields
45/// cannot hold an unassigned value at all, in either direction, and the encode
46/// paths need no check of their own.
47/// [`SubgroupObjectMeta::status`] deliberately keeps the raw code — it is a
48/// decode-only view that never feeds an encoder — but it is filtered through
49/// here too, so the two readers agree byte for byte on what parses.
50fn decoded_status(code: u64) -> Result<ObjectStatus, CodecError> {
51    ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)
52}
53
54/// Refuse an object that would be written with extension headers beside a
55/// status other than Normal.
56///
57/// Draft-15 Section 10.2.1.2: "Any Object with status Normal can have extension
58/// headers. If an endpoint receives extension headers on Objects with status
59/// that is not Normal, it MUST close the session with a PROTOCOL_VIOLATION."
60///
61/// Wider than the rule drafts 11 through 14 carry. There, at
62/// draft-14 Section 10.2.1.2, the sentence read "Any Object may have extension
63/// headers except those with Object Status 'Object Does Not Exist'", which left
64/// extensions beside End of Group and End of Track legal; draft-15 is where the
65/// single exception became the general case. Using the earlier wording here
66/// would let an End of Group object carry metadata draft-15 says to close the
67/// session over.
68///
69/// **Called from [`DatagramHeader::encode_checked`] and nowhere else, and
70/// deliberately so.** A frame carrying extensions beside a non-Normal status is
71/// well formed — every length is honest and every field parses — so a decoder
72/// can hand it back intact, and a tool that reproduces a capture has to.
73/// Refusing it on decode would make a captured violation unreadable, which
74/// loses the one artifact anybody debugging it needs.
75///
76/// The same argument bars it from a carrier's only writer.
77/// [`SubgroupObjectReader::write_object`] and
78/// [`FetchObjectReader::write_object_header`] are the sole way to write their
79/// objects, so a refusal there would leave a captured violation impossible to
80/// re-emit; the corpus ships exactly such a subgroup frame. A datagram is the
81/// one carrier with two encoders, so `encode_checked` can refuse while
82/// [`DatagramHeader::encode`] still reproduces bytes verbatim. That is the
83/// whole of the rule: opt-in strictness where an unchecked path exists, and a
84/// predicate — [`SubgroupObject::extensions_permitted`],
85/// [`SubgroupObjectMeta::extensions_permitted`],
86/// [`DatagramHeader::extensions_permitted`],
87/// [`FetchObjectHeader::extensions_permitted`] — everywhere else.
88///
89/// That is why drafts 11 through 14 look different and should stay different:
90/// they state only the narrow Object Does Not Exist form, they apply it on both
91/// sides, and no vector exercises it. Drafts 15 through 19 state the broad form
92/// and enforce it on encode alone. The split is intentional, not an
93/// inconsistency to harmonise away.
94///
95/// `status` is the code the object resolves to, not the field as it appears on
96/// the wire. An object whose framing omits the status field has status Normal —
97/// on a subgroup or fetch stream because its payload length is non-zero, on a
98/// datagram because its type byte leaves the status bit clear — and such an
99/// object may carry extensions. Passing `None` says exactly that.
100fn check_extensions_against_status_on_encode(
101    status: Option<u64>,
102    extension_headers_len: u64,
103) -> Result<(), CodecError> {
104    if extension_headers_len != 0
105        && matches!(status, Some(code) if code != ObjectStatus::Normal.as_u64())
106    {
107        return Err(CodecError::InvalidField);
108    }
109    Ok(())
110}
111
112// ── Payload permission ─────────────────────────────────────
113
114/// Whether an object carrying a given status may hold a non-empty payload.
115///
116/// Draft-15 Section 10.2.1.1 states the rule in one sentence — "Any object with
117/// a status code other than zero MUST have an empty payload" — and Section
118/// 10.2.1 says the same from the other side, listing the Object Payload as
119/// "Only present when 'Object Status' is Normal (0x0)". On this draft the
120/// answer therefore falls out of the code being zero or not, and every status
121/// but [`ObjectStatus::Normal`] forbids a payload.
122///
123/// It is worth a type all the same, because that arithmetic is not something a
124/// consumer can safely perform on a raw wire code. A code draft-15 does not
125/// assign is not *non-zero, and therefore forbidden*: it is a code with no
126/// meaning at all, and the draft's answer to it is to terminate the session,
127/// not to infer a payload rule. Handing back a `PayloadPermission` keeps the
128/// two apart, and lets a caller ask the question without restating the rule —
129/// or, worse, restating it slightly differently.
130#[derive(Debug, Clone, Copy, PartialEq, Eq)]
131pub enum PayloadPermission {
132    /// The status permits a payload but does not require one: a zero-length
133    /// object with such a status is well formed, and draft-15's encodings have
134    /// a way to spell it.
135    Permitted,
136    /// An object with such a status has an empty payload, and one carrying
137    /// bytes is malformed.
138    Forbidden,
139}
140
141impl PayloadPermission {
142    /// The permission draft-15 gives objects carrying `status`.
143    ///
144    /// Written as a match over every assigned status rather than as a test for
145    /// zero, so that a status added to [`ObjectStatus`] later cannot quietly
146    /// inherit *not Normal, therefore forbidden* — it stops the crate compiling
147    /// until its own answer is written down. Drafts after 15 moved this rule
148    /// into a column of the Object Status registry, where a future status may
149    /// well permit a payload; the shape here does not have to change when a
150    /// caller crosses that boundary.
151    pub fn for_status(status: ObjectStatus) -> Self {
152        match status {
153            ObjectStatus::Normal => PayloadPermission::Permitted,
154            ObjectStatus::ObjectDoesNotExist => PayloadPermission::Forbidden,
155            ObjectStatus::EndOfGroup => PayloadPermission::Forbidden,
156            ObjectStatus::EndOfTrack => PayloadPermission::Forbidden,
157        }
158    }
159
160    /// `true` for [`PayloadPermission::Permitted`].
161    ///
162    /// The permission answers on its own, with no payload length in hand,
163    /// which is the point of asking the status rather than the framing.
164    pub fn permits(self) -> bool {
165        matches!(self, PayloadPermission::Permitted)
166    }
167}
168
169// ── Subgroup streams ───────────────────────────────────────
170
171/// Subgroup stream header for draft-15.
172///
173/// Draft-15 Section 10.4.2 Table 6 assigns twenty-four stream types, built from
174/// a base of `0x10` or `0x30` and three independent fields:
175/// - `& 0x01`: extensions present on objects
176/// - `& 0x06`: how the Subgroup ID is carried — `0x00` it is zero, `0x02` it
177///   is the Object ID of the first object on the stream and is not
178///   transmitted, `0x04` it is a field on the wire. `0x06` is not a row
179/// - `& 0x08`: the last object on the stream ends its group
180/// - `& 0x20`: no publisher priority; the object inherits the one the
181///   subscription established
182///
183/// **These two bits are read together, not as two independent flags**, and the
184/// table is what settles it: it gives them as a pair of columns, Subgroup ID
185/// Field Present and Subgroup ID Value, and no row carries both a present
186/// field and a value taken from the first object. Reading `0x02` as an
187/// end-of-group marker — which is what this did — answers the wrong question
188/// in both directions: a real end-of-group stream (`0x18`-`0x1D`,
189/// `0x38`-`0x3D`) reports `false`, and a first-object stream (`0x12`, `0x13`)
190/// reports `true`. Neither is a framing error, so nothing downstream notices.
191///
192/// Draft-16 later folds the same three choices into a named SUBGROUP_ID_MODE
193/// field with a fourth, reserved value. Draft-15 has no such name and no such
194/// value: `0x16`, `0x17`, `0x1E`, `0x1F`, `0x36`, `0x37`, `0x3E` and `0x3F`
195/// are absent from Table 6 rather than reserved by it, and the word does not
196/// appear here in that sense at all. The two drafts agree on every byte and
197/// differ only in how they say why, so borrowing the later vocabulary reads
198/// as though draft-15 states something it does not.
199#[derive(Debug, Clone, PartialEq, Eq)]
200pub struct SubgroupHeader {
201    pub header_type: u8,
202    pub track_alias: VarInt,
203    pub group_id: VarInt,
204    pub subgroup_id: VarInt,
205    pub publisher_priority: Option<u8>,
206}
207
208/// Whether draft-15 Section 10.4.2 Table 6 assigns this subgroup stream type.
209///
210/// Twenty-four values are assigned, counted off the table itself: `0x10`-`0x15`
211/// and `0x18`-`0x1D`, then `0x30`-`0x35` and `0x38`-`0x3D`. A type is valid
212/// when its base is `0x10` or `0x30` and its `0x06` bits name one of the three
213/// ways the table carries a Subgroup ID. That leaves `0x16`, `0x17`, `0x1E`,
214/// `0x1F`, `0x36`, `0x37`, `0x3E` and `0x3F` with no row.
215///
216/// They are **unassigned, not reserved.** Draft-16 reserves the same eight by
217/// name; draft-15 reaches them by omission and says nothing about them at all.
218/// The test is the same either way, which is why the distinction is only worth
219/// a sentence — but the sentence keeps the next reader from carrying draft-16's
220/// vocabulary back into a draft that does not use it.
221///
222/// Section 10 requires closing the session on a stream type the draft does not
223/// define, so accepting one and inventing a framing for it — the decoder read a
224/// Subgroup ID field for these, because `0x04` happens to be set — is not a
225/// harmless leniency.
226fn subgroup_type_is_assigned(ty: u64) -> bool {
227    let Ok(ty) = u8::try_from(ty) else {
228        return false;
229    };
230    // The `0x20` bit is the one difference between the two assigned bases, and
231    // masking with `0xD0` rather than `0xF0` drops it — so `0x30` through `0x3F`
232    // fold onto `0x10` and one comparison covers both. Every other high nibble
233    // survives the mask as something other than `0x10` and is refused: `0x20`
234    // folds to `0x00`, `0x50` and `0x70` to `0x50`, and so on up.
235    (ty & 0xD0) == 0x10 && (ty & 0x06) != 0x06
236}
237
238impl SubgroupHeader {
239    pub fn has_extensions(&self) -> bool {
240        self.header_type & 0x01 != 0
241    }
242
243    /// When set, the Subgroup ID is the Object ID of the first object on the
244    /// stream and is not transmitted. The `0x02` row of the `0x06` bits.
245    pub fn subgroup_id_from_first_object(&self) -> bool {
246        self.header_type & 0x06 == 0x02
247    }
248
249    pub fn has_explicit_subgroup_id(&self) -> bool {
250        self.header_type & 0x06 == 0x04
251    }
252
253    pub fn has_end_of_group(&self) -> bool {
254        self.header_type & 0x08 != 0
255    }
256
257    pub fn has_priority(&self) -> bool {
258        self.header_type & 0x20 == 0
259    }
260
261    /// Encode the header, writing whichever fields the type byte announces.
262    ///
263    /// Field presence follows `header_type`, because that is what the peer
264    /// reads. A `publisher_priority` of `None` under a type whose `0x20` bit is
265    /// clear writes a zero rather than dropping the byte: omitting it would
266    /// leave the peer reading the first object's Object ID Delta as a priority
267    /// and desync the whole stream. Use [`encode_checked`](Self::encode_checked)
268    /// to be told about the disagreement instead of having it papered over.
269    pub fn encode(&self, buf: &mut impl BufMut) {
270        VarInt::from_usize(self.header_type as usize).encode(buf);
271        self.track_alias.encode(buf);
272        self.group_id.encode(buf);
273        if self.has_explicit_subgroup_id() {
274            self.subgroup_id.encode(buf);
275        }
276        if self.has_priority() {
277            buf.put_u8(self.publisher_priority.unwrap_or(0));
278        }
279    }
280
281    /// Encode, refusing a header whose fields disagree with its own type byte.
282    ///
283    /// [`encode`](Self::encode) is driven by `header_type` and
284    /// [`decode`](Self::decode) reads the same byte, so the two agree on the
285    /// wire whatever the struct holds —
286    /// but a caller that sets `publisher_priority` beside a type whose `0x20`
287    /// bit is set, or leaves it `None` beside one whose bit is clear, has built
288    /// a header that does not mean what it says. The fetch object writer
289    /// already refuses that shape; this is the same check one layer up.
290    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
291        if self.has_priority() != self.publisher_priority.is_some() {
292            return Err(CodecError::InvalidField);
293        }
294        if !subgroup_type_is_assigned(self.header_type as u64) {
295            return Err(CodecError::InvalidField);
296        }
297        self.encode(buf);
298        Ok(())
299    }
300
301    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
302        // Compared as the full varint rather than truncated to `u8`: a type of
303        // `0x110` narrows to `0x10` and would be accepted as a plain subgroup
304        // header, so a stream this draft does not define would be parsed as one
305        // it does instead of closing the session.
306        let type_value = VarInt::decode(buf)?.into_inner();
307        if !subgroup_type_is_assigned(type_value) {
308            return Err(stream_type_error(type_value));
309        }
310        let header_type = type_value as u8;
311        let track_alias = VarInt::decode(buf)?;
312        let group_id = VarInt::decode(buf)?;
313        // `0x04` alone carries the Subgroup ID on the wire. `0x02` takes it
314        // from the first object, which the stream reader resolves, and `0x00`
315        // defines it as zero.
316        let subgroup_id = if type_value as u8 & 0x06 == 0x04 {
317            VarInt::decode(buf)?
318        } else {
319            VarInt::from_usize(0)
320        };
321        let publisher_priority = if header_type & 0x20 == 0 {
322            if buf.remaining() < 1 {
323                return Err(CodecError::UnexpectedEnd);
324            }
325            Some(buf.get_u8())
326        } else {
327            None
328        };
329        Ok(Self { header_type, track_alias, group_id, subgroup_id, publisher_priority })
330    }
331}
332
333// ── Subgroup objects (stateful) ─────────────────────────────
334
335/// One object within a draft-15 subgroup stream with its Object ID
336/// already resolved from the delta encoding.
337///
338/// Draft-15 object framing requires context from the enclosing
339/// [`SubgroupHeader`] (specifically, whether extension headers are
340/// present and the running delta state), so decoding/encoding uses a
341/// stateful [`SubgroupObjectReader`] rather than a standalone method.
342#[derive(Debug, Clone, PartialEq, Eq)]
343pub struct SubgroupObject {
344    /// Resolved absolute Object ID.
345    pub object_id: VarInt,
346    /// Raw extension-header bytes, excluding the byte-length prefix that
347    /// precedes them on the wire. Empty when the stream header does not
348    /// set the extensions-present bit, or when the block is present but
349    /// zero-length. Opaque: [`SubgroupObjectReader::write_object`] re-emits
350    /// the prefix and these bytes verbatim.
351    pub extension_headers: Vec<u8>,
352    /// Payload length as encoded on the wire. Zero when the object is
353    /// a status-only object.
354    pub payload_length: VarInt,
355    /// Object status; `Some` when `payload_length == 0`.
356    ///
357    /// The wire field is a varint, so it can carry any value up to 2^62-1;
358    /// draft-15 Section 10.2.1.1 assigns four of them and says a peer SHOULD
359    /// treat the rest as a protocol error. This field is typed to the assigned
360    /// set, so it refuses to hold the codes the draft leaves unassigned —
361    /// 0x2, and everything from 0x5 up. That makes the refusal a property of
362    /// the struct rather than of any one code path: an encoder cannot be
363    /// handed a status the draft does not define, and does not have to check.
364    ///
365    /// `None` on a zero-length object means the same thing as
366    /// [`ObjectStatus::Normal`] and encodes as it; the wire field is not
367    /// optional once `payload_length` is zero.
368    pub object_status: Option<ObjectStatus>,
369    /// Payload bytes; empty when `object_status` is `Some`.
370    pub payload: Vec<u8>,
371}
372
373impl SubgroupObject {
374    /// The status this object resolves to.
375    ///
376    /// The wire carries a status field only on a zero-length object, so an
377    /// object holding bytes is [`ObjectStatus::Normal`] whatever
378    /// [`Self::object_status`] says — draft-15 Section 10.2.1.1: "This status is
379    /// implicit for any non-zero length object."
380    pub fn status(&self) -> ObjectStatus {
381        if self.payload_length.into_inner() == 0 {
382            self.object_status.unwrap_or(ObjectStatus::Normal)
383        } else {
384            ObjectStatus::Normal
385        }
386    }
387
388    /// Whether this object's status is allowed to carry the extension headers
389    /// it has.
390    ///
391    /// Draft-15 Section 10.2.1.2: "Any Object with status Normal can have
392    /// extension headers. If an endpoint receives extension headers on Objects
393    /// with status that is not Normal, it MUST close the session with a
394    /// PROTOCOL_VIOLATION."
395    ///
396    /// So this is `false` for exactly one shape: a non-empty extension block on
397    /// an object whose status is not [`ObjectStatus::Normal`]. An object with no
398    /// extensions is fine at any status, and an object at Normal may carry any
399    /// extensions.
400    ///
401    /// [`SubgroupObjectReader::read_object`] and
402    /// [`SubgroupObjectReader::write_object`] refuse an object this answers
403    /// `false` for; [`SubgroupObjectReader::read_object_meta`] frames it.
404    pub fn extensions_permitted(&self) -> bool {
405        self.extension_headers.is_empty() || self.status() == ObjectStatus::Normal
406    }
407}
408
409/// The framing of one draft-15 subgroup object, without its payload.
410///
411/// Produced by [`SubgroupObjectReader::read_object_meta`] for callers that
412/// forward an object's bytes verbatim and never inspect the payload.
413#[derive(Debug, Clone, Copy, PartialEq, Eq)]
414pub struct SubgroupObjectMeta {
415    /// Resolved absolute Object ID.
416    pub object_id: u64,
417    /// Byte length of the extension-header block's contents, excluding its
418    /// length prefix.
419    pub extension_headers_len: u64,
420    /// Declared payload length. Zero when `status` is `Some`.
421    pub payload_length: u64,
422    /// Object status wire code, present only when the payload is empty.
423    ///
424    /// Kept as the raw code, unlike [`SubgroupObject::object_status`]: a meta
425    /// is produced by [`SubgroupObjectReader::read_object_meta`] and is never
426    /// an encode input, and a relay that reads a status on one draft may hand
427    /// it to a draft that numbers the same value differently. The code is
428    /// still one draft-15 assigns — `read_object_meta` refuses the others.
429    pub status: Option<u64>,
430    /// Total bytes this object occupies on the wire, prefix fields included.
431    pub wire_len: u64,
432}
433
434impl SubgroupObjectMeta {
435    /// Whether this object is permitted a non-empty payload, or `None` when
436    /// [`Self::status`] holds a code draft-15 does not assign.
437    ///
438    /// [`Self::status`] is a raw wire code, so the rule of draft-15 Section
439    /// 10.2.1.1 — "Any object with a status code other than zero MUST have an
440    /// empty payload" — cannot be read off it without first deciding what the
441    /// code means. This does that decision once: a consumer asking whether an
442    /// object may carry bytes gets an answer instead of a number and a rule to
443    /// apply to it.
444    ///
445    /// An absent status answers [`PayloadPermission::Permitted`] rather than
446    /// `None`. A meta has no status only when its payload length is non-zero,
447    /// and Section 10.2.1.1 says of Normal that "This status is implicit for
448    /// any non-zero length object" — the object has a status, the encoding just
449    /// does not spell it.
450    ///
451    /// `None` means the code is one the draft leaves unassigned, and so one it
452    /// gives no payload rule for. That is not reachable through
453    /// [`SubgroupObjectReader::read_object_meta`], which refuses such a code
454    /// before it can reach the field, but the fields here are public and a meta
455    /// assembled by hand — by a relay carrying a status across from a draft that
456    /// numbers them differently, say — can hold anything a varint can. The
457    /// answer there is that draft-15 has none, not that the payload is
458    /// forbidden.
459    pub fn payload_permission(&self) -> Option<PayloadPermission> {
460        match self.status {
461            None => Some(PayloadPermission::Permitted),
462            Some(code) => ObjectStatus::from_u64(code).map(PayloadPermission::for_status),
463        }
464    }
465
466    /// Whether this object's status is allowed to carry the extension block it
467    /// declares.
468    ///
469    /// The same rule [`SubgroupObject::extensions_permitted`] answers, from the
470    /// declared length rather than from the bytes. A relay that forwards an
471    /// object verbatim reads it through
472    /// [`SubgroupObjectReader::read_object_meta`] and never copies the block, so
473    /// asking this must not require having it — draft-15 Section 10.2.1.2 turns
474    /// on whether the block is empty, and the length says that on its own.
475    ///
476    /// A status this draft does not assign answers `None` rather than `false`,
477    /// for the reason [`Self::payload_permission`] gives: the draft states no
478    /// rule for a code it does not define, and a meta assembled by hand can
479    /// hold one. `read_object_meta` refuses such a code before it reaches the
480    /// field.
481    ///
482    /// An absent status answers from Normal. A meta has no status only when its
483    /// payload length is non-zero, and such an object is Normal by Section
484    /// 10.2.1.1, so its extensions are permitted.
485    pub fn extensions_permitted(&self) -> Option<bool> {
486        if self.extension_headers_len == 0 {
487            return Some(true);
488        }
489        match self.status {
490            None => Some(true),
491            Some(code) => ObjectStatus::from_u64(code).map(|s| s == ObjectStatus::Normal),
492        }
493    }
494}
495
496/// Prior Group ID Gap, Extension Header Type 0x3C (Section 11.1).
497pub const PRIOR_GROUP_ID_GAP: u64 = 0x3C;
498/// Prior Object ID Gap, Extension Header Type 0x3E (Section 11.2).
499pub const PRIOR_OBJECT_ID_GAP: u64 = 0x3E;
500
501/// The Prior Group ID Gap and Prior Object ID Gap one Object carries.
502#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
503pub struct PriorGaps {
504    /// Prior Group ID Gap, if present.
505    pub group: Option<u64>,
506    /// Prior Object ID Gap, if present.
507    pub object: Option<u64>,
508}
509
510impl PriorGaps {
511    /// Read the two gaps from an Object's parsed extension headers, refusing
512    /// the conditions one Object shows on its own.
513    ///
514    /// Section 11.1 makes the Track malformed when "An Object contains more
515    /// than one instance of Prior Group ID Gap" and when "An Object has a Prior
516    /// Group ID Gap larger than the Group ID". Section 11.2 says the same two
517    /// for Prior Object ID Gap against the Object ID. Each is
518    /// [`CodecError::MalformedTrack`].
519    pub fn from_properties(
520        properties: &[ObjectProperty<'_>],
521        group_id: u64,
522        object_id: u64,
523    ) -> Result<Self, CodecError> {
524        let mut gaps = Self::default();
525        collect_gaps(properties, &mut gaps)?;
526        if let Some(gap) = gaps.group.filter(|&gap| gap > group_id) {
527            return Err(CodecError::MalformedTrack {
528                detail: format!("Prior Group ID Gap {gap} is larger than Group ID {group_id}"),
529            });
530        }
531        if let Some(gap) = gaps.object.filter(|&gap| gap > object_id) {
532            return Err(CodecError::MalformedTrack {
533                detail: format!("Prior Object ID Gap {gap} is larger than Object ID {object_id}"),
534            });
535        }
536        Ok(gaps)
537    }
538}
539
540fn collect_gaps(properties: &[ObjectProperty<'_>], gaps: &mut PriorGaps) -> Result<(), CodecError> {
541    for property in properties {
542        let slot = match (&property.value, property.property_type) {
543            (PropertyValue::Immutable { properties, .. }, _) => {
544                collect_gaps(properties, gaps)?;
545                continue;
546            }
547            (PropertyValue::Varint(value), PRIOR_GROUP_ID_GAP) => {
548                (&mut gaps.group, *value, "Group")
549            }
550            (PropertyValue::Varint(value), PRIOR_OBJECT_ID_GAP) => {
551                (&mut gaps.object, *value, "Object")
552            }
553            _ => continue,
554        };
555        let (field, value, which) = slot;
556        if field.replace(value).is_some() {
557            return Err(CodecError::MalformedTrack {
558                detail: format!("an Object contains more than one Prior {which} ID Gap"),
559            });
560        }
561    }
562    Ok(())
563}
564
565/// The Prior Group ID Gap and Prior Object ID Gap conditions of
566/// draft-15 Section 11.1 and draft-15 Section 11.2 that one fetch stream
567/// shows.
568///
569/// [`PriorGaps::from_properties`] refuses what one Object shows. Across the
570/// stream this refuses "A Group contains more than one Object with different
571/// values for Prior Group ID Gap" for the Objects of one Group, "An endpoint
572/// receives an Object with a Prior Group ID Gap covering an Object it
573/// previously received", "An endpoint receives an Object with a Group ID
574/// within a previously communicated gap", and the same two for Prior Object ID
575/// Gap, for the Group and the Object before on the stream.
576/// draft-15 Section 9.16.3: "Within each group, objects are sent in Object ID
577/// order; subgroup ID is not used for ordering."
578///
579/// [`FetchObjectReader::read_object_header`] applies it to each Object it reads.
580#[derive(Debug, Clone, PartialEq, Eq)]
581// Boxed so that the fetch readers holding one stay small.
582pub struct FetchGaps(Box<crate::prior_gaps::FetchGapTracker>);
583
584impl Default for FetchGaps {
585    fn default() -> Self {
586        Self::new()
587    }
588}
589
590impl FetchGaps {
591    /// Track the Objects of one fetch stream.
592    pub fn new() -> Self {
593        Self(Box::new(crate::prior_gaps::FetchGapTracker::new(
594            crate::prior_gaps::GapRules::Draft14To15,
595        )))
596    }
597
598    /// Check the next Object on the stream, from its resolved Group ID and
599    /// Object ID and its extension block, refusing a gap condition with
600    /// [`CodecError::MalformedTrack`] and a block that does not parse as
601    /// [`crate::object_properties::parse_object_properties`] does. Records the
602    /// Object only when it passes.
603    pub fn check(
604        &mut self,
605        group_id: u64,
606        object_id: u64,
607        extension_block: &[u8],
608    ) -> Result<PriorGaps, CodecError> {
609        let properties = crate::object_properties::parse_object_properties(
610            crate::version::DraftVersion::Draft15,
611            extension_block,
612        )?;
613        let gaps = PriorGaps::from_properties(&properties, group_id, object_id)?;
614        self.0.check(group_id, object_id, gaps.group.as_slice(), gaps.object)?;
615        Ok(gaps)
616    }
617}
618
619/// Stateful reader/writer for draft-15 subgroup objects.
620///
621/// Carries the running delta state for object IDs and remembers whether
622/// extension headers are present on this stream.
623#[derive(Debug, Clone)]
624pub struct SubgroupObjectReader {
625    extensions_present: bool,
626    prev_object_id: Option<u64>,
627    group_id: u64,
628    group_gap: Option<u64>,
629}
630
631impl SubgroupObjectReader {
632    /// Build a reader seeded from the enclosing subgroup header.
633    pub fn new(header: &SubgroupHeader) -> Self {
634        Self {
635            extensions_present: header.has_extensions(),
636            prev_object_id: None,
637            group_id: header.group_id.into_inner(),
638            group_gap: None,
639        }
640    }
641
642    /// Decode the next object whole, payload included.
643    ///
644    /// Refuses, with [`CodecError::InvalidField`], extension headers on an
645    /// object whose status is not Normal: Section 10.2.1.2, "If an endpoint
646    /// receives extension headers on Objects with status that is not Normal, it
647    /// MUST close the session with a PROTOCOL_VIOLATION."
648    /// [`Self::read_object_meta`] frames such an object without judging it.
649    ///
650    /// Refuses, with [`CodecError::MalformedTrack`], the Prior Group ID Gap
651    /// and Prior Object ID Gap conditions of Section 11.1 and Section 11.2
652    /// this stream shows: those [`PriorGaps::from_properties`] refuses, "A
653    /// Group contains more than one Object with different values for Prior
654    /// Group ID Gap", and "An endpoint receives an Object with a Prior Object
655    /// ID Gap covering an Object it previously received" for an Object earlier
656    /// on this stream. [`Self::read_object_meta`] does not read extension
657    /// headers and applies none of them.
658    pub fn read_object(&mut self, buf: &mut impl Buf) -> Result<SubgroupObject, CodecError> {
659        let delta = VarInt::decode(buf)?.into_inner();
660        let prior_object_id = self.prev_object_id;
661        // The first object's field is its absolute Object ID; every later
662        // object encodes the gap to its predecessor, biased by one because
663        // two objects on a subgroup stream cannot share an ID.
664        let object_id_val = match self.prev_object_id {
665            None => delta,
666            Some(prev) => prev
667                .checked_add(1)
668                .and_then(|v| v.checked_add(delta))
669                .ok_or(CodecError::InvalidField)?,
670        };
671        self.prev_object_id = Some(object_id_val);
672        let object_id = VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?;
673
674        let extension_headers = if self.extensions_present {
675            // Draft-15+: extensions are a byte-length-prefixed opaque
676            // blob. We copy the blob verbatim; callers that want
677            // structured extensions can parse the returned bytes.
678            let ext_len = VarInt::decode(buf)?.into_inner() as usize;
679            crate::types::read_bytes(buf, ext_len)?
680        } else {
681            Vec::new()
682        };
683
684        let payload_length_vi = VarInt::decode(buf)?;
685        let payload_length_val = payload_length_vi.into_inner() as usize;
686        let (object_status, payload) = if payload_length_val == 0 {
687            let status = decoded_status(VarInt::decode(buf)?.into_inner())?;
688            (Some(status), Vec::new())
689        } else {
690            let payload = crate::types::read_bytes(buf, payload_length_val)?;
691            (None, payload)
692        };
693
694        let properties = crate::object_properties::parse_object_properties(
695            crate::version::DraftVersion::Draft15,
696            &extension_headers,
697        )?;
698        let gaps = PriorGaps::from_properties(&properties, self.group_id, object_id_val)?;
699        self.check_stream_gaps(gaps, prior_object_id, object_id_val)?;
700        let object = SubgroupObject {
701            object_id,
702            extension_headers,
703            payload_length: payload_length_vi,
704            object_status,
705            payload,
706        };
707        if !object.extensions_permitted() {
708            return Err(CodecError::InvalidField);
709        }
710        Ok(object)
711    }
712
713    fn check_stream_gaps(
714        &mut self,
715        gaps: PriorGaps,
716        prior_object_id: Option<u64>,
717        object_id: u64,
718    ) -> Result<(), CodecError> {
719        if let Some(gap) = gaps.group {
720            match self.group_gap {
721                Some(seen) if seen != gap => {
722                    return Err(CodecError::MalformedTrack {
723                        detail: format!(
724                            "two Objects in Group {} carry different Prior Group ID Gap values",
725                            self.group_id
726                        ),
727                    });
728                }
729                _ => self.group_gap = Some(gap),
730            }
731        }
732        // Object IDs rise along a subgroup stream, so the latest earlier
733        // Object is the only one a gap below this Object can reach.
734        if let (Some(gap), Some(prior)) = (gaps.object, prior_object_id) {
735            if gap > 0 && prior >= object_id - gap {
736                return Err(CodecError::MalformedTrack {
737                    detail: format!(
738                        "Prior Object ID Gap {gap} of Object {object_id} covers Object {prior}"
739                    ),
740                });
741            }
742        }
743        Ok(())
744    }
745
746    /// Decode the next object's framing without copying its payload.
747    ///
748    /// Consumes exactly the bytes [`Self::read_object`] consumes and leaves
749    /// the same delta state behind, so the two are interchangeable on a
750    /// given stream.
751    pub fn read_object_meta(
752        &mut self,
753        buf: &mut impl Buf,
754    ) -> Result<SubgroupObjectMeta, CodecError> {
755        let start = buf.remaining();
756        let delta = VarInt::decode(buf)?.into_inner();
757        let object_id_val = match self.prev_object_id {
758            None => delta,
759            Some(prev) => prev
760                .checked_add(1)
761                .and_then(|v| v.checked_add(delta))
762                .ok_or(CodecError::InvalidField)?,
763        };
764        self.prev_object_id = Some(object_id_val);
765        let object_id =
766            VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?.into_inner();
767
768        let extension_headers_len = if self.extensions_present {
769            let ext_len = VarInt::decode(buf)?.into_inner();
770            skip(buf, ext_len)?;
771            ext_len
772        } else {
773            0
774        };
775
776        let payload_length = VarInt::decode(buf)?.into_inner();
777        let status = if payload_length == 0 {
778            let code = VarInt::decode(buf)?.into_inner();
779            decoded_status(code)?;
780            Some(code)
781        } else {
782            skip(buf, payload_length)?;
783            None
784        };
785
786        Ok(SubgroupObjectMeta {
787            object_id,
788            extension_headers_len,
789            payload_length,
790            status,
791            wire_len: (start - buf.remaining()) as u64,
792        })
793    }
794
795    /// Serialize an object, producing the correct delta encoding.
796    ///
797    /// Errors with [`CodecError::InvalidField`] when `object.object_id` is
798    /// not strictly greater than the previously written object's ID, since
799    /// no valid delta exists for that case.
800    ///
801    /// A zero `payload_length` makes this a status object, and the status
802    /// field is then mandatory on the wire: an absent
803    /// [`SubgroupObject::object_status`] is written as
804    /// [`ObjectStatus::Normal`]. The code written is always one draft-15
805    /// assigns, because the field cannot hold any other, so the bytes this
806    /// produces are always bytes [`SubgroupObjectReader::read_object`] accepts.
807    ///
808    /// Errors with [`CodecError::InvalidField`] when `payload_length` is not
809    /// exactly `payload.len()`. The declared length is written ahead of the
810    /// payload, so a mismatch is a frame [`Self::read_object`] cannot parse
811    /// and one no caller could fix by appending bytes.
812    ///
813    /// Does NOT refuse extension headers beside a status other than Normal,
814    /// though draft-15 Section 10.2.1.2 forbids them. Such a frame is well
815    /// formed and merely non-conforming, this is the only writer a subgroup
816    /// object has, and the corpus ships one — so refusing here would leave a
817    /// captured violation impossible to reproduce.
818    /// [`SubgroupObject::extensions_permitted`] reports it instead.
819    pub fn write_object(
820        &mut self,
821        object: &SubgroupObject,
822        buf: &mut impl BufMut,
823    ) -> Result<(), CodecError> {
824        // A declared length that disagrees with the payload framed under it
825        // produces bytes no reader can parse and no caller can repair: the
826        // length is already on the wire ahead of the payload. Checked before
827        // anything is written, so a refused object leaves `buf` untouched
828        // rather than half an object the next read would run into.
829        //
830        // Zero is not "an empty payload" here; it is the marker that puts a
831        // status code where the payload would go, so an object carrying bytes
832        // under it is asking for two framings at once.
833        let declared = object.payload_length.into_inner();
834        if declared != object.payload.len() as u64 {
835            return Err(CodecError::InvalidField);
836        }
837
838        // Extension headers on a non-Normal status are a frame `read_object`
839        // refuses (Section 10.2.1.2), so they are not written either.
840        if !object.extensions_permitted() {
841            return Err(CodecError::InvalidField);
842        }
843
844        let oid = object.object_id.into_inner();
845        let delta = match self.prev_object_id {
846            None => oid,
847            Some(prev) => oid
848                .checked_sub(prev)
849                .and_then(|v| v.checked_sub(1))
850                .ok_or(CodecError::InvalidField)?,
851        };
852        VarInt::from_u64(delta).map_err(|_| CodecError::InvalidField)?.encode(buf);
853        if self.extensions_present {
854            let ext_len = object.extension_headers.len();
855            VarInt::from_usize(ext_len).encode(buf);
856            buf.put_slice(&object.extension_headers);
857        }
858        object.payload_length.encode(buf);
859        if object.payload_length.into_inner() == 0 {
860            // Zero-length means a status object, and the status is not
861            // optional on the wire; an unset one is Normal.
862            let status = object.object_status.unwrap_or(ObjectStatus::Normal);
863            VarInt::from_usize(status.as_u64() as usize).encode(buf);
864        } else {
865            buf.put_slice(&object.payload);
866        }
867        self.prev_object_id = Some(oid);
868        Ok(())
869    }
870}
871
872// ── Datagram headers ───────────────────────────────────────
873
874/// Whether draft-15 Section 10.3.1 Table 5 assigns this datagram type.
875///
876/// Twenty-four values are assigned: `0x00`-`0x0F`, and `0x20`, `0x21`, `0x24`,
877/// `0x25`, `0x28`, `0x29`, `0x2C`, `0x2D`. Two rules generate that set. A type
878/// may only use the bits the draft defines — `0x01` extensions, `0x02` end of
879/// group, `0x04` no object ID, `0x08` default priority, `0x20` status — so any
880/// other bit makes it undefined. And a status datagram cannot also mark the end
881/// of a group, which rules out every type setting `0x20` and `0x02` together.
882///
883/// Section 10 requires closing the session on a datagram type the draft does
884/// not define. Accepting one means inferring field presence from bits that
885/// carry no meaning, which is how an undefined type gets parsed as a
886/// well-formed object.
887///
888/// Note that Section 10 Table 4 lists only ten datagram types and describes
889/// them as the whole set. That table is draft-14 text left behind: Table 5 in
890/// Section 10.3.1 is the normative field-presence table, it lists twenty-four,
891/// and draft-16 follows it.
892fn datagram_type_is_assigned(ty: u64) -> bool {
893    let Ok(ty) = u8::try_from(ty) else {
894        return false;
895    };
896    ty & 0xD0 == 0 && ty & 0x22 != 0x22
897}
898
899/// Datagram header for draft-15.
900///
901/// The `datagram_type` byte encodes flags:
902/// - `0x02`: end-of-group
903/// - `0x04`: no object_id (object_id = 0 implied)
904/// - `0x20`: status datagram (carries object_status instead of payload)
905#[derive(Debug, Clone, PartialEq, Eq)]
906pub struct DatagramHeader {
907    /// Raw datagram-type byte encoding flags + kind.
908    pub datagram_type: u8,
909    /// Track alias identifying the track.
910    pub track_alias: VarInt,
911    /// Group ID for the contained object.
912    pub group_id: VarInt,
913    /// Object ID (zero when the `no-object-id` flag is set).
914    pub object_id: VarInt,
915    /// Publisher priority, present only when the type byte leaves the
916    /// default-priority bit (`0x08`) clear.
917    ///
918    /// `None` means the object inherits the priority the control message that
919    /// established the subscription specified — draft-15 Section 10.3.1. This
920    /// is optional here and not on draft-14 because draft-15 is the draft that
921    /// made it so.
922    pub publisher_priority: Option<u8>,
923    /// Opaque extension-headers blob (only when the `0x01` flag is set).
924    pub extension_headers: Vec<u8>,
925    /// Object status (only when the `0x20` status flag is set).
926    ///
927    /// The wire field is a varint and can carry any value up to 2^62-1;
928    /// draft-15 Section 10.2.1.1 assigns four of them and says a peer SHOULD
929    /// treat the rest as a protocol error. This field is typed to the assigned
930    /// set, so it cannot hold 0x2 or anything from 0x5 up — [`Self::encode`]
931    /// therefore needs no check and cannot emit a datagram that
932    /// [`Self::decode`] would reject.
933    ///
934    /// `None` while the status flag is set encodes as
935    /// [`ObjectStatus::Normal`]: once the flag is set the field is present on
936    /// the wire, so there is nothing for `None` to mean but the default.
937    pub object_status: Option<ObjectStatus>,
938}
939
940impl DatagramHeader {
941    /// Whether the datagram carries an explicit object_id.
942    pub fn has_object_id(&self) -> bool {
943        self.datagram_type & 0x04 == 0
944    }
945
946    /// Whether this datagram marks the end of its group.
947    pub fn is_end_of_group(&self) -> bool {
948        self.datagram_type & 0x02 != 0
949    }
950
951    /// Whether this datagram carries an object_status instead of payload.
952    pub fn is_status(&self) -> bool {
953        self.datagram_type & 0x20 != 0
954    }
955
956    /// Whether this datagram carries extension headers.
957    pub fn has_extensions(&self) -> bool {
958        self.datagram_type & 0x01 != 0
959    }
960
961    /// Whether the publisher priority is omitted and inherited.
962    ///
963    /// Draft-15 Section 10.3.1: with Priority Present set to No the field is
964    /// absent and "this Object inherits the Publisher Priority specified in the
965    /// control message that established the subscription". New in draft-15;
966    /// draft-14 always carries the byte.
967    pub fn has_default_priority(&self) -> bool {
968        self.datagram_type & 0x08 != 0
969    }
970
971    /// Encode the datagram header, refusing a status the framing cannot carry.
972    ///
973    /// A datagram states a status only when its type byte sets the status flag
974    /// (0x20). With the flag clear there is no status field on the wire, so an
975    /// `object_status` of anything but [`ObjectStatus::Normal`] has nowhere to
976    /// go: [`Self::encode`] drops it, and the datagram parses back as an
977    /// ordinary payload object. An End of Group marker written that way does
978    /// not arrive late or malformed — it does not arrive at all, and the
979    /// receiver sees a normal object in its place.
980    ///
981    /// Draft-15 Section 10.3.1 puts the framing side plainly — "The Object
982    /// Status field and Object Payload are mutually exclusive" — and Section
983    /// 10.2.1.1 the conformance side: "Any object with a status code other than
984    /// zero MUST have an empty payload." Between them there is no datagram that
985    /// carries a non-zero status and a payload, so the pair being refused here
986    /// is not one this encoder merely declines to spell.
987    ///
988    /// [`ObjectStatus::Normal`] with the flag clear is not that case and is
989    /// accepted. It is the status the encoding elides for every datagram that
990    /// carries a payload, so stating it asks for exactly the bytes leaving it
991    /// out asks for, and nothing is lost.
992    ///
993    /// Errors with [`CodecError::InvalidField`] on the lossy combination,
994    /// before any byte is written, so a refused header leaves `buf` untouched.
995    /// It also refuses a type byte Section 10.3.1 Table 5 does not assign, so
996    /// this encoder cannot emit a datagram its own decoder — or a conforming
997    /// peer — must close the session over.
998    ///
999    /// The extension block is refused on the same terms the type byte governs
1000    /// it, in both directions. A type announcing extensions must carry some:
1001    /// Section 10.3.1 states that "If an endpoint receives a datagram with
1002    /// Extensions Present as 'Yes' and a Extension Headers Length of 0, it MUST
1003    /// close the session with a PROTOCOL_VIOLATION", so a zero-length block
1004    /// under that type is a datagram no peer may accept. A type announcing none
1005    /// cannot carry any, because [`Self::encode`] would drop the bytes in
1006    /// silence.
1007    ///
1008    /// That first rule belongs to datagrams alone. Section 10.4.2 says the
1009    /// opposite of a subgroup stream — "Objects with no extensions set Extension
1010    /// Headers Length to 0" — because there the type byte is fixed for the whole
1011    /// stream and a zero-length block is the only way one object among many can
1012    /// say it has no extensions. Applying the datagram rule to a subgroup object
1013    /// would refuse frames the draft spells out.
1014    ///
1015    /// Also refused: an extension block beside a status other than Normal, per
1016    /// Section 10.2.1.2. [`Self::decode`] still parses such a datagram, so a
1017    /// capture containing one stays readable; see
1018    /// `check_extensions_against_status_on_encode`.
1019    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1020        if !self.is_status() && matches!(self.object_status, Some(s) if s != ObjectStatus::Normal) {
1021            return Err(CodecError::InvalidField);
1022        }
1023        if !datagram_type_is_assigned(self.datagram_type as u64) {
1024            return Err(CodecError::UnknownDatagramType(self.datagram_type as u64));
1025        }
1026        if self.has_extensions() {
1027            if self.extension_headers.is_empty() {
1028                return Err(CodecError::InvalidField);
1029            }
1030        } else if !self.extension_headers.is_empty() {
1031            return Err(CodecError::InvalidField);
1032        }
1033        check_extensions_against_status_on_encode(
1034            self.effective_status().map(|s| s.as_u64()),
1035            if self.has_extensions() { self.extension_headers.len() as u64 } else { 0 },
1036        )?;
1037        self.encode(buf);
1038        Ok(())
1039    }
1040
1041    /// The status this datagram resolves to, or `None` when its framing gives
1042    /// it none to resolve.
1043    ///
1044    /// The type byte decides. With the status bit set the field is on the wire
1045    /// and an unset [`Self::object_status`] is written as
1046    /// [`ObjectStatus::Normal`]; with the bit clear the datagram carries a
1047    /// payload, and the status of an object that carries a payload is Normal —
1048    /// draft-15 Section 10.2.1.1 says "This status is implicit for any non-zero
1049    /// length object". `None` is that implicit Normal, which is why a caller
1050    /// asking what a datagram's status is may not read [`Self::object_status`]
1051    /// directly.
1052    fn effective_status(&self) -> Option<ObjectStatus> {
1053        if self.is_status() {
1054            Some(self.object_status.unwrap_or(ObjectStatus::Normal))
1055        } else {
1056            None
1057        }
1058    }
1059
1060    /// Whether this datagram's status is allowed to carry the extension headers
1061    /// it has.
1062    ///
1063    /// The datagram half of the rule [`SubgroupObject::extensions_permitted`]
1064    /// answers for a subgroup object; draft-15 Section 10.3.1 builds the
1065    /// datagram's extension block out of the same structure Section 10.2.1.2
1066    /// defines, so the rule covers both carriers.
1067    ///
1068    /// [`Self::decode`] reports this rather than refusing, because the datagram
1069    /// is well framed and a codec that could not read one could not reproduce a
1070    /// capture containing it. [`Self::encode_checked`] does refuse it — that is
1071    /// the one direction with no such excuse, and a plain [`Self::encode`]
1072    /// stands beside it when verbatim reproduction is what is wanted.
1073    pub fn extensions_permitted(&self) -> bool {
1074        if !self.has_extensions() || self.extension_headers.is_empty() {
1075            return true;
1076        }
1077        self.effective_status().unwrap_or(ObjectStatus::Normal) == ObjectStatus::Normal
1078    }
1079
1080    /// Encode the datagram header to `buf`.
1081    ///
1082    /// When the type byte sets the status flag the status field is written
1083    /// unconditionally, defaulting to [`ObjectStatus::Normal`]. Omitting it
1084    /// would truncate the datagram: [`Self::decode`] reads a status whenever
1085    /// the flag is set, and answers [`CodecError::UnexpectedEnd`] when the
1086    /// bytes stop first.
1087    ///
1088    /// The type byte is taken as the authority on framing, which is what makes
1089    /// this infallible — and what makes it lossy when the struct disagrees with
1090    /// itself. An `object_status` set while the type byte leaves the status
1091    /// flag clear is discarded here without a word. Prefer
1092    /// [`Self::encode_checked`], which refuses that combination instead of
1093    /// resolving it.
1094    pub fn encode(&self, buf: &mut impl BufMut) {
1095        VarInt::from_usize(self.datagram_type as usize).encode(buf);
1096        self.track_alias.encode(buf);
1097        self.group_id.encode(buf);
1098        if self.has_object_id() {
1099            self.object_id.encode(buf);
1100        }
1101        if !self.has_default_priority() {
1102            // The type byte is the authority on presence. A `None` here under a
1103            // type whose `0x08` bit is clear writes a zero rather than dropping
1104            // the byte: omitting it would leave the peer reading the first
1105            // extension-length or status varint as a priority.
1106            buf.put_u8(self.publisher_priority.unwrap_or(0));
1107        }
1108        if self.has_extensions() {
1109            VarInt::from_usize(self.extension_headers.len()).encode(buf);
1110            buf.put_slice(&self.extension_headers);
1111        }
1112        if self.is_status() {
1113            let status = self.object_status.unwrap_or(ObjectStatus::Normal);
1114            VarInt::from_usize(status.as_u64() as usize).encode(buf);
1115        }
1116    }
1117
1118    /// Decode a datagram header from `buf`.
1119    ///
1120    /// [`Self::decode_object`] is the read of a whole datagram, and applies
1121    /// the Object rules this header-only read does not.
1122    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1123        // Compared as the full varint, not truncated to `u8`: a type of `0x100`
1124        // narrows to `0x00` and would be read as an ordinary object datagram, so
1125        // a greased or future type would be silently mis-parsed rather than
1126        // closing the session.
1127        let type_value = VarInt::decode(buf)?.into_inner();
1128        if !datagram_type_is_assigned(type_value) {
1129            return Err(CodecError::UnknownDatagramType(type_value));
1130        }
1131        let datagram_type = type_value as u8;
1132        let track_alias = VarInt::decode(buf)?;
1133        let group_id = VarInt::decode(buf)?;
1134        let object_id =
1135            if datagram_type & 0x04 == 0 { VarInt::decode(buf)? } else { VarInt::from_usize(0) };
1136        let publisher_priority = if datagram_type & 0x08 == 0 {
1137            if buf.remaining() < 1 {
1138                return Err(CodecError::UnexpectedEnd);
1139            }
1140            Some(buf.get_u8())
1141        } else {
1142            None
1143        };
1144        let extension_headers = if datagram_type & 0x01 != 0 {
1145            let ext_len = VarInt::decode(buf)?.into_inner() as usize;
1146            // A datagram whose type says extensions are present must actually carry
1147            // some: receiving one with an Extension Headers Length of 0 closes the
1148            // session. The opposite holds on a subgroup stream, where the type byte is
1149            // fixed for the whole stream and an object with no extensions has no other
1150            // way to say so, which is why this check belongs to the datagram readers
1151            // alone.
1152            if ext_len == 0 {
1153                return Err(CodecError::InvalidField);
1154            }
1155            crate::types::read_bytes(buf, ext_len)?
1156        } else {
1157            Vec::new()
1158        };
1159        let object_status = if datagram_type & 0x20 != 0 {
1160            Some(decoded_status(VarInt::decode(buf)?.into_inner())?)
1161        } else {
1162            None
1163        };
1164        Ok(Self {
1165            datagram_type,
1166            track_alias,
1167            group_id,
1168            object_id,
1169            publisher_priority,
1170            extension_headers,
1171            object_status,
1172        })
1173    }
1174
1175    /// Decode one whole datagram: the header, then the payload that runs to the
1176    /// end of `buf`, which must hold exactly one transport datagram.
1177    ///
1178    /// Unlike [`Self::decode`], this refuses an extension block that does not
1179    /// parse, as [`crate::object_properties::parse_object_properties`] does,
1180    /// and, with [`CodecError::MalformedTrack`], the Prior Group ID Gap and
1181    /// Prior Object ID Gap conditions one Object shows (draft-15 Section 11.1,
1182    /// draft-15 Section 11.2), through [`PriorGaps::from_properties`].
1183    ///
1184    /// It also refuses bytes after a status datagram's Object Status, with
1185    /// [`CodecError::PayloadNotPermitted`]. draft-15 Section 10.3.1: "For Type
1186    /// values 0x20, 0x21, 0x24 and 0x25 the Object Status field is present and
1187    /// there is no Object Payload."
1188    pub fn decode_object(buf: &mut impl Buf) -> Result<(Self, Vec<u8>), CodecError> {
1189        let header = Self::decode(buf)?;
1190        let properties = crate::object_properties::parse_object_properties(
1191            crate::version::DraftVersion::Draft15,
1192            &header.extension_headers,
1193        )?;
1194        PriorGaps::from_properties(
1195            &properties,
1196            header.group_id.into_inner(),
1197            header.object_id.into_inner(),
1198        )?;
1199        let payload = crate::types::read_bytes(buf, buf.remaining())?;
1200        if let Some(status) = header.object_status {
1201            if !payload.is_empty() {
1202                return Err(CodecError::PayloadNotPermitted {
1203                    status: status.as_u64(),
1204                    len: payload.len(),
1205                    detail: "its type states a status in place of a payload",
1206                });
1207            }
1208        }
1209        Ok((header, payload))
1210    }
1211}
1212
1213// ── Fetch stream headers ───────────────────────────────────
1214
1215/// Fetch stream header for draft-15.
1216///
1217/// Stream type is 0x05. Only contains a request_id.
1218#[derive(Debug, Clone, PartialEq, Eq)]
1219pub struct FetchHeader {
1220    pub request_id: VarInt,
1221}
1222
1223/// The unidirectional stream type draft-15 Section 10.4.4 gives a fetch stream.
1224const FETCH_STREAM_TYPE: u64 = 0x05;
1225
1226/// Which failure a leading unidirectional stream type that is not the one a
1227/// reader wants is.
1228///
1229/// Section 10: "An endpoint that receives an unknown stream or datagram type
1230/// MUST close the session." One sentence, two tables. The stream table assigns
1231/// FETCH_HEADER and the subgroup types [`subgroup_type_is_assigned`] describes;
1232/// everything outside them is unknown at the head of a stream, and the session
1233/// ends.
1234///
1235/// Both assigned kinds are what the [`CodecError::InvalidField`] arm is for: a
1236/// fetch stream reaching the subgroup reader, or a subgroup stream reaching the
1237/// fetch reader, is a value this draft defines, and the disagreement is with
1238/// the reader that was called rather than with the draft. Reporting it as
1239/// unknown would end sessions over streams draft-15 permits.
1240///
1241/// The datagram reader needs no such helper. Its table shares no value with the
1242/// stream table, so every type it rejects is one no table assigns and the
1243/// answer is always [`CodecError::UnknownDatagramType`].
1244fn stream_type_error(raw: u64) -> CodecError {
1245    if raw == FETCH_STREAM_TYPE || subgroup_type_is_assigned(raw) {
1246        CodecError::InvalidField
1247    } else {
1248        CodecError::UnknownStreamType(raw)
1249    }
1250}
1251
1252impl FetchHeader {
1253    pub fn encode(&self, buf: &mut impl BufMut) {
1254        VarInt::from_usize(FETCH_STREAM_TYPE as usize).encode(buf);
1255        self.request_id.encode(buf);
1256    }
1257
1258    /// Decode the header.
1259    ///
1260    /// Errors with [`CodecError::UnknownStreamType`] when the stream table does
1261    /// not assign the leading type, which this draft answers with a close, and
1262    /// with [`CodecError::InvalidField`] for the subgroup types, which it does
1263    /// assign. `stream_type_error` draws that line.
1264    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1265        let stream_type = VarInt::decode(buf)?.into_inner();
1266        if stream_type != FETCH_STREAM_TYPE {
1267            return Err(stream_type_error(stream_type));
1268        }
1269        let request_id = VarInt::decode(buf)?;
1270        Ok(Self { request_id })
1271    }
1272}
1273
1274// ── Fetch objects (stateful) ───────────────────────────────
1275
1276/// How a draft-15 fetch object carries its Subgroup ID.
1277///
1278/// The two least significant bits of the Serialization Flags are one field,
1279/// not two independent flags: draft-15 Section 10.4.4, Table 7 gives all four
1280/// of their values a meaning, and only one of them puts a Subgroup ID on the
1281/// wire. Reading either bit on its own gets the wrong answer for half the
1282/// values.
1283#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1284pub enum SubgroupIdEncoding {
1285    /// `0x00`: the Subgroup ID is zero, whatever the preceding object's was.
1286    Zero,
1287    /// `0x01`: the Subgroup ID is the prior object's.
1288    SameAsPrior,
1289    /// `0x02`: the Subgroup ID is the prior object's plus one.
1290    PriorPlusOne,
1291    /// `0x03`: the Subgroup ID follows as a varint.
1292    Present,
1293}
1294
1295impl SubgroupIdEncoding {
1296    /// Read the two-bit field out of a Serialization Flags byte.
1297    pub fn from_flags(flags: u8) -> Self {
1298        match flags & 0x03 {
1299            0x00 => SubgroupIdEncoding::Zero,
1300            0x01 => SubgroupIdEncoding::SameAsPrior,
1301            0x02 => SubgroupIdEncoding::PriorPlusOne,
1302            _ => SubgroupIdEncoding::Present,
1303        }
1304    }
1305
1306    /// The bit pattern this encoding occupies in a Serialization Flags byte.
1307    pub fn as_bits(self) -> u8 {
1308        match self {
1309            SubgroupIdEncoding::Zero => 0x00,
1310            SubgroupIdEncoding::SameAsPrior => 0x01,
1311            SubgroupIdEncoding::PriorPlusOne => 0x02,
1312            SubgroupIdEncoding::Present => 0x03,
1313        }
1314    }
1315
1316    /// Whether resolving a Subgroup ID under this encoding needs the object
1317    /// before it on the stream.
1318    ///
1319    /// [`SubgroupIdEncoding::Zero`] does not, which is what makes it the one
1320    /// implicit form a stream's first object may use.
1321    pub fn references_prior(self) -> bool {
1322        matches!(self, SubgroupIdEncoding::SameAsPrior | SubgroupIdEncoding::PriorPlusOne)
1323    }
1324}
1325
1326/// One object on a draft-15 fetch stream, without its payload.
1327///
1328/// Draft-15 Section 10.4.4 gives fetch objects a leading Serialization Flags
1329/// byte that says which of the object's fields are on the wire; every field it
1330/// omits is taken from, or counted from, the object before it on the same
1331/// stream. An object therefore cannot be decoded on its own, and the fields
1332/// below are the resolved absolute values rather than whatever the wire spelled
1333/// out — reading them needs the running state a [`FetchObjectReader`] carries.
1334///
1335/// `serialization_flags` is kept beside the resolved values so that an object
1336/// re-encodes to the bytes it was decoded from. One object has as many
1337/// encodings as there are flag bytes that resolve to it, and choosing one on
1338/// the caller's behalf would rewrite a stream a relay is meant to forward
1339/// unchanged.
1340///
1341/// The bits, from Section 10.4.4, Tables 7 and 8:
1342/// - `& 0x03`: how the Subgroup ID is carried; see [`SubgroupIdEncoding`]
1343/// - `& 0x04`: Object ID field present, else the prior object's ID plus one
1344/// - `& 0x08`: Group ID field present, else the prior object's Group ID
1345/// - `& 0x10`: Publisher Priority field present, else the prior object's
1346/// - `& 0x20`: Extensions field present
1347/// - `& 0xc0`: unassigned, and Table 8 makes either bit a protocol violation
1348///
1349/// The payload is deliberately not part of this type: an object's declared
1350/// length is the last thing before its bytes, so a caller that forwards
1351/// payloads verbatim can read the framing and then move `payload_length` bytes
1352/// without ever copying them.
1353#[derive(Debug, Clone, PartialEq, Eq)]
1354pub struct FetchObjectHeader {
1355    /// Raw Serialization Flags byte, as decoded or as it is to be written.
1356    pub serialization_flags: u8,
1357    /// Resolved absolute Group ID.
1358    pub group_id: VarInt,
1359    /// Resolved absolute Subgroup ID.
1360    pub subgroup_id: VarInt,
1361    /// Resolved absolute Object ID.
1362    pub object_id: VarInt,
1363    /// Publisher priority for delivery ordering.
1364    pub publisher_priority: u8,
1365    /// Raw extension-header bytes, excluding the byte-length prefix that
1366    /// precedes them on the wire. Empty when the flags do not set the
1367    /// extensions bit, or when the block is present but zero-length. Opaque:
1368    /// [`FetchObjectReader::write_object_header`] re-emits the prefix and these
1369    /// bytes verbatim.
1370    pub extension_headers: Vec<u8>,
1371    /// Payload length as encoded on the wire. Zero when the object is a
1372    /// status-only object.
1373    pub payload_length: VarInt,
1374    /// Object status; `Some` when `payload_length == 0`.
1375    ///
1376    /// Section 10.4.4: "The Object Status field is only present if the Object
1377    /// Payload Length is zero." Typed to the statuses draft-15 assigns for the
1378    /// same reason [`SubgroupObject::object_status`] is: the field cannot hold
1379    /// a code the draft leaves unassigned, so the encoder needs no check and
1380    /// cannot emit an object the decoder would refuse.
1381    ///
1382    /// `None` on a zero-length object means the same as
1383    /// [`ObjectStatus::Normal`] and encodes as it; the wire field is not
1384    /// optional once `payload_length` is zero.
1385    pub object_status: Option<ObjectStatus>,
1386}
1387
1388impl FetchObjectHeader {
1389    /// How this object's Subgroup ID is carried, from the low two flag bits.
1390    pub fn subgroup_id_encoding(&self) -> SubgroupIdEncoding {
1391        SubgroupIdEncoding::from_flags(self.serialization_flags)
1392    }
1393
1394    /// Whether the Object ID is on the wire, rather than the prior object's
1395    /// ID plus one.
1396    pub fn has_object_id(&self) -> bool {
1397        self.serialization_flags & 0x04 != 0
1398    }
1399
1400    /// Whether the Group ID is on the wire, rather than the prior object's.
1401    pub fn has_group_id(&self) -> bool {
1402        self.serialization_flags & 0x08 != 0
1403    }
1404
1405    /// Whether the Publisher Priority is on the wire, rather than the prior
1406    /// object's.
1407    pub fn has_priority(&self) -> bool {
1408        self.serialization_flags & 0x10 != 0
1409    }
1410
1411    /// Whether an extensions block is on the wire.
1412    pub fn has_extensions(&self) -> bool {
1413        self.serialization_flags & 0x20 != 0
1414    }
1415
1416    /// Whether any of this object's fields is taken from the object before it
1417    /// on the stream.
1418    ///
1419    /// Draft-15 Section 10.4.4: "If the first Object in the FETCH response uses
1420    /// a flag that references fields in the prior Object, the Subscriber MUST
1421    /// close the session with a PROTOCOL_VIOLATION." Four of the flags do so —
1422    /// two of the Subgroup ID encodings, and the cleared state of the Object
1423    /// ID, Group ID and Priority bits, each of which means "the prior
1424    /// object's". The extensions bit does not: it is present or it is not, and
1425    /// nothing is inherited either way.
1426    pub fn references_prior_object(&self) -> bool {
1427        self.subgroup_id_encoding().references_prior()
1428            || !self.has_object_id()
1429            || !self.has_group_id()
1430            || !self.has_priority()
1431    }
1432
1433    /// The status this object resolves to.
1434    ///
1435    /// Section 10.4.4: "The Object Status field is only present if the Object
1436    /// Payload Length is zero." An object declaring a length is therefore
1437    /// [`ObjectStatus::Normal`] whatever [`Self::object_status`] holds, on the
1438    /// reading Section 10.2.1.1 gives Normal: "This status is implicit for any
1439    /// non-zero length object."
1440    pub fn status(&self) -> ObjectStatus {
1441        if self.payload_length.into_inner() == 0 {
1442            self.object_status.unwrap_or(ObjectStatus::Normal)
1443        } else {
1444            ObjectStatus::Normal
1445        }
1446    }
1447
1448    /// Whether this object's status is allowed to carry the extension headers
1449    /// it has.
1450    ///
1451    /// The fetch half of the rule [`SubgroupObject::extensions_permitted`]
1452    /// answers, and the same one: Section 10.4.4 builds a fetch object's
1453    /// Extensions field out of the structure Section 10.2.1.2 defines, and that
1454    /// section is where the rule sits — "Any Object with status Normal can have
1455    /// extension headers. If an endpoint receives extension headers on Objects
1456    /// with status that is not Normal, it MUST close the session with a
1457    /// PROTOCOL_VIOLATION."
1458    /// Draft-15 is the only draft where a fetch object can state this
1459    /// violation, which is why no counterpart to this exists on drafts 16 and
1460    /// later rather than one that is always `true`. Section 10.4.4 gives this
1461    /// draft's fetch object an Object Status field — "The Object Status field
1462    /// is only present if the Object Payload Length is zero" — and its
1463    /// extensions bit, Table 8's `0x20`, is independent of every other flag, so
1464    /// the two can appear together. Drafts 16 and later remove the field
1465    /// outright, draft-16 Section 10.2.1.1: "The Object Status is a field that
1466    /// is only present in objects that are delivered via a SUBSCRIPTION, and is
1467    /// absent in Objects delivered via a FETCH." A fetch object there has no
1468    /// status to disagree with, so the rule has nothing to bite on.
1469    ///
1470    /// [`FetchObjectReader`] reports this rather than refusing it, for the
1471    /// reason [`SubgroupObject::extensions_permitted`] sets out: the frame is
1472    /// well formed and merely non-conforming, and a reader that refused it
1473    /// could not reproduce a capture containing one.
1474    pub fn extensions_permitted(&self) -> bool {
1475        self.extension_headers.is_empty() || self.status() == ObjectStatus::Normal
1476    }
1477
1478    /// Serialize this Object's framing, writing only the fields its own
1479    /// Serialization Flags announce.
1480    ///
1481    /// The inverse of [`FetchObjectReader::read_object_header`], and the reason
1482    /// it is fallible where the subgroup form's is not: this type holds every
1483    /// field resolved, so the flags rather than the values decide what goes on
1484    /// the wire. A Group ID held with the 0x08 bit clear is *not* written and
1485    /// the reader takes the Object as sharing its predecessor's group — which
1486    /// is silent data loss when the two differ, and correct when they do not.
1487    /// Nothing here can tell those apart, so the caller settles it by choosing
1488    /// the flags, and [`FetchObjectWriter`] is what chooses them against a
1489    /// predecessor.
1490    ///
1491    /// # Errors
1492    ///
1493    /// [`CodecError::InvalidField`] for a flags byte with either of the two
1494    /// bits Section 10.4.4 leaves unassigned, which is the same value
1495    /// [`FetchObjectReader::read_object_header`] refuses to read — the field is
1496    /// one fixed byte and not a variable-length integer, so 0x40 and 0x80 are
1497    /// not wider spellings of anything.
1498    pub fn encode(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1499        if self.serialization_flags & 0xc0 != 0 {
1500            return Err(CodecError::InvalidField);
1501        }
1502        buf.put_u8(self.serialization_flags);
1503        // Field order is Section 10.4.4's, Figure 30: Group ID, Subgroup ID,
1504        // Object ID, Priority, Extensions.
1505        if self.has_group_id() {
1506            self.group_id.encode(buf);
1507        }
1508        if self.subgroup_id_encoding() == SubgroupIdEncoding::Present {
1509            self.subgroup_id.encode(buf);
1510        }
1511        if self.has_object_id() {
1512            self.object_id.encode(buf);
1513        }
1514        if self.has_priority() {
1515            buf.put_u8(self.publisher_priority);
1516        }
1517        if self.has_extensions() {
1518            VarInt::from_usize(self.extension_headers.len()).encode(buf);
1519            buf.put_slice(&self.extension_headers);
1520        }
1521        self.payload_length.encode(buf);
1522        // A zero-length Object carries a status varint after the length, and a
1523        // non-zero-length one carries none: the reader reads the field on
1524        // exactly that test, so writing it on any other would put a byte on the
1525        // wire the reader would take for payload.
1526        if self.payload_length.into_inner() == 0 {
1527            VarInt::from_u64(self.object_status.unwrap_or(ObjectStatus::Normal).as_u64())
1528                .map_err(|_| CodecError::InvalidField)?
1529                .encode(buf);
1530        }
1531        Ok(())
1532    }
1533}
1534
1535/// Re-encodes resolved fetch Objects onto one FETCH stream.
1536///
1537/// The exact inverse of [`FetchObjectReader`], and it exists for one caller:
1538/// something that has read a stream and is writing a different stream from the
1539/// same Objects. Draft-15 Section 10.4.4 lets an Object leave out its Group
1540/// ID, Object ID, Subgroup ID and Priority and take the prior Object's, so
1541/// removing an Object changes what the Objects after it are read against: a
1542/// field the survivor left off has to appear, and a flag bit with it.
1543///
1544/// Every field draft-15 puts on the wire is the absolute value rather than a
1545/// difference — the deltas arrive at draft-18. What is stateful is the
1546/// *omission*, and that alone makes removal a re-encode rather than a deletion.
1547///
1548/// # Why this is not a general encoder
1549///
1550/// Every Object it writes came off a stream, so the caller holds the Object's
1551/// own header with both its resolved values and the flags it arrived under.
1552/// Those flags are the preference: wherever the original shape still says the
1553/// same thing against the new predecessor it is kept, so a stream with nothing
1554/// removed is reproduced byte for byte.
1555#[derive(Debug, Clone, Default)]
1556pub struct FetchObjectWriter {
1557    prior: Option<PriorFetchObject>,
1558}
1559
1560impl FetchObjectWriter {
1561    /// A writer positioned before the first Object of a fetch stream, with no
1562    /// prior Object for anything to be written against.
1563    pub fn new() -> Self {
1564        Self::default()
1565    }
1566
1567    /// The header that encodes `original`'s resolved values against everything
1568    /// written so far.
1569    ///
1570    /// Does not advance the writer — [`Self::write_object_header`] is the call
1571    /// that does both.
1572    ///
1573    /// # Errors
1574    ///
1575    /// [`CodecError::InvalidField`] for a flags byte with an unassigned bit.
1576    pub fn header_for(
1577        &self,
1578        original: &FetchObjectHeader,
1579    ) -> Result<FetchObjectHeader, CodecError> {
1580        if original.serialization_flags & 0xc0 != 0 {
1581            return Err(CodecError::InvalidField);
1582        }
1583        let group_id = original.group_id.into_inner();
1584        let subgroup_id = original.subgroup_id.into_inner();
1585        let object_id = original.object_id.into_inner();
1586
1587        // Each field is written when the Object wrote it, and written anyway
1588        // when leaving it off would now say something else. Keeping the
1589        // Object's own choice is what reproduces an untouched stream byte for
1590        // byte; the second half is what a removal forces.
1591        let mut flags = 0u8;
1592        if original.has_group_id() || self.prior.map(|p| p.group_id) != Some(group_id) {
1593            flags |= 0x08;
1594        }
1595        if original.has_object_id()
1596            || self.prior.and_then(|p| p.object_id.checked_add(1)) != Some(object_id)
1597        {
1598            flags |= 0x04;
1599        }
1600        if original.has_priority()
1601            || self.prior.map(|p| p.publisher_priority) != Some(original.publisher_priority)
1602        {
1603            flags |= 0x10;
1604        }
1605        if original.has_extensions() {
1606            flags |= 0x20;
1607        }
1608        flags |= self.subgroup_bits(original, subgroup_id).as_bits();
1609
1610        Ok(FetchObjectHeader {
1611            serialization_flags: flags,
1612            group_id: original.group_id,
1613            subgroup_id: original.subgroup_id,
1614            object_id: original.object_id,
1615            publisher_priority: original.publisher_priority,
1616            extension_headers: original.extension_headers.clone(),
1617            payload_length: original.payload_length,
1618            object_status: original.object_status,
1619        })
1620    }
1621
1622    /// Which Subgroup ID encoding says `subgroup_id` against this predecessor.
1623    ///
1624    /// The Object's own encoding is tried first, so a run that inherited its
1625    /// Subgroup ID keeps inheriting it and its bytes do not move. Only when the
1626    /// predecessor changed under it is a different one chosen, and then the
1627    /// cheapest that names the right number.
1628    fn subgroup_bits(&self, original: &FetchObjectHeader, subgroup_id: u64) -> SubgroupIdEncoding {
1629        let inherits = self.prior.map(|p| p.subgroup_id) == Some(subgroup_id);
1630        let successor =
1631            self.prior.is_some_and(|p| p.subgroup_id.checked_add(1) == Some(subgroup_id));
1632        let kept = match original.subgroup_id_encoding() {
1633            SubgroupIdEncoding::Zero if subgroup_id == 0 => Some(SubgroupIdEncoding::Zero),
1634            SubgroupIdEncoding::SameAsPrior if inherits => Some(SubgroupIdEncoding::SameAsPrior),
1635            SubgroupIdEncoding::PriorPlusOne if successor => Some(SubgroupIdEncoding::PriorPlusOne),
1636            SubgroupIdEncoding::Present => Some(SubgroupIdEncoding::Present),
1637            _ => None,
1638        };
1639        match kept {
1640            Some(encoding) => encoding,
1641            None if subgroup_id == 0 => SubgroupIdEncoding::Zero,
1642            None if inherits => SubgroupIdEncoding::SameAsPrior,
1643            None if successor => SubgroupIdEncoding::PriorPlusOne,
1644            None => SubgroupIdEncoding::Present,
1645        }
1646    }
1647
1648    /// Encode `original`'s resolved values against everything written so far
1649    /// and advance.
1650    ///
1651    /// Writes the framing only. The payload is `payload_length` bytes and is
1652    /// the caller's to copy, unchanged.
1653    ///
1654    /// # Errors
1655    ///
1656    /// [`CodecError::InvalidField`] for a flags byte with an unassigned bit;
1657    /// the writer is left untouched when this happens.
1658    pub fn write_object_header(
1659        &mut self,
1660        original: &FetchObjectHeader,
1661        out: &mut impl BufMut,
1662    ) -> Result<FetchObjectHeader, CodecError> {
1663        let header = self.header_for(original)?;
1664        header.encode(out)?;
1665        self.advance(&header);
1666        Ok(header)
1667    }
1668
1669    /// Record what was written as the predecessor of whatever comes next.
1670    ///
1671    /// Every field draft-15 puts on a fetch object's wire is the absolute
1672    /// value, so this reads them straight off the header rather than resolving
1673    /// anything.
1674    ///
1675    /// Public because a re-emitting caller has a second way of putting a frame
1676    /// on the wire: when the framing it arrived in still encodes the same
1677    /// meaning against the frame before it, its own bytes are forwarded
1678    /// untouched — no header is produced and nothing is copied. The writer
1679    /// still has to move, or the frame after it is encoded against a
1680    /// predecessor one frame stale. `written` is then the Object's own header, which is
1681    /// what was put on the wire.
1682    pub fn advance(&mut self, written: &FetchObjectHeader) {
1683        self.prior = Some(PriorFetchObject {
1684            group_id: written.group_id.into_inner(),
1685            subgroup_id: written.subgroup_id.into_inner(),
1686            object_id: written.object_id.into_inner(),
1687            publisher_priority: written.publisher_priority,
1688        });
1689    }
1690}
1691
1692/// The fields a draft-15 fetch object leaves to its successor to inherit.
1693#[derive(Debug, Clone, Copy)]
1694struct PriorFetchObject {
1695    group_id: u64,
1696    subgroup_id: u64,
1697    object_id: u64,
1698    publisher_priority: u8,
1699}
1700
1701/// Stateful reader/writer for the objects on one draft-15 fetch stream.
1702///
1703/// Holds the fields of the object last read or written, which is what the next
1704/// object's Serialization Flags may refer to. One reader belongs to one stream:
1705/// draft-15 Section 10.4.4 counts "the prior Object" along the stream, so
1706/// sharing a reader between streams, or restarting one mid-stream, resolves
1707/// later objects onto the wrong group, subgroup, ID or priority without
1708/// producing an error anywhere.
1709///
1710/// A fresh reader has no prior object, which is exactly the state in which the
1711/// draft's protocol violation applies — see
1712/// [`FetchObjectHeader::references_prior_object`].
1713#[derive(Debug, Clone, Default)]
1714pub struct FetchObjectReader {
1715    prior: Option<PriorFetchObject>,
1716    /// The Prior gap conditions the stream's Objects show.
1717    gaps: FetchGaps,
1718}
1719
1720impl FetchObjectReader {
1721    /// Build a reader for the objects following a fetch stream's header.
1722    pub fn new() -> Self {
1723        Self::default()
1724    }
1725
1726    /// Decode the next object's framing, leaving its payload in `buf`.
1727    ///
1728    /// Consumes the Serialization Flags byte, whichever of the Group ID,
1729    /// Subgroup ID, Object ID, Priority and Extensions fields that byte says are
1730    /// present, the Object Payload Length, and the Object Status when that
1731    /// length is zero. The declared payload bytes are left where they are, so a
1732    /// caller can forward them without a copy; skipping them is the caller's
1733    /// job, and skipping the wrong number of them desynchronises every later
1734    /// object on the stream.
1735    ///
1736    /// Errors with [`CodecError::InvalidField`] when the flags set either bit
1737    /// draft-15 Section 10.4.4, Table 8 leaves unassigned, when the stream's
1738    /// first object inherits from an object that does not exist, when an
1739    /// inherited value cannot be represented, or when the Object Status is a
1740    /// code the draft does not assign.
1741    ///
1742    /// Refuses, with [`CodecError::MalformedTrack`], the Prior gap conditions an
1743    /// Object shows alone or after the Objects before it on the stream, as
1744    /// [`FetchGaps`] does, and an extension block that does not parse as
1745    /// [`crate::object_properties::parse_object_properties`] does.
1746    pub fn read_object_header(
1747        &mut self,
1748        buf: &mut impl Buf,
1749    ) -> Result<FetchObjectHeader, CodecError> {
1750        let header = self.read_frame_header(buf)?;
1751        self.gaps.check(
1752            header.group_id.into_inner(),
1753            header.object_id.into_inner(),
1754            &header.extension_headers,
1755        )?;
1756        Ok(header)
1757    }
1758
1759    /// [`Self::read_object_header`] without the Prior gap conditions and without reading
1760    /// inside the extension block, for the facade's frame and meta reads, which
1761    /// a forwarder uses.
1762    pub(crate) fn read_frame_header(
1763        &mut self,
1764        buf: &mut impl Buf,
1765    ) -> Result<FetchObjectHeader, CodecError> {
1766        if buf.remaining() < 1 {
1767            return Err(CodecError::UnexpectedEnd);
1768        }
1769        // Section 10.4.4 writes the field as "Serialization Flags (8)": one
1770        // fixed byte, not a varint. The two readings only part on the values
1771        // Table 8 forbids, which is what makes the difference easy to miss —
1772        // 0x40 and 0x80 are the two- and four-byte varint prefixes, so a varint
1773        // reader consumes the fields after them as part of the flags and
1774        // reports a plausible object instead of the violation.
1775        let serialization_flags = buf.get_u8();
1776        if serialization_flags & 0xc0 != 0 {
1777            return Err(CodecError::InvalidField);
1778        }
1779
1780        let subgroup_encoding = SubgroupIdEncoding::from_flags(serialization_flags);
1781        let has_object_id = serialization_flags & 0x04 != 0;
1782        let has_group_id = serialization_flags & 0x08 != 0;
1783        let has_priority = serialization_flags & 0x10 != 0;
1784        let has_extensions = serialization_flags & 0x20 != 0;
1785
1786        // The first object on the stream has nothing to inherit from, and the
1787        // draft's answer to being asked anyway is to close the session rather
1788        // than to invent a zero.
1789        let inherits = subgroup_encoding.references_prior()
1790            || !has_object_id
1791            || !has_group_id
1792            || !has_priority;
1793        if inherits && self.prior.is_none() {
1794            return Err(CodecError::InvalidField);
1795        }
1796        let prior = self.prior;
1797
1798        // Field order is Section 10.4.4's, Figure 30: Group ID, then Subgroup
1799        // ID, then Object ID, then Priority, then Extensions. Only the fields
1800        // the flags announce are on the wire, so resolving out of order would
1801        // read one field's bytes as another's.
1802        let group_id = if has_group_id {
1803            VarInt::decode(buf)?
1804        } else {
1805            let prior = prior.ok_or(CodecError::InvalidField)?;
1806            VarInt::from_u64(prior.group_id).map_err(|_| CodecError::InvalidField)?
1807        };
1808
1809        let subgroup_id = match subgroup_encoding {
1810            SubgroupIdEncoding::Zero => VarInt::from_usize(0),
1811            SubgroupIdEncoding::SameAsPrior => {
1812                let prior = prior.ok_or(CodecError::InvalidField)?;
1813                VarInt::from_u64(prior.subgroup_id).map_err(|_| CodecError::InvalidField)?
1814            }
1815            SubgroupIdEncoding::PriorPlusOne => {
1816                let prior = prior.ok_or(CodecError::InvalidField)?;
1817                let next = prior.subgroup_id.checked_add(1).ok_or(CodecError::InvalidField)?;
1818                VarInt::from_u64(next).map_err(|_| CodecError::InvalidField)?
1819            }
1820            SubgroupIdEncoding::Present => VarInt::decode(buf)?,
1821        };
1822
1823        let object_id = if has_object_id {
1824            VarInt::decode(buf)?
1825        } else {
1826            let prior = prior.ok_or(CodecError::InvalidField)?;
1827            let next = prior.object_id.checked_add(1).ok_or(CodecError::InvalidField)?;
1828            VarInt::from_u64(next).map_err(|_| CodecError::InvalidField)?
1829        };
1830
1831        let publisher_priority = if has_priority {
1832            if buf.remaining() < 1 {
1833                return Err(CodecError::UnexpectedEnd);
1834            }
1835            buf.get_u8()
1836        } else {
1837            prior.ok_or(CodecError::InvalidField)?.publisher_priority
1838        };
1839
1840        let extension_headers = if has_extensions {
1841            // Section 10.4.4 defers to Section 10.2.1.2 here, so the block is
1842            // the same byte-length-prefixed opaque blob subgroup objects carry.
1843            let ext_len = VarInt::decode(buf)?.into_inner() as usize;
1844            crate::types::read_bytes(buf, ext_len)?
1845        } else {
1846            Vec::new()
1847        };
1848
1849        let payload_length = VarInt::decode(buf)?;
1850        let object_status = if payload_length.into_inner() == 0 {
1851            Some(decoded_status(VarInt::decode(buf)?.into_inner())?)
1852        } else {
1853            None
1854        };
1855        self.prior = Some(PriorFetchObject {
1856            group_id: group_id.into_inner(),
1857            subgroup_id: subgroup_id.into_inner(),
1858            object_id: object_id.into_inner(),
1859            publisher_priority,
1860        });
1861
1862        Ok(FetchObjectHeader {
1863            serialization_flags,
1864            group_id,
1865            subgroup_id,
1866            object_id,
1867            publisher_priority,
1868            extension_headers,
1869            payload_length,
1870            object_status,
1871        })
1872    }
1873
1874    /// Serialize an object's framing, leaving its payload to the caller.
1875    ///
1876    /// Writes exactly the fields `header.serialization_flags` announces, and
1877    /// stops after the Object Payload Length — or, when that length is zero,
1878    /// after the Object Status. The caller appends `payload_length` payload
1879    /// bytes; the length is already on the wire by then, so appending a
1880    /// different number of them frames an object no reader can parse.
1881    ///
1882    /// The flags are taken as the authority on what reaches the wire, which
1883    /// means a header whose resolved fields disagree with its own flags cannot
1884    /// be written faithfully. Every such disagreement is refused with
1885    /// [`CodecError::InvalidField`] rather than resolved:
1886    ///
1887    /// - a field the flags omit whose value is not the one the omission
1888    ///   implies — a Group ID that is not the prior object's, a Subgroup ID
1889    ///   that is not what the two-bit encoding resolves to, an Object ID that
1890    ///   is not the prior object's plus one, a Priority that is not the prior
1891    ///   object's. Written anyway, each would arrive as the value the flags
1892    ///   imply, and the object the peer sees would be a different object.
1893    /// - extension bytes with the extensions bit clear, which would be dropped
1894    ///   in silence. Draft-15 Section 10.2.1.2 requires relays to forward
1895    ///   extensions they do not understand unchanged, so dropping them is not
1896    ///   a smaller loss than mis-stating an ID.
1897    /// - a status other than [`ObjectStatus::Normal`] on an object with a
1898    ///   non-zero payload length. Section 10.4.4 puts the status field on the
1899    ///   wire only when that length is zero, and Section 10.2.1.1 the
1900    ///   conformance side: "Any object with a status code other than zero MUST
1901    ///   have an empty payload." There is no such object to write.
1902    ///   [`ObjectStatus::Normal`] alongside a payload is not that case and is
1903    ///   accepted: it is the status the encoding elides for every object that
1904    ///   carries bytes, so stating it asks for exactly the bytes leaving it out
1905    ///   asks for.
1906    ///
1907    /// Not refused: extension bytes on an object whose resolved status is not
1908    /// [`ObjectStatus::Normal`], which Section 10.2.1.2 forbids. As on a
1909    /// subgroup stream this is the only writer a fetch object has, and the
1910    /// frame encodes and reads back exactly, so refusing it would cost the
1911    /// ability to reproduce a capture rather than prevent anything.
1912    ///
1913    /// Also refused: flags setting either bit Table 8 leaves unassigned, and a
1914    /// first object on the stream that inherits from an object that does not
1915    /// exist.
1916    ///
1917    /// Every check runs before a byte is written, so a refused header leaves
1918    /// `buf` untouched rather than half an object the next write would run
1919    /// into, and leaves the reader's prior-object state as it was.
1920    pub fn write_object_header(
1921        &mut self,
1922        header: &FetchObjectHeader,
1923        buf: &mut impl BufMut,
1924    ) -> Result<(), CodecError> {
1925        if header.serialization_flags & 0xc0 != 0 {
1926            return Err(CodecError::InvalidField);
1927        }
1928        if header.payload_length.into_inner() != 0
1929            && matches!(header.object_status, Some(s) if s != ObjectStatus::Normal)
1930        {
1931            return Err(CodecError::InvalidField);
1932        }
1933        if !header.has_extensions() && !header.extension_headers.is_empty() {
1934            return Err(CodecError::InvalidField);
1935        }
1936        // Extension headers beside a non-Normal status are not refused here:
1937        // they encode and read back exactly, this is the only writer a fetch
1938        // object has, and a capture containing one has to stay reproducible.
1939        if header.references_prior_object() && self.prior.is_none() {
1940            return Err(CodecError::InvalidField);
1941        }
1942        let prior = self.prior;
1943        if !header.has_group_id() {
1944            let prior = prior.ok_or(CodecError::InvalidField)?;
1945            if header.group_id.into_inner() != prior.group_id {
1946                return Err(CodecError::InvalidField);
1947            }
1948        }
1949        let subgroup_id = header.subgroup_id.into_inner();
1950        match header.subgroup_id_encoding() {
1951            SubgroupIdEncoding::Zero => {
1952                if subgroup_id != 0 {
1953                    return Err(CodecError::InvalidField);
1954                }
1955            }
1956            SubgroupIdEncoding::SameAsPrior => {
1957                let prior = prior.ok_or(CodecError::InvalidField)?;
1958                if subgroup_id != prior.subgroup_id {
1959                    return Err(CodecError::InvalidField);
1960                }
1961            }
1962            SubgroupIdEncoding::PriorPlusOne => {
1963                let prior = prior.ok_or(CodecError::InvalidField)?;
1964                let next = prior.subgroup_id.checked_add(1).ok_or(CodecError::InvalidField)?;
1965                if subgroup_id != next {
1966                    return Err(CodecError::InvalidField);
1967                }
1968            }
1969            SubgroupIdEncoding::Present => {}
1970        }
1971        if !header.has_object_id() {
1972            let prior = prior.ok_or(CodecError::InvalidField)?;
1973            let next = prior.object_id.checked_add(1).ok_or(CodecError::InvalidField)?;
1974            if header.object_id.into_inner() != next {
1975                return Err(CodecError::InvalidField);
1976            }
1977        }
1978        if !header.has_priority() {
1979            let prior = prior.ok_or(CodecError::InvalidField)?;
1980            if header.publisher_priority != prior.publisher_priority {
1981                return Err(CodecError::InvalidField);
1982            }
1983        }
1984
1985        buf.put_u8(header.serialization_flags);
1986        if header.has_group_id() {
1987            header.group_id.encode(buf);
1988        }
1989        if header.subgroup_id_encoding() == SubgroupIdEncoding::Present {
1990            header.subgroup_id.encode(buf);
1991        }
1992        if header.has_object_id() {
1993            header.object_id.encode(buf);
1994        }
1995        if header.has_priority() {
1996            buf.put_u8(header.publisher_priority);
1997        }
1998        if header.has_extensions() {
1999            VarInt::from_usize(header.extension_headers.len()).encode(buf);
2000            buf.put_slice(&header.extension_headers);
2001        }
2002        header.payload_length.encode(buf);
2003        if header.payload_length.into_inner() == 0 {
2004            // Zero length means a status object, and the status is not optional
2005            // on the wire; an unset one is Normal.
2006            let status = header.object_status.unwrap_or(ObjectStatus::Normal);
2007            VarInt::from_usize(status.as_u64() as usize).encode(buf);
2008        }
2009
2010        self.prior = Some(PriorFetchObject {
2011            group_id: header.group_id.into_inner(),
2012            subgroup_id,
2013            object_id: header.object_id.into_inner(),
2014            publisher_priority: header.publisher_priority,
2015        });
2016        Ok(())
2017    }
2018}
2019
2020#[cfg(test)]
2021mod tests {
2022    use super::*;
2023
2024    /// Canonically encoded subgroup stream vectors from
2025    /// `test-vectors/transport/draft15/codec/data-streams/subgroup.json`.
2026    /// `subgroup-explicit-subgroup-id` is omitted: it encodes group_id 100 as a
2027    /// two-byte varint, which does not survive a minimal-width re-encode.
2028    const VECTORS: &[&str] = &[
2029        // subgroup-single-object
2030        "100100800004deadbeef",
2031        // subgroup-two-objects
2032        "100100800004deadbeef0002cafe",
2033        // subgroup-no-priority
2034        "3001000004deadbeef",
2035        // subgroup-with-extensions
2036        "11010080000004deadbeef",
2037        // subgroup-nonempty-extensions
2038        "1101058000023c0104deadbeef",
2039        // subgroup-status-end-of-group
2040        "100105800004deadbeef000003",
2041        // subgroup-status-end-of-track
2042        "10010a800004deadbeef000004",
2043        // subgroup-extensions-two-objects-empty
2044        "11010080000004deadbeef000002cafe",
2045        // subgroup-extensions-two-objects-nonempty
2046        "1101058000023c0204deadbeef00023c0202cafe",
2047        // subgroup-end-of-group
2048        "120105800004deadbeef",
2049    ];
2050
2051    fn vi(v: u64) -> VarInt {
2052        VarInt::from_u64(v).unwrap()
2053    }
2054
2055    fn hex(s: &str) -> Vec<u8> {
2056        (0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap()).collect()
2057    }
2058
2059    /// Decode a whole subgroup stream: the header, then every object up to
2060    /// the end of the buffer.
2061    fn decode_all(bytes: &[u8]) -> (SubgroupHeader, Vec<SubgroupObject>) {
2062        let mut cursor = bytes;
2063        let header = SubgroupHeader::decode(&mut cursor)
2064            .unwrap_or_else(|e| panic!("header decode failed: {e:?}"));
2065        let mut reader = SubgroupObjectReader::new(&header);
2066        let mut objects = Vec::new();
2067        while cursor.has_remaining() {
2068            objects.push(
2069                reader
2070                    .read_object(&mut cursor)
2071                    .unwrap_or_else(|e| panic!("object {} decode failed: {e:?}", objects.len())),
2072            );
2073        }
2074        (header, objects)
2075    }
2076
2077    fn encode_all(header: &SubgroupHeader, objects: &[SubgroupObject]) -> Vec<u8> {
2078        let mut buf = Vec::new();
2079        header.encode(&mut buf);
2080        let mut writer = SubgroupObjectReader::new(header);
2081        for o in objects {
2082            writer.write_object(o, &mut buf).unwrap_or_else(|e| panic!("write failed: {e:?}"));
2083        }
2084        buf
2085    }
2086
2087    fn object(id: u64, extensions: Vec<u8>, payload: Vec<u8>) -> SubgroupObject {
2088        SubgroupObject {
2089            object_id: vi(id),
2090            extension_headers: extensions,
2091            payload_length: vi(payload.len() as u64),
2092            object_status: None,
2093            payload,
2094        }
2095    }
2096
2097    // ── Object ID deltas ────────────────────────────────────
2098
2099    #[test]
2100    fn two_objects_with_extensions_have_distinct_ids() {
2101        // Vector `subgroup-extensions-two-objects-empty`: two objects, each
2102        // carrying an empty extensions block and a delta of 0. The delta is
2103        // biased by one whether or not the extensions bit is set, so the IDs
2104        // are 0 and 1 — not 0 and 0.
2105        let bytes = hex("11010080000004deadbeef000002cafe");
2106        let (header, objects) = decode_all(&bytes);
2107        assert!(header.has_extensions());
2108        assert_eq!(objects.len(), 2);
2109        assert_eq!(objects[0].object_id.into_inner(), 0);
2110        assert_eq!(objects[1].object_id.into_inner(), 1);
2111        assert_eq!(objects[0].payload, hex("deadbeef"));
2112        assert_eq!(objects[1].payload, hex("cafe"));
2113        assert!(objects.iter().all(|o| o.extension_headers.is_empty()));
2114    }
2115
2116    #[test]
2117    fn deltas_resolve_sparse_ids() {
2118        let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
2119        let objects: Vec<_> =
2120            [3u64, 4, 40].iter().map(|&id| object(id, vec![], vec![0xAA, id as u8])).collect();
2121        let (_, decoded) = decode_all(&encode_all(&header, &objects));
2122        let ids: Vec<u64> = decoded.iter().map(|o| o.object_id.into_inner()).collect();
2123        assert_eq!(ids, vec![3, 4, 40]);
2124    }
2125
2126    #[test]
2127    fn write_rejects_non_increasing_ids() {
2128        let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
2129        let mut writer = SubgroupObjectReader::new(&header);
2130        let mut buf = Vec::new();
2131        writer.write_object(&object(7, vec![], vec![0x01]), &mut buf).unwrap();
2132        for id in [7u64, 6, 0] {
2133            let err = writer.write_object(&object(id, vec![], vec![0x01]), &mut buf).unwrap_err();
2134            assert!(matches!(err, CodecError::InvalidField), "id {id} gave {err:?}");
2135        }
2136    }
2137
2138    #[test]
2139    fn eliding_an_object_renumbers_its_successor() {
2140        let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
2141        let all: Vec<_> = (0..5u64).map(|id| object(id, vec![], vec![id as u8])).collect();
2142        for elided in 0..5u64 {
2143            let kept: Vec<_> =
2144                all.iter().filter(|o| o.object_id.into_inner() != elided).cloned().collect();
2145            let (_, decoded) = decode_all(&encode_all(&header, &kept));
2146            let ids: Vec<u64> = decoded.iter().map(|o| o.object_id.into_inner()).collect();
2147            let expected: Vec<u64> = (0..5u64).filter(|&i| i != elided).collect();
2148            assert_eq!(ids, expected, "eliding object {elided}");
2149        }
2150    }
2151
2152    // ── Extension blocks ──────────────────────────────
2153
2154    #[test]
2155    fn extensions_blob_excludes_its_length_prefix() {
2156        // Vector `subgroup-extensions-two-objects-nonempty`: each
2157        // object carries a two-byte block, so the blob is those two bytes
2158        // with the `02` length prefix stripped.
2159        let bytes = hex("1101058000023c0204deadbeef00023c0202cafe");
2160        let (_, objects) = decode_all(&bytes);
2161        assert_eq!(objects.len(), 2);
2162        assert_eq!(objects[0].object_id.into_inner(), 0);
2163        assert_eq!(objects[1].object_id.into_inner(), 1);
2164        assert_eq!(objects[0].extension_headers, hex("3c02"));
2165        assert_eq!(objects[1].extension_headers, hex("3c02"));
2166        assert_eq!(objects[0].payload, hex("deadbeef"));
2167        assert_eq!(objects[1].payload, hex("cafe"));
2168    }
2169
2170    /// Read every object on a subgroup stream, returning the first error.
2171    fn first_error(stream: &str) -> Option<CodecError> {
2172        let bytes = hex(stream);
2173        let mut cursor = &bytes[..];
2174        let header = SubgroupHeader::decode(&mut cursor).unwrap();
2175        let mut reader = SubgroupObjectReader::new(&header);
2176        while cursor.has_remaining() {
2177            if let Err(e) = reader.read_object(&mut cursor) {
2178                return Some(e);
2179            }
2180        }
2181        None
2182    }
2183
2184    #[test]
2185    fn prior_gaps_are_checked_per_object_and_along_the_stream() {
2186        let malformed =
2187            |stream| matches!(first_error(stream), Some(CodecError::MalformedTrack { .. }));
2188        // Section 11.1: "An Object contains more than one instance of Prior
2189        // Group ID Gap".
2190        assert!(malformed("1101058000043c013c0104deadbeef"));
2191        // Section 11.2: Object 2's gap of 2 covers Object 0, received earlier.
2192        assert!(malformed("11010580000004deadbeef01023e0202cafe"));
2193        // Object 3's gap of 2 covers Objects 1 and 2, neither received.
2194        assert_eq!(first_error("11010580000004deadbeef02023e0202cafe"), None);
2195        // Equal Prior Group ID Gap values in one Group are fine.
2196        assert_eq!(first_error("1101058000023c0204deadbeef00023c0202cafe"), None);
2197    }
2198
2199    #[test]
2200    fn status_object_carries_its_extensions_block() {
2201        // Section 10.2.1.2 forbids extension headers on a non-Normal status, so
2202        // the whole-object reader refuses the frame and the meta reader frames it.
2203        let bytes = hex("1101058000023c010003");
2204        let mut cursor = &bytes[..];
2205        let header = SubgroupHeader::decode(&mut cursor).unwrap();
2206        let mut reader = SubgroupObjectReader::new(&header);
2207        assert!(matches!(reader.read_object(&mut &cursor[..]), Err(CodecError::InvalidField)));
2208        let meta = SubgroupObjectReader::new(&header).read_object_meta(&mut &cursor[..]).unwrap();
2209        assert_eq!(meta.extension_headers_len, 2);
2210        assert_eq!(meta.status, Some(3));
2211    }
2212
2213    // ── Object status ───────────────────────────────────────
2214
2215    /// A subgroup header with neither extensions nor an explicit subgroup ID,
2216    /// so an object on this stream is just `delta, payload_length, [status]`.
2217    fn plain_header() -> SubgroupHeader {
2218        SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap()
2219    }
2220
2221    /// A status object on a [`plain_header`] stream: object 0, empty payload,
2222    /// `code` as the status. Every code used here is one wire byte.
2223    fn subgroup_status_body(code: u64) -> Vec<u8> {
2224        let mut buf = vec![0x00, 0x00];
2225        VarInt::from_u64(code).unwrap().encode(&mut buf);
2226        buf
2227    }
2228
2229    fn status_object(status: Option<ObjectStatus>) -> SubgroupObject {
2230        SubgroupObject {
2231            object_id: vi(0),
2232            extension_headers: vec![],
2233            payload_length: vi(0),
2234            object_status: status,
2235            payload: vec![],
2236        }
2237    }
2238
2239    /// A status datagram carrying `code`. Type 0x20 sets the status flag and
2240    /// leaves the object-id (0x04) and extensions (0x01) flags clear, so the
2241    /// layout is `type, track_alias, group_id, object_id, priority, status`.
2242    fn datagram_status_bytes(code: u64) -> Vec<u8> {
2243        let mut buf = vec![0x20, 0x01, 0x02, 0x03, 0x80];
2244        VarInt::from_u64(code).unwrap().encode(&mut buf);
2245        buf
2246    }
2247
2248    fn status_datagram(status: Option<ObjectStatus>) -> DatagramHeader {
2249        DatagramHeader {
2250            datagram_type: 0x20,
2251            track_alias: vi(1),
2252            group_id: vi(2),
2253            object_id: vi(3),
2254            publisher_priority: Some(0x80),
2255            extension_headers: vec![],
2256            object_status: status,
2257        }
2258    }
2259
2260    /// Every status the type can hold reaches the wire as its own code and
2261    /// comes back unchanged, through both subgroup readers and the datagram.
2262    /// Fails if `write_object` writes any status other than the object's own.
2263    #[test]
2264    fn assigned_statuses_round_trip() {
2265        let header = plain_header();
2266        for &status in ObjectStatus::ALL {
2267            let mut bytes = Vec::new();
2268            SubgroupObjectReader::new(&header)
2269                .write_object(&status_object(Some(status)), &mut bytes)
2270                .unwrap();
2271            assert_eq!(
2272                bytes,
2273                subgroup_status_body(status.as_u64()),
2274                "{status:?} on the subgroup wire"
2275            );
2276
2277            let decoded = SubgroupObjectReader::new(&header)
2278                .read_object(&mut &bytes[..])
2279                .unwrap_or_else(|e| panic!("{status:?} was written and then refused: {e:?}"));
2280            assert_eq!(decoded.object_status, Some(status), "{status:?} through read_object");
2281
2282            let meta = SubgroupObjectReader::new(&header)
2283                .read_object_meta(&mut &bytes[..])
2284                .unwrap_or_else(|e| panic!("{status:?} was written and then refused: {e:?}"));
2285            assert_eq!(meta.status, Some(status.as_u64()), "{status:?} through read_object_meta");
2286
2287            let datagram = status_datagram(Some(status));
2288            let mut bytes = Vec::new();
2289            datagram.encode(&mut bytes);
2290            assert_eq!(
2291                bytes,
2292                datagram_status_bytes(status.as_u64()),
2293                "{status:?} on the datagram wire"
2294            );
2295            let decoded = DatagramHeader::decode(&mut &bytes[..]).unwrap_or_else(|e| {
2296                panic!("{status:?} datagram was written and then refused: {e:?}")
2297            });
2298            assert_eq!(decoded, datagram, "{status:?} datagram round trip");
2299        }
2300    }
2301
2302    /// A datagram whose type byte sets the status flag always carries a status
2303    /// field, because the flag is what puts the field on the wire — an unset
2304    /// `object_status` writes Normal rather than nothing. Fails if such a
2305    /// datagram stops before its status field.
2306    #[test]
2307    fn a_status_datagram_without_a_status_encodes_normal() {
2308        let mut bytes = Vec::new();
2309        status_datagram(None).encode(&mut bytes);
2310        assert_eq!(bytes, datagram_status_bytes(ObjectStatus::Normal.as_u64()));
2311        let decoded = DatagramHeader::decode(&mut &bytes[..])
2312            .unwrap_or_else(|e| panic!("own output refused: {e:?}"));
2313        assert_eq!(decoded.object_status, Some(ObjectStatus::Normal));
2314    }
2315
2316    /// The codes this draft's decoders accept are exactly the codes its
2317    /// encoders can emit.
2318    ///
2319    /// The sweep covers `0x00..=0x3f`, the whole one-byte varint range, so it
2320    /// contains every code draft-15 assigns, the gap inside that range (0x2)
2321    /// and the codes later drafts moved (0x1, 0x5). The expected set is read
2322    /// from [`ObjectStatus::ALL`] rather than written out here, so reassigning
2323    /// a code moves both halves of the test at once.
2324    ///
2325    /// Fails if `ObjectStatus::from_u64` answers `Some` for a code not in
2326    /// `ALL`, such as 0x05, which drafts 07-10 assigned and draft-15 does not.
2327    #[test]
2328    fn the_wire_accepts_exactly_what_the_type_can_hold() {
2329        let header = plain_header();
2330        for code in 0x00u64..=0x3f {
2331            let assigned = ObjectStatus::ALL.iter().any(|s| s.as_u64() == code);
2332            let body = subgroup_status_body(code);
2333
2334            let read = SubgroupObjectReader::new(&header).read_object(&mut &body[..]);
2335            assert_eq!(
2336                read.is_ok(),
2337                assigned,
2338                "subgroup read_object on status {code:#x}: {read:?}"
2339            );
2340
2341            let meta = SubgroupObjectReader::new(&header).read_object_meta(&mut &body[..]);
2342            assert_eq!(
2343                meta.is_ok(),
2344                assigned,
2345                "subgroup read_object_meta on status {code:#x}: {meta:?}"
2346            );
2347
2348            let datagram = DatagramHeader::decode(&mut &datagram_status_bytes(code)[..]);
2349            assert_eq!(
2350                datagram.is_ok(),
2351                assigned,
2352                "status datagram on status {code:#x}: {datagram:?}"
2353            );
2354
2355            // The other direction: an accepted code is one the encoders can
2356            // reach, and they reach it with exactly these bytes.
2357            if assigned {
2358                let status = ObjectStatus::from_u64(code).unwrap();
2359                let mut bytes = Vec::new();
2360                SubgroupObjectReader::new(&header)
2361                    .write_object(&status_object(Some(status)), &mut bytes)
2362                    .unwrap();
2363                assert_eq!(bytes, body, "write_object on status {code:#x}");
2364                let mut bytes = Vec::new();
2365                status_datagram(Some(status)).encode(&mut bytes);
2366                assert_eq!(
2367                    bytes,
2368                    datagram_status_bytes(code),
2369                    "datagram encode on status {code:#x}"
2370                );
2371            }
2372        }
2373    }
2374
2375    // ── Re-encoding ─────────────────────────────────────────
2376
2377    #[test]
2378    fn vectors_re_encode_byte_identically() {
2379        for vector in VECTORS {
2380            let bytes = hex(vector);
2381            let (header, objects) = decode_all(&bytes);
2382            assert_eq!(encode_all(&header, &objects), bytes, "[{vector}] re-encode");
2383        }
2384    }
2385
2386    // ── Payload-free framing ────────────────────────────────
2387
2388    #[test]
2389    fn meta_matches_read_object() {
2390        for vector in VECTORS {
2391            let bytes = hex(vector);
2392            let mut cursor = &bytes[..];
2393            let header = SubgroupHeader::decode(&mut cursor).unwrap();
2394            let mut full_reader = SubgroupObjectReader::new(&header);
2395            let mut meta_reader = SubgroupObjectReader::new(&header);
2396            let mut full_cursor = cursor;
2397            let mut meta_cursor = cursor;
2398            while meta_cursor.has_remaining() {
2399                let before = meta_cursor.remaining();
2400                let object = full_reader.read_object(&mut full_cursor).unwrap();
2401                let meta = meta_reader.read_object_meta(&mut meta_cursor).unwrap();
2402                assert_eq!(meta.object_id, object.object_id.into_inner(), "[{vector}]");
2403                assert_eq!(
2404                    meta.extension_headers_len,
2405                    object.extension_headers.len() as u64,
2406                    "[{vector}]"
2407                );
2408                assert_eq!(meta.payload_length, object.payload_length.into_inner(), "[{vector}]");
2409                assert_eq!(
2410                    meta.status,
2411                    object.object_status.map(ObjectStatus::as_u64),
2412                    "[{vector}]"
2413                );
2414                assert_eq!(meta.wire_len, (before - meta_cursor.remaining()) as u64, "[{vector}]");
2415                assert_eq!(full_cursor.remaining(), meta_cursor.remaining(), "[{vector}]");
2416            }
2417        }
2418    }
2419
2420    /// A fetch object reports the extensions-beside-a-status rule the subgroup
2421    /// and datagram carriers already reported.
2422    /// Draft-15 is the only draft on which this is expressible: from
2423    /// draft-16 Section 10.2.1.1 the Object Status field is "absent in Objects
2424    /// delivered via a FETCH", so there is no status for extensions to sit
2425    /// beside. All four objects below are well formed and all four decode — the
2426    /// predicate is the only thing that separates them, which is the whole
2427    /// point of reporting rather than refusing.
2428    ///
2429    /// Inverting the predicate's `||` to `&&` fails this at the second case,
2430    /// `[1c0000000003] End of Group with no extensions is fine: expected true,
2431    /// got false`, and fails
2432    /// `fetch_object_status_is_normal_whenever_a_payload_is_declared` beside
2433    /// it.
2434    #[test]
2435    fn fetch_objects_report_extensions_beside_a_non_normal_status() {
2436        // flags 0x3c: Subgroup ID zero, Group ID / Object ID / Priority and an
2437        // extensions block all present. flags 0x1c is the same without the
2438        // extensions block.
2439        let cases: [(&str, bool, &str); 4] = [
2440            (
2441                "3c0000000210010003",
2442                false,
2443                "a two-byte extension block beside End of Group is the violation",
2444            ),
2445            ("1c0000000003", true, "End of Group with no extensions is fine"),
2446            (
2447                "3c000000000003",
2448                true,
2449                "a present but zero-length block carries nothing, so nothing is beside the status",
2450            ),
2451            ("3c00000002100104", true, "a normal object may carry extensions"),
2452        ];
2453
2454        for (vector, permitted, why) in cases {
2455            let bytes = hex(vector);
2456            let mut cursor = &bytes[..];
2457            let header = FetchObjectReader::new()
2458                .read_object_header(&mut cursor)
2459                .unwrap_or_else(|e| panic!("[{vector}] {why}: decode failed with {e:?}"));
2460            assert_eq!(
2461                header.extensions_permitted(),
2462                permitted,
2463                "[{vector}] {why}: expected {permitted}, got {}",
2464                header.extensions_permitted(),
2465            );
2466        }
2467    }
2468
2469    /// A fetch object's status resolves the way a subgroup object's does: the
2470    /// field is on the wire only under a zero payload length, so an object
2471    /// declaring bytes is Normal whatever the field would have said.
2472    #[test]
2473    fn fetch_object_status_is_normal_whenever_a_payload_is_declared() {
2474        let bytes = hex("3c00000002100104");
2475        let mut cursor = &bytes[..];
2476        let header = FetchObjectReader::new().read_object_header(&mut cursor).unwrap();
2477        assert_eq!(header.object_status, None, "no status field follows a non-zero length");
2478        assert_eq!(header.status(), ObjectStatus::Normal);
2479
2480        // Assembled by hand rather than decoded: the wire cannot put a status
2481        // beside a payload, but the struct's fields are public and a caller
2482        // porting an object across drafts can set both.
2483        let contradictory =
2484            FetchObjectHeader { object_status: Some(ObjectStatus::EndOfGroup), ..header };
2485        assert_eq!(
2486            contradictory.status(),
2487            ObjectStatus::Normal,
2488            "a declared payload wins over a status the wire could not have carried",
2489        );
2490        assert!(contradictory.extensions_permitted());
2491    }
2492
2493    #[test]
2494    fn short_buffers_report_unexpected_end() {
2495        let bytes = hex("1101058000023c0204deadbeef00023c0202cafe");
2496        let mut cursor = &bytes[..];
2497        let header = SubgroupHeader::decode(&mut cursor).unwrap();
2498        let objects_start = bytes.len() - cursor.len();
2499        for cut in objects_start..bytes.len() {
2500            let mut reader = SubgroupObjectReader::new(&header);
2501            let mut meta_reader = SubgroupObjectReader::new(&header);
2502            let mut cursor = &bytes[objects_start..cut];
2503            let mut meta_cursor = cursor;
2504            while cursor.has_remaining() {
2505                if let Err(err) = reader.read_object(&mut cursor) {
2506                    assert!(
2507                        matches!(err, CodecError::UnexpectedEnd | CodecError::VarInt(_)),
2508                        "cut {cut} gave {err:?}"
2509                    );
2510                    break;
2511                }
2512            }
2513            while meta_cursor.has_remaining() {
2514                if let Err(err) = meta_reader.read_object_meta(&mut meta_cursor) {
2515                    assert!(
2516                        matches!(err, CodecError::UnexpectedEnd | CodecError::VarInt(_)),
2517                        "cut {cut} gave {err:?}"
2518                    );
2519                    break;
2520                }
2521            }
2522        }
2523    }
2524}
2525
2526#[cfg(test)]
2527mod datagram_gap_tests {
2528    use super::*;
2529
2530    fn datagram(group: usize, object: usize, block: &[u8]) -> Vec<u8> {
2531        let mut bytes = Vec::new();
2532        DatagramHeader {
2533            datagram_type: 0x01,
2534            track_alias: VarInt::from_usize(1),
2535            group_id: VarInt::from_usize(group),
2536            object_id: VarInt::from_usize(object),
2537            publisher_priority: Some(0x80),
2538            extension_headers: block.to_vec(),
2539            object_status: None,
2540        }
2541        .encode(&mut bytes);
2542        bytes.push(0xAA);
2543        bytes
2544    }
2545
2546    fn is_malformed_track(bytes: &[u8]) -> bool {
2547        matches!(
2548            DatagramHeader::decode_object(&mut &bytes[..]),
2549            Err(CodecError::MalformedTrack { .. })
2550        )
2551    }
2552
2553    #[test]
2554    fn decode_object_refuses_the_gap_conditions_one_datagram_shows() {
2555        for bytes in [
2556            datagram(1, 0, &[0x3C, 2]),
2557            datagram(5, 0, &[0x3E, 1]),
2558            datagram(5, 5, &[0x3C, 1, 0x3C, 1]),
2559        ] {
2560            assert!(is_malformed_track(&bytes));
2561            assert!(DatagramHeader::decode(&mut &bytes[..]).is_ok());
2562        }
2563        let (_, payload) =
2564            DatagramHeader::decode_object(&mut &datagram(2, 1, &[0x3C, 2])[..]).unwrap();
2565        assert_eq!(payload, vec![0xAA]);
2566    }
2567
2568    #[test]
2569    fn decode_object_refuses_a_payload_after_a_status_datagram() {
2570        let mut wire = Vec::new();
2571        DatagramHeader {
2572            datagram_type: 0x20,
2573            track_alias: VarInt::from_usize(1),
2574            group_id: VarInt::from_usize(0),
2575            object_id: VarInt::from_usize(0),
2576            publisher_priority: Some(0x80),
2577            extension_headers: Vec::new(),
2578            object_status: Some(ObjectStatus::Normal),
2579        }
2580        .encode(&mut wire);
2581        let (_, payload) = DatagramHeader::decode_object(&mut &wire[..]).unwrap();
2582        assert!(payload.is_empty());
2583        wire.push(0xAA);
2584        assert!(matches!(
2585            DatagramHeader::decode_object(&mut &wire[..]),
2586            Err(CodecError::PayloadNotPermitted { len: 1, .. })
2587        ));
2588    }
2589}