feat(net): count lite-07 group streams in SUBSCRIBE_END

drafts/draft-lcurley-moq-lite.md

@@ -447,17 +447,16 @@ A receiver SHOULD NOT cache refusals; rate limiting is the advertiser's concern.
 A subscriber opens Subscribe Streams to request a Track.
 
 The subscriber MUST start a Subscribe Stream with a SUBSCRIBE message followed by any number of SUBSCRIBE_UPDATE messages.
-The publisher replies with a SUBSCRIBE_OK message once the start group is resolved, followed by any number of SUBSCRIBE_END and SUBSCRIBE_DROP messages.
+The publisher replies with a SUBSCRIBE_OK message once the start group is resolved, followed by a SUBSCRIBE_END message once the subscription ends.
 For a live track the publisher MAY withhold SUBSCRIBE_OK until the first matching group resolves the start; if the track has already ended with no matching groups, it sends SUBSCRIBE_END with no preceding SUBSCRIBE_OK.
 A rejection is a stream reset: a publisher that cannot serve the subscription (no such track, an ended broadcast, or any other refusal) MUST promptly reset the stream rather than leave it pending, so a subscriber distinguishes "pending" from "refused" by the reset, not by a timeout.
 A route claims capability rather than inventory, so a subscription for a covered path that names nothing is refused this way too.
 
 The track's immutable publisher properties are not carried here; they are fetched once via a [Track Stream](#track-stream).
 The subscriber needs the track's TRACK_INFO (notably its timescale) to interpret FRAME messages, and MAY open the Track and Subscribe streams concurrently, buffering frames until it arrives.
 
-The publisher sends SUBSCRIBE_OK once the absolute start position is resolved, and SUBSCRIBE_END once no further groups will be produced (see [SUBSCRIBE_OK](#subscribe-ok) and [SUBSCRIBE_END](#subscribe-end)).
-The publisher closes the stream (FIN) only once every group from start to end has been accounted for, either via a Group Stream (completed or reset) or a SUBSCRIBE_DROP message.
-This MAY occur after SUBSCRIBE_END, since stragglers within the range can still be dropped.
+The publisher sends SUBSCRIBE_OK once the absolute start position is resolved, and SUBSCRIBE_END once no further groups will be produced and every Group Stream it opens for the subscription has been opened (see [SUBSCRIBE_OK](#subscribe-ok) and [SUBSCRIBE_END](#subscribe-end)).
+The publisher closes the stream (FIN) after SUBSCRIBE_END, once every counted Group Stream has finished or been reset.
 Unbounded subscriptions stay open until SUBSCRIBE_END, and either endpoint MAY reset the stream at any time.
 
 ### Fetch
@@ -1116,7 +1115,7 @@ 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 and implicitly dropped, with no separate SUBSCRIBE_DROP required.
+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.
 
 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)).
@@ -1136,6 +1135,7 @@ SUBSCRIBE_END Message {
   Type (i) = 0x1
   Message Length (i)
   Group (i)
+  Stream Count (i)
 }
 ~~~
 
@@ -1147,34 +1147,14 @@ The exclusive end of the range: the absolute sequence number of the first group
 A value of 0 means the track ended before producing any groups.
 The subscriber MUST NOT wait for any group at or after this sequence.
 
-SUBSCRIBE_END bounds the range but does not by itself end the stream: the publisher MAY still send SUBSCRIBE_DROP for groups below this sequence that it cannot deliver, and FINs the stream only once every group below this sequence has been accounted for.
+**Stream Count**:
+The number of Group Streams the publisher opened for this subscription, whether they finished or were reset.
+A group that was skipped, never produced, delivered only as a datagram, or given up before its stream opened is not counted.
+A relay counts the Group Streams it opened itself, never the count it received upstream.
 
-## SUBSCRIBE_DROP
-A SUBSCRIBE_DROP message is sent by the publisher on the Subscribe Stream when groups cannot be served.
-It MAY arrive at any point after the subscription is opened, including after SUBSCRIBE_END for stragglers within the resolved range (a leading range is instead dropped implicitly by SUBSCRIBE_OK).
-
-~~~
-SUBSCRIBE_DROP Message {
-  Type (i) = 0x2
-  Message Length (i)
-  Group Start (i)
-  Group End (i)
-  Error Code (i)
-}
-~~~
-
-**Type**:
-Set to 0x2 to indicate a SUBSCRIBE_DROP message.
-
-**Group Start**:
-The first absolute group sequence in the dropped range.
-
-**Group End**:
-The last absolute group sequence in the dropped range (inclusive).
-
-**Error Code**:
-An application-specific error code.
-A value of 0 indicates no error; the groups are simply unavailable.
+The publisher MUST NOT send SUBSCRIBE_END until every Group Stream it will open for the subscription has been opened, so the count is final; it does not wait for them to finish.
+The subscriber has received every Group Stream once it has read the header of `Stream Count` of them, which MAY happen after SUBSCRIBE_END or the FIN since streams are not ordered.
+A Group Stream reset before its header arrived is never seen, so a subscriber SHOULD bound how long it waits for the rest, for example by `Subscriber Max Age`.
 
 ## FETCH
 FETCH is sent by a subscriber to request a single group from a track.
@@ -1332,6 +1312,9 @@ The `Message Length` describes the payload size on the wire.
 
 - Assigned `moq-lite-07-wip` as this draft's protocol identifier until it is finalized as `moq-lite-07`.
 - Hid routes with a `.`-prefixed segment below the requested prefix from announce discovery, and added the ANNOUNCE_REQUEST `Hidden` field to opt in.
+- Added `Stream Count` to SUBSCRIBE_END: the number of Group Streams opened for the subscription. SUBSCRIBE_END is now sent once every counted Group Stream has opened, rather than as soon as the final group is known.
+- Removed SUBSCRIBE_DROP and its type 0x2; a group without a Group Stream is not counted.
+- The Subscribe Stream FIN now follows once every counted Group Stream has finished or been reset.
 
 ## moq-lite-06
 

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

@@ -15,8 +15,8 @@ import { Group as GroupMessage } from "./group.ts";
 import { sendOrder } from "./priority.ts";
 import { Probe as ProbeMessage } from "./probe.ts";
 import { Publisher } from "./publisher.ts";
-import { decodeSubscribeResponse, Subscribe, SubscribeUpdate } from "./subscribe.ts";
-import { ALPN_05, ALPN_06, Version } from "./version.ts";
+import { decodeSubscribeResponse, Subscribe, type SubscribeEnd, SubscribeUpdate } from "./subscribe.ts";
+import { ALPN_05, ALPN_06, ALPN_07_WIP, Version } from "./version.ts";
 
 function publish(origin: OriginProducer, path: Path.Valid) {
 	const broadcast = origin.createBroadcast(path);
@@ -85,11 +85,11 @@ test.each([Version.DRAFT_01, Version.DRAFT_03, Version.DRAFT_06])(
 );
 
 // Delivers `sequences` in the given order, finishes the track, and returns the
-// SUBSCRIBE_END boundary the publisher put on the wire.
-async function subscribeEnd(sequences: number[]): Promise<number> {
-	const pair = createMockTransportPair(ALPN_05);
+// SUBSCRIBE_END the publisher put on the wire.
+async function subscribeEnd(sequences: number[], version: Version = Version.DRAFT_05): Promise<SubscribeEnd> {
+	const pair = createMockTransportPair(version === Version.DRAFT_07 ? ALPN_07_WIP : ALPN_05);
 	const origin = new OriginProducer();
-	const publisher = new Publisher(pair.server, Version.DRAFT_05, randomHop(), origin.consume());
+	const publisher = new Publisher(pair.server, version, randomHop(), origin.consume());
 
 	const broadcast = publish(origin, Path.from("test"));
 	const track = broadcast.createTrack("video");
@@ -118,8 +118,8 @@ async function subscribeEnd(sequences: number[]): Promise<number> {
 
 	try {
 		for (;;) {
-			const resp = await decodeSubscribeResponse(client.reader, Version.DRAFT_05);
-			if ("end" in resp) return resp.end.group;
+			const resp = await decodeSubscribeResponse(client.reader, version);
+			if ("end" in resp) return resp.end;
 		}
 	} finally {
 		publisher.close();
@@ -972,19 +972,153 @@ test("lite draft-05: teardown unwinds with an undelivered update queued", async
 // A Rust subscriber feeds this value straight into `track::Producer::finish_at`, which is
 // exclusive, so an inclusive bound here silently truncates the final group across languages.
 test("lite draft-05: subscribe end is the exclusive boundary", async () => {
-	expect(await subscribeEnd([0, 1, 2])).toBe(3);
+	expect((await subscribeEnd([0, 1, 2])).group).toBe(3);
 });
 
 // recvGroup is arrival-ordered, so the boundary has to clear the max sequence delivered,
 // not the last one seen. Otherwise the boundary lands on a group already on the wire.
 test("lite draft-05: subscribe end clears the max sequence when groups arrive out of order", async () => {
-	expect(await subscribeEnd([0, 2, 1])).toBe(3);
+	expect((await subscribeEnd([0, 2, 1])).group).toBe(3);
 });
 
 // 0 is the only encoding for "no groups at all"; an inclusive bound cannot express it
 // without colliding with a track whose sole group was sequence 0.
 test("lite draft-05: subscribe end is 0 when no groups were produced", async () => {
-	expect(await subscribeEnd([])).toBe(0);
+	expect((await subscribeEnd([])).group).toBe(0);
+});
+
+// The count is of group streams opened, not of groups below the end: a group the track
+// never produced has no stream and is not counted.
+test("lite draft-07: subscribe end counts the group streams opened", async () => {
+	const end = await subscribeEnd([0, 2], Version.DRAFT_07);
+	expect([end.group, end.streams]).toEqual([3, 2]);
+});
+
+test("lite draft-07: subscribe end counts zero streams when no groups were produced", async () => {
+	const end = await subscribeEnd([], Version.DRAFT_07);
+	expect([end.group, end.streams]).toEqual([0, 0]);
+});
+
+// finishAt names the end while groups below it are still being produced. The count
+// cannot include a stream that has not opened, so SUBSCRIBE_END waits for them.
+test("lite draft-07: subscribe end waits for groups below a declared finish", async () => {
+	const pair = createMockTransportPair(ALPN_07_WIP);
+	const origin = new OriginProducer();
+	const publisher = new Publisher(pair.server, Version.DRAFT_07, randomHop(), origin.consume());
+	const broadcast = publish(origin, Path.from("test"));
+	const track = broadcast.createTrack("video");
+
+	const client = await Stream.open(pair.client);
+	const server = await Stream.accept(pair.server);
+	if (!server) throw new Error("publisher never accepted the subscribe stream");
+	void publisher.runSubscribe(
+		new Subscribe({ id: 0n, broadcast: Path.from("test"), track: "video", priority: 0 }),
+		server,
+	);
+
+	try {
+		const first = new GroupProducer(0);
+		first.writeString("hello");
+		first.close();
+		track.writeGroup(first);
+		track.finishAt(2);
+
+		const start = await decodeSubscribeResponse(client.reader, Version.DRAFT_07);
+		expect("start" in start).toBe(true);
+		const pending = decodeSubscribeResponse(client.reader, Version.DRAFT_07);
+		const early = await Promise.race([pending, new Promise((resolve) => setTimeout(resolve, IDLE_MS))]);
+		expect(early).toBeUndefined();
+
+		const second = new GroupProducer(1);
+		second.writeString("hello");
+		second.close();
+		track.writeGroup(second);
+		track.close();
+
+		const resp = await pending;
+		if (!("end" in resp)) throw new Error("expected SUBSCRIBE_END");
+		expect([resp.end.group, resp.end.streams]).toEqual([2, 2]);
+	} finally {
+		publisher.close();
+		client.close();
+	}
+});
+
+// Serves one group with its stream open held until `open(ok)`, and returns the pending
+// SUBSCRIBE_END plus the call that lets the open succeed or fail.
+async function heldOpenEnd() {
+	const pair = createMockTransportPair(ALPN_07_WIP);
+	const origin = new OriginProducer();
+	const publisher = new Publisher(pair.server, Version.DRAFT_07, randomHop(), origin.consume());
+	const broadcast = publish(origin, Path.from("test"));
+	const track = broadcast.createTrack("video");
+
+	let open!: (ok: boolean) => void;
+	const opened = new Promise<boolean>((resolve) => {
+		open = resolve;
+	});
+	const createUni = pair.server.createUnidirectionalStream.bind(pair.server);
+	spyOn(pair.server, "createUnidirectionalStream").mockImplementation(async (options) => {
+		if (!(await opened)) throw new Error("no stream credit");
+		return createUni(options);
+	});
+
+	const client = await Stream.open(pair.client);
+	const server = await Stream.accept(pair.server);
+	if (!server) throw new Error("publisher never accepted the subscribe stream");
+	void publisher.runSubscribe(
+		new Subscribe({ id: 0n, broadcast: Path.from("test"), track: "video", priority: 0 }),
+		server,
+	);
+
+	const group = new GroupProducer(0);
+	group.writeString("hello");
+	group.close();
+	track.writeGroup(group);
+	track.close();
+
+	const start = await decodeSubscribeResponse(client.reader, Version.DRAFT_07);
+	expect("start" in start).toBe(true);
+	const end = decodeSubscribeResponse(client.reader, Version.DRAFT_07);
+
+	return {
+		end,
+		open,
+		close() {
+			publisher.close();
+			client.close();
+		},
+	};
+}
+
+// The count is final only once no served group is still waiting for its stream, so
+// SUBSCRIBE_END waits for the open.
+test("lite draft-07: subscribe end waits for every group stream to open", async () => {
+	const held = await heldOpenEnd();
+	try {
+		const early = await Promise.race([held.end, new Promise((resolve) => setTimeout(resolve, IDLE_MS))]);
+		expect(early).toBeUndefined();
+
+		held.open(true);
+		const resp = await held.end;
+		if (!("end" in resp)) throw new Error("expected SUBSCRIBE_END");
+		expect([resp.end.group, resp.end.streams]).toEqual([1, 1]);
+	} finally {
+		held.close();
+	}
+});
+
+// A group that never gets a stream owes the subscriber nothing, so it is not counted.
+test("lite draft-07: a group whose stream never opened is not counted", async () => {
+	const held = await heldOpenEnd();
+	try {
+		held.open(false);
+		const resp = await held.end;
+		if (!("end" in resp)) throw new Error("expected SUBSCRIBE_END");
+		expect([resp.end.group, resp.end.streams]).toEqual([1, 0]);
+	} finally {
+		held.close();
+	}
 });
 
 /** One group stream the publisher put on the wire. */