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