feat(net): lite-07 GROUP Frame End continues bounded groups

drafts/draft-lcurley-moq-lite.md

@@ -785,7 +785,7 @@ Receiving padding does not change session or subscription state.
 This allows bandwidth probes to measure delivery without requiring a media subscription.
 Versions before lite-07 do not define this stream type and reset only the stream, as required for unknown stream types.
 
-### Group
+### Group {#group-stream}
 A publisher creates Group Streams in response to a Subscribe Stream.
 
 A Group Stream MUST start with a GROUP message and MAY be followed by any number of FRAME messages.
@@ -795,7 +795,8 @@ A frame MAY contain an empty payload, potentially indicating a gap in the group.
 The GROUP message's `Frame Start` gives the index of the first FRAME on the stream, so a stream carrying only part of a Group is self-describing.
 It is redundant with the subscription's own `Frame Start` in the steady state, and carried anyway because a SUBSCRIBE_UPDATE that moves the bound races the Group Streams already in flight: the two travel on different streams with no ordering between them, so the receiver would otherwise have to guess which bound a given Group Stream was opened under.
 
-A publisher MUST NOT send two Group Streams for the same Group Sequence on one subscription.
+A publisher MUST NOT send two Group Streams for the same Group Sequence on one subscription, except to continue one that stopped at its `Frame End`.
+When a SUBSCRIBE_UPDATE moves the end past such a Group, the publisher sends the rest on a new Group Stream whose `Frame Start` is the earlier stream's `Frame End`.
 A subscriber assembling a Group from more than one publisher does so across separate subscriptions, and is responsible for concatenating the runs in index order and for ignoring any frame it has already received.
 
 Both the publisher and subscriber MAY reset the stream at any time.
@@ -1293,6 +1294,7 @@ SUBSCRIBE_UPDATE Message {
 
 See [SUBSCRIBE](#subscribe) for information about each field.
 Moving `Frame Start` forward within a group that is already being delivered does not retract frames the publisher has already sent; like `Group Start`, it only bounds what the publisher sends from here on.
+Moving the end past a Group whose stream stopped at its `Frame End` sends the rest of that Group (see [Group](#group-stream)).
 
 
 **Wanted Group**:
@@ -1732,6 +1734,7 @@ GROUP Message {
   Subscribe ID (i)
   Group Sequence (i)
   Frame Start (i)
+  Frame End (i)
 }
 ~~~
 
@@ -1750,6 +1753,12 @@ A plain absolute index; 0 means the stream carries the group from its beginning,
 A non-zero value means the leading frames are not on this stream, either because the subscription started partway into this group or because the publisher only holds part of it.
 The subscriber MUST NOT assume it will receive the missing frames on another stream; they are a gap unless a separate subscription or FETCH covers them.
 
+**Frame End**:
+The subscription's end frame this stream stops at, encoded as the index + 1 like SUBSCRIBE's `Frame End`, or 0 when the stream runs to the end of the Group.
+A FIN before `Frame End`, or any FIN when it is 0, ends the Group.
+A FIN at `Frame End` ends only the stream: the Group may have more frames, which a later Group Stream delivers if the subscription's end moves past them (see [Group](#group-stream)).
+A non-zero `Frame End` at or below `Frame Start` is malformed, as is a FRAME at or past it.
+
 
 ## FRAME
 The FRAME message is a payload within a group.
@@ -1789,6 +1798,7 @@ The `Message Length` describes the payload size on the wire.
 ## moq-lite-07
 
 - Allow a negotiated extension to preserve timestamps with explicitly unchecked provenance.
+- Added `Frame End` to GROUP: a FIN at it ends only the stream, and a later Group Stream continues the Group once the subscription's end moves past it.
 
 - Added Padding Stream (0x2), read and discarded incrementally by either endpoint.
 

js/net/src/lite/group.ts

@@ -5,7 +5,7 @@ import { hooks } from "../internal.ts";
 import type { Cursor, Reader, Writer } from "../stream.ts";
 import * as Time from "../time.ts";
 import * as Message from "./message.ts";
-import { hasFrameBounds, type Version } from "./version.ts";
+import { hasFrameBounds, hasFrameEnd, type Version } from "./version.ts";
 
 export class Group {
 	subscribe: bigint;
@@ -21,18 +21,28 @@ export class Group {
 	 */
 	frameStart: number;
 
+	/**
+	 * The exclusive frame this stream stops at because the subscription ends there, or
+	 * undefined when it runs to the end of the group. A FIN at it ends only the stream, so
+	 * a later stream may continue the group. Lite-07+.
+	 */
+	frameEnd?: number;
+
 	constructor({
 		subscribe,
 		sequence,
 		frameStart = 0,
+		frameEnd,
 	}: {
 		subscribe: bigint;
 		sequence: number;
 		frameStart?: number;
+		frameEnd?: number;
 	}) {
 		this.subscribe = subscribe;
 		this.sequence = sequence;
 		this.frameStart = frameStart;
+		this.frameEnd = frameEnd;
 	}
 
 	async #encode(w: Writer, version: Version) {
@@ -44,13 +54,21 @@ export class Group {
 			// The peer would number the frames from 0 and silently misalign the group.
 			throw new Error("frame offsets not supported for this version");
 		}
+		if (hasFrameEnd(version)) {
+			await w.u53(this.frameEnd ?? 0);
+		} else if (this.frameEnd !== undefined) {
+			// The peer would read the FIN as the group's end.
+			throw new Error("frame end not supported for this version");
+		}
 	}
 
 	static async #decode(r: Reader, version: Version): Promise<Group> {
 		const subscribe = await r.u62();
 		const sequence = await r.u53();
 		const frameStart = hasFrameBounds(version) ? await r.u53() : 0;
-		return new Group({ subscribe, sequence, frameStart });
+		const frameEnd = hasFrameEnd(version) ? await r.u53() : 0;
+		if (frameEnd !== 0 && frameEnd <= frameStart) throw new Error("frame end at or below frame start");
+		return new Group({ subscribe, sequence, frameStart, frameEnd: frameEnd === 0 ? undefined : frameEnd });
 	}
 
 	async encode(w: Writer, version: Version): Promise<void> {

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

@@ -1,6 +1,7 @@
 import { expect, test } from "bun:test";
 import { Reader, Writer } from "../stream.ts";
 import { AnnounceInit } from "./announce.ts";
+import { Group } from "./group.ts";
 import { Subscribe } from "./subscribe.ts";
 import { Version } from "./version.ts";
 
@@ -34,3 +35,38 @@ test("control message length 65,535 passes the prefix check", async () => {
 	const wire = await prefix(65_535, version);
 	await expect(Subscribe.decode(new Reader(undefined, wire, version), version)).rejects.not.toThrow("too large");
 });
+
+async function encodeGroup(group: Group, version: Version): Promise<Uint8Array> {
+	const written: Uint8Array[] = [];
+	const writer = new Writer(
+		new WritableStream<Uint8Array>({ write: (chunk) => void written.push(new Uint8Array(chunk)) }),
+		version,
+	);
+	await group.encode(writer, version);
+	writer.close();
+	await writer.closed;
+	const wire = new Uint8Array(written.reduce((size, chunk) => size + chunk.byteLength, 0));
+	let offset = 0;
+	for (const chunk of written) {
+		wire.set(chunk, offset);
+		offset += chunk.byteLength;
+	}
+	return wire;
+}
+
+// Lite-07 carries where a capped stream stops; lite-06 cannot, so it refuses one rather than
+// let the FIN read as the group's end. An end at or below the start is malformed.
+test("GROUP Frame End is lite-07 only", async () => {
+	const capped = new Group({ subscribe: 1n, sequence: 7, frameStart: 4, frameEnd: 6 });
+	const wire = await encodeGroup(capped, Version.DRAFT_07);
+	const decoded = await Group.decode(new Reader(undefined, wire, Version.DRAFT_07), Version.DRAFT_07);
+	expect([decoded.frameStart, decoded.frameEnd]).toEqual([4, 6]);
+
+	await expect(encodeGroup(capped, Version.DRAFT_06)).rejects.toThrow("frame end");
+
+	const inverted = await encodeGroup(new Group({ subscribe: 1n, sequence: 7, frameStart: 4 }), Version.DRAFT_07);
+	inverted[inverted.length - 1] = 4;
+	await expect(Group.decode(new Reader(undefined, inverted, Version.DRAFT_07), Version.DRAFT_07)).rejects.toThrow(
+		"frame end",
+	);
+});

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

@@ -651,7 +651,8 @@ async function servedSubscription(
 ) {
 	const version = options.version ?? Version.DRAFT_05;
 	const frames = options.frames ?? ["hello"];
-	const pair = createMockTransportPair(version === Version.DRAFT_06 ? ALPN_06 : ALPN_05);
+	const alpn = { [Version.DRAFT_06]: ALPN_06, [Version.DRAFT_07]: ALPN_07_WIP }[version as number] ?? ALPN_05;
+	const pair = createMockTransportPair(alpn);
 	const origin = new OriginProducer();
 	const publisher = new Publisher(pair.server, version, randomHop(), origin.consume());
 
@@ -755,7 +756,7 @@ async function servedSubscription(
 				await reader.u62();
 				payloads.push(new TextDecoder().decode(await reader.read(await reader.u53())));
 			}
-			return { sequence: header.sequence, frameStart: header.frameStart, payloads };
+			return { sequence: header.sequence, frameStart: header.frameStart, frameEnd: header.frameEnd, payloads };
 		},
 		// The sequence alone, for tests that only care which groups reached the wire.
 		async servedSequence(): Promise<number | undefined> {
@@ -1331,6 +1332,34 @@ test("lite draft-06: a queued frame floor applies before the next group pop", as
 	}
 });
 
+// A lite-07 stream stopped at the end frame says where, and each update moving the end past
+// it continues the group on a new stream from there.
+test("lite draft-07: a capped group continues once the end moves past it", async () => {
+	const sub = await servedSubscription({
+		version: Version.DRAFT_07,
+		startGroup: 0,
+		endGroup: 0,
+		endFrame: 1,
+		frames: ["a", "b", "c"],
+	});
+	try {
+		sub.serve(0);
+		expect(await sub.servedGroup()).toEqual({ sequence: 0, frameStart: 0, frameEnd: 2, payloads: ["a", "b"] });
+
+		await replayUpdate({ priority: 0, startGroup: 0, endGroup: 0, endFrame: 2 }).encode(
+			sub.client.writer,
+			Version.DRAFT_07,
+		);
+		expect(await sub.servedGroup()).toEqual({ sequence: 0, frameStart: 2, frameEnd: 3, payloads: ["c"] });
+
+		// The group ends where the last stream stopped, which only an empty continuation says.
+		await replayUpdate({ priority: 0, startGroup: 0 }).encode(sub.client.writer, Version.DRAFT_07);
+		expect(await sub.servedGroup()).toEqual({ sequence: 0, frameStart: 3, payloads: [] });
+	} finally {
+		await sub.close();
+	}
+});
+
 // The update and group are both ready when the write unblocks. Control is drained first, then
 // the group is popped and positioned synchronously under the new frame cap.
 test("lite draft-06: a queued frame update applies before the next group pop", async () => {
@@ -1838,7 +1867,7 @@ test("blocked DROP responses stop opening later groups", async () => {
 });
 
 /** One group stream the publisher put on the wire. */
-type Served = { sequence: number; frameStart: number; payloads: string[] };
+type Served = { sequence: number; frameStart: number; frameEnd?: number; payloads: string[] };
 
 /**
  * Serves `groups` (frame payloads per group, keyed by sequence) under `bounds`, and