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