fix(net): accept an End of Group status on a marked subgroup stream (backport #5081)

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();
 		}

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.

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<Timescale>,
 	version: Version,
 	prior_object: Option<u64>,
@@ -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]