fix(net): backport an explicit group 0 floor on pre-06 wires (#5000)

drafts/draft-lcurley-moq-lite.md

@@ -1053,7 +1053,7 @@ A value of 0 means the whole group (default).
 A non-zero value is the absolute frame index + 1, matching `Group End`.
 MUST be 0 when `Group End` is 0, since an unbounded subscription has no end group to qualify.
 
-`Group Start` and `Group End` are offset by 1 only so 0 can mean "absent"; every other group field in this document is a plain absolute sequence.
+`Group End` and `Frame End` are offset by 1 so 0 can mean "absent". `Group Start` is an absolute sequence, and 0 is group 0.
 
 ## SUBSCRIBE_UPDATE
 A subscriber can modify a subscription with a SUBSCRIBE_UPDATE message.
@@ -1150,9 +1150,13 @@ SUBSCRIBE_OK Message {
 Set to 0x0 to indicate a SUBSCRIBE_OK message.
 
 **Group**:
-The absolute sequence number of the first group that will be delivered.
-It MUST be greater than or equal to the requested start group; any groups in between are unavailable.
-A subscriber that requested the latest group learns the resolved sequence here.
+The absolute sequence number where delivery starts when the subscription resolves.
+Group streams can arrive out of order, so it need not be the first group sent.
+It MUST be greater than or equal to the requested `Group Start`.
+This group is not a new floor.
+A subscriber SHOULD NOT wait for a group between the requested floor and this group.
+The publisher still delivers such a group if it is created or becomes available later, while it is within `Subscriber Max Age`.
+A subscriber whose `Subscriber Max Age` resolves the start to the latest group learns that sequence here.
 
 There is no matching frame field, because the start frame is never in doubt: a partial group is only delivered when it was asked for, so the subscription starts either exactly where it asked or at the beginning of a later group (see [Positions](#positions)).
 The subscriber derives the start frame from `Group` and its own request:
@@ -1346,6 +1350,8 @@ The `Message Length` describes the payload size on the wire.
 
 ## moq-lite-07
 
+- SUBSCRIBE_OK `Group` names where delivery starts when the subscription resolves, which need not be the first group sent. It is not a new floor. A subscriber does not wait for a group between the requested floor and it, but a publisher still delivers one that arrives later within Subscriber Max Age.
+- Corrected the SUBSCRIBE note that offset `Group Start` by 1. Only `Group End` and `Frame End` are offset so 0 can mean absent. `Group Start` is an absolute sequence, and 0 is group 0.
 - Assigned `moq-lite-07-wip` as this draft's protocol identifier until it is finalized as `moq-lite-07`.
 - Switched every variable-length integer, including SETUP parameter values, from QUIC's two-bit length prefix to moq-transport's leading-ones encoding, widening the range to 64 bits.
 - Hid routes with a `.`-prefixed segment below the requested prefix from announce discovery, and added the ANNOUNCE_REQUEST `Hidden` field to opt in.

js/net/src/integration.test.ts

@@ -164,8 +164,7 @@ test("integration: lite subscription options and updates reach the publisher", a
 	})();
 
 	const remote = wireOf(client).consume(Path.from("test"));
-	// A floor of 1, not 0: a pre-06 wire folds a vacuous floor of 0 back to absent, since
-	// its encoding of group 0 means "replay from the beginning" instead.
+	// A floor of group 1, a concrete group on every draft.
 	const subscriber = remote.track("video").subscribe({
 		priority: 3,
 		maxAge: Milli(250),
@@ -223,8 +222,7 @@ test("integration: lite carries a fractional maxAge as a whole millisecond", asy
 
 	const remote = wireOf(client).consume(Path.from("test"));
 	// A varint cannot encode 38.75, so an unrounded value fails the SUBSCRIBE outright and
-	// nothing resubscribes. The publisher must see the budget rounded up instead. A floor
-	// of 1, not 0: a pre-06 wire folds a vacuous floor of 0 back to absent.
+	// nothing resubscribes. The publisher must see the budget rounded up instead.
 	const subscriber = remote.track("video").subscribe({ maxAge: Milli(38.75), groups: { start: { included: 1 } } });
 
 	// A failed subscribe never reaches the publisher, so race its closure to report the

js/net/src/lite/publisher.test.ts

@@ -644,9 +644,9 @@ test("lite draft-05: a late-arriving older group is still served", async () => {
 	}
 });
 
-// SUBSCRIBE_START promises nothing below the announced sequence will be delivered, so
-// the floor is pinned there: a straggler below the first served group is dropped even
-// though arrival-order serving would otherwise surface it.
+// An unfloored pre-06 subscribe joins at the publisher's start. SUBSCRIBE_START makes
+// that group the floor: a straggler below it is dropped even though arrival-order
+// serving would otherwise surface it.
 test("lite draft-05: a straggler below the announced start group is not served", async () => {
 	const sub = await servedSubscription();
 	try {
@@ -668,6 +668,71 @@ test("lite draft-05: a straggler below the announced start group is not served",
 	}
 });
 
+// An explicit floor stays where it was named. SUBSCRIBE_START still reports the first
+// served group, and a later group at or above the floor is delivered.
+test("lite draft-05: an explicit floor still serves a group below the first one", async () => {
+	const sub = await servedSubscription({ startGroup: 0 });
+	try {
+		sub.serve(2);
+		expect(await sub.servedSequence()).toBe(2);
+
+		const resp = await decodeSubscribeResponse(sub.client.reader, Version.DRAFT_05);
+		if (!("start" in resp)) throw new Error("expected SUBSCRIBE_START");
+		expect(resp.start.group).toBe(2);
+
+		sub.serve(0);
+		expect(await sub.servedSequence()).toBe(0);
+	} finally {
+		await sub.close();
+	}
+});
+
+// Pre-06 an absent Group Start is the latest group. An update that drops a named floor
+// re-pins there, so a group created below it afterwards is not served.
+test("lite draft-05: an update dropping the floor re-pins to the latest group", async () => {
+	const sub = await servedSubscription({ startGroup: 0 });
+	try {
+		sub.serve(2);
+		expect(await sub.servedSequence()).toBe(2);
+
+		await replayUpdate({ priority: 0 }).encode(sub.client.writer, Version.DRAFT_05);
+		await flush();
+		sub.serve(0);
+		sub.serve(3);
+		expect(await sub.servedSequence()).toBe(3);
+	} finally {
+		await sub.close();
+	}
+});
+
+// On an empty track the re-pin has no latest group to name, so the first group the
+// track produces is served, even below the dropped floor.
+test("lite draft-05: an update dropping the floor on an empty track serves the next group", async () => {
+	const sub = await servedSubscription({ startGroup: 5 });
+	try {
+		await replayUpdate({ priority: 0 }).encode(sub.client.writer, Version.DRAFT_05);
+		await flush();
+		sub.serve(0);
+		expect(await sub.servedSequence()).toBe(0);
+	} finally {
+		await sub.close();
+	}
+});
+
+// Lite-06 encodes a floor of group 0 as 0, including a subscribe that omitted Group Start.
+// That is a floor, so a later group 0 is still served.
+test("lite draft-06: an omitted group start still serves a group below the first one", async () => {
+	const sub = await servedSubscription({ version: Version.DRAFT_06 });
+	try {
+		sub.serve(2);
+		expect(await sub.servedSequence()).toBe(2);
+		sub.serve(0);
+		expect(await sub.servedSequence()).toBe(0);
+	} finally {
+		await sub.close();
+	}
+});
+
 // A group pop is the linearization point. Once the publisher reaches the SUBSCRIBE_START write,
 // the group and its range have already been decided, so an update queued during the write applies
 // to the next pop rather than reaching backward into this one.
@@ -687,7 +752,8 @@ test("lite draft-05: a group popped before a cap update is still served", async
 		sub.release();
 		expect(await sub.servedSequence()).toBe(1);
 		while (ranges.mock.calls.length === 0) await flush();
-		expect(lastGroups(ranges)).toEqual({ start: undefined, end: { included: 0 } });
+		// Dropping the named floor re-pins to the latest group, as an unfloored SUBSCRIBE does.
+		expect(lastGroups(ranges)).toEqual({ start: { included: 1 }, end: { included: 0 } });
 	} finally {
 		ranges.mockRestore();
 		sub.release();
@@ -709,8 +775,9 @@ test("lite draft-06: a queued floor update applies before the next group pop", a
 		await replayUpdate({ priority: 0, startGroup: 2 }).encode(sub.client.writer, Version.DRAFT_06);
 		sub.serve(2);
 		await flush();
-		expect(ranges).toHaveBeenCalledTimes(1);
-		expect(lastGroups(ranges)).toEqual({ start: { included: 0 }, end: undefined });
+		// Draft-06 group 0 is a floor, so the first group does not raise the cursor.
+		// The queued update is still waiting on the parked write.
+		expect(ranges).not.toHaveBeenCalled();
 
 		sub.release();
 
@@ -761,8 +828,9 @@ test("lite draft-06: a queued frame floor applies before the next group pop", as
 
 		await replayUpdate({ priority: 0, startGroup: 1, startFrame: 2 }).encode(sub.client.writer, Version.DRAFT_06);
 		await flush();
-		expect(ranges).toHaveBeenCalledTimes(1);
-		expect(lastGroups(ranges)).toEqual({ start: { included: 0 }, end: undefined });
+		// Draft-06 group 0 is a floor, so serving the first group does not move the cursor.
+		// The queued update is still waiting on the parked write.
+		expect(ranges).not.toHaveBeenCalled();
 
 		sub.release();
 

js/net/src/lite/publisher.ts

@@ -790,8 +790,17 @@ export class Publisher {
 						case "update": {
 							const update = control.update;
 							console.debug(`subscribe update: broadcast=${broadcast} track=${track.name}`);
+							// `positionCursor` pins an unfloored pre-06 SUBSCRIBE to the latest
+							// group, so an update that drops a named floor re-pins there too. On
+							// an empty track the next group is the latest, whatever its sequence.
+							const start =
+								update.startGroup === undefined &&
+								bounds.startGroup !== undefined &&
+								!resolvesStart(this.version)
+									? (track.latest() ?? 0)
+									: update.startGroup;
 							hooks.replaceGroups(track, {
-								start: update.startGroup === undefined ? undefined : { included: update.startGroup },
+								start: start === undefined ? undefined : { included: start },
 								end: update.endGroup === undefined ? undefined : { included: update.endGroup },
 							});
 							bounds.startGroup = update.startGroup;
@@ -851,13 +860,17 @@ export class Publisher {
 
 				if (emitRange && !startSent) {
 					startSent = true;
-					// SUBSCRIBE_START promises nothing below this sequence will be delivered.
-					// Arrival-order serving could later surface a straggler below the first
-					// group, so pin the floor to what was announced.
-					hooks.replaceGroups(track, {
-						start: { included: group.sequence },
-						end: bounds.endGroup === undefined ? undefined : { included: bounds.endGroup },
-					});
+					// SUBSCRIBE_START names the first group served now. A later group at or
+					// above an explicit floor is still delivered, so the cursor stays put.
+					// A pre-06 subscribe that named no group is the exception: an absent
+					// Group Start is the latest group, and this sequence becomes the floor.
+					// Lite-06 encodes a floor of group 0 as 0, so an omitted start is not pinned.
+					if (bounds.startGroup === undefined && !resolvesStart(this.version)) {
+						hooks.replaceGroups(track, {
+							start: { included: group.sequence },
+							end: bounds.endGroup === undefined ? undefined : { included: bounds.endGroup },
+						});
+					}
 					if (
 						!(await controls.response(
 							encodeSubscribeResponse(

js/net/src/lite/subscribe.test.ts

@@ -111,14 +111,23 @@ test("Subscribe round-trips every option including startGroup 0", async () => {
 	expect(resumed.startFrame).toBe(4);
 	message.startFrame = 0;
 
-	// A pre-06 wire folds the vacuous floor back to absent: an explicit group 0 there
-	// would mean "replay from the beginning", which is not what a floor of 0 asks for.
-	const folded = await Subscribe.decode(
+	// A pre-06 wire encodes an explicit group 0 as sequence + 1, so it round-trips.
+	// Omitting the floor is still the latest group, not group 0.
+	const explicit = await Subscribe.decode(
 		new Reader(undefined, await encodeMessage(Version.DRAFT_05, message), Version.DRAFT_05),
 		Version.DRAFT_05,
 	);
-	expect(folded.startGroup).toBeUndefined();
-	expect(folded.endGroup).toBe(9);
+	expect(explicit.startGroup).toBe(0);
+	expect(explicit.endGroup).toBe(9);
+	const absent = await Subscribe.decode(
+		new Reader(
+			undefined,
+			await encodeMessage(Version.DRAFT_05, new Subscribe({ ...message, startGroup: undefined })),
+			Version.DRAFT_05,
+		),
+		Version.DRAFT_05,
+	);
+	expect(absent.startGroup).toBeUndefined();
 });
 
 test("SubscribeUpdate round-trips every option including startGroup 0", async () => {
@@ -137,13 +146,45 @@ test("SubscribeUpdate round-trips every option including startGroup 0", async ()
 	expect(got.startGroup).toBeUndefined();
 	expect(got.endGroup).toBe(12);
 
-	// The same fold as SUBSCRIBE on a pre-06 wire.
-	const folded = await SubscribeUpdate.decode(
+	// The same pre-06 encoding as SUBSCRIBE: explicit group 0 round-trips, and an
+	// omitted floor stays omitted.
+	const explicit = await SubscribeUpdate.decode(
 		new Reader(undefined, await encodeMessage(Version.DRAFT_05, message), Version.DRAFT_05),
 		Version.DRAFT_05,
 	);
-	expect(folded.startGroup).toBeUndefined();
-	expect(folded.endGroup).toBe(12);
+	expect(explicit.startGroup).toBe(0);
+	expect(explicit.endGroup).toBe(12);
+	const absent = await SubscribeUpdate.decode(
+		new Reader(
+			undefined,
+			await encodeMessage(Version.DRAFT_05, new SubscribeUpdate({ ...message, startGroup: undefined })),
+			Version.DRAFT_05,
+		),
+		Version.DRAFT_05,
+	);
+	expect(absent.startGroup).toBeUndefined();
+});
+
+// The same wire values as Rust's `encode_start_group`: pre-06 writes an omitted floor as 0
+// and group 0 as 1, while lite-06 writes 0 for both.
+test("Group Start encodes an omitted floor and group 0 apart before draft-06", async () => {
+	for (const make of [
+		(startGroup?: number) =>
+			new Subscribe({ id: 1n, broadcast: Path.from("test"), track: "video", priority: 0, startGroup }),
+		(startGroup?: number) => new SubscribeUpdate({ priority: 0, startGroup }),
+	]) {
+		const absent = await encodeMessage(Version.DRAFT_05, make(undefined));
+		const zero = await encodeMessage(Version.DRAFT_05, make(0));
+		expect(zero.byteLength).toBe(absent.byteLength);
+		const diff = [...absent.keys()].filter((i) => absent[i] !== zero[i]);
+		expect(diff).toHaveLength(1);
+		expect(absent[diff[0]]).toBe(0);
+		expect(zero[diff[0]]).toBe(1);
+
+		expect(await encodeMessage(Version.DRAFT_06, make(0))).toEqual(
+			await encodeMessage(Version.DRAFT_06, make(undefined)),
+		);
+	}
 });
 
 test("SubscribeStart round-trips on draft-05", async () => {

js/net/src/lite/subscribe.ts

@@ -6,17 +6,16 @@ import { hasFrameBounds, hasGroupOrder, hasStreamCount, resolvesStart, Version }
 /**
  * Encode the `Group Start` field shared by SUBSCRIBE and SUBSCRIBE_UPDATE.
  *
- * Lite-06 writes the raw floor (`undefined` and 0 are the same absence of a constraint),
- * while a pre-06 wire encodes the sequence + 1 and gets a vacuous floor folded back to
- * absent: an explicit group 0 there means "replay from the beginning", which is not what
- * a floor of 0 asks for.
+ * Lite-06 writes the raw floor (`undefined` and 0 are both group 0). A pre-06 wire
+ * encodes the sequence + 1: omitting the floor is 0 (the latest group, where the
+ * publisher starts) and an explicit 0 is 1, replay from the beginning.
  */
 async function encodeStartGroup(w: Writer, version: Version, startGroup?: number) {
 	if (resolvesStart(version)) {
 		await w.u53(startGroup ?? 0);
 		return;
 	}
-	await w.u53(startGroup !== undefined && startGroup > 0 ? startGroup + 1 : 0);
+	await w.u53(startGroup === undefined ? 0 : startGroup + 1);
 }
 
 /**

js/net/src/lite/subscriber.ts

@@ -47,6 +47,7 @@ import {
 	hasDatagrams,
 	hasProbeRtt,
 	hasStreamCount,
+	resolvesStart,
 	restartSupported,
 	Version,
 } from "./version.ts";
@@ -892,8 +893,8 @@ export class Subscriber {
 
 			if ("start" in resp) {
 				entry.start = resp.start.group;
-				// The groups the SUBSCRIBE asked for below it are unavailable, whatever the
-				// demand asks later.
+				// The groups the SUBSCRIBE asked for below it are not waited for, whatever the
+				// demand asks later. One that still arrives is delivered.
 				if (entry.requested !== undefined) entry.tail.account(entry.requested, entry.start);
 			} else if ("end" in resp) {
 				if (entry.end !== undefined) throw new ProtocolViolation("duplicate SUBSCRIBE_END");
@@ -1012,9 +1013,14 @@ export class Subscriber {
 	#sameSubscription(a: track.Subscription, b: track.Subscription): boolean {
 		const ag = groupBounds(a.groups);
 		const bg = groupBounds(b.groups);
+		// `groupBounds` reads an omitted start as 0. A pre-06 wire tells them apart: omitted
+		// joins at the publisher's start, and 0 is group 0. Lite-06 encodes both as 0.
+		const sameStart =
+			resolvesStart(this.version) || (a.groups?.start === undefined) === (b.groups?.start === undefined);
 		return (
 			(a.priority ?? 0) === (b.priority ?? 0) &&
 			(a.maxAge ?? 0) === (b.maxAge ?? 0) &&
+			sameStart &&
 			ag.start === bg.start &&
 			ag.end === bg.end
 		);

js/net/src/track.test.ts

@@ -323,6 +323,15 @@ test("multiple subscriber options aggregate like Rust", async () => {
 	expect(await none).toBeUndefined();
 });
 
+// A resume floor above the live edge must not hide the latest group from a reader that
+// joins after it: on a quiet track that group may be the only one for a long time.
+test("an unfloored subscriber clears a resume floor above the live edge", () => {
+	const producer = new TrackProducer("test");
+	producer.subscribe({ groups: { start: { included: 4 } } });
+	producer.subscribe({});
+	expect(producer.subscription.peek()?.groups?.start).toBeUndefined();
+});
+
 test("the producer aggregate is clamped without changing subscriber options", async () => {
 	const producer = new TrackProducer("test");
 	const track = producer.subscribe({ maxAge: Milli(10_000) });

quest/m1/README.md

@@ -45,7 +45,6 @@ QUIC studies there on that rule.
 - [WebTransport close upstream](/quest/m1/wt-close-upstream.md) - web-transport-moq delivers the close capsule itself, and moq-tokio's `CLOSE_LINGER` is deleted
 - [Browser close code](/quest/m1/browser-close-code.md) - a playwright case proves the page reads a relay's close code and reason, on every web-transport backend
 - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - every stream group in a lite subscription arrives or is dropped by name, and lite-07 drops its stream count for it
-- [Late lower groups](/quest/m1/lite-late-lower-group.md) - a moq-lite subscriber with a floor receives a group created below the first served one, as moq-transport does
 - [Parked reads wake](/quest/m1/parked-read-wakes.md) - a read parked on an evicted or aborted group wakes and re-judges, for resume successors and plain tracks alike
 - [Reader end parity](/quest/m1/reader-end-parity.md) - JS readers see a track's end once the newest group reaches the declared end, as Rust readers do
 - [JS cache window](/quest/m1/js-prune-cache-window.md) - js/net ages idle groups on a cache window and keeps max_age as media-time staleness, as Rust does