From 64eee3d29674d0f4a52ff25cbfd7062beb6fb716 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 10:04:26 -0700 Subject: [PATCH] fix(net): accept an End of Group status on a marked subgroup stream (#5081) Backport to release. The Rust regression test writes the draft-18 status object inline with release's encode form and runs under tokio::test, since release lacks the end_marker helper and moq_net_sim. Quest edits dropped. (cherry picked from commit 52a2a8953910feb1e627883524fdd5ff2bbf19f7) Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/object.ts | 5 +++- js/net/src/ietf/subscriber.test.ts | 28 +++++++++++++++++++++ rs/moq-net/src/ietf/subscriber.rs | 39 +++++++++++++++++++++++++++--- 3 files changed, 68 insertions(+), 4 deletions(-) diff --git a/js/net/src/ietf/object.ts b/js/net/src/ietf/object.ts index 3f6f86f6a3..511c39e853 100644 --- a/js/net/src/ietf/object.ts +++ b/js/net/src/ietf/object.ts @@ -343,11 +343,14 @@ export class Frame { // Defined on every implemented draft, whether or not the header marks the group's end. if (status === END_OF_TRACK) return new Frame({ endOfTrack: true }); + // Allowed even when the header marks the group's end: that bit only lets a FIN imply it, + // and imquic sends both. + if (status === GROUP_END) return new Frame(); if (flags.hasEnd) { // Empty frame if (status === 0) return new Frame({ payload: new Uint8Array(0), timestamp }); - } else if (status === 0 || status === GROUP_END) { + } else if (status === 0) { // TODO status === 0 should be an empty frame, but moq-rs seems to be sending it incorrectly on group end. return new Frame(); } diff --git a/js/net/src/ietf/subscriber.test.ts b/js/net/src/ietf/subscriber.test.ts index ab3048e54b..c8e1b73894 100644 --- a/js/net/src/ietf/subscriber.test.ts +++ b/js/net/src/ietf/subscriber.test.ts @@ -1005,6 +1005,34 @@ test("a clear FIRST_OBJECT at object 0 is the whole group", async () => { track.close(); }); +/** + * The END_OF_GROUP header bit only lets a FIN imply the group's end, so an explicit + * END_OF_GROUP status on the same stream ends the group there rather than failing it. + */ +test("an END_OF_GROUP status on a marked stream finishes the group", async () => { + const { subscriber, track } = await subscribeTrack(); + + const flags = groupFlags(true); + expect(flags.hasEnd).toBe(true); + // Objects 0..4, then object 5 as a zero-length END_OF_GROUP (0x3) status. + const objects = new Uint8Array([...encodeObjects([0, 0, 0, 0, 0]), 0, 0, 0x3]); + await subscriber.handleGroup( + new GroupMessage({ trackAlias: ALIAS, groupId: 3, subGroupId: 0, publisherPriority: 0, flags }), + new Reader(undefined, objects, VERSION), + ); + + const group = await track.ordered().nextGroup(); + expect(group?.sequence).toBe(3); + if (!group) return; + for (let i = 0; i < 5; i++) { + expect(await group.readString()).toBe("object 0"); + } + expect(await group.readFrame()).toBeUndefined(); + expect(await group.closed).toBeNull(); + + 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. diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 1a3e128b7b..1f4bc2854c 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -2647,7 +2647,6 @@ where struct GroupIngest { runtime: crate::time::Clock, has_extensions: bool, - has_end: bool, timescale: Option, version: Version, prior_object: Option, @@ -2683,7 +2682,6 @@ impl GroupIngest { Self { runtime: subscriber.runtime.clone(), has_extensions: group.flags.has_extensions, - has_end: group.flags.has_end, timescale, version: subscriber.version, prior_object: None, @@ -3505,7 +3503,9 @@ impl GroupIngest { let frame = group.create_frame_owned(frame::Info { size: 0, timestamp }, &self.budget)?; frame.finish()?; self.phase = IngestPhase::Delta; - } else if status == END_OF_GROUP && !self.has_end { + } else if status == END_OF_GROUP { + // Allowed even when the header marks the group's end: that bit only + // lets a FIN imply it, and imquic sends both. self.phase = IngestPhase::Finished(Ended::Group); } else if status == END_OF_TRACK { // Defined on every implemented draft, whether or not the header marks @@ -7050,6 +7050,39 @@ mod stitch_tests { assert_eq!(frames[1].1, b"next"); } + /// The END_OF_GROUP header bit only lets a FIN imply the group's end, so an explicit + /// END_OF_GROUP status on the same stream ends the group there rather than failing it. + #[tokio::test] + async fn an_end_of_group_status_on_a_marked_stream_finishes_the_group() { + use futures::FutureExt; + + let payloads: [&[u8]; 5] = [b"o0", b"o1", b"o2", b"o3", b"o4"]; + let mut bytes = draft18_subgroup(SEQUENCE, true, 0, &payloads); + // Object 5: the id delta, an empty extension block, a zero size, and the status. + for value in [0u64, 0, 0, END_OF_GROUP] { + value.encode(&mut bytes, Version::Draft18).unwrap(); + } + let flags = ietf::GroupFlags::decode(u64::from(bytes[0]), Version::Draft18).unwrap(); + assert!(flags.has_end, "the header marks the group's end"); + + 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"); + + // The stream was read to its end, so everything is already delivered. + let mut group = consumer.recv_group().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(group.sequence, SEQUENCE); + for payload in payloads { + let frame = group.read_frame().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(frame.payload.as_ref(), payload); + } + let end = group.read_frame().now_or_never().expect("the group is closed"); + assert!(matches!(end, Ok(None)), "the group is finished, not aborted: {end:?}"); + } + /// 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]