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