diff --git a/js/net/src/ietf/object.ts b/js/net/src/ietf/object.ts index 769c933252..3f6f86f6a3 100644 --- a/js/net/src/ietf/object.ts +++ b/js/net/src/ietf/object.ts @@ -20,9 +20,9 @@ const PROP_TIMESTAMP = 0x10n; const PROP_TIMESTAMP_DRAFT03 = 0x06n; // draft-18 adds bit 0x40 (FIRST_OBJECT) to the subgroup header type per spec -// 11.4.2. Set on emit unless a filter trimmed the group's head; on parse a -// cleared bit is the peer saying its stream starts partway through a group, -// which the subscriber drops. +// 11.4.2. Set on emit unless a filter trimmed the group's head. On parse a +// cleared bit is the peer's claim that the stream starts partway through; +// object 0 is still the head, and any other first ID is dropped. const FIRST_OBJECT_BIT = 0x40; function hasFirstObjectBit(version: IetfVersion): boolean { diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index 5b89d78170..ab3048e54b 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -982,6 +982,54 @@ test("a group served from partway through is dropped", async () => { track.close(); }); +/** + * The first Object ID is absolute whatever FIRST_OBJECT says, and IDs start at 0, so a + * draft-18 stream that leaves the bit clear and starts at object 0 is the whole group. + * The publisher is out of spec on the bit, not missing a head. + */ +test("a clear FIRST_OBJECT at object 0 is the whole group", async () => { + const { subscriber, track } = await subscribeTrack(); + + const flags = groupFlags(false); + await subscriber.handleGroup( + new GroupMessage({ trackAlias: ALIAS, groupId: 3, subGroupId: 0, publisherPriority: 0, flags }), + new Reader(undefined, encodeObjects([0, 0]), VERSION), + ); + + const group = await track.ordered().nextGroup(); + expect(group?.sequence).toBe(3); + if (!group) return; + expect(await group.readString()).toBe("object 0"); + expect(group.frameCount).toBe(2); + + track.close(); +}); + +/** + * A clear FIRST_OBJECT whose first ID is not 0 still has a hole at the front, so the + * stream is dropped and the track resumes at the next group. + */ +test("a clear FIRST_OBJECT past object 0 is dropped", async () => { + const { subscriber, track } = await subscribeTrack(); + + const flags = groupFlags(false); + await subscriber.handleGroup( + new GroupMessage({ trackAlias: ALIAS, groupId: 3, subGroupId: 0, publisherPriority: 0, flags }), + new Reader(undefined, encodeObjects([3]), VERSION), + ); + + const whole = groupFlags(true); + await subscriber.handleGroup( + new GroupMessage({ trackAlias: ALIAS, groupId: 4, subGroupId: 0, publisherPriority: 0, flags: whole }), + new Reader(undefined, encodeObjects([0]), VERSION), + ); + + const group = await track.ordered().nextGroup(); + expect(group?.sequence).toBe(4); + + track.close(); +}); + /** * FIRST_OBJECT is the publisher's claim, and the object ids are what actually happened. A * peer that sets the bit and then starts at object 5 is contradicting itself, so the group diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index bbe17715d0..473af22e40 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -7,7 +7,7 @@ import * as netGroup from "../group.ts"; import { Cost, type Route, randomHop, routesEqual, stampHops, UNKNOWN_HOP } from "../hop.ts"; import { hiddenBelow, hooks, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts"; import * as Path from "../path.ts"; -import type { Cursor, Reader, Stream } from "../stream.ts"; +import { type Cursor, type Reader, type Stream, UnexpectedEnd } from "../stream.ts"; import { Tail } from "../tail.ts"; import { type Timescale, Timestamp } from "../time.ts"; import type * as track from "../track.ts"; @@ -1040,20 +1040,28 @@ export class Subscriber { }; try { - // FIRST_OBJECT clear says this stream starts partway through the group, which the - // draft lets a publisher do to answer a filter. Nothing above here can use it: the - // objects that would arrive are not decodable without the missing head, and a group - // is the unit an application resyncs on. Drop it and pick up at the next group, the - // same degradation as a publisher that no longer holds the head. + // FIRST_OBJECT clear is the publisher's claim that the stream starts partway + // through the group. The first Object ID is absolute either way, and IDs start + // at 0, so a clear bit on object 0 is still the whole group. Any other first ID, + // or a stream with no object, has a hole at the front: drop it and pick up at + // the next group. // // This only saves reading a stream we would throw away. The bit is the publisher's // claim, so what is enforced is the object ids themselves: `Frame.decode` holds every // object to starting at 0 and incrementing by 1, whatever the header said and on the // drafts that have no such bit to read. if (!group.flags.firstObject) { - console.debug(`dropping a group with no head: alias=${group.trackAlias} group=${group.groupId}`); - stream.stop(new Error("a group must start at object 0")); - return; + let id: bigint | undefined; + try { + id = await stream.peekU62(); + } catch (err: unknown) { + if (!(err instanceof UnexpectedEnd)) throw err; + } + if (id !== 0n) { + console.debug(`dropping a group with no head: alias=${group.trackAlias} group=${group.groupId}`); + stream.stop(new Error("a group must start at object 0")); + return; + } } // The alias binds after SUBSCRIBE_OK commits the track property; an omitted diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index a8a7dad47e..1a3e128b7b 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -2258,28 +2258,10 @@ where group.publisher_priority = super::priority::to_wire(track.publisher_priority()); } - // FIRST_OBJECT clear says this stream starts partway through the group, which the - // draft lets a publisher do to answer a filter. Without a head it is unusable: the - // objects are not decodable without the ones missing in front, and a group is the - // unit an application resyncs on. Drop it and pick up at the next group, the same - // degradation as a publisher that no longer holds the head. + // Whether a FIRST_OBJECT-clear stream is a group with no head is decided in + // [`Self::open_group`], after it peeks the first Object ID. The bit is only the + // publisher's claim; [`next_object_id`] holds the sequence after that. // - // A fill we asked for is the exception, since its fetch stream is carrying exactly - // that head for [`Self::open_group`] to stitch this onto. - // - // The bit is only the publisher's claim, so what is enforced is the object ids - // themselves: [`next_object_id`] holds every object to starting where the head - // stopped and incrementing by 1, whatever the header said and on the drafts that - // have no such bit to read. - if !group.flags.first_object && !fill.read().outstanding() { - tracing::debug!( - track_alias = %group.track_alias, - group = %group.group_id, - "dropping a group with no head" - ); - return Err(Error::Unsupported); - } - // The peek inside blocks until the publisher produces the group's first object, so // race it against the subscription going away the same way the group read below is. // Otherwise dropping the local subscriber cannot end this handler. @@ -2550,6 +2532,25 @@ where return Ok(Opened::EndOfTrack); } + // FIRST_OBJECT clear is the publisher's claim that the stream starts partway + // through the group. The first Object ID is absolute either way, and IDs start + // at 0, so a clear bit on object 0 is still the head: the publisher is out of + // spec, not missing a keyframe. Any other first ID, or a stream with no object, + // has a hole at the front. Drop it before a group exists and resume at the next + // group, the same as a publisher that no longer holds the head. + // + // A fill we asked for is the exception: its fetch stream is carrying the head + // this tail stitches onto. + if !header.flags.first_object && !fill.read().outstanding() && first.is_none_or(|first| first.id != 0) { + tracing::debug!( + track_alias = %header.track_alias, + group = %header.group_id, + object = first.map(|first| first.id), + "dropping a group with no head" + ); + return Err(Error::Unsupported); + } + if !fill.read().outstanding() { return Ok(Opened::Group(create(track)?, 0)); } @@ -6458,6 +6459,56 @@ mod stitch_tests { buf.to_vec() } + /// A draft-18 subgroup stream whose FIRST_OBJECT bit is independent of the first + /// object's absolute ID. A strict publisher sets the bit exactly when `start` is 0; + /// one that is out of spec can leave it clear and still start there. Each object + /// carries the Timestamp of its Object ID, which the harness's timed track requires. + fn draft18_subgroup(sequence: u64, first_object: bool, start: u64, payloads: &[&[u8]]) -> Vec { + let version = Version::Draft18; + let mut buf = bytes::BytesMut::new(); + ietf::GroupHeader { + track_alias: ALIAS, + group_id: sequence, + sub_group_id: 0, + publisher_priority: 0, + flags: ietf::GroupFlags { + first_object, + has_extensions: true, + ..Default::default() + }, + } + .encode(&mut buf, version) + .unwrap(); + + for (index, payload) in payloads.iter().enumerate() { + // The first object's delta is its absolute Object ID; every later one counts + // the objects skipped, so zero is the next one. + let delta = match index { + 0 => start, + _ => 0, + }; + delta.encode(&mut buf, version).unwrap(); + let mut properties = bytes::BytesMut::new(); + let object = usize::try_from(start).unwrap() + index; + ietf::encode_object_time(&mut properties, timestamp(object), Timescale::MICRO, version).unwrap(); + (properties.len() as u64).encode(&mut buf, version).unwrap(); + buf.put_slice(&properties); + (payload.len() as u64).encode(&mut buf, version).unwrap(); + buf.put_slice(payload); + } + + buf.to_vec() + } + + /// A reader over the next scripted stream, decoded as draft-18. + async fn read_draft18( + session: &ScriptedSession, + ) -> Reader<::RecvStream, Version> { + let mut session = session.clone(); + let (_, recv) = session.open_bi().await.unwrap(); + Reader::new(recv, Version::Draft18) + } + /// A subscriber holding one draft-20 subscription, as its SUBSCRIBE_OK left it: the /// alias bound, the timescale declared, and `fill` waiting on its fetch stream. struct Harness { @@ -6973,6 +7024,63 @@ mod stitch_tests { assert!(matches!(delivered, Err(_) | Ok(None)), "no frame is delivered"); } + /// Object IDs are absolute whatever FIRST_OBJECT says, and they start at 0, so a + /// draft-18 stream that leaves the bit clear and starts at object 0 is the whole + /// group. The publisher is out of spec on the bit, not missing a head. + #[tokio::test] + async fn a_clear_first_object_at_zero_delivers_the_group() { + let bytes = draft18_subgroup(SEQUENCE, false, 0, &[b"whole", b"next"]); + assert_eq!( + u64::from(bytes[0]) & ietf::GroupFlags::FIRST_OBJECT_BIT, + 0, + "the bit is clear" + ); + + let mut h = Harness::new(Fill::Done, vec![bytes]); + h.subscriber.version = Version::Draft18; + let mut consumer = h.track.subscribe(None); + let mut stream = read_draft18(&h.session).await; + + h.subscriber.clone().recv_group(&mut stream).await.expect("the group"); + + let (sequence, frames) = read_group(&mut consumer).await; + assert_eq!(sequence, SEQUENCE); + assert_eq!(frames.len(), 2); + assert_eq!(frames[0].1, b"whole"); + assert_eq!(frames[1].1, b"next"); + } + + /// A clear FIRST_OBJECT whose first ID is not 0 still has a hole at the front, so + /// the stream is dropped and the track resumes at the next group. + #[tokio::test] + async fn a_clear_first_object_past_zero_is_dropped() { + let partial = draft18_subgroup(SEQUENCE, false, 3, &[b"tail-3"]); + let whole = draft18_subgroup(SEQUENCE + 1, true, 0, &[b"next"]); + assert_eq!(u64::from(partial[0]) & ietf::GroupFlags::FIRST_OBJECT_BIT, 0); + + let mut h = Harness::new(Fill::Done, vec![partial, whole]); + h.subscriber.version = Version::Draft18; + let mut consumer = h.track.subscribe(None); + + let mut partial = read_draft18(&h.session).await; + assert!(matches!( + h.subscriber.clone().recv_group(&mut partial).await, + Err(Error::Unsupported) + )); + + let mut whole = read_draft18(&h.session).await; + h.subscriber + .clone() + .recv_group(&mut whole) + .await + .expect("the next group"); + + let (sequence, frames) = read_group(&mut consumer).await; + assert_eq!(sequence, SEQUENCE + 1); + assert_eq!(frames[0].1, b"next"); + assert!(h.session.log.closes().is_empty()); + } + /// A head that stops short of where the tail starts would leave a hole in the middle of /// the group, which the model cannot express. Both halves go, and the head is published /// as the prefix it is.