Skip to main content

moqtap_codec/draft16/
data_stream.rs

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