moqtap_codec/draft23/data_stream.rs
1//! Draft-23 data stream header encoding and decoding.
2//!
3//! Subgroup Type Flags: form 0b0XX1XXXX, so bit 4 is set and bit 7 is
4//! clear; the ranges are 0x10..0x1F, 0x30..0x3F, 0x50..0x5F, 0x70..0x7F.
5//! - bit 0 (0x01): PROPERTIES
6//! - bits 1-2 (0x06): SUBGROUP_ID_MODE (0=zero, 1=first_obj, 2=explicit, 3=reserved)
7//! - bit 3 (0x08): END_OF_GROUP
8//! - bit 5 (0x20): DEFAULT_PRIORITY (no priority byte)
9//! - bit 6 (0x40): FIRST_OBJECT
10//!
11//! Datagram Type Flags: 0b00X0XXXX, so bits 4, 6 and 7 are clear; the ranges
12//! are 0x00..0x0F and 0x20..0x2F.
13//! - bit 0 (0x01): PROPERTIES
14//! - bit 1 (0x02): END_OF_GROUP
15//! - bit 2 (0x04): ZERO_OBJECT_ID (object_id=0, field omitted)
16//! - bit 3 (0x08): DEFAULT_PRIORITY (no priority byte)
17//! - bit 5 (0x20): STATUS (status byte replaces payload)
18//!
19//! Section 11.3.1 and Section 11.2.1 each list the Type Flags values an
20//! endpoint refuses with INVALID_STREAM_HEADER, and the two lists differ: bit 4
21//! MUST be 1 on a subgroup header and 0 on a datagram. A non-minimally encoded
22//! Type Flags value is accepted on receive and never emitted.
23//!
24//! An Object has an empty payload unless its status is registered as
25//! permitting one (Section 11.1.1); [`SubgroupObjectReader::write_object`]
26//! refuses a payload its status forbids.
27//!
28//! Fetch stream: stream type 0x05 + Request ID, then entries whose leading
29//! Serialization Flags say which fields follow (Section 11.4.1).
30//! [`FetchObjectHeader`] decodes one entry and [`FetchObjectReader`] resolves
31//! its IDs. Table 8 has six indicators: four End of Range values and End of
32//! Group and End of Track.
33
34use bytes::{Buf, BufMut};
35
36use super::types::{ObjectStatus, PayloadPermission};
37use crate::error::CodecError;
38use crate::object_properties::{ObjectProperty, PropertyValue};
39use crate::varint::{Moqt18 as Wire, VarInt};
40
41/// Advance `buf` past `len` bytes without copying them.
42fn skip(buf: &mut impl Buf, len: u64) -> Result<(), CodecError> {
43 let len = usize::try_from(len).map_err(|_| CodecError::UnexpectedEnd)?;
44 if buf.remaining() < len {
45 return Err(CodecError::UnexpectedEnd);
46 }
47 buf.advance(len);
48 Ok(())
49}
50
51/// Turn a wire Object Status code into an [`ObjectStatus`], refusing one
52/// draft-23 does not assign.
53///
54/// Draft-23 Section 11.1.1 lists the codes an object may carry — the three
55/// rows of the Object Status registry in Section 16.9 — and says: "If an
56/// endpoint receives any other value, it SHOULD close the session with
57/// INVALID_OBJECT_STATUS (Section 12.2)." Every place this module reads a status
58/// converts it here, so a decoded [`SubgroupObject::object_status`] or
59/// [`DatagramHeader::object_status`] is always a status the draft assigns, and
60/// [`SubgroupObjectMeta::status`] — which stays a raw code because a relay may
61/// carry it to a draft that numbers the set differently — holds one because it
62/// comes from the same conversion.
63fn decoded_status(code: u64) -> Result<ObjectStatus, CodecError> {
64 ObjectStatus::from_u64(code).ok_or(CodecError::UnknownObjectStatus(code))
65}
66
67// ── Stream and datagram types ─────────────────────────────────
68
69/// Unidirectional stream type for padding, draft-23 Section 11.5.1: "An
70/// endpoint MAY open a unidirectional stream with a stream type of 0x132B3E28
71/// to send padding data. The stream begins with the stream type, followed by
72/// zero or more bytes that MUST all be set to zero."
73///
74/// Named here because the value is what tells a padding stream from a data
75/// stream, and nothing else in this module would otherwise say so. Under the
76/// draft-23 variable-length integer encoding the value takes five bytes, the
77/// first of which is 0xF0 — an octet with bit 7 set, which the subgroup form
78/// leaves clear. A reader judging the Type by that first byte alone would find
79/// the form broken and call the stream unknown, which Section 6.4.1 answers by
80/// ending the session. Table 2 assigns the value, so that would be a close over
81/// traffic this draft permits.
82pub const PADDING_STREAM_TYPE: u64 = 0x132B_3E28;
83
84/// Datagram type for padding, draft-23 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, and
90/// assigned the same way, so the same reasoning applies to a datagram reader.
91pub const PADDING_DATAGRAM_TYPE: u64 = 0x132B_3E29;
92
93/// The unidirectional stream Type draft-23 Section 9.1 gives the control
94/// stream.
95const SETUP_STREAM_TYPE: u64 = 0x2F00;
96
97// ── Subgroup ──────────────────────────────────────────────────
98
99const SUBGROUP_PROPERTIES_BIT: u8 = 0x01;
100const SUBGROUP_ID_MODE_MASK: u8 = 0x06;
101const SUBGROUP_END_OF_GROUP_BIT: u8 = 0x08;
102const SUBGROUP_BASE_BIT: u8 = 0x10;
103const SUBGROUP_DEFAULT_PRIORITY_BIT: u8 = 0x20;
104const SUBGROUP_FIRST_OBJECT_BIT: u8 = 0x40;
105/// The SUBGROUP_ID_MODE value draft-23 reserves, once the mask is applied and
106/// the field shifted down.
107const SUBGROUP_ID_MODE_RESERVED: u8 = 0b11;
108
109/// Refuse a subgroup header `Type Flags` value draft-23 Section 11.3.1 lists
110/// as invalid.
111///
112/// The section gives **three** conditions: "If an endpoint receives a stream
113/// header with any of these values, it MUST close the session with
114/// INVALID_STREAM_HEADER":
115///
116/// 1. "Values with SUBGROUP_ID_MODE set to 0b11. This mode is reserved for
117/// future use." That is 0x16, 0x17, 0x1E, 0x1F and the same four offsets
118/// in each higher range.
119/// 2. "Values where bit 4 is not set. Bit 4 MUST be 1 for SUBGROUP_HEADER."
120/// 3. "Values of 128 or greater (i.e., any value that requires more than a
121/// one-byte variable-length integer encoding)."
122///
123/// Computing the three gives 0x10-0x15, 0x18-0x1D, 0x30-0x35, 0x38-0x3D,
124/// 0x50-0x55, 0x58-0x5D, 0x70-0x75 and 0x78-0x7D.
125///
126/// **These are not the datagram's rules.** Section 11.2.1 reserves bit 4 and
127/// requires it to be *zero*, and adds an unspecified-bit condition that this
128/// list does not have, because bits 0 through 6 are all specified here. The two
129/// sets are read from their own sections and neither is derived from the other.
130///
131/// The reserved mode is worth separating from a mere unassigned code point,
132/// because it is not decodable rather than merely unknown. The other three
133/// modes each say whether a Subgroup ID field follows the Group ID; 0b11 says
134/// nothing, so a decoder has to guess, and a wrong guess shifts every
135/// subsequent field by the width of that varint.
136fn validate_subgroup_type(raw: u64) -> Result<(), CodecError> {
137 if subgroup_type_is_valid(raw) {
138 Ok(())
139 } else {
140 Err(stream_type_error(raw))
141 }
142}
143
144/// Whether `raw` is a subgroup `Type Flags` value draft-23 admits: below 128,
145/// bit 4 set, and not the reserved SUBGROUP_ID_MODE.
146fn subgroup_type_is_valid(raw: u64) -> bool {
147 // `raw < 0x80` is Section 11.3.1's third condition, and it subsumes the
148 // form's bit-7-clear requirement.
149 raw < 0x80 && {
150 let t = raw as u8;
151 t & SUBGROUP_BASE_BIT != 0 && (t & SUBGROUP_ID_MODE_MASK) >> 1 != SUBGROUP_ID_MODE_RESERVED
152 }
153}
154
155/// Which failure a leading unidirectional stream Type that is not the one a
156/// reader wants is.
157///
158/// Draft-23 states two rules about such a Type and answers both with a close,
159/// and telling them apart is the whole job of this function.
160///
161/// Section 6.4.1 is about the table: "An endpoint that receives an unknown stream
162/// type MUST close the session with INVALID_STREAM_HEADER." A Type Table 2 does
163/// not assign is
164/// [`CodecError::UnknownStreamType`].
165///
166/// Section 11.3.1 is about the subgroup form specifically, and gives three
167/// conditions on a `Type Flags` value *inside* it. Those values are not unknown
168/// — the registry pattern 0b0XX1XXXX assigns the space and the draft rules
169/// particular values out within it — but they are unreadable, and they are
170/// [`CodecError::InvalidStreamTypeValue`], one arm per condition so a log names which.
171///
172/// Table 2 assigns four things, and two of them carry no Objects at all:
173/// FETCH_HEADER, the subgroup form, SETUP and PADDING. A subgroup reader handed
174/// any of them refuses it as [`CodecError::InvalidField`] — the value is one
175/// this draft defines, the disagreement is with the reader that was called, and
176/// the session survives it. See [`PADDING_STREAM_TYPE`] for why that case in
177/// particular is worth the trouble.
178///
179/// # Where the boundary between the two rules sits
180///
181/// Section 11.3.1's conditions are about a one-byte flags field, so they are
182/// applied to values that fit one — `raw <= 0xFF`. Above that the value is out
183/// of the space the section describes and the question is Table 2's instead:
184/// an assigned non-subgroup stream type is refused as the wrong reader, and
185/// anything else is unknown. That split is what keeps a datagram's
186/// `Type Flags` of 0x20 reported as an invalid subgroup value — which is what
187/// Section 11.3.1's bit-4 condition calls it — while a stray 0x5000 stays an
188/// unknown stream type.
189fn stream_type_error(raw: u64) -> CodecError {
190 if raw == FETCH_STREAM_TYPE
191 || raw == SETUP_STREAM_TYPE
192 || raw == PADDING_STREAM_TYPE
193 || subgroup_type_is_valid(raw)
194 {
195 return CodecError::InvalidField;
196 }
197 if raw > 0xFF {
198 return CodecError::UnknownStreamType(raw);
199 }
200 let t = raw as u8;
201 // Third condition first: it is the coarsest, and 128 or more can also set
202 // the other two bits in ways that would misreport it.
203 if raw >= 0x80 {
204 CodecError::InvalidStreamTypeValue {
205 raw,
206 detail: "SUBGROUP_HEADER Type Flags of 128 or greater are invalid",
207 }
208 } else if t & SUBGROUP_BASE_BIT == 0 {
209 CodecError::InvalidStreamTypeValue { raw, detail: "bit 4 must be 1 for SUBGROUP_HEADER" }
210 } else {
211 CodecError::InvalidStreamTypeValue {
212 raw,
213 detail: "its SUBGROUP_ID_MODE is 0b11, which this draft reserves",
214 }
215 }
216}
217
218#[derive(Debug, Clone)]
219pub struct SubgroupHeader {
220 pub header_type: u8,
221 pub track_alias: VarInt,
222 pub group_id: VarInt,
223 pub subgroup_id: VarInt,
224 pub publisher_priority: Option<u8>,
225}
226
227impl SubgroupHeader {
228 /// Decode a subgroup header, `Type Flags` field included.
229 ///
230 /// The field is read as a whole varint of whatever width the sender used,
231 /// and the value it decodes to is what the three Section 11.3.1 conditions
232 /// are applied to. **A non-minimal spelling of a valid value is
233 /// accepted**: `0x8010` decodes to 16, which is a legal `Type Flags`, and
234 /// this reads it. See the module documentation for why the draft leaves
235 /// that open and why permissive-on-receive is the choice here; the encoder
236 /// is the strict half and always writes the one-byte form.
237 ///
238 /// The whole field is read, which is what
239 /// lets `stream_type_error` tell a padding or SETUP stream — both assigned,
240 /// both several bytes wide — from a Type Table 2 does not assign.
241 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
242 let raw = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
243 validate_subgroup_type(raw)?;
244 // Narrowing is safe: `validate_subgroup_type` has already refused
245 // everything at 128 or above.
246 let header_type = raw as u8;
247
248 let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
249 let group_id = VarInt::decode_moqt::<Wire>(buf)?;
250
251 let subgroup_id_mode = (header_type & SUBGROUP_ID_MODE_MASK) >> 1;
252 let subgroup_id = match subgroup_id_mode {
253 0 => VarInt::from_u64_moqt(0),
254 2 => VarInt::decode_moqt::<Wire>(buf)?,
255 // Mode 1 puts no Subgroup ID on the wire either: it is the first
256 // object's ID, which this header cannot see. Store 0 until a
257 // caller resolves it. Mode 3 never reaches here — the type check
258 // above refuses it.
259 _ => VarInt::from_u64_moqt(0),
260 };
261
262 let publisher_priority = if header_type & SUBGROUP_DEFAULT_PRIORITY_BIT == 0 {
263 if buf.remaining() < 1 {
264 return Err(CodecError::UnexpectedEnd);
265 }
266 Some(buf.get_u8())
267 } else {
268 None
269 };
270
271 Ok(SubgroupHeader { header_type, track_alias, group_id, subgroup_id, publisher_priority })
272 }
273
274 /// Serialize the header exactly as its Type byte describes it.
275 ///
276 /// Infallible, and so willing to write a Type draft-23 Section 11.3.1
277 /// tells an endpoint to reject — including a reserved-mode Type this
278 /// module's own [`Self::decode`] refuses to read back. Prefer
279 /// [`Self::encode_checked`], which refuses those Types instead.
280 pub fn encode(&self, buf: &mut impl BufMut) {
281 buf.put_u8(self.header_type);
282 self.track_alias.encode_moqt::<Wire>(buf);
283 self.group_id.encode_moqt::<Wire>(buf);
284
285 let subgroup_id_mode = (self.header_type & SUBGROUP_ID_MODE_MASK) >> 1;
286 if subgroup_id_mode == 2 {
287 self.subgroup_id.encode_moqt::<Wire>(buf);
288 }
289
290 if self.header_type & SUBGROUP_DEFAULT_PRIORITY_BIT == 0 {
291 buf.put_u8(self.publisher_priority.unwrap_or(128));
292 }
293 }
294
295 /// Serialize the header, refusing a Type value draft-23 forbids.
296 ///
297 /// Errors with [`CodecError::InvalidField`] for exactly the Types draft-23
298 /// Section 11.3.1 lists as invalid (the same set [`Self::decode`] refuses),
299 /// before any byte is written, so a refused header leaves `buf` untouched.
300 /// Everything else is written by [`Self::encode`].
301 ///
302 /// The check belongs on the encode side as well as the decode side because
303 /// the two halves would otherwise disagree about which streams exist: a
304 /// reserved-mode header written by [`Self::encode`] cannot be read back by
305 /// [`Self::decode`], and a codec used to rewrite captured traffic would
306 /// emit a stream it could not then parse.
307 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
308 validate_subgroup_type(self.header_type as u64)?;
309 self.encode(buf);
310 Ok(())
311 }
312
313 pub fn has_properties(&self) -> bool {
314 self.header_type & SUBGROUP_PROPERTIES_BIT != 0
315 }
316
317 /// The subgroup-ID mode: `(header_type & 0x06) >> 1`.
318 ///
319 /// `0` = no subgroup ID on the wire and it is zero; `1` = the subgroup ID
320 /// is the first object's ID; `2` = an explicit ID follows the Group ID;
321 /// `3` = reserved. Exposed because the mask is module-private and
322 /// `dispatch::AnySubgroupHeader::subgroup_id_mode` cannot read it.
323 ///
324 /// `3` never comes back from [`Self::decode`], which refuses the Types
325 /// carrying it. `header_type` is a public field, so a hand-built header
326 /// can still report it; [`Self::encode_checked`] is what refuses to put
327 /// one on the wire.
328 pub fn subgroup_id_mode(&self) -> u8 {
329 (self.header_type & SUBGROUP_ID_MODE_MASK) >> 1
330 }
331
332 pub fn is_end_of_group(&self) -> bool {
333 self.header_type & SUBGROUP_END_OF_GROUP_BIT != 0
334 }
335
336 /// `true` when the FIRST_OBJECT bit (0x40) is set, signaling the first
337 /// object on this stream is the original publisher's first object in the
338 /// subgroup.
339 pub fn is_first_object(&self) -> bool {
340 self.header_type & SUBGROUP_FIRST_OBJECT_BIT != 0
341 }
342}
343
344// ── Subgroup objects (stateful) ───────────────────────────────
345
346/// One object within a draft-23 subgroup stream. Object IDs are
347/// delta-encoded; whether a per-object "properties" block (the draft-23
348/// equivalent of extension headers) is present depends on the PROPERTIES
349/// bit on the enclosing [`SubgroupHeader`]. Use [`SubgroupObjectReader`]
350/// to encode/decode.
351#[derive(Debug, Clone)]
352pub struct SubgroupObject {
353 pub object_id: VarInt,
354 /// Raw properties bytes, excluding the byte-length prefix that precedes
355 /// them on the wire. Empty unless the subgroup header sets the
356 /// PROPERTIES bit, or when the block is present but zero-length.
357 /// Opaque: [`SubgroupObjectReader::write_object`] re-emits the prefix
358 /// and these bytes verbatim.
359 pub extension_headers: Vec<u8>,
360 pub payload_length: VarInt,
361 /// The object's status, carried on the wire only when `payload_length` is
362 /// zero. `None` with a zero `payload_length` is written as
363 /// [`ObjectStatus::Normal`].
364 ///
365 /// `None` is not "no status": every Object has one. It means the status is
366 /// the one the encoding elides — [`ObjectStatus::Normal`], the only row of
367 /// the Object Status registry (draft-23 Section 16.9) that permits the
368 /// payload such an object carries. Decoding a payload-bearing object leaves
369 /// this `None` for that reason; [`Self::status`] resolves it either way.
370 ///
371 /// Typed rather than a raw code. The wire field is a varint with room for
372 /// any value, and draft-23 assigns three of them; the decoder refuses the
373 /// rest, and this type is that same refusal on the encode side — 0x1 and
374 /// 0x2 cannot be named here, so [`SubgroupObjectReader::write_object`]
375 /// cannot emit a status this module's own decoder would reject.
376 ///
377 /// A status and a payload can be held here together, which the wire has no
378 /// way to express. That combination is what draft-23's registry rules on:
379 /// [`SubgroupObjectReader::write_object`] accepts it when the status is
380 /// registered as permitting a payload and refuses it otherwise.
381 pub object_status: Option<ObjectStatus>,
382 pub payload: Vec<u8>,
383}
384
385impl SubgroupObject {
386 /// The object's status, with the one draft-23's encoding elides filled in.
387 ///
388 /// A subgroup object states its status only when its Object Payload Length
389 /// is zero. An object that carries bytes therefore has no status field, and
390 /// its status is [`ObjectStatus::Normal`] — the sole row of the Object
391 /// Status registry (draft-23 Section 16.9) permitting a payload, so the
392 /// only status such an object could have had.
393 pub fn status(&self) -> ObjectStatus {
394 self.object_status.unwrap_or(ObjectStatus::Normal)
395 }
396
397 /// Whether the Object Status registry permits this object a non-empty
398 /// payload, per draft-23 Section 16.9.
399 ///
400 /// Answered from the status alone. `payload` and `payload_length` are not
401 /// consulted: on a status that permits a payload they say only whether this
402 /// particular object took the offer, and on one that forbids a payload a
403 /// non-empty payload is the malformation this reports, not evidence about
404 /// the rule.
405 pub fn permits_payload(&self) -> bool {
406 self.status().permits_payload()
407 }
408
409 /// Whether this object's status is allowed to carry the properties it has.
410 ///
411 /// Draft-23 Section 11.1.2: "Any Object with status Normal can have
412 /// properties (Section 8.4). If an endpoint receives properties on an
413 /// Object with status that is not Normal, it MUST close the session with
414 /// INVALID_OBJECT_STATUS."
415 ///
416 /// So this is `false` for exactly one shape: a non-empty properties block
417 /// on an object whose status is not [`ObjectStatus::Normal`]. An object
418 /// with no properties is fine at any status, and an object at Normal may
419 /// carry any properties.
420 ///
421 /// [`SubgroupObjectReader::read_object`] and
422 /// [`SubgroupObjectReader::write_object`] refuse an object this answers
423 /// `false` for; [`SubgroupObjectReader::read_object_meta`] frames it.
424 pub fn properties_permitted(&self) -> bool {
425 self.extension_headers.is_empty() || self.status() == ObjectStatus::Normal
426 }
427}
428
429/// The framing of one draft-23 subgroup object, without its payload.
430///
431/// Produced by [`SubgroupObjectReader::read_object_meta`] for callers that
432/// forward an object's bytes verbatim and never inspect the payload.
433#[derive(Debug, Clone, Copy, PartialEq, Eq)]
434pub struct SubgroupObjectMeta {
435 /// Resolved absolute Object ID.
436 pub object_id: u64,
437 /// Byte length of the properties block's contents, excluding its length
438 /// prefix.
439 pub extension_headers_len: u64,
440 /// Declared payload length. Zero when `status` is `Some`.
441 pub payload_length: u64,
442 /// Object status wire code, present only when the payload is empty.
443 pub status: Option<u64>,
444 /// Total bytes this object occupies on the wire, prefix fields included.
445 pub wire_len: u64,
446}
447
448impl SubgroupObjectMeta {
449 /// What the Object Status registry says about this object's payload, or
450 /// `None` if the registry has no row for its status.
451 ///
452 /// Draft-23 Section 16.9, Table 17 gives the registry a "Payload" column,
453 /// and Section 11.1.1 makes it the rule: an Object has an empty payload
454 /// unless its status is registered as permitting one. Table 17 fills the
455 /// column in for the three statuses draft-23 assigns — 0x0 Normal is
456 /// "Yes", 0x3 End of Group and 0x4 End of Track are "No" — and requires
457 /// every future registration to fill it in too.
458 ///
459 /// `status` is `None` for an object whose payload length is non-zero,
460 /// because the encoding puts a status field only where the payload does
461 /// not go. Such an object's status is Normal, the one row permitting the
462 /// payload it is carrying, so this answers
463 /// [`PayloadPermission::Permitted`] rather than `None`.
464 ///
465 /// The `None` this does return means something else entirely: a status
466 /// code with no row in the registry, for which the draft supplies no
467 /// answer and this must not invent one. [`SubgroupObjectReader`] never
468 /// produces such a meta — it refuses an unassigned code while decoding —
469 /// but every field here is public, so a caller that assembled a meta by
470 /// hand, or carried a status across from a draft numbering the set
471 /// differently, can hold one.
472 ///
473 /// Answered from the status alone. `payload_length` is not consulted: on a
474 /// status that permits a payload it says only whether this particular
475 /// object took the offer, and on one that forbids a payload a non-zero
476 /// length is the malformation a caller uses this to detect, not evidence
477 /// about the rule.
478 pub fn payload_permission(&self) -> Option<PayloadPermission> {
479 match self.status {
480 None => Some(ObjectStatus::Normal.payload_permission()),
481 Some(code) => ObjectStatus::from_u64(code).map(ObjectStatus::payload_permission),
482 }
483 }
484}
485
486/// Prior Group ID Gap, Object Property Type 0x3C (Section 10.8).
487pub const PRIOR_GROUP_ID_GAP: u64 = 0x3C;
488/// Prior Object ID Gap, Object Property Type 0x3E (Section 10.9).
489pub const PRIOR_OBJECT_ID_GAP: u64 = 0x3E;
490
491/// The Prior Group ID Gap and Prior Object ID Gap one Object carries.
492#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
493pub struct PriorGaps {
494 /// Prior Group ID Gap, if present.
495 pub group: Option<u64>,
496 /// Prior Object ID Gap, if present.
497 pub object: Option<u64>,
498}
499
500impl PriorGaps {
501 /// Read the two gaps from an Object's parsed properties, Immutable
502 /// Properties included, refusing the conditions one Object shows on its
503 /// own.
504 ///
505 /// Section 10.8 and Section 10.9 make the Track malformed when "An Object
506 /// contains more than one instance of Prior Group ID Gap", when "An Object
507 /// has a Prior Group ID Gap larger than the Group ID", and the same two
508 /// for Prior Object ID Gap against the Object ID. Each is
509 /// [`CodecError::MalformedTrack`].
510 pub fn from_properties(
511 properties: &[ObjectProperty<'_>],
512 group_id: u64,
513 object_id: u64,
514 ) -> Result<Self, CodecError> {
515 let mut gaps = Self::default();
516 collect_gaps(properties, &mut gaps)?;
517 if let Some(gap) = gaps.group.filter(|&gap| gap > group_id) {
518 return Err(CodecError::MalformedTrack {
519 detail: format!("Prior Group ID Gap {gap} is larger than Group ID {group_id}"),
520 });
521 }
522 if let Some(gap) = gaps.object.filter(|&gap| gap > object_id) {
523 return Err(CodecError::MalformedTrack {
524 detail: format!("Prior Object ID Gap {gap} is larger than Object ID {object_id}"),
525 });
526 }
527 Ok(gaps)
528 }
529}
530
531fn collect_gaps(properties: &[ObjectProperty<'_>], gaps: &mut PriorGaps) -> Result<(), CodecError> {
532 for property in properties {
533 let slot = match (&property.value, property.property_type) {
534 (PropertyValue::Immutable { properties, .. }, _) => {
535 collect_gaps(properties, gaps)?;
536 continue;
537 }
538 (PropertyValue::Varint(value), PRIOR_GROUP_ID_GAP) => {
539 (&mut gaps.group, *value, "Group")
540 }
541 (PropertyValue::Varint(value), PRIOR_OBJECT_ID_GAP) => {
542 (&mut gaps.object, *value, "Object")
543 }
544 _ => continue,
545 };
546 let (field, value, which) = slot;
547 if field.replace(value).is_some() {
548 return Err(CodecError::MalformedTrack {
549 detail: format!("an Object contains more than one Prior {which} ID Gap"),
550 });
551 }
552 }
553 Ok(())
554}
555
556/// The Prior Group ID Gap and Prior Object ID Gap conditions of
557/// draft-23 Section 10.8 and draft-23 Section 10.9 that one fetch stream
558/// shows.
559///
560/// [`PriorGaps::from_properties`] refuses what one Object shows. Across the
561/// stream this refuses "A Group contains more than one Object with different
562/// values for Prior Group ID Gap" for the Objects of one Group, "An endpoint
563/// receives an Object with a Prior Group ID Gap covering an Object it
564/// previously received", "An endpoint receives an Object with a Group ID
565/// within a previously communicated gap", and the same two for Prior Object ID
566/// Gap, for the Group and the Object before on the stream.
567/// draft-23 Section 3.2.1: "Within each group, objects are sent in Object ID
568/// order; subgroup ID is not used for ordering."
569///
570/// [`FetchObjectReader::read_object_header`] applies it to each Object it reads.
571///
572/// End of Range, End of Group and End of Track indicators are not Objects;
573/// the caller passes Objects only.
574#[derive(Debug, Clone, PartialEq, Eq)]
575// Boxed so that the fetch readers holding one stay small.
576pub struct FetchGaps(Box<crate::prior_gaps::FetchGapTracker>);
577
578impl Default for FetchGaps {
579 fn default() -> Self {
580 Self::new()
581 }
582}
583
584impl FetchGaps {
585 /// Track the Objects of one fetch stream.
586 pub fn new() -> Self {
587 Self(Box::new(crate::prior_gaps::FetchGapTracker::new(
588 crate::prior_gaps::GapRules::Draft23On,
589 )))
590 }
591
592 /// Check the next Object on the stream, from its resolved Group ID and
593 /// Object ID and its property block, refusing a gap condition with
594 /// [`CodecError::MalformedTrack`] and a block that does not parse as
595 /// [`crate::object_properties::parse_object_properties`] does. Records the
596 /// Object only when it passes.
597 pub fn check(
598 &mut self,
599 group_id: u64,
600 object_id: u64,
601 property_block: &[u8],
602 ) -> Result<PriorGaps, CodecError> {
603 let properties = crate::object_properties::parse_object_properties(
604 crate::version::DraftVersion::Draft23,
605 property_block,
606 )?;
607 let gaps = PriorGaps::from_properties(&properties, group_id, object_id)?;
608 self.0.check(group_id, object_id, gaps.group.as_slice(), gaps.object)?;
609 Ok(gaps)
610 }
611}
612
613#[derive(Debug, Clone)]
614pub struct SubgroupObjectReader {
615 extensions_present: bool,
616 prev_object_id: Option<u64>,
617 group_id: u64,
618 group_gap: Option<u64>,
619}
620
621impl SubgroupObjectReader {
622 pub fn new(header: &SubgroupHeader) -> Self {
623 Self {
624 extensions_present: header.has_properties(),
625 prev_object_id: None,
626 group_id: header.group_id.into_inner(),
627 group_gap: None,
628 }
629 }
630
631 /// Decode the next object whole, payload included.
632 ///
633 /// Refuses, with [`CodecError::UnknownObjectStatus`], a status Section
634 /// 11.1.1 does not list.
635 ///
636 /// Refuses, with [`CodecError::InvalidField`], properties on an object
637 /// whose status is not Normal: Section 11.1.2, "If an endpoint receives
638 /// properties on an Object with status that is not Normal, it MUST close
639 /// the session with INVALID_OBJECT_STATUS." [`Self::read_object_meta`]
640 /// frames such an object without judging it, as
641 /// [`DatagramHeader::decode`] does a datagram.
642 ///
643 /// Refuses, with [`CodecError::MalformedTrack`], the Prior Group ID Gap
644 /// and Prior Object ID Gap conditions of Section 10.8 and Section 10.9 this
645 /// stream shows: those [`PriorGaps::from_properties`] refuses, "A Group
646 /// contains more than one Object with different values for Prior Group ID
647 /// Gap", and "An endpoint receives an Object with a Prior Object ID Gap
648 /// covering an Object it previously received" for an Object earlier on
649 /// this stream. [`Self::read_object_meta`] does not read properties and
650 /// applies none of them.
651 pub fn read_object(&mut self, buf: &mut impl Buf) -> Result<SubgroupObject, CodecError> {
652 let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
653 let prior_object_id = self.prev_object_id;
654 let object_id_val = match self.prev_object_id {
655 None => delta,
656 Some(prev) => prev
657 .checked_add(1)
658 .and_then(|v| v.checked_add(delta))
659 .ok_or(CodecError::ObjectIdOverflow(prev, delta))?,
660 };
661 self.prev_object_id = Some(object_id_val);
662 let object_id = VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?;
663
664 // The properties block is a byte-length-prefixed opaque blob. We
665 // copy the blob verbatim; callers that want structured properties
666 // can parse the returned bytes.
667 let extension_headers = if self.extensions_present {
668 let ext_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
669 crate::types::read_bytes(buf, ext_len)?
670 } else {
671 Vec::new()
672 };
673
674 let payload_length_vi = VarInt::decode_moqt::<Wire>(buf)?;
675 let payload_length_val = payload_length_vi.into_inner() as usize;
676 let (object_status, payload) = if payload_length_val == 0 {
677 let status = VarInt::decode_moqt::<Wire>(buf)?;
678 (Some(decoded_status(status.into_inner())?), Vec::new())
679 } else {
680 let payload = crate::types::read_bytes(buf, payload_length_val)?;
681 (None, payload)
682 };
683
684 let properties = crate::object_properties::parse_object_properties(
685 crate::version::DraftVersion::Draft23,
686 &extension_headers,
687 )?;
688 let gaps = PriorGaps::from_properties(&properties, self.group_id, object_id_val)?;
689 self.check_stream_gaps(gaps, prior_object_id, object_id_val)?;
690 let object = SubgroupObject {
691 object_id,
692 extension_headers,
693 payload_length: payload_length_vi,
694 object_status,
695 payload,
696 };
697 if !object.properties_permitted() {
698 return Err(CodecError::InvalidField);
699 }
700 Ok(object)
701 }
702
703 fn check_stream_gaps(
704 &mut self,
705 gaps: PriorGaps,
706 prior_object_id: Option<u64>,
707 object_id: u64,
708 ) -> Result<(), CodecError> {
709 if let Some(gap) = gaps.group {
710 match self.group_gap {
711 Some(seen) if seen != gap => {
712 return Err(CodecError::MalformedTrack {
713 detail: format!(
714 "two Objects in Group {} carry different Prior Group ID Gap values",
715 self.group_id
716 ),
717 });
718 }
719 _ => self.group_gap = Some(gap),
720 }
721 }
722 // Object IDs rise along a subgroup stream, so the latest earlier
723 // Object is the only one a gap below this Object can reach.
724 if let (Some(gap), Some(prior)) = (gaps.object, prior_object_id) {
725 if gap > 0 && prior >= object_id - gap {
726 return Err(CodecError::MalformedTrack {
727 detail: format!(
728 "Prior Object ID Gap {gap} of Object {object_id} covers Object {prior}"
729 ),
730 });
731 }
732 }
733 Ok(())
734 }
735
736 /// Decode the next object's framing without copying its payload.
737 ///
738 /// Consumes exactly the bytes [`Self::read_object`] consumes and leaves
739 /// the same delta state behind, so the two are interchangeable on a
740 /// given stream.
741 pub fn read_object_meta(
742 &mut self,
743 buf: &mut impl Buf,
744 ) -> Result<SubgroupObjectMeta, CodecError> {
745 let start = buf.remaining();
746 let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
747 let object_id_val = match self.prev_object_id {
748 None => delta,
749 Some(prev) => prev
750 .checked_add(1)
751 .and_then(|v| v.checked_add(delta))
752 .ok_or(CodecError::ObjectIdOverflow(prev, delta))?,
753 };
754 self.prev_object_id = Some(object_id_val);
755 let object_id =
756 VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?.into_inner();
757
758 let extension_headers_len = if self.extensions_present {
759 let ext_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
760 skip(buf, ext_len)?;
761 ext_len
762 } else {
763 0
764 };
765
766 let payload_length = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
767 let status = if payload_length == 0 {
768 let code = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
769 Some(decoded_status(code)?.as_u64())
770 } else {
771 skip(buf, payload_length)?;
772 None
773 };
774
775 Ok(SubgroupObjectMeta {
776 object_id,
777 extension_headers_len,
778 payload_length,
779 status,
780 wire_len: (start - buf.remaining()) as u64,
781 })
782 }
783
784 /// Serialize an object, producing the correct delta encoding.
785 ///
786 /// A zero `payload_length` writes the object's status, with
787 /// [`SubgroupObject::status`] filling in [`ObjectStatus::Normal`] when the
788 /// `object_status` field is `None`. The status is typed, so every value
789 /// that can reach this method is one draft-23 assigns and one
790 /// [`Self::read_object`] accepts; no unassigned code can be written.
791 ///
792 /// Errors with [`CodecError::InvalidField`] when the Object Status registry
793 /// (draft-23 Section 16.9) does not permit the object's status to carry a
794 /// payload and a payload was handed over anyway. Draft-23 Section 11.1.1
795 /// makes that object malformed, and the encoding has no way to state it:
796 /// the status field appears only where the payload does not. Writing it
797 /// would mean silently discarding one of the two — emitting the bytes as a
798 /// Normal object and losing an End of Group marker, say — so it is refused
799 /// instead. A status the registry does permit a payload is written as an
800 /// ordinary payload object, its status implicit, since that is how the
801 /// encoding spells it.
802 ///
803 /// Errors with [`CodecError::InvalidField`] when `object.object_id` is
804 /// not strictly greater than the previously written object's ID, since
805 /// no valid delta exists for that case.
806 ///
807 /// Errors with [`CodecError::InvalidField`] when `payload_length` is not
808 /// exactly `payload.len()`. The declared length is written ahead of the
809 /// payload, so a mismatch is a frame [`Self::read_object`] cannot parse
810 /// and one no caller could fix by appending bytes.
811 pub fn write_object(
812 &mut self,
813 object: &SubgroupObject,
814 buf: &mut impl BufMut,
815 ) -> Result<(), CodecError> {
816 // A declared length that disagrees with the payload framed under it
817 // produces bytes no reader can parse and no caller can repair: the
818 // length is already on the wire ahead of the payload. Checked before
819 // anything is written, so a refused object leaves `buf` untouched
820 // rather than half an object the next read would run into.
821 //
822 // A zero declared length is also what puts a status code where the
823 // payload would go, so an object carrying bytes under it is asking for
824 // two framings at once.
825 let declared = object.payload_length.into_inner();
826 if declared != object.payload.len() as u64 {
827 return Err(CodecError::InvalidField);
828 }
829
830 // The registry decides whether these two fields may be filled in at
831 // once, rather than the length deciding on its own which of them gets
832 // written. A status marked "Payload: No" is refused a payload; a status
833 // marked "Yes" keeps it, and gets no status field on the wire because
834 // the encoding elides the status of any object that carries bytes.
835 // Checked alongside the length above, before any byte is emitted.
836 if declared != 0 && !object.permits_payload() {
837 return Err(CodecError::PayloadNotPermitted {
838 status: object.status().as_u64(),
839 len: object.payload.len(),
840 detail: "its status is registered as forbidding one",
841 });
842 }
843
844 // Properties on a non-Normal status are a frame `read_object` refuses
845 // (Section 11.1.2), so they are not written either.
846 if !object.properties_permitted() {
847 return Err(CodecError::InvalidField);
848 }
849
850 let oid = object.object_id.into_inner();
851 let delta = match self.prev_object_id {
852 None => oid,
853 Some(prev) => oid
854 .checked_sub(prev)
855 .and_then(|v| v.checked_sub(1))
856 .ok_or(CodecError::InvalidField)?,
857 };
858 VarInt::from_u64(delta).map_err(|_| CodecError::InvalidField)?.encode_moqt::<Wire>(buf);
859 if self.extensions_present {
860 VarInt::from_u64(object.extension_headers.len() as u64)
861 .map_err(|_| CodecError::InvalidField)?
862 .encode_moqt::<Wire>(buf);
863 buf.put_slice(&object.extension_headers);
864 }
865 object.payload_length.encode_moqt::<Wire>(buf);
866 if declared == 0 {
867 VarInt::from_u64_moqt(object.status().as_u64()).encode_moqt::<Wire>(buf);
868 } else {
869 buf.put_slice(&object.payload);
870 }
871 self.prev_object_id = Some(oid);
872 Ok(())
873 }
874}
875
876// ── Datagram ──────────────────────────────────────────────────
877
878const DATAGRAM_PROPERTIES_BIT: u8 = 0x01;
879const DATAGRAM_END_OF_GROUP_BIT: u8 = 0x02;
880const DATAGRAM_ZERO_OBJECT_ID_BIT: u8 = 0x04;
881const DATAGRAM_DEFAULT_PRIORITY_BIT: u8 = 0x08;
882const DATAGRAM_STATUS_BIT: u8 = 0x20;
883/// Bit 4. Draft-23 Section 11.2.1: "This bit is reserved and MUST be zero."
884/// The opposite of Section 11.3.1, where the same bit MUST be 1.
885const DATAGRAM_RESERVED_BIT: u8 = 0x10;
886
887/// Refuse a datagram `Type Flags` value draft-23 Section 11.2.1 lists as
888/// invalid.
889///
890/// The section states the rule twice, once in the paragraph introducing the
891/// field — "If a received value has bit 4 set, or has a bit set whose meaning
892/// is not specified, the endpoint MUST close the session with
893/// INVALID_STREAM_HEADER" — and once as a list of three conditions, each of
894/// which MUST close the session with INVALID_STREAM_HEADER:
895///
896/// 1. "Values with both the STATUS bit (0x20) and END_OF_GROUP bit (0x02)
897/// set": 0x22, 0x23, 0x26, 0x27, 0x2A, 0x2B, 0x2E and 0x2F. The reason is
898/// that an object status message cannot signal end of group — the two bits
899/// ask for one datagram to be both a status and an end-of-group marker for
900/// an object it does not carry.
901/// 2. "Values with bit 4 (0x10) set. This bit is reserved and MUST be zero."
902/// 3. "Values with a bit set whose meaning is not specified." Bits 0, 1, 2, 3
903/// and 5 are specified; bit 6 and everything above it are not, so this
904/// catches 0x40 and, with it, every value of 128 or greater.
905///
906/// The valid set is 0x00..0x0F and 0x20..0x2F.
907///
908/// **These are not the subgroup header's rules.** Section 11.3.1 requires bit 4
909/// to be *set*, has no unspecified-bit condition, and states an explicit
910/// "values of 128 or greater" condition that this list does not need. The two
911/// sets are read from their own sections and neither is derived from the other.
912fn validate_datagram_type(raw: u64) -> Result<(), CodecError> {
913 if datagram_type_is_valid(raw) {
914 Ok(())
915 } else {
916 Err(datagram_type_error(raw))
917 }
918}
919
920/// Every bit draft-23 Section 11.2.1 gives a datagram `Type Flags` value a
921/// meaning for: PROPERTIES, END_OF_GROUP, ZERO_OBJECT_ID, DEFAULT_PRIORITY and
922/// STATUS. Bit 4 is reserved and bits 6 and up are unspecified, so both fail
923/// the mask.
924const DATAGRAM_SPECIFIED_BITS: u64 = (DATAGRAM_PROPERTIES_BIT
925 | DATAGRAM_END_OF_GROUP_BIT
926 | DATAGRAM_ZERO_OBJECT_ID_BIT
927 | DATAGRAM_DEFAULT_PRIORITY_BIT
928 | DATAGRAM_STATUS_BIT) as u64;
929
930/// Whether `raw` is a datagram `Type Flags` value draft-23 admits.
931fn datagram_type_is_valid(raw: u64) -> bool {
932 // Every bit outside the specified set fails, which covers bit 4 (reserved),
933 // bit 6 (unspecified) and everything at 128 or above in one test.
934 raw & !DATAGRAM_SPECIFIED_BITS == 0
935 && !(raw & DATAGRAM_STATUS_BIT as u64 != 0 && raw & DATAGRAM_END_OF_GROUP_BIT as u64 != 0)
936}
937
938/// Which failure a leading datagram `Type Flags` value that is not one a reader
939/// wants is.
940///
941/// The same split as `stream_type_error`, read against the datagram table.
942/// Section 11.2.1's three conditions each get an arm, so a log names which one
943/// a value fell into rather than reporting all three as one complaint; a value
944/// too wide to be a one-byte flags field at all is Section 6.4.1's
945/// [`CodecError::UnknownDatagramType`] instead.
946///
947/// The padding datagram is why the [`CodecError::InvalidField`] arm exists.
948/// [`PADDING_DATAGRAM_TYPE`] is assigned, so a datagram carrying it is not
949/// unknown; it simply carries no Object, and refusing it must not end the
950/// session.
951fn datagram_type_error(raw: u64) -> CodecError {
952 if raw == PADDING_DATAGRAM_TYPE || datagram_type_is_valid(raw) {
953 return CodecError::InvalidField;
954 }
955 if raw > 0xFF {
956 return CodecError::UnknownDatagramType(raw);
957 }
958 let t = raw as u8;
959 if t & DATAGRAM_RESERVED_BIT != 0 {
960 CodecError::InvalidDatagramTypeValue {
961 raw,
962 detail: "bit 4 (0x10) is reserved for a datagram and must be zero",
963 }
964 } else if t & DATAGRAM_STATUS_BIT != 0 && t & DATAGRAM_END_OF_GROUP_BIT != 0 {
965 CodecError::InvalidDatagramTypeValue {
966 raw,
967 detail: "it sets both the STATUS bit and the END_OF_GROUP bit",
968 }
969 } else {
970 CodecError::InvalidDatagramTypeValue {
971 raw,
972 detail: "it sets a bit whose meaning is not specified for a datagram",
973 }
974 }
975}
976
977#[derive(Debug, Clone)]
978pub struct DatagramHeader {
979 pub datagram_type: u8,
980 pub track_alias: VarInt,
981 pub group_id: VarInt,
982 pub object_id: VarInt,
983 pub publisher_priority: Option<u8>,
984 /// Raw properties bytes, excluding the byte-length prefix that precedes
985 /// them on the wire. Present only when `datagram_type` sets the PROPERTIES
986 /// bit (0x01), and empty otherwise — the bit is what puts the block on the
987 /// wire, so contents held here with the bit clear are not written.
988 ///
989 /// Opaque: [`Self::encode`] re-emits the prefix and these bytes verbatim,
990 /// and [`Self::decode`] copies them out the same way, so a datagram can be
991 /// decoded and re-encoded without understanding what its properties mean.
992 /// The block sits between the publisher priority and the status field, so
993 /// leaving it out of the struct would put the status where the decoder
994 /// looks for the properties length.
995 pub properties: Vec<u8>,
996 /// The object's status, carried on the wire only when `datagram_type` sets
997 /// the STATUS bit (0x20): such a datagram holds a one-byte status code in
998 /// place of a payload. `None` with the bit set is written as
999 /// [`ObjectStatus::Normal`]; a status with the bit clear is not written at
1000 /// all, because the bit is what puts the field on the wire.
1001 ///
1002 /// `None` is not "no status": a datagram whose type leaves the STATUS bit
1003 /// clear carries a payload, and the status of an Object that carries a
1004 /// payload is [`ObjectStatus::Normal`], the only row of the Object Status
1005 /// registry (draft-23 Section 16.9) permitting one. [`Self::status`]
1006 /// resolves the field either way.
1007 ///
1008 /// Typed rather than a bare byte. The wire field is one octet with 256
1009 /// values, and draft-23 Section 11.1.1 assigns three of them; the
1010 /// decoder refuses the other 253, and this type is that same refusal on
1011 /// the encode side — [`Self::encode`] is infallible precisely because a
1012 /// status it could not legally write cannot be built.
1013 pub object_status: Option<ObjectStatus>,
1014}
1015
1016impl DatagramHeader {
1017 /// Decode a datagram header, `Type Flags` field included.
1018 ///
1019 /// The field is read as a whole varint of whatever width the sender used,
1020 /// and a non-minimal spelling of a valid value is accepted, for the reason
1021 /// given on [`SubgroupHeader::decode`] and in the module documentation.
1022 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1023 let raw = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
1024 validate_datagram_type(raw)?;
1025 // Narrowing is safe: `validate_datagram_type` has already refused every
1026 // value with a bit set above the specified five.
1027 let datagram_type = raw as u8;
1028
1029 let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
1030 let group_id = VarInt::decode_moqt::<Wire>(buf)?;
1031
1032 let object_id = if datagram_type & DATAGRAM_ZERO_OBJECT_ID_BIT != 0 {
1033 VarInt::from_usize(0)
1034 } else {
1035 VarInt::decode_moqt::<Wire>(buf)?
1036 };
1037
1038 let publisher_priority = if datagram_type & DATAGRAM_DEFAULT_PRIORITY_BIT == 0 {
1039 if buf.remaining() < 1 {
1040 return Err(CodecError::UnexpectedEnd);
1041 }
1042 Some(buf.get_u8())
1043 } else {
1044 None
1045 };
1046
1047 let properties = if datagram_type & DATAGRAM_PROPERTIES_BIT != 0 {
1048 let props_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1049 crate::types::read_bytes(buf, props_len)?
1050 } else {
1051 Vec::new()
1052 };
1053
1054 let object_status = if datagram_type & DATAGRAM_STATUS_BIT != 0 {
1055 if buf.remaining() < 1 {
1056 return Err(CodecError::UnexpectedEnd);
1057 }
1058 let status = buf.get_u8();
1059 Some(decoded_status(status as u64)?)
1060 } else {
1061 None
1062 };
1063
1064 // Two rules of draft-23 Section 11.2.1 reach the properties block just
1065 // read, and neither is applied here — [`Self::properties_permitted`]
1066 // and [`Self::properties_block_well_formed`] report them instead, and
1067 // [`Self::encode_checked`] refuses to write either shape:
1068 //
1069 // - "If an endpoint receives a datagram with the PROPERTIES bit set
1070 // and an Properties Length of 0, it MUST close the session with
1071 // MALFORMED_OBJECT."
1072 // - "If an Object Datagram includes both the STATUS bit and
1073 // PROPERTIES bit, and the Object Status is not Normal (0x0), the
1074 // endpoint MUST close the session with INVALID_OBJECT_STATUS,
1075 // because only Normal Objects can have Properties."
1076 //
1077 // Both describe a datagram that is well framed and non-conforming: the
1078 // fields are all where the layout puts them and every one of them
1079 // parses, so a decoder can read the datagram back exactly as it
1080 // arrived. Refusing here would leave this module unable to reproduce a
1081 // capture containing one, and both rules address an endpoint receiving
1082 // such a datagram, so the endpoint is where they are enforced — the
1083 // same division [`SubgroupObject::properties_permitted`] explains.
1084 //
1085 // The Type rules above are the contrast, and the contrast is what
1086 // decides it: an invalid Type names no layout at all, so reading on
1087 // invents the fields behind it rather than reporting them.
1088
1089 Ok(DatagramHeader {
1090 datagram_type,
1091 track_alias,
1092 group_id,
1093 object_id,
1094 publisher_priority,
1095 properties,
1096 object_status,
1097 })
1098 }
1099
1100 /// Decode one whole datagram: the header, then the payload that runs to the
1101 /// end of `buf`.
1102 ///
1103 /// `buf` must hold exactly one transport datagram and nothing else, since
1104 /// that boundary is the only thing that delimits the payload — draft-23
1105 /// Section 11.2.1: "There is no explicit length field for the Object
1106 /// Payload; the entirety of the transport datagram following the Object
1107 /// header contains the payload."
1108 ///
1109 /// Which is why the refusal lives here and not in [`Self::decode`]. A
1110 /// datagram whose type sets the STATUS bit has no payload at all — the same
1111 /// section: "When set to 1, the Object Status field is present and there is
1112 /// no Object Payload" — so trailing bytes after its status are not a short
1113 /// payload or an odd one, they are bytes the frame does not define. A
1114 /// decoder that stops at the header cannot see them, and a caller that
1115 /// treats whatever is left as the payload hands the application content the
1116 /// publisher never framed as content. That is the case this refuses, and it
1117 /// bites hardest on a status datagram carrying the Normal code 0x0, whose
1118 /// status alone would report a payload as permitted.
1119 ///
1120 /// The same refusal covers a status the registry forbids a payload to, per
1121 /// Section 11.1.1 and the "Payload" column of Section 16.9.
1122 ///
1123 /// Errors with [`CodecError::PayloadNotPermitted`] when bytes remain and
1124 /// the header forbids them, naming which of the two rules refused them.
1125 ///
1126 /// # The two Properties rules are enforced here too
1127 ///
1128 /// Section 11.2.1 states them of a receiving endpoint, and this is the
1129 /// endpoint's read:
1130 ///
1131 /// * "If an endpoint receives a datagram with the PROPERTIES bit set and an
1132 /// Properties Length of 0, it MUST close the session with
1133 /// MALFORMED_OBJECT." The bit and a zero length are two ways to spell
1134 /// *no properties* and a datagram may use only the first, because a
1135 /// datagram with none has a Type Flags value that says so and the block
1136 /// costs bytes the flags already saved. A subgroup stream says the
1137 /// opposite in Section 11.3.1 — there the PROPERTIES bit is fixed for the
1138 /// whole stream, so an object with no properties has nowhere else to say
1139 /// so and a zero-length block is the required spelling.
1140 /// * "If an Object Datagram includes both the STATUS bit and PROPERTIES
1141 /// bit, and the Object Status is not Normal (0x0), the endpoint MUST close
1142 /// the session with INVALID_OBJECT_STATUS, because only Normal Objects can
1143 /// have Properties."
1144 ///
1145 /// Both errors are [`CodecError::InvalidField`].
1146 ///
1147 /// [`Self::decode`] does **not** apply them, and the split is deliberate.
1148 /// Both describe a datagram that is well framed and non-conforming: every
1149 /// field is where the layout puts it and every one of them parses, so the
1150 /// header reads back exactly as it arrived and a tool reproducing a capture
1151 /// can re-emit it. What it may not do is hand such a datagram to an
1152 /// application as an ordinary Object, which is what this entry point would
1153 /// be doing. [`Self::properties_block_well_formed`] and
1154 /// [`Self::properties_permitted`] report the two for a caller that wants
1155 /// the header without the judgement, and [`Self::encode_checked`] refuses
1156 /// to write either shape.
1157 ///
1158 /// A property block that does not parse is refused by
1159 /// [`crate::object_properties::parse_object_properties`], and the Prior
1160 /// Group ID Gap and Prior Object ID Gap conditions one Object shows by
1161 /// [`PriorGaps::from_properties`].
1162 pub fn decode_object(buf: &mut impl Buf) -> Result<(Self, Vec<u8>), CodecError> {
1163 let header = Self::decode(buf)?;
1164 if !header.properties_block_well_formed() || !header.properties_permitted() {
1165 return Err(CodecError::InvalidField);
1166 }
1167 let properties = crate::object_properties::parse_object_properties(
1168 crate::version::DraftVersion::Draft23,
1169 &header.properties,
1170 )?;
1171 PriorGaps::from_properties(
1172 &properties,
1173 header.group_id.into_inner(),
1174 header.object_id.into_inner(),
1175 )?;
1176 let payload = crate::types::read_bytes(buf, buf.remaining())?;
1177 if !payload.is_empty() && !header.permits_payload() {
1178 return Err(CodecError::PayloadNotPermitted {
1179 status: header.status().as_u64(),
1180 len: payload.len(),
1181 detail: if header.has_status() {
1182 "its type states a status in place of a payload"
1183 } else {
1184 "its status is registered as forbidding one"
1185 },
1186 });
1187 }
1188 Ok((header, payload))
1189 }
1190
1191 /// Serialize the header, refusing a status the framing cannot carry.
1192 ///
1193 /// A datagram states a status only when its type byte sets the STATUS bit
1194 /// (0x20). With the bit clear there is no status field on the wire, so an
1195 /// `object_status` of anything but [`ObjectStatus::Normal`] has nowhere to
1196 /// go: [`Self::encode`] drops it, and the datagram parses back as an
1197 /// ordinary payload object. An End of Group marker written that way does
1198 /// not arrive late or malformed — it does not arrive at all, and the
1199 /// receiver sees a normal object in its place.
1200 ///
1201 /// [`ObjectStatus::Normal`] with the bit clear is not that case and is
1202 /// accepted. It is the status the encoding elides for every object that
1203 /// carries a payload, so stating it asks for exactly the bytes leaving it
1204 /// out asks for, and nothing is lost.
1205 ///
1206 /// Errors with [`CodecError::InvalidField`] on the lossy combination,
1207 /// before any byte is written, so a refused header leaves `buf` untouched.
1208 /// This is the datagram half of the rule
1209 /// [`SubgroupObjectReader::write_object`] applies on a subgroup stream.
1210 ///
1211 /// Also errors with [`CodecError::InvalidField`] for a Type value draft-23
1212 /// Section 11.2.1 lists as invalid, and for the same reason: a datagram
1213 /// written with one could not be read back by [`Self::decode`], and a
1214 /// codec whose two halves disagree about which datagrams exist cannot be
1215 /// used to rewrite captured traffic.
1216 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1217 validate_datagram_type(self.datagram_type as u64)?;
1218 if !self.has_status() && matches!(self.object_status, Some(s) if s != ObjectStatus::Normal)
1219 {
1220 return Err(CodecError::InvalidField);
1221 }
1222 // The two properties rules of Section 11.2.1. [`Self::decode`] reports
1223 // both rather than refusing them, because the datagrams they describe
1224 // are well framed and a codec that could not read one could not
1225 // reproduce a capture containing it. Writing one is the other
1226 // direction and has no such excuse: a conforming peer answers either
1227 // with a session close, so emitting one costs the session and not
1228 // merely the datagram.
1229 if !self.properties_block_well_formed() || !self.properties_permitted() {
1230 return Err(CodecError::InvalidField);
1231 }
1232 self.encode(buf);
1233 Ok(())
1234 }
1235
1236 /// Serialize the header exactly as its type byte describes it.
1237 ///
1238 /// Every field the type byte announces is written, in the order
1239 /// [`Self::decode`] reads them, so the bytes this produces always parse
1240 /// back. The properties block in particular has to be written here: it
1241 /// sits ahead of the status field, and a datagram that skipped it would
1242 /// offer the status byte where the decoder reads the block's length.
1243 ///
1244 /// The type byte is taken as the authority on framing, which is what makes
1245 /// this infallible — and what makes it lossy when the struct disagrees with
1246 /// itself. An `object_status` set while the type byte leaves the STATUS bit
1247 /// clear is discarded here without a word, and a Type value draft-23
1248 /// forbids is written out as readily as one it assigns, including Types
1249 /// this module's own [`Self::decode`] refuses. Prefer
1250 /// [`Self::encode_checked`], which refuses both instead of resolving them.
1251 pub fn encode(&self, buf: &mut impl BufMut) {
1252 buf.put_u8(self.datagram_type);
1253 self.track_alias.encode_moqt::<Wire>(buf);
1254 self.group_id.encode_moqt::<Wire>(buf);
1255
1256 if self.datagram_type & DATAGRAM_ZERO_OBJECT_ID_BIT == 0 {
1257 self.object_id.encode_moqt::<Wire>(buf);
1258 }
1259
1260 if self.datagram_type & DATAGRAM_DEFAULT_PRIORITY_BIT == 0 {
1261 buf.put_u8(self.publisher_priority.unwrap_or(128));
1262 }
1263
1264 if self.datagram_type & DATAGRAM_PROPERTIES_BIT != 0 {
1265 VarInt::from_usize(self.properties.len()).encode_moqt::<Wire>(buf);
1266 buf.put_slice(&self.properties);
1267 }
1268
1269 if self.datagram_type & DATAGRAM_STATUS_BIT != 0 {
1270 buf.put_u8(self.object_status.unwrap_or(ObjectStatus::Normal).as_u8());
1271 }
1272 }
1273
1274 pub fn is_end_of_group(&self) -> bool {
1275 self.datagram_type & DATAGRAM_END_OF_GROUP_BIT != 0
1276 }
1277
1278 pub fn has_status(&self) -> bool {
1279 self.datagram_type & DATAGRAM_STATUS_BIT != 0
1280 }
1281
1282 /// `true` when the type byte sets the PROPERTIES bit (0x01), which is what
1283 /// puts the properties block on the wire.
1284 ///
1285 /// Reports the framing, not the contents. A decoded datagram with this set
1286 /// always has a non-empty [`Self::properties`], because [`Self::decode`]
1287 /// refuses a zero-length block; a header built by hand can hold the two
1288 /// apart, and [`Self::encode_checked`] is what refuses that.
1289 pub fn has_properties(&self) -> bool {
1290 self.datagram_type & DATAGRAM_PROPERTIES_BIT != 0
1291 }
1292
1293 /// The datagram's object status, with the one the encoding elides filled
1294 /// in.
1295 ///
1296 /// A datagram states a status only when its type sets the STATUS bit, and
1297 /// such a datagram has no payload. One without the bit is all payload, and
1298 /// its status is [`ObjectStatus::Normal`] — the sole row of the Object
1299 /// Status registry (draft-23 Section 16.9) permitting a payload, so the
1300 /// only status it could have had.
1301 pub fn status(&self) -> ObjectStatus {
1302 self.object_status.unwrap_or(ObjectStatus::Normal)
1303 }
1304
1305 /// Whether the bytes after this datagram's header are allowed to exist.
1306 ///
1307 /// Two independent rules forbid them, and this reports both:
1308 ///
1309 /// - The framing. Draft-23 Section 11.2.1: "The STATUS bit (0x20)
1310 /// indicates whether the datagram contains an Object Status or Object
1311 /// Payload. When set to 1, the Object Status field is present and there
1312 /// is no Object Payload." A datagram that states a status has no payload
1313 /// field at all, whichever status it states — so a STATUS datagram
1314 /// carrying the Normal code 0x0 has no more room for bytes than one
1315 /// carrying End of Group.
1316 /// - The status. Section 11.1.1 and the Object Status registry's
1317 /// "Payload" column, Section 16.9: an Object has an empty payload unless
1318 /// its status is registered as permitting one. This half reaches a
1319 /// datagram whose type byte leaves the STATUS bit clear while the value
1320 /// claims a status that forbids a payload — a disagreement
1321 /// [`Self::encode_checked`] refuses to write, and one a decoded header
1322 /// never shows.
1323 ///
1324 /// The first is the rule a decoded datagram can actually trip, and reading
1325 /// the registry alone misses it: `Some(ObjectStatus::Normal)` under a type
1326 /// byte with the STATUS bit set is exactly the case where the payload the
1327 /// draft says does not exist would otherwise be handed to the application
1328 /// as the object's content, because Normal is the one status the registry
1329 /// marks as permitting a payload.
1330 ///
1331 /// Distinct from [`Self::has_status`], which reports how the datagram is
1332 /// framed rather than whether a payload may follow. A caller holding the
1333 /// bytes after the header wants this one; [`Self::decode_object`] applies
1334 /// it for a caller who would rather the decode simply fail.
1335 pub fn permits_payload(&self) -> bool {
1336 if self.has_status() {
1337 return false;
1338 }
1339 self.status().permits_payload()
1340 }
1341
1342 /// Whether this datagram's status is allowed to carry the properties it
1343 /// has.
1344 ///
1345 /// The same rule the subgroup form obeys. Draft-23 Section 11.2.1 builds
1346 /// the datagram's Properties field out of "the Object Properties structure
1347 /// defined in Section 11.1.2", and that section is where the rule sits:
1348 /// "If an endpoint receives properties on an Object with status that is not
1349 /// Normal, it MUST close the session with INVALID_OBJECT_STATUS."
1350 ///
1351 /// [`Self::decode`] reports this and [`Self::decode_object`] refuses it, as
1352 /// the subgroup readers split it.
1353 pub fn properties_permitted(&self) -> bool {
1354 self.properties.is_empty() || self.status() == ObjectStatus::Normal
1355 }
1356
1357 /// Whether the properties block is framed the way a datagram may frame it.
1358 ///
1359 /// Draft-23 Section 11.2.1: "If an endpoint receives a datagram with the
1360 /// PROPERTIES bit set and an Properties Length of 0, it MUST close the
1361 /// session with MALFORMED_OBJECT."
1362 ///
1363 /// The bit and a zero length are two ways to spell "no properties", and on
1364 /// a datagram they are not interchangeable: a datagram with none has a type
1365 /// byte that says so, and the block costs bytes the type byte already
1366 /// saved. This rule is the datagram's alone. A subgroup stream says the
1367 /// opposite in Section 11.3.1 — "Objects with no properties or non-Normal
1368 /// status set Properties Length to 0" — because there the PROPERTIES bit is fixed for the whole
1369 /// stream, so an object with no properties has nowhere else to say so and a
1370 /// zero-length block is the required spelling rather than a violation.
1371 ///
1372 /// The mirror case is not a wire state but is a state this struct can hold:
1373 /// properties with the bit clear. [`Self::encode`] drops them without a
1374 /// word, so this reports that too, and [`Self::encode_checked`] refuses
1375 /// both.
1376 pub fn properties_block_well_formed(&self) -> bool {
1377 self.has_properties() != self.properties.is_empty()
1378 }
1379}
1380
1381// ── Fetch Header ──────────────────────────────────────────────
1382
1383const FETCH_STREAM_TYPE: u64 = 0x05;
1384
1385/// The head of a fetch stream: the stream type `0x05` and a Request ID.
1386///
1387/// Section 11.4.1: "all objects on the stream belong to the track requested in
1388/// the message identified by Request ID". On a fill fetch stream that is the
1389/// subscription's Request ID, a PUBLISH's included, or a REQUEST_UPDATE's
1390/// (Section 3.4), not a FETCH's.
1391#[derive(Debug, Clone)]
1392pub struct FetchHeader {
1393 pub request_id: VarInt,
1394}
1395
1396impl FetchHeader {
1397 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1398 let stream_type = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
1399 if stream_type != FETCH_STREAM_TYPE {
1400 return Err(stream_type_error(stream_type));
1401 }
1402 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1403 Ok(FetchHeader { request_id })
1404 }
1405
1406 pub fn encode(&self, buf: &mut impl BufMut) {
1407 VarInt::from_usize(FETCH_STREAM_TYPE as usize).encode_moqt::<Wire>(buf);
1408 self.request_id.encode_moqt::<Wire>(buf);
1409 }
1410}
1411
1412// ── Fetch objects ─────────────────────────────────────────────
1413
1414/// Serialization Flags bits 0-1, the Subgroup ID encoding
1415/// (draft-23 Section 11.4.1.1, Table 9).
1416const FETCH_SUBGROUP_ID_MODE_MASK: u64 = 0x03;
1417/// Subgroup ID mode 0b11: an explicit Subgroup ID field is on the wire.
1418const FETCH_SUBGROUP_ID_EXPLICIT: u64 = 0b11;
1419/// Table 10 flag: an Object ID Delta field is present.
1420const FETCH_OBJECT_ID_DELTA_BIT: u64 = 0x04;
1421/// Table 10 flag: a Group ID Delta field is present.
1422const FETCH_GROUP_ID_DELTA_BIT: u64 = 0x08;
1423/// Table 10 flag: a Publisher Priority field is present.
1424const FETCH_PRIORITY_BIT: u64 = 0x10;
1425/// Table 10 flag: a Properties field is present.
1426const FETCH_PROPERTIES_BIT: u64 = 0x20;
1427/// Table 10 flag: the Object's Delivery Mode is Datagram, so it has no
1428/// Subgroup ID and the two low bits are to be ignored.
1429const FETCH_DATAGRAM_BIT: u64 = 0x40;
1430/// The largest Serialization Flags value whose bits are flags. Section 11.4.1:
1431/// "When less than 128, the bits represent flags described in Section
1432/// 11.4.1.1."
1433const FETCH_FLAGS_MAX: u64 = 0x7F;
1434/// Table 8: End of Group.
1435const FETCH_END_OF_GROUP: u64 = 0x80;
1436/// Table 8: End of Track.
1437const FETCH_END_OF_TRACK: u64 = 0x81;
1438/// Table 8: End of Non-Existent Range.
1439const FETCH_END_OF_NON_EXISTENT_RANGE: u64 = 0x8C;
1440/// Table 8: End of Unknown Range.
1441const FETCH_END_OF_UNKNOWN_RANGE: u64 = 0x10C;
1442/// Table 8: End of Timed-Out Range.
1443const FETCH_END_OF_TIMED_OUT_RANGE: u64 = 0x20C;
1444/// Table 8: End of Gone Range. The two-byte `84 0C`.
1445const FETCH_END_OF_GONE_RANGE: u64 = 0x40C;
1446
1447/// What an End of Range indicator says about the Locations it covers.
1448///
1449/// Section 11.4.2: "All Objects with Locations between the last serialized
1450/// Object, if any, and the Location decoded from this indicator, inclusive,
1451/// either do not exist (when Serialization Flags is 0x8C), are Unknown
1452/// (0x10C), are timed out (0x20C), or are Gone (0x40C)."
1453#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1454pub enum FetchEndOfRange {
1455 /// Serialization Flags 0x8C. The Objects in the range do not exist.
1456 NonExistent,
1457 /// Serialization Flags 0x10C. The Objects in the range are Unknown.
1458 Unknown,
1459 /// Serialization Flags 0x20C. The Objects in the range timed out.
1460 TimedOut,
1461 /// Serialization Flags 0x40C. The Objects in the range are Gone: Section
1462 /// 2.1.2, "The original publisher cannot deliver the Object and will not
1463 /// be able to, but does not know whether it exists."
1464 Gone,
1465}
1466
1467/// An indicator that ends a Group or the Track rather than a range.
1468///
1469/// Section 11.4.2.2: "When Serialization Flags is 0x80 (End of Group) or 0x81
1470/// (End of Track), no other fields are present. The prior Object is the last
1471/// Object in its Group, or in the Track, respectively. End of Track also
1472/// indicates End of Group. The prior Object ID becomes 2^64-1."
1473#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1474pub enum FetchEndMarker {
1475 /// Serialization Flags 0x80.
1476 EndOfGroup,
1477 /// Serialization Flags 0x81.
1478 EndOfTrack,
1479}
1480
1481/// Which fields a Serialization Flags value puts on the wire.
1482///
1483/// Derived once by [`FetchObjectHeader::layout`] and then used by both
1484/// [`FetchObjectHeader::decode`] and [`FetchObjectHeader::encode`], so the two
1485/// cannot drift into disagreeing about a shape.
1486#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1487struct FetchObjectLayout {
1488 group_id_delta: bool,
1489 subgroup_id: bool,
1490 object_id_delta: bool,
1491 publisher_priority: bool,
1492 properties: bool,
1493 payload_length: bool,
1494}
1495
1496/// One entry on a draft-23 fetch stream, up to but not including its payload.
1497///
1498/// ```text
1499/// {
1500/// Serialization Flags (vi64),
1501/// [Group ID Delta (vi64),]
1502/// [Subgroup ID (vi64),]
1503/// [Object ID Delta (vi64),]
1504/// [Publisher Priority (8),]
1505/// [Properties (..),]
1506/// Object Payload Length (vi64),
1507/// [Object Payload (..),]
1508/// }
1509/// ```
1510///
1511/// This holds what is on the wire: the deltas, not the IDs they resolve to.
1512/// [`FetchObjectReader`] resolves them (Section 11.4.1.1 and Section 11.4.2.1).
1513///
1514/// An entry may be an indicator rather than an Object: [`Self::end_of_range`]
1515/// and [`Self::end_marker`] say which.
1516#[derive(Debug, Clone, PartialEq, Eq)]
1517pub struct FetchObjectHeader {
1518 /// The raw Serialization Flags value, kept whole because it also names an
1519 /// indicator and because re-encoding must reproduce the publisher's value.
1520 pub serialization_flags: VarInt,
1521 /// Group ID Delta, present when the flags set 0x08 and on every End of
1522 /// Range indicator.
1523 pub group_id_delta: Option<VarInt>,
1524 /// An explicit Subgroup ID, present only when the two low flag bits are
1525 /// 0b11 and the Datagram bit is clear.
1526 pub subgroup_id: Option<VarInt>,
1527 /// Object ID Delta, present when the flags set 0x04 and on every End of
1528 /// Range indicator.
1529 pub object_id_delta: Option<VarInt>,
1530 /// Publisher Priority, present when the flags set 0x10.
1531 pub publisher_priority: Option<u8>,
1532 /// Raw properties bytes, excluding their length prefix, and `None` when
1533 /// the flags leave 0x20 clear.
1534 pub properties: Option<Vec<u8>>,
1535 /// Object Payload Length. Zero, and not on the wire, for End of Group and
1536 /// End of Track.
1537 pub payload_length: VarInt,
1538}
1539
1540impl FetchObjectHeader {
1541 /// The Serialization Flags as a plain integer.
1542 pub fn flags(&self) -> u64 {
1543 self.serialization_flags.into_inner()
1544 }
1545
1546 /// Which End of Range indicator this is, or `None` for anything else.
1547 pub fn end_of_range(&self) -> Option<FetchEndOfRange> {
1548 match self.flags() {
1549 FETCH_END_OF_NON_EXISTENT_RANGE => Some(FetchEndOfRange::NonExistent),
1550 FETCH_END_OF_UNKNOWN_RANGE => Some(FetchEndOfRange::Unknown),
1551 FETCH_END_OF_TIMED_OUT_RANGE => Some(FetchEndOfRange::TimedOut),
1552 FETCH_END_OF_GONE_RANGE => Some(FetchEndOfRange::Gone),
1553 _ => None,
1554 }
1555 }
1556
1557 /// Which of End of Group and End of Track this is, or `None` for anything
1558 /// else.
1559 pub fn end_marker(&self) -> Option<FetchEndMarker> {
1560 match self.flags() {
1561 FETCH_END_OF_GROUP => Some(FetchEndMarker::EndOfGroup),
1562 FETCH_END_OF_TRACK => Some(FetchEndMarker::EndOfTrack),
1563 _ => None,
1564 }
1565 }
1566
1567 /// `true` for any Table 8 indicator, `false` for an Object.
1568 pub fn is_indicator(&self) -> bool {
1569 self.end_of_range().is_some() || self.end_marker().is_some()
1570 }
1571
1572 /// The two-bit Subgroup ID mode, `flags & 0x03` (Section 11.4.1.1,
1573 /// Table 9). Meaningless when [`Self::is_datagram`] is true.
1574 pub fn subgroup_id_mode(&self) -> u8 {
1575 (self.flags() & FETCH_SUBGROUP_ID_MODE_MASK) as u8
1576 }
1577
1578 /// `true` when the flags set 0x40, marking an Object whose Delivery Mode
1579 /// is Datagram, which has no Subgroup ID.
1580 pub fn is_datagram(&self) -> bool {
1581 self.flags() < FETCH_END_OF_GROUP && self.flags() & FETCH_DATAGRAM_BIT != 0
1582 }
1583
1584 /// Which fields `flags` puts on the wire.
1585 ///
1586 /// Values below 128 are flags. Table 8 defines six values above that, and
1587 /// Section 11.4.1: "If an endpoint receives any other value greater than
1588 /// 127, it MUST close the session with MALFORMED_OBJECT."
1589 ///
1590 /// Section 11.4.2: on an End of Range indicator "the Group ID Delta and
1591 /// Object ID Delta fields are always present. Subgroup ID, Priority and
1592 /// Properties are not present." Figure 29 makes Object Payload Length
1593 /// mandatory and Section 11.4.2 does not list it among the absent fields,
1594 /// so it is present, and written as 0. On End of Group and End of Track "no
1595 /// other fields are present", so there is no Object Payload Length either.
1596 fn layout(flags: u64) -> Result<FetchObjectLayout, CodecError> {
1597 match flags {
1598 FETCH_END_OF_NON_EXISTENT_RANGE
1599 | FETCH_END_OF_UNKNOWN_RANGE
1600 | FETCH_END_OF_TIMED_OUT_RANGE
1601 | FETCH_END_OF_GONE_RANGE => {
1602 return Ok(FetchObjectLayout {
1603 group_id_delta: true,
1604 subgroup_id: false,
1605 object_id_delta: true,
1606 publisher_priority: false,
1607 properties: false,
1608 payload_length: true,
1609 });
1610 }
1611 FETCH_END_OF_GROUP | FETCH_END_OF_TRACK => {
1612 return Ok(FetchObjectLayout {
1613 group_id_delta: false,
1614 subgroup_id: false,
1615 object_id_delta: false,
1616 publisher_priority: false,
1617 properties: false,
1618 payload_length: false,
1619 });
1620 }
1621 _ => {}
1622 }
1623 if flags > FETCH_FLAGS_MAX {
1624 return Err(CodecError::MalformedObject {
1625 detail: "its Serialization Flags are above 127 and not a Table 8 indicator",
1626 });
1627 }
1628 Ok(FetchObjectLayout {
1629 group_id_delta: flags & FETCH_GROUP_ID_DELTA_BIT != 0,
1630 subgroup_id: flags & FETCH_DATAGRAM_BIT == 0
1631 && flags & FETCH_SUBGROUP_ID_MODE_MASK == FETCH_SUBGROUP_ID_EXPLICIT,
1632 object_id_delta: flags & FETCH_OBJECT_ID_DELTA_BIT != 0,
1633 publisher_priority: flags & FETCH_PRIORITY_BIT != 0,
1634 properties: flags & FETCH_PROPERTIES_BIT != 0,
1635 payload_length: true,
1636 })
1637 }
1638
1639 /// Decode one fetch entry's header, leaving the payload in `buf`.
1640 ///
1641 /// The flags are validated before any field is read, because they are what
1642 /// says where the fields are.
1643 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1644 let serialization_flags = VarInt::decode_moqt::<Wire>(buf)?;
1645 let layout = Self::layout(serialization_flags.into_inner())?;
1646
1647 let group_id_delta =
1648 layout.group_id_delta.then(|| VarInt::decode_moqt::<Wire>(buf)).transpose()?;
1649 let subgroup_id =
1650 layout.subgroup_id.then(|| VarInt::decode_moqt::<Wire>(buf)).transpose()?;
1651 let object_id_delta =
1652 layout.object_id_delta.then(|| VarInt::decode_moqt::<Wire>(buf)).transpose()?;
1653
1654 let publisher_priority = if layout.publisher_priority {
1655 if buf.remaining() < 1 {
1656 return Err(CodecError::UnexpectedEnd);
1657 }
1658 Some(buf.get_u8())
1659 } else {
1660 None
1661 };
1662
1663 let properties = if layout.properties {
1664 let len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1665 Some(crate::types::read_bytes(buf, len)?)
1666 } else {
1667 None
1668 };
1669
1670 let payload_length = if layout.payload_length {
1671 VarInt::decode_moqt::<Wire>(buf)?
1672 } else {
1673 VarInt::from_u64_moqt(0)
1674 };
1675
1676 Ok(FetchObjectHeader {
1677 serialization_flags,
1678 group_id_delta,
1679 subgroup_id,
1680 object_id_delta,
1681 publisher_priority,
1682 properties,
1683 payload_length,
1684 })
1685 }
1686
1687 /// Serialize the header, refusing one whose fields disagree with its own
1688 /// Serialization Flags.
1689 ///
1690 /// [`CodecError::InvalidField`] when an optional field is present while its
1691 /// flag says absent or the reverse, and for an End of Group or End of
1692 /// Track with a non-zero payload length. Checked before any byte is
1693 /// written.
1694 pub fn encode(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1695 let layout = Self::layout(self.flags())?;
1696 if layout.group_id_delta != self.group_id_delta.is_some()
1697 || layout.subgroup_id != self.subgroup_id.is_some()
1698 || layout.object_id_delta != self.object_id_delta.is_some()
1699 || layout.publisher_priority != self.publisher_priority.is_some()
1700 || layout.properties != self.properties.is_some()
1701 || (!layout.payload_length && self.payload_length.into_inner() != 0)
1702 {
1703 return Err(CodecError::InvalidField);
1704 }
1705
1706 self.serialization_flags.encode_moqt::<Wire>(buf);
1707 // Wire order, from Figure 29.
1708 for field in
1709 [self.group_id_delta, self.subgroup_id, self.object_id_delta].into_iter().flatten()
1710 {
1711 field.encode_moqt::<Wire>(buf);
1712 }
1713 if let Some(priority) = self.publisher_priority {
1714 buf.put_u8(priority);
1715 }
1716 if let Some(properties) = &self.properties {
1717 VarInt::from_usize(properties.len()).encode_moqt::<Wire>(buf);
1718 buf.put_slice(properties);
1719 }
1720 if layout.payload_length {
1721 self.payload_length.encode_moqt::<Wire>(buf);
1722 }
1723 Ok(())
1724 }
1725}
1726
1727/// The order a FETCH response's Groups arrive in, which decides how a Group ID
1728/// Delta is applied.
1729///
1730/// Draft-23 Section 11.4.1.1: "If the Group Order is Ascending, the Group ID is
1731/// the prior Object's Group ID plus the Group ID Delta + 1. If the Group Order
1732/// is Descending, the Group ID is the prior Object's Group ID minus the (Group
1733/// ID Delta + 1)." The order is settled by the FETCH's GROUP_ORDER parameter
1734/// (Section 9.19.8), not by the data stream.
1735#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1736pub enum GroupOrder {
1737 /// Group IDs increase: a delta adds to the prior Group ID.
1738 Ascending,
1739 /// Group IDs decrease: a delta subtracts from the prior Group ID.
1740 Descending,
1741}
1742
1743/// One entry from a FETCH stream with its delta-encoded fields resolved.
1744#[derive(Debug, Clone, PartialEq, Eq)]
1745pub struct FetchObject {
1746 /// The entry as it appeared on the wire.
1747 pub header: FetchObjectHeader,
1748 /// Resolved Group ID. On an End of Range indicator, the Group ID of the
1749 /// Location it names; on End of Group or End of Track, the prior Object's.
1750 pub group_id: u64,
1751 /// Resolved Subgroup ID. `None` for an indicator and for an Object whose
1752 /// Delivery Mode is Datagram.
1753 pub subgroup_id: Option<u64>,
1754 /// Resolved Object ID. On an End of Range indicator, the Object ID of the
1755 /// Location it names; on End of Group or End of Track, the prior Object's,
1756 /// which is the last in its Group.
1757 pub object_id: u64,
1758 /// The Publisher Priority in force for this entry, and `None` while no
1759 /// Object has stated one. An indicator carries none of its own.
1760 pub publisher_priority: Option<u8>,
1761}
1762
1763fn malformed(detail: &'static str) -> CodecError {
1764 CodecError::MalformedObject { detail }
1765}
1766
1767fn overflow(detail: &'static str) -> CodecError {
1768 CodecError::DeltaDecodingOverflow { detail }
1769}
1770
1771/// Resolves the delta-encoded fields of the entries on one FETCH stream.
1772///
1773/// Section 11.4.2.2 keeps two kinds of prior state apart. After an End of
1774/// Range indicator, "Prior Group ID and prior Object ID: The decoded values
1775/// from the End of Range indicator", while the prior Subgroup ID and Priority
1776/// come from "the last actual Object before the End of Range indicator". After
1777/// End of Group or End of Track, "The prior Object ID becomes 2^64-1".
1778#[derive(Debug, Clone)]
1779pub struct FetchObjectReader {
1780 group_order: GroupOrder,
1781 /// Group ID and Object ID the next entry resolves against.
1782 prior_location: Option<(u64, u64)>,
1783 /// Subgroup ID of the last actual Object that had one.
1784 prior_subgroup_id: Option<u64>,
1785 /// Publisher Priority of the last actual Object.
1786 prior_publisher_priority: Option<u8>,
1787 /// The Prior gap conditions the stream's Objects show.
1788 gaps: FetchGaps,
1789}
1790
1791impl FetchObjectReader {
1792 /// A reader for a stream whose Groups arrive in `group_order`.
1793 pub fn new(group_order: GroupOrder) -> Self {
1794 Self {
1795 group_order,
1796 prior_location: None,
1797 prior_subgroup_id: None,
1798 prior_publisher_priority: None,
1799 gaps: FetchGaps::new(),
1800 }
1801 }
1802
1803 /// Decode the next entry's header and resolve its fields.
1804 ///
1805 /// Consumes the header only; the Object Payload stays in `buf`.
1806 ///
1807 /// [`CodecError::MalformedObject`] for an entry that references a prior
1808 /// Object that does not exist. Section 11.4.1.1: "If the first Object in
1809 /// the FETCH response uses a flag that references fields in the prior
1810 /// Object, the Subscriber MUST close the session with MALFORMED_OBJECT."
1811 /// End of Group or End of Track as the first entry names a prior Object
1812 /// too, and is refused the same way.
1813 ///
1814 /// [`CodecError::DeltaDecodingOverflow`] for an ID below 0 or above
1815 /// 2^64-1 (Section 11.4.1.1 and Section 11.4.2.1).
1816 ///
1817 /// Refuses, with [`CodecError::MalformedTrack`], the Prior gap conditions an
1818 /// Object shows alone or after the Objects before it on the stream, as
1819 /// [`FetchGaps`] does, and a property block that does not parse as
1820 /// [`crate::object_properties::parse_object_properties`] does.
1821 pub fn read_object_header(&mut self, buf: &mut impl Buf) -> Result<FetchObject, CodecError> {
1822 let object = self.read_frame_header(buf)?;
1823 if !object.header.is_indicator() {
1824 self.gaps.check(
1825 object.group_id,
1826 object.object_id,
1827 object.header.properties.as_deref().unwrap_or_default(),
1828 )?;
1829 }
1830 Ok(object)
1831 }
1832
1833 /// [`Self::read_object_header`] without the Prior gap conditions and without reading
1834 /// inside the property block, for the facade's frame and meta reads, which
1835 /// a forwarder uses.
1836 pub(crate) fn read_frame_header(
1837 &mut self,
1838 buf: &mut impl Buf,
1839 ) -> Result<FetchObject, CodecError> {
1840 let header = FetchObjectHeader::decode(buf)?;
1841
1842 if header.end_marker().is_some() {
1843 let (group_id, object_id) = self
1844 .prior_location
1845 .ok_or_else(|| malformed("End of Group or End of Track has no prior Object"))?;
1846 self.prior_location = Some((group_id, u64::MAX));
1847 return Ok(FetchObject {
1848 header,
1849 group_id,
1850 subgroup_id: None,
1851 object_id,
1852 publisher_priority: self.prior_publisher_priority,
1853 });
1854 }
1855
1856 if header.end_of_range().is_some() {
1857 let (group_id, object_id) = self.range_location(&header)?;
1858 self.prior_location = Some((group_id, object_id));
1859 return Ok(FetchObject {
1860 header,
1861 group_id,
1862 subgroup_id: None,
1863 object_id,
1864 publisher_priority: self.prior_publisher_priority,
1865 });
1866 }
1867
1868 let (group_id, object_id) = self.object_location(&header)?;
1869
1870 let subgroup_id = if header.is_datagram() {
1871 None
1872 } else {
1873 Some(match header.subgroup_id_mode() {
1874 0x00 => 0,
1875 0x01 => self
1876 .prior_subgroup_id
1877 .ok_or_else(|| malformed("it inherits a Subgroup ID with no prior Object"))?,
1878 0x02 => self
1879 .prior_subgroup_id
1880 .ok_or_else(|| malformed("it inherits a Subgroup ID with no prior Object"))?
1881 .checked_add(1)
1882 .ok_or_else(|| overflow("Subgroup ID"))?,
1883 // Mode 0b11, the only value left: the field is on the wire.
1884 _ => header.subgroup_id.ok_or(CodecError::InvalidField)?.into_inner(),
1885 })
1886 };
1887
1888 let publisher_priority = match header.publisher_priority {
1889 Some(p) => p,
1890 None => self
1891 .prior_publisher_priority
1892 .ok_or_else(|| malformed("it inherits a Priority with no prior Object"))?,
1893 };
1894
1895 self.prior_location = Some((group_id, object_id));
1896 // A Datagram-forwarded Object has no Subgroup ID to leave behind, so
1897 // the running one survives it.
1898 if let Some(subgroup_id) = subgroup_id {
1899 self.prior_subgroup_id = Some(subgroup_id);
1900 }
1901 self.prior_publisher_priority = Some(publisher_priority);
1902
1903 Ok(FetchObject {
1904 header,
1905 group_id,
1906 subgroup_id,
1907 object_id,
1908 publisher_priority: Some(publisher_priority),
1909 })
1910 }
1911
1912 /// An Object's Location, Section 11.4.1.1.
1913 fn object_location(&self, header: &FetchObjectHeader) -> Result<(u64, u64), CodecError> {
1914 let Some((prior_group, prior_object)) = self.prior_location else {
1915 // "The first Object MUST include a Group ID Delta and Object ID
1916 // Delta, and these values are the absolute Group ID and Object ID."
1917 return match (header.group_id_delta, header.object_id_delta) {
1918 (Some(g), Some(o)) => Ok((g.into_inner(), o.into_inner())),
1919 _ => Err(malformed("the first Object leaves an ID to a prior Object")),
1920 };
1921 };
1922 let group_id = match header.group_id_delta {
1923 None => prior_group,
1924 Some(delta) => step_group(self.group_order, prior_group, delta.into_inner(), 1)?,
1925 };
1926 let object_id = match (header.group_id_delta.is_some(), header.object_id_delta) {
1927 // "When the Group ID Delta field is present, the Object ID is the
1928 // value of Object ID Delta if present."
1929 (true, Some(delta)) => delta.into_inner(),
1930 // "When the Group ID Delta field is not present, the Object ID is
1931 // the prior Object's ID plus the Object ID Delta + 2 if present."
1932 (false, Some(delta)) => prior_object
1933 .checked_add(delta.into_inner())
1934 .and_then(|v| v.checked_add(2))
1935 .ok_or_else(|| overflow("Object ID"))?,
1936 // "If Object ID Delta is not present, the Object ID is the prior
1937 // Object's ID plus one, regardless of which group it belongs to."
1938 (_, None) => prior_object.checked_add(1).ok_or_else(|| overflow("Object ID"))?,
1939 };
1940 Ok((group_id, object_id))
1941 }
1942
1943 /// An End of Range indicator's Location, Section 11.4.2.1.
1944 fn range_location(&self, header: &FetchObjectHeader) -> Result<(u64, u64), CodecError> {
1945 let (Some(group_delta), Some(object_delta)) =
1946 (header.group_id_delta, header.object_id_delta)
1947 else {
1948 return Err(CodecError::InvalidField);
1949 };
1950 let (group_delta, object_delta) = (group_delta.into_inner(), object_delta.into_inner());
1951 let Some((prior_group, prior_object)) = self.prior_location else {
1952 // "If this End of Range indicator is the first entry after the
1953 // FETCH_HEADER, both fields encode absolute IDs."
1954 return Ok((group_delta, object_delta));
1955 };
1956 // "the Group ID is the prior Group ID plus Group ID Delta for Ascending
1957 // Group Order, or minus Group ID Delta for Descending Group Order. If
1958 // Group ID Delta is zero, the Object ID is the prior Object ID plus
1959 // Object ID Delta + 1. When Group ID Delta is nonzero, the Object ID
1960 // Delta field contains the absolute Object ID in the new Group."
1961 let group_id = step_group(self.group_order, prior_group, group_delta, 0)?;
1962 let object_id = if group_delta == 0 {
1963 prior_object
1964 .checked_add(object_delta)
1965 .and_then(|v| v.checked_add(1))
1966 .ok_or_else(|| overflow("End of Range Object ID"))?
1967 } else {
1968 object_delta
1969 };
1970 Ok((group_id, object_id))
1971 }
1972}
1973
1974/// `prior` moved by `delta + extra` in `order`'s direction.
1975fn step_group(order: GroupOrder, prior: u64, delta: u64, extra: u64) -> Result<u64, CodecError> {
1976 let step = delta.checked_add(extra).ok_or_else(|| overflow("Group ID"))?;
1977 match order {
1978 GroupOrder::Ascending => prior.checked_add(step),
1979 GroupOrder::Descending => prior.checked_sub(step),
1980 }
1981 .ok_or_else(|| overflow("Group ID"))
1982}
1983
1984/// Re-encodes resolved fetch entries onto one FETCH stream.
1985///
1986/// The inverse of [`FetchObjectReader`], for a caller writing a different
1987/// stream from the entries it read: removing an entry changes what the ones
1988/// after it are encoded against. The entry's own header is the preference;
1989/// wherever its shape still encodes the same meaning against the new
1990/// predecessor it is kept, so a stream with nothing removed is reproduced byte
1991/// for byte.
1992///
1993/// [`CodecError::InvalidField`] where no encoding exists: a Group ID that moves
1994/// against the Group Order, an Object ID that does not advance, an Object with
1995/// neither a Subgroup ID nor the Datagram bit, an End of Group or End of Track
1996/// whose Object is not the one written before it, and the arithmetic
1997/// overflows.
1998#[derive(Debug, Clone)]
1999pub struct FetchObjectWriter {
2000 group_order: GroupOrder,
2001 /// Group ID and Object ID the next entry is encoded against.
2002 prior_location: Option<(u64, u64)>,
2003 /// Subgroup ID of the last actual Object written that had one.
2004 prior_subgroup_id: Option<u64>,
2005 /// Publisher Priority of the last actual Object written.
2006 prior_publisher_priority: Option<u8>,
2007}
2008
2009impl FetchObjectWriter {
2010 /// A writer for a stream whose Groups are being written in `group_order`,
2011 /// which has to be the order the FETCH was opened with.
2012 pub fn new(group_order: GroupOrder) -> Self {
2013 Self {
2014 group_order,
2015 prior_location: None,
2016 prior_subgroup_id: None,
2017 prior_publisher_priority: None,
2018 }
2019 }
2020
2021 /// The header that encodes `frame` against everything written so far.
2022 /// Does not advance the writer.
2023 pub fn header_for(&self, frame: &FetchObject) -> Result<FetchObjectHeader, CodecError> {
2024 let original = &frame.header;
2025
2026 if original.end_marker().is_some() {
2027 // Section 11.4.2.2: "The prior Object is the last Object in its
2028 // Group, or in the Track, respectively." The claim is about the
2029 // Object it follows, so it has no encoding after any other.
2030 if self.prior_location != Some((frame.group_id, frame.object_id)) {
2031 return Err(CodecError::InvalidField);
2032 }
2033 return Ok(FetchObjectHeader {
2034 serialization_flags: original.serialization_flags,
2035 group_id_delta: None,
2036 subgroup_id: None,
2037 object_id_delta: None,
2038 publisher_priority: None,
2039 properties: None,
2040 payload_length: VarInt::from_u64_moqt(0),
2041 });
2042 }
2043
2044 if original.end_of_range().is_some() {
2045 let (group_id_delta, object_id_delta) = self.range_fields(frame)?;
2046 return Ok(FetchObjectHeader {
2047 serialization_flags: original.serialization_flags,
2048 group_id_delta: Some(group_id_delta),
2049 subgroup_id: None,
2050 object_id_delta: Some(object_id_delta),
2051 publisher_priority: None,
2052 properties: None,
2053 payload_length: original.payload_length,
2054 });
2055 }
2056
2057 let (group_id_delta, object_id_delta) = self.identity_fields(frame, original)?;
2058 let (subgroup_mode, subgroup_id) = self.subgroup_field(frame, original)?;
2059 let publisher_priority = self.priority_field(frame, original)?;
2060
2061 let mut flags = subgroup_mode;
2062 if original.is_datagram() {
2063 flags |= FETCH_DATAGRAM_BIT;
2064 }
2065 if group_id_delta.is_some() {
2066 flags |= FETCH_GROUP_ID_DELTA_BIT;
2067 }
2068 if object_id_delta.is_some() {
2069 flags |= FETCH_OBJECT_ID_DELTA_BIT;
2070 }
2071 if publisher_priority.is_some() {
2072 flags |= FETCH_PRIORITY_BIT;
2073 }
2074 if original.properties.is_some() {
2075 flags |= FETCH_PROPERTIES_BIT;
2076 }
2077
2078 Ok(FetchObjectHeader {
2079 serialization_flags: VarInt::from_u64_moqt(flags),
2080 group_id_delta,
2081 subgroup_id,
2082 object_id_delta,
2083 publisher_priority,
2084 properties: original.properties.clone(),
2085 payload_length: original.payload_length,
2086 })
2087 }
2088
2089 /// An End of Range indicator's two fields against this predecessor,
2090 /// Section 11.4.2.1.
2091 fn range_fields(&self, frame: &FetchObject) -> Result<(VarInt, VarInt), CodecError> {
2092 let Some((prior_group, prior_object)) = self.prior_location else {
2093 return Ok((
2094 VarInt::from_u64_moqt(frame.group_id),
2095 VarInt::from_u64_moqt(frame.object_id),
2096 ));
2097 };
2098 let step = match self.group_order {
2099 GroupOrder::Ascending => frame.group_id.checked_sub(prior_group),
2100 GroupOrder::Descending => prior_group.checked_sub(frame.group_id),
2101 }
2102 .ok_or(CodecError::InvalidField)?;
2103 if step != 0 {
2104 return Ok((VarInt::from_u64_moqt(step), VarInt::from_u64_moqt(frame.object_id)));
2105 }
2106 let advance = frame
2107 .object_id
2108 .checked_sub(prior_object)
2109 .and_then(|a| a.checked_sub(1))
2110 .ok_or(CodecError::InvalidField)?;
2111 Ok((VarInt::from_u64_moqt(0), VarInt::from_u64_moqt(advance)))
2112 }
2113
2114 /// An Object's Group ID Delta and Object ID Delta fields against this
2115 /// predecessor, Section 11.4.1.1.
2116 fn identity_fields(
2117 &self,
2118 frame: &FetchObject,
2119 original: &FetchObjectHeader,
2120 ) -> Result<(Option<VarInt>, Option<VarInt>), CodecError> {
2121 let Some((prior_group, prior_object)) = self.prior_location else {
2122 return Ok((
2123 Some(VarInt::from_u64_moqt(frame.group_id)),
2124 Some(VarInt::from_u64_moqt(frame.object_id)),
2125 ));
2126 };
2127
2128 if frame.group_id != prior_group {
2129 let step = match self.group_order {
2130 GroupOrder::Ascending => frame.group_id.checked_sub(prior_group),
2131 GroupOrder::Descending => prior_group.checked_sub(frame.group_id),
2132 };
2133 let delta = step.and_then(|s| s.checked_sub(1)).ok_or(CodecError::InvalidField)?;
2134 // Without an Object ID Delta "the Object ID is the prior Object's
2135 // ID plus one, regardless of which group it belongs to", which is
2136 // kept where the frame arrived that way and still says the same.
2137 if original.object_id_delta.is_none()
2138 && prior_object != u64::MAX
2139 && frame.object_id == prior_object + 1
2140 {
2141 return Ok((Some(VarInt::from_u64_moqt(delta)), None));
2142 }
2143 return Ok((
2144 Some(VarInt::from_u64_moqt(delta)),
2145 Some(VarInt::from_u64_moqt(frame.object_id)),
2146 ));
2147 }
2148
2149 // Same group: plus one with no delta, plus `delta + 2` with one, so an
2150 // advance of one has only the first form.
2151 let advance = frame.object_id.checked_sub(prior_object).ok_or(CodecError::InvalidField)?;
2152 match advance {
2153 0 => Err(CodecError::InvalidField),
2154 1 => Ok((None, None)),
2155 _ => Ok((None, Some(VarInt::from_u64_moqt(advance - 2)))),
2156 }
2157 }
2158
2159 /// The Subgroup ID mode bits and the explicit field, if one is needed.
2160 /// The frame's own mode is tried first.
2161 fn subgroup_field(
2162 &self,
2163 frame: &FetchObject,
2164 original: &FetchObjectHeader,
2165 ) -> Result<(u64, Option<VarInt>), CodecError> {
2166 // Section 11.4.1.1 has the subscriber ignore these bits on a
2167 // Datagram-forwarded Object, so they are carried across untouched.
2168 if original.is_datagram() {
2169 return Ok((original.flags() & FETCH_SUBGROUP_ID_MODE_MASK, None));
2170 }
2171
2172 let subgroup_id = frame.subgroup_id.ok_or(CodecError::InvalidField)?;
2173 let inherits = self.prior_subgroup_id == Some(subgroup_id);
2174 let successor =
2175 self.prior_subgroup_id.is_some_and(|p| p.checked_add(1) == Some(subgroup_id));
2176
2177 let kept = match original.flags() & FETCH_SUBGROUP_ID_MODE_MASK {
2178 0x00 if subgroup_id == 0 => Some((0x00, None)),
2179 0x01 if inherits => Some((0x01, None)),
2180 0x02 if successor => Some((0x02, None)),
2181 FETCH_SUBGROUP_ID_EXPLICIT => Some((FETCH_SUBGROUP_ID_EXPLICIT, Some(subgroup_id))),
2182 _ => None,
2183 };
2184 let (mode, explicit) = match kept {
2185 Some(pair) => pair,
2186 None if subgroup_id == 0 => (0x00, None),
2187 None if inherits => (0x01, None),
2188 None if successor => (0x02, None),
2189 None => (FETCH_SUBGROUP_ID_EXPLICIT, Some(subgroup_id)),
2190 };
2191 Ok((mode, explicit.map(VarInt::from_u64_moqt)))
2192 }
2193
2194 /// The Publisher Priority field, or `None` when the predecessor already
2195 /// carries it.
2196 fn priority_field(
2197 &self,
2198 frame: &FetchObject,
2199 original: &FetchObjectHeader,
2200 ) -> Result<Option<u8>, CodecError> {
2201 let priority = frame.publisher_priority.ok_or(CodecError::InvalidField)?;
2202 if original.publisher_priority.is_some() || self.prior_publisher_priority != Some(priority)
2203 {
2204 return Ok(Some(priority));
2205 }
2206 Ok(None)
2207 }
2208
2209 /// Encode `frame` against everything written so far and advance. Writes
2210 /// the header only; the writer is left untouched on an error.
2211 pub fn write_object_header(
2212 &mut self,
2213 frame: &FetchObject,
2214 out: &mut impl BufMut,
2215 ) -> Result<FetchObjectHeader, CodecError> {
2216 let header = self.header_for(frame)?;
2217 header.encode(out)?;
2218 self.advance(frame);
2219 Ok(header)
2220 }
2221
2222 /// Record `frame` as the predecessor of whatever is written next.
2223 pub fn advance(&mut self, frame: &FetchObject) {
2224 if frame.header.end_marker().is_some() {
2225 self.prior_location = Some((frame.group_id, u64::MAX));
2226 return;
2227 }
2228 self.prior_location = Some((frame.group_id, frame.object_id));
2229 if let Some(subgroup_id) = frame.subgroup_id {
2230 self.prior_subgroup_id = Some(subgroup_id);
2231 }
2232 if frame.header.end_of_range().is_none() {
2233 if let Some(priority) = frame.publisher_priority {
2234 self.prior_publisher_priority = Some(priority);
2235 }
2236 }
2237 }
2238}
2239
2240#[cfg(test)]
2241mod tests {
2242 use super::*;
2243
2244 /// Canonically encoded subgroup stream vectors from
2245 /// `test-vectors/transport/draft23/codec/data-streams/subgroup.json`.
2246 /// `subgroup-explicit-subgroup-id` is omitted: it encodes group_id 100 as a
2247 /// two-byte varint, which does not survive a minimal-width re-encode.
2248 const VECTORS: &[&str] = &[
2249 // subgroup-single-object
2250 "100100800004deadbeef",
2251 // subgroup-two-objects
2252 "100100800004deadbeef0002cafe",
2253 // subgroup-no-priority
2254 "3001000004deadbeef",
2255 // subgroup-with-extensions
2256 "11010080000004deadbeef",
2257 // subgroup-end-of-group
2258 "180105800004deadbeef",
2259 // subgroup-id-mode-01
2260 "120100800504deadbeef",
2261 // subgroup-with-object-properties
2262 "1101058001043c02020104deadbeef",
2263 // subgroup-object-status-end-of-group
2264 "100100800004deadbeef000003",
2265 // subgroup-object-status-end-of-track
2266 "10010080000004",
2267 // subgroup-properties-two-objects-empty
2268 "11010080000004deadbeef000002cafe",
2269 // subgroup-properties-two-objects-nonempty
2270 "1101058000023c0204deadbeef00023c0202cafe",
2271 // subgroup-first-object-bit
2272 "500100800004deadbeef",
2273 // subgroup-first-object-and-end-of-group
2274 "580102800002cafe",
2275 ];
2276
2277 fn vi(v: u64) -> VarInt {
2278 VarInt::from_u64_moqt(v)
2279 }
2280
2281 fn hex(s: &str) -> Vec<u8> {
2282 (0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap()).collect()
2283 }
2284
2285 /// Decode a whole subgroup stream: the header, then every object up to
2286 /// the end of the buffer.
2287 fn decode_all(bytes: &[u8]) -> (SubgroupHeader, Vec<SubgroupObject>) {
2288 let mut cursor = bytes;
2289 let header = SubgroupHeader::decode(&mut cursor)
2290 .unwrap_or_else(|e| panic!("header decode failed: {e:?}"));
2291 let mut reader = SubgroupObjectReader::new(&header);
2292 let mut objects = Vec::new();
2293 while cursor.has_remaining() {
2294 objects.push(
2295 reader
2296 .read_object(&mut cursor)
2297 .unwrap_or_else(|e| panic!("object {} decode failed: {e:?}", objects.len())),
2298 );
2299 }
2300 (header, objects)
2301 }
2302
2303 fn encode_all(header: &SubgroupHeader, objects: &[SubgroupObject]) -> Vec<u8> {
2304 let mut buf = Vec::new();
2305 header.encode(&mut buf);
2306 let mut writer = SubgroupObjectReader::new(header);
2307 for o in objects {
2308 writer.write_object(o, &mut buf).unwrap_or_else(|e| panic!("write failed: {e:?}"));
2309 }
2310 buf
2311 }
2312
2313 fn object(id: u64, extensions: Vec<u8>, payload: Vec<u8>) -> SubgroupObject {
2314 SubgroupObject {
2315 object_id: vi(id),
2316 extension_headers: extensions,
2317 payload_length: vi(payload.len() as u64),
2318 object_status: None,
2319 payload,
2320 }
2321 }
2322
2323 // ── Object ID deltas ────────────────────────────────────
2324
2325 #[test]
2326 fn two_objects_with_properties_have_distinct_ids() {
2327 // Vector `subgroup-properties-two-objects-empty`: two objects, each
2328 // carrying an empty properties block and a delta of 0. The delta is
2329 // biased by one whether or not the properties bit is set, so the IDs
2330 // are 0 and 1 — not 0 and 0.
2331 let bytes = hex("11010080000004deadbeef000002cafe");
2332 let (header, objects) = decode_all(&bytes);
2333 assert!(header.has_properties());
2334 assert_eq!(objects.len(), 2);
2335 assert_eq!(objects[0].object_id.into_inner(), 0);
2336 assert_eq!(objects[1].object_id.into_inner(), 1);
2337 assert_eq!(objects[0].payload, hex("deadbeef"));
2338 assert_eq!(objects[1].payload, hex("cafe"));
2339 assert!(objects.iter().all(|o| o.extension_headers.is_empty()));
2340 }
2341
2342 #[test]
2343 fn deltas_resolve_sparse_ids() {
2344 let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
2345 let objects: Vec<_> =
2346 [3u64, 4, 40].iter().map(|&id| object(id, vec![], vec![0xAA, id as u8])).collect();
2347 let (_, decoded) = decode_all(&encode_all(&header, &objects));
2348 let ids: Vec<u64> = decoded.iter().map(|o| o.object_id.into_inner()).collect();
2349 assert_eq!(ids, vec![3, 4, 40]);
2350 }
2351
2352 #[test]
2353 fn write_rejects_non_increasing_ids() {
2354 let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
2355 let mut writer = SubgroupObjectReader::new(&header);
2356 let mut buf = Vec::new();
2357 writer.write_object(&object(7, vec![], vec![0x01]), &mut buf).unwrap();
2358 for id in [7u64, 6, 0] {
2359 let err = writer.write_object(&object(id, vec![], vec![0x01]), &mut buf).unwrap_err();
2360 assert!(matches!(err, CodecError::InvalidField), "id {id} gave {err:?}");
2361 }
2362 }
2363
2364 #[test]
2365 fn eliding_an_object_renumbers_its_successor() {
2366 let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
2367 let all: Vec<_> = (0..5u64).map(|id| object(id, vec![], vec![id as u8])).collect();
2368 for elided in 0..5u64 {
2369 let kept: Vec<_> =
2370 all.iter().filter(|o| o.object_id.into_inner() != elided).cloned().collect();
2371 let (_, decoded) = decode_all(&encode_all(&header, &kept));
2372 let ids: Vec<u64> = decoded.iter().map(|o| o.object_id.into_inner()).collect();
2373 let expected: Vec<u64> = (0..5u64).filter(|&i| i != elided).collect();
2374 assert_eq!(ids, expected, "eliding object {elided}");
2375 }
2376 }
2377
2378 // ── Properties blocks ──────────────────────────────
2379
2380 #[test]
2381 fn properties_blob_excludes_its_length_prefix() {
2382 // Vector `subgroup-properties-two-objects-nonempty`: each
2383 // object carries a two-byte block, so the blob is those two bytes
2384 // with the `02` length prefix stripped.
2385 let bytes = hex("1101058000023c0204deadbeef00023c0202cafe");
2386 let (_, objects) = decode_all(&bytes);
2387 assert_eq!(objects.len(), 2);
2388 assert_eq!(objects[0].object_id.into_inner(), 0);
2389 assert_eq!(objects[1].object_id.into_inner(), 1);
2390 assert_eq!(objects[0].extension_headers, hex("3c02"));
2391 assert_eq!(objects[1].extension_headers, hex("3c02"));
2392 assert_eq!(objects[0].payload, hex("deadbeef"));
2393 assert_eq!(objects[1].payload, hex("cafe"));
2394 }
2395
2396 /// Read every object on a subgroup stream, returning the first error.
2397 fn first_error(stream: &str) -> Option<CodecError> {
2398 let bytes = hex(stream);
2399 let mut cursor = &bytes[..];
2400 let header = SubgroupHeader::decode(&mut cursor).unwrap();
2401 let mut reader = SubgroupObjectReader::new(&header);
2402 while cursor.has_remaining() {
2403 if let Err(e) = reader.read_object(&mut cursor) {
2404 return Some(e);
2405 }
2406 }
2407 None
2408 }
2409
2410 #[test]
2411 fn prior_gaps_are_checked_per_object_and_along_the_stream() {
2412 let malformed =
2413 |stream| matches!(first_error(stream), Some(CodecError::MalformedTrack { .. }));
2414 // Section 10.8: "An Object contains more than one instance of Prior
2415 // Group ID Gap", at the top level and inside Immutable Properties.
2416 assert!(malformed("1101058000043c01000104deadbeef"));
2417 assert!(malformed("1101058000060b023c01310104deadbeef"));
2418 // Section 10.9: Object 2's gap of 2 covers Object 0, received earlier.
2419 assert!(malformed("11010580000004deadbeef01023e0202cafe"));
2420 // Object 3's gap of 2 covers Objects 1 and 2, neither received.
2421 assert_eq!(first_error("11010580000004deadbeef02023e0202cafe"), None);
2422 // Equal Prior Group ID Gap values in one Group are fine.
2423 assert_eq!(first_error("1101058000023c0204deadbeef00023c0202cafe"), None);
2424 }
2425
2426 #[test]
2427 fn status_object_carries_its_properties_block() {
2428 // Section 11.1.2 forbids properties on a non-Normal status, so the
2429 // whole-object reader refuses the frame and the meta reader frames it.
2430 let bytes = hex("1101058000023c010003");
2431 let mut cursor = &bytes[..];
2432 let header = SubgroupHeader::decode(&mut cursor).unwrap();
2433 let mut reader = SubgroupObjectReader::new(&header);
2434 assert!(matches!(reader.read_object(&mut &cursor[..]), Err(CodecError::InvalidField)));
2435 let meta = SubgroupObjectReader::new(&header).read_object_meta(&mut &cursor[..]).unwrap();
2436 assert_eq!(meta.extension_headers_len, 2);
2437 assert_eq!(meta.status, Some(3));
2438 }
2439
2440 // ── Re-encoding ─────────────────────────────────────────
2441
2442 #[test]
2443 fn vectors_re_encode_byte_identically() {
2444 for vector in VECTORS {
2445 let bytes = hex(vector);
2446 let (header, objects) = decode_all(&bytes);
2447 assert_eq!(encode_all(&header, &objects), bytes, "[{vector}] re-encode");
2448 }
2449 }
2450
2451 // ── Payload-free framing ────────────────────────────────
2452
2453 #[test]
2454 fn meta_matches_read_object() {
2455 for vector in VECTORS {
2456 let bytes = hex(vector);
2457 let mut cursor = &bytes[..];
2458 let header = SubgroupHeader::decode(&mut cursor).unwrap();
2459 let mut full_reader = SubgroupObjectReader::new(&header);
2460 let mut meta_reader = SubgroupObjectReader::new(&header);
2461 let mut full_cursor = cursor;
2462 let mut meta_cursor = cursor;
2463 while meta_cursor.has_remaining() {
2464 let before = meta_cursor.remaining();
2465 let object = full_reader.read_object(&mut full_cursor).unwrap();
2466 let meta = meta_reader.read_object_meta(&mut meta_cursor).unwrap();
2467 assert_eq!(meta.object_id, object.object_id.into_inner(), "[{vector}]");
2468 assert_eq!(
2469 meta.extension_headers_len,
2470 object.extension_headers.len() as u64,
2471 "[{vector}]"
2472 );
2473 assert_eq!(meta.payload_length, object.payload_length.into_inner(), "[{vector}]");
2474 assert_eq!(
2475 meta.status,
2476 object.object_status.map(ObjectStatus::as_u64),
2477 "[{vector}]"
2478 );
2479 assert_eq!(meta.wire_len, (before - meta_cursor.remaining()) as u64, "[{vector}]");
2480 assert_eq!(full_cursor.remaining(), meta_cursor.remaining(), "[{vector}]");
2481 }
2482 }
2483 }
2484
2485 #[test]
2486 fn short_buffers_report_unexpected_end() {
2487 let bytes = hex("1101058000023c0204deadbeef00023c0202cafe");
2488 let mut cursor = &bytes[..];
2489 let header = SubgroupHeader::decode(&mut cursor).unwrap();
2490 let objects_start = bytes.len() - cursor.len();
2491 for cut in objects_start..bytes.len() {
2492 let mut reader = SubgroupObjectReader::new(&header);
2493 let mut meta_reader = SubgroupObjectReader::new(&header);
2494 let mut cursor = &bytes[objects_start..cut];
2495 let mut meta_cursor = cursor;
2496 while cursor.has_remaining() {
2497 if let Err(err) = reader.read_object(&mut cursor) {
2498 assert!(
2499 matches!(err, CodecError::UnexpectedEnd | CodecError::VarInt(_)),
2500 "cut {cut} gave {err:?}"
2501 );
2502 break;
2503 }
2504 }
2505 while meta_cursor.has_remaining() {
2506 if let Err(err) = meta_reader.read_object_meta(&mut meta_cursor) {
2507 assert!(
2508 matches!(err, CodecError::UnexpectedEnd | CodecError::VarInt(_)),
2509 "cut {cut} gave {err:?}"
2510 );
2511 break;
2512 }
2513 }
2514 }
2515 }
2516
2517 // ── Object status ───────────────────────────────────────
2518
2519 /// A one-object subgroup stream whose object carries `status` in place of
2520 /// a payload: header type 0x10 (no properties, subgroup-ID mode 0, no
2521 /// FIRST_OBJECT bit), track alias 1, group 0, publisher priority 128; then
2522 /// an Object ID delta of 0, a payload length of 0, and the status code.
2523 fn subgroup_status_stream(status: u64) -> Vec<u8> {
2524 vec![0x10, 0x01, 0x00, 0x80, 0x00, 0x00, status as u8]
2525 }
2526
2527 /// A status datagram carrying `status`: type 0x20 (STATUS bit set,
2528 /// explicit Object ID, explicit priority), track alias 1, group 0, object
2529 /// 0, priority 128, then the status byte.
2530 fn status_datagram(status: u64) -> Vec<u8> {
2531 vec![0x20, 0x01, 0x00, 0x00, 0x80, status as u8]
2532 }
2533
2534 /// The object [`subgroup_status_stream`] describes, as a value.
2535 fn status_object(status: Option<ObjectStatus>) -> SubgroupObject {
2536 SubgroupObject {
2537 object_id: vi(0),
2538 extension_headers: Vec::new(),
2539 payload_length: vi(0),
2540 object_status: status,
2541 payload: Vec::new(),
2542 }
2543 }
2544
2545 /// An object carrying both `payload` and, in the caller's hands, `status`.
2546 /// The wire has no room for both, which is what the registry rules on.
2547 fn payload_object(status: Option<ObjectStatus>, payload: Vec<u8>) -> SubgroupObject {
2548 SubgroupObject {
2549 object_id: vi(0),
2550 extension_headers: Vec::new(),
2551 payload_length: vi(payload.len() as u64),
2552 object_status: status,
2553 payload,
2554 }
2555 }
2556
2557 /// The datagram [`status_datagram`] describes, as a value.
2558 fn status_datagram_header(status: Option<ObjectStatus>) -> DatagramHeader {
2559 DatagramHeader {
2560 datagram_type: 0x20,
2561 track_alias: vi(1),
2562 group_id: vi(0),
2563 object_id: vi(0),
2564 publisher_priority: Some(128),
2565 properties: Vec::new(),
2566 object_status: status,
2567 }
2568 }
2569
2570 /// Every status draft-23 assigns can be written and read back as the same
2571 /// status, on both a subgroup stream and a status datagram.
2572 ///
2573 /// The set is read from `ObjectStatus::ALL` — the three rows of the Object
2574 /// Status registry — rather than restated here, so this moves with the
2575 /// draft if a code is ever reassigned. It is the gate on typing the two
2576 /// `object_status` fields: a typed field that silently narrowed or
2577 /// renumbered the set would fail here even though it still compiled.
2578 ///
2579 /// Writing `ObjectStatus::Normal` when a zero-length object's status is
2580 /// `None` is checked too — without it the encoder emits an object whose
2581 /// declared payload length promises a status field that never arrives.
2582 ///
2583 /// Fails if `write_object` or `DatagramHeader::encode` writes a status
2584 /// other than the object's own.
2585 #[test]
2586 fn every_assigned_status_survives_a_round_trip() {
2587 let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
2588 for &status in ObjectStatus::ALL {
2589 let mut buf = Vec::new();
2590 SubgroupObjectReader::new(&header)
2591 .write_object(&status_object(Some(status)), &mut buf)
2592 .unwrap_or_else(|e| panic!("write_object refused {status:?}: {e:?}"));
2593
2594 let mut cursor = &buf[..];
2595 let object =
2596 SubgroupObjectReader::new(&header).read_object(&mut cursor).unwrap_or_else(|e| {
2597 panic!("read_object refused the bytes written for {status:?}: {e:?}")
2598 });
2599 assert_eq!(object.object_status, Some(status), "subgroup object status");
2600 assert!(!cursor.has_remaining(), "{status:?}: bytes left over after read_object");
2601
2602 let meta =
2603 SubgroupObjectReader::new(&header).read_object_meta(&mut &buf[..]).unwrap_or_else(
2604 |e| panic!("read_object_meta refused the bytes written for {status:?}: {e:?}"),
2605 );
2606 assert_eq!(meta.status, Some(status.as_u64()), "subgroup meta status");
2607
2608 let mut datagram = Vec::new();
2609 status_datagram_header(Some(status)).encode(&mut datagram);
2610 let decoded = DatagramHeader::decode(&mut &datagram[..]).unwrap_or_else(|e| {
2611 panic!("datagram decode refused the bytes written for {status:?}: {e:?}")
2612 });
2613 assert_eq!(decoded.object_status, Some(status), "datagram object status");
2614 }
2615
2616 let mut buf = Vec::new();
2617 SubgroupObjectReader::new(&header).write_object(&status_object(None), &mut buf).unwrap();
2618 let object = SubgroupObjectReader::new(&header)
2619 .read_object(&mut &buf[..])
2620 .expect("a zero-length object with no status must still decode");
2621 assert_eq!(object.object_status, Some(ObjectStatus::Normal));
2622
2623 let mut datagram = Vec::new();
2624 status_datagram_header(None).encode(&mut datagram);
2625 let decoded = DatagramHeader::decode(&mut &datagram[..])
2626 .expect("a status datagram with no status must still decode");
2627 assert_eq!(decoded.object_status, Some(ObjectStatus::Normal));
2628 }
2629
2630 /// The encoder writes exactly the frames the decoder accepts.
2631 ///
2632 /// Sweeps every status code `0x00..=0x3f` — one wire byte under both the
2633 /// varint on a subgroup stream and the bare byte on a datagram, and wide
2634 /// enough to contain the unassigned `0x1` and `0x2`. For
2635 /// a code the draft assigns, the hand-built frame must decode *and* the
2636 /// encoder handed that status must reproduce those exact bytes. For a code
2637 /// it does not assign, the same frame must be refused at all three decode
2638 /// sites — and no `ObjectStatus` exists to hand the encoder, so the frame
2639 /// has no way to be produced in the first place.
2640 ///
2641 /// This is a gate on the status code points only. Which of those statuses
2642 /// may carry a payload is the separate question
2643 /// [`the_registry_decides_which_statuses_may_carry_a_payload`] gates.
2644 ///
2645 /// Fails if either encoder writes a status other than the one it was
2646 /// handed, or if the decoder accepts a code outside `ObjectStatus::ALL`.
2647 ///
2648 /// The encode-side refusal is a type, not an assertion: `object_status` is
2649 /// typed, so there is no runtime path that offers the encoder the `0x2`
2650 /// that this module's decoder refuses.
2651 #[test]
2652 fn the_encoder_writes_exactly_the_frames_the_decoder_accepts() {
2653 for code in 0x00u64..=0x3f {
2654 let assigned = ObjectStatus::ALL.iter().copied().find(|s| s.as_u64() == code);
2655
2656 let stream = subgroup_status_stream(code);
2657 let mut cursor: &[u8] = &stream;
2658 let header = SubgroupHeader::decode(&mut cursor).unwrap();
2659 let objects = cursor;
2660 let read = SubgroupObjectReader::new(&header).read_object(&mut { objects });
2661 let meta = SubgroupObjectReader::new(&header).read_object_meta(&mut { objects });
2662
2663 let datagram = status_datagram(code);
2664 let decoded = DatagramHeader::decode(&mut &datagram[..]);
2665
2666 match assigned {
2667 Some(status) => {
2668 let object = read.unwrap_or_else(|e| {
2669 panic!(
2670 "read_object refused status {code:#x}, which the draft assigns: {e:?}"
2671 )
2672 });
2673 assert_eq!(object.object_status, Some(status));
2674 assert_eq!(meta.unwrap().status, Some(code));
2675 assert_eq!(decoded.unwrap().object_status, Some(status));
2676
2677 let mut written = Vec::new();
2678 header.encode(&mut written);
2679 SubgroupObjectReader::new(&header)
2680 .write_object(&status_object(Some(status)), &mut written)
2681 .unwrap();
2682 assert_eq!(
2683 written, stream,
2684 "the encoder must produce the frame the decoder accepted for status {code:#x}"
2685 );
2686
2687 let mut written = Vec::new();
2688 status_datagram_header(Some(status)).encode(&mut written);
2689 assert_eq!(
2690 written, datagram,
2691 "the encoder must produce the datagram the decoder accepted for status {code:#x}"
2692 );
2693 }
2694 None => {
2695 for (site, result) in [
2696 ("subgroup read_object", read.map(|_| ())),
2697 ("subgroup read_object_meta", meta.map(|_| ())),
2698 ("status datagram", decoded.map(|_| ())),
2699 ] {
2700 match result {
2701 Ok(()) => panic!(
2702 "{site} accepted status {code:#x}, which the draft does not assign"
2703 ),
2704 Err(error) => assert!(
2705 matches!(error, CodecError::UnknownObjectStatus(c) if c == code),
2706 "{site} refused status {code:#x} with {error:?}, not \
2707 UnknownObjectStatus"
2708 ),
2709 }
2710 }
2711 }
2712 }
2713 }
2714 }
2715
2716 // ── The registry's payload column ───────────────────────
2717
2718 /// Draft-23 Section 16.9, Table 17, "Payload" column: Normal "Yes", End of
2719 /// Group "No", End of Track "No".
2720 ///
2721 /// Restated here rather than read from [`ObjectStatus::payload_permission`]
2722 /// so the gate holds its own copy of the registry. A codec that changed a
2723 /// row would still agree with itself; only a second copy notices.
2724 const PAYLOAD_COLUMN: &[(ObjectStatus, bool)] = &[
2725 (ObjectStatus::Normal, true),
2726 (ObjectStatus::EndOfGroup, false),
2727 (ObjectStatus::EndOfTrack, false),
2728 ];
2729
2730 /// Which statuses may carry a payload is decided by the registry, not by a
2731 /// payload length.
2732 ///
2733 /// Draft-23 Section 11.1.1: "An Object MUST have an empty payload unless
2734 /// its Object Status value is registered as permitting a payload", the
2735 /// permission being a column of the Object Status registry in Section
2736 /// 16.9. What this gate observes is that the answers come from the rows.
2737 ///
2738 /// Each row is driven both ways. A zero-length object with that status must
2739 /// encode and read back — Normal included, since it permits a payload
2740 /// without requiring one and so has to stay expressible with none. An
2741 /// object handed a payload under that status must be accepted exactly when
2742 /// the row permits one, and otherwise refused with nothing written.
2743 ///
2744 /// Fails if `write_object` writes an End of Group object with a payload as
2745 /// a plain payload object, if a status reports a payload permission its
2746 /// registry row does not give, or if the permission picks the framing:
2747 /// the draft frames the status on payload length alone, so a zero-length
2748 /// Normal object must keep its status field.
2749 #[test]
2750 fn the_registry_decides_which_statuses_may_carry_a_payload() {
2751 let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
2752 assert_eq!(
2753 PAYLOAD_COLUMN.len(),
2754 ObjectStatus::ALL.len(),
2755 "every assigned status needs a row in the payload column"
2756 );
2757
2758 for &(status, permitted) in PAYLOAD_COLUMN {
2759 assert!(ObjectStatus::ALL.contains(&status), "{status:?} is not an assigned status");
2760
2761 // A zero-length object is legal under every row, and is the only
2762 // framing that states a status on a subgroup stream.
2763 let mut empty = Vec::new();
2764 SubgroupObjectReader::new(&header)
2765 .write_object(&status_object(Some(status)), &mut empty)
2766 .unwrap_or_else(|e| panic!("write_object refused a zero-length {status:?}: {e:?}"));
2767 let object = SubgroupObjectReader::new(&header)
2768 .read_object(&mut &empty[..])
2769 .unwrap_or_else(|e| panic!("read_object refused a zero-length {status:?}: {e:?}"));
2770 assert_eq!(object.status(), status, "zero-length {status:?} lost its status");
2771 assert!(object.payload.is_empty(), "zero-length {status:?} gained a payload");
2772 assert_eq!(
2773 object.permits_payload(),
2774 permitted,
2775 "decoded {status:?} reports the wrong payload permission"
2776 );
2777
2778 // A status datagram is the one carrier where the registry is not
2779 // the last word. Section 11.2.1 puts the status field in the
2780 // payload's place — "When set to 1, the Object Status field is
2781 // present and there is no Object Payload" — so no status makes
2782 // trailing bytes part of such a datagram, Normal included, and
2783 // `permitted` is not the expected answer here.
2784 let datagram = DatagramHeader::decode(&mut &status_datagram(status.as_u64())[..])
2785 .unwrap_or_else(|e| panic!("datagram decode refused {status:?}: {e:?}"));
2786 assert!(
2787 !datagram.permits_payload(),
2788 "a status datagram has no Object Payload field, so {status:?} permits no bytes"
2789 );
2790 assert_eq!(
2791 datagram.status().permits_payload(),
2792 permitted,
2793 "decoded {status:?} datagram reports the wrong registry row"
2794 );
2795 let mut whole = status_datagram(status.as_u64());
2796 whole.extend_from_slice(&hex("deadbeef"));
2797 let trailing = DatagramHeader::decode_object(&mut &whole[..]);
2798 assert!(
2799 matches!(trailing, Err(CodecError::PayloadNotPermitted { .. })),
2800 "trailing bytes on a {status:?} status datagram must be refused; got {trailing:?}"
2801 );
2802
2803 // The same status, handed a payload the wire cannot frame beside it.
2804 let mut written = Vec::new();
2805 let result = SubgroupObjectReader::new(&header)
2806 .write_object(&payload_object(Some(status), hex("deadbeef")), &mut written);
2807
2808 if permitted {
2809 result.unwrap_or_else(|e| {
2810 panic!(
2811 "write_object refused a payload under {status:?}, \
2812 which the registry permits: {e:?}"
2813 )
2814 });
2815 let object = SubgroupObjectReader::new(&header)
2816 .read_object(&mut &written[..])
2817 .unwrap_or_else(|e| {
2818 panic!("read_object refused its own output for {status:?}: {e:?}")
2819 });
2820 assert_eq!(object.payload, hex("deadbeef"), "{status:?} lost its payload");
2821 assert_eq!(object.status(), status, "{status:?} came back as another status");
2822 } else {
2823 assert!(
2824 matches!(result, Err(CodecError::PayloadNotPermitted { .. })),
2825 "write_object must refuse a payload under {status:?}, \
2826 which the registry forbids one; got {result:?}"
2827 );
2828 assert!(written.is_empty(), "a refused {status:?} object still wrote {written:?}");
2829 }
2830 }
2831
2832 // A datagram without the STATUS bit is all payload, and the status its
2833 // framing leaves out is the one row that permits a payload.
2834 let plain = DatagramHeader::decode(&mut &[0x00u8, 0x01, 0x00, 0x00, 0x80][..])
2835 .expect("a datagram with no status field must decode");
2836 assert_eq!(plain.status(), ObjectStatus::Normal);
2837 assert!(plain.permits_payload(), "a payload-carrying datagram must be permitted one");
2838 }
2839}