fix(net): accept a clear FIRST_OBJECT that starts at object 0 (backport #5027)

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 {

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

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

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<u8> {
+		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<<ScriptedSession as web_transport_trait::poll::Session>::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.