fix(js): drop a headless subgroup before FIRST_OBJECT (backport #5019)

js/net/src/ietf/object.ts

@@ -25,7 +25,8 @@ const PROP_TIMESTAMP_DRAFT03 = 0x06n;
 // object 0 is still the head, and any other first ID is dropped.
 const FIRST_OBJECT_BIT = 0x40;
 
-function hasFirstObjectBit(version: IetfVersion): boolean {
+/** Whether the subgroup header carries FIRST_OBJECT: draft-18 and later, not drafts 14-17. */
+export function hasFirstObjectBit(version: IetfVersion): boolean {
 	switch (version) {
 		case Version.DRAFT_14:
 		case Version.DRAFT_15:

js/net/src/ietf/subscriber.test.ts

@@ -15,7 +15,7 @@ import { RequestError, RequestOk } from "./request.ts";
 import { Subscribe, SubscribeOk, Unsubscribe } from "./subscribe.ts";
 import { SubscribeNamespace, SubscribeNamespaceEntry, SubscribeNamespaceEntryDone } from "./subscribe_namespace.ts";
 import { Subscriber } from "./subscriber.ts";
-import { ALPN, Version } from "./version.ts";
+import { ALPN, type IetfVersion, Version } from "./version.ts";
 
 const VERSION = Version.DRAFT_19;
 
@@ -860,20 +860,24 @@ function encodeObjects(deltas: number[]): Uint8Array {
  * A subscriber with one track subscribed and answered, which is what registers {@link ALIAS}
  * and lets a group stream naming it be handled.
  */
-async function subscribeTrack(): Promise<{ subscriber: Subscriber; track: track.Subscriber }> {
-	const pair = createMockTransportPair(ALPN.DRAFT_19);
-	const session = new NativeSession(pair.server, VERSION, true);
+async function subscribeTrack(
+	version: IetfVersion = VERSION,
+): Promise<{ subscriber: Subscriber; track: track.Subscriber }> {
+	const pair = createMockTransportPair(version === Version.DRAFT_16 ? ALPN.DRAFT_16 : ALPN.DRAFT_19);
+	const session = new NativeSession(pair.server, version, true);
 	const subscriber = new Subscriber({ session });
 
 	const track = subscriber.consume(Path.from("room")).track("video").subscribe();
 
 	const peer = await nextStream(pair.client);
 	if (!peer) throw new Error("the subscriber never opened a subscribe stream");
+	peer.reader.version = version;
+	peer.writer.version = version;
 
 	expect(await peer.reader.u53()).toBe(Subscribe.id);
-	const request = await Subscribe.decode(peer.reader, VERSION);
+	const request = await Subscribe.decode(peer.reader, version);
 	await peer.writer.u53(SubscribeOk.id);
-	await new SubscribeOk({ requestId: request.requestId, trackAlias: ALIAS }).encode(peer.writer, VERSION);
+	await new SubscribeOk({ requestId: request.requestId, trackAlias: ALIAS }).encode(peer.writer, version);
 
 	return { subscriber, track };
 }
@@ -1030,6 +1034,49 @@ test("a clear FIRST_OBJECT past object 0 is dropped", async () => {
 	track.close();
 });
 
+/**
+ * Drafts 14-17 have no FIRST_OBJECT bit, so a subgroup that starts at the live edge
+ * arrives with `firstObject` forced on and a non-zero first delta. That stream is the
+ * in-progress group: drop it, keep the subscription, and deliver the next group, which
+ * starts at object 0. A gap after an object was delivered still fails that group.
+ */
+test("a draft without FIRST_OBJECT drops a subgroup that starts mid-group", async () => {
+	const version = Version.DRAFT_16;
+	const { subscriber, track } = await subscribeTrack(version);
+
+	// The header cannot say otherwise on this draft: decode reports firstObject.
+	const whole = groupFlags(true);
+	await subscriber.handleGroup(
+		new GroupMessage({ trackAlias: ALIAS, groupId: 3, subGroupId: 0, publisherPriority: 0, flags: whole }),
+		new Reader(undefined, encodeObjects([2, 0]), version),
+	);
+	expect(track.latest()).toBeUndefined();
+	expect(track.closed.peek()).toBeUndefined();
+
+	await subscriber.handleGroup(
+		new GroupMessage({ trackAlias: ALIAS, groupId: 4, subGroupId: 0, publisherPriority: 0, flags: whole }),
+		new Reader(undefined, encodeObjects([0, 0]), version),
+	);
+
+	const ordered = track.ordered();
+	const group = await ordered.nextGroup();
+	expect(group?.sequence).toBe(4);
+	expect(await group?.readString()).toBe("object 0");
+
+	await subscriber.handleGroup(
+		new GroupMessage({ trackAlias: ALIAS, groupId: 5, subGroupId: 0, publisherPriority: 0, flags: whole }),
+		new Reader(undefined, encodeObjects([0, 5]), version),
+	);
+	const gapped = await ordered.nextGroup();
+	expect(gapped?.sequence).toBe(5);
+	expect(await gapped?.readString()).toBe("object 0");
+	await expect(gapped?.readFrameSequence()).rejects.toThrow(/object IDs must start at 0/);
+	expect(track.closed.peek()).toBeUndefined();
+
+	ordered.close();
+	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

@@ -17,7 +17,7 @@ import type { Session } from "./adapter.ts";
 import { DuplicateTrackAlias, RetiredTrackAlias, TrackAliases } from "./aliases.ts";
 import * as Cluster from "./cluster.ts";
 import { requestReason, toRequestCode } from "./error.ts";
-import { Frame, type Group as GroupMessage } from "./object.ts";
+import { Frame, type Group as GroupMessage, hasFirstObjectBit } from "./object.ts";
 import { fromWire, toWire } from "./priority.ts";
 import { type Publish, PublishDone, PublishError, publishDoneClean } from "./publish.ts";
 import {
@@ -1046,18 +1046,19 @@ export class Subscriber {
 			// 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) {
+			// Drafts before the bit cannot say this in the header, so the first ID is the only
+			// signal: a non-zero one is the same hole and is dropped the same way, but an empty
+			// stream there is still a group. A later gap, or a header that claimed the group
+			// starts at object 0, still fails it: `Frame.decode` refuses every non-zero delta.
+			const legacy = !hasFirstObjectBit(this.#session.version);
+			if (!group.flags.firstObject || legacy) {
 				let id: bigint | undefined;
 				try {
 					id = await stream.peekU62();
 				} catch (err: unknown) {
 					if (!(err instanceof UnexpectedEnd)) throw err;
 				}
-				if (id !== 0n) {
+				if (id !== 0n && !(legacy && id === undefined)) {
 					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;