fix(net): deliver a track's tail up to its declared end

doc/lib/js/net.md

@@ -52,6 +52,7 @@ for (;;) {
 - **Bandwidth** (`Bandwidth.Allocator`) divides the connection's send-rate estimate by track priority, max-min fair within a tier. An idle track claims nothing. The receive side is untouched.
 - **Discovery** by any pattern scope (`origin.announced(scope)`, such as `room/*/chat`; default everything). Each event's `prefix` is the covered prefix relative to the origin, `captures` reports what the scope's wildcards matched when the prefix pins them, and `kind` says whether it was announced, updated, or retracted. The consumer is an async iterable. `origin.broadcasts(scope)` is a live `Getter<ReadonlyMap<Path.Valid, Route>>` of the same covered prefixes for UIs that need the current set. A borrowed `Connection.origin` also exposes `dynamic(prefix, route)` for serving paths on demand.
 - **Subscriptions** carry a priority, a `Time.Milli` max age, and optional `groups` bounds. Groups arrive out of order and are read frame by frame, with `Error.TooFarBehind` when a reader asks for a frame the group never held and `Error.GroupTooLarge` when a write exceeds the cache budget and aborts the group.
+- **Track ends**: `close()` ends a track at its live edge, while `finishAt(n)` declares the exclusive end ahead of it and still accepts the groups below. A subscriber reads the end with `final()` or awaits `finished()`. A remote track ends only once every group below its end has arrived or was dropped; one reset before its header arrived is skipped after the subscription's max age on moq-lite (one second without one), or after one second on IETF.
 - **Datagrams** on moq-lite 05+ and fetch-by-sequence for history.
 - **Errors** live under one namespace: a stream reset throws `Error.Stream` with a `StreamCode`, while a session close gives `Error.Session` with a `SessionCode`. The registries are disjoint, so the same number means different things in each, and 64+ is yours. Named conditions such as `Error.TooFarBehind`, `Error.FrameTooLarge`, and `Error.GroupTooLarge` subclass `Error.Stream`, so one `code` check handles a condition raised here or reported by the peer. IETF streams use their own mapping: cancellation sends CANCELLED, other local failures send INTERNAL\_ERROR, and received codes remain opaque.
 - **Paths** with `Path.relative` for the cross-broadcast catalog references hang uses. Path patterns (`Path.Pattern`, `Path.Patterns`) are re-exported from [`@moq/pattern`](https://www.npmjs.com/package/@moq/pattern). Literal `Path` stays a coordinate.

js/net/src/ietf/adapter.ts

@@ -58,7 +58,7 @@ export class NativeSession implements Session {
 const Route = {
 	NewRequest: 0, // Create virtual bidi stream, push initial message
 	Response: 1, // Push message to existing stream (keep open)
-	ErrorResponse: 2, // Push message to existing stream, then close
+	ErrorResponse: 2, // Push a final message to existing stream, then close
 	CloseStream: 3, // Close stream recv (no bytes pushed)
 	FollowUp: 4, // Push follow-up message to existing stream
 	MaxRequestId: 5, // Update flow control
@@ -658,9 +658,9 @@ export class ControlStreamAdapter implements Session {
 				return { route: Route.CloseStream, requestId };
 			}
 			case 0x0b: {
-				// PublishDone
+				// PublishDone: the subscriber reads its status and stream count before the end.
 				const requestId = await readRequestId();
-				return { route: Route.CloseStream, requestId };
+				return { route: Route.ErrorResponse, requestId };
 			}
 			case 0x17: {
 				// FetchCancel

js/net/src/ietf/object.ts

@@ -3,6 +3,7 @@ import { Timescale, Timestamp } from "../time.ts";
 import { type IetfVersion, Version } from "./version.ts";
 
 const GROUP_END = 0x03;
+const END_OF_TRACK = 0x04;
 
 // MOQ Object Property ids, shared with draft-ietf-moq-loc-04.
 const PROP_TIMESCALE = 0x08n;
@@ -259,14 +260,24 @@ export class Group {
 
 /** A moq-transport object inside a group stream. */
 export class Frame {
-	/** The object payload, or `undefined` for the end of group marker. */
+	/** The object payload, or `undefined` for an end of group or end of track marker. */
 	payload?: Uint8Array;
 	/** The presentation timestamp carried in object properties, when present. */
 	timestamp?: Timestamp;
+	/**
+	 * An END_OF_TRACK marker: no object at or past its location exists. At object 0 its group
+	 * does not exist either, so the track ends at that group; later in a group it ends after it.
+	 */
+	endOfTrack: boolean;
 
-	constructor({ payload, timestamp }: { payload?: Uint8Array; timestamp?: Timestamp } = {}) {
+	constructor({
+		payload,
+		timestamp,
+		endOfTrack = false,
+	}: { payload?: Uint8Array; timestamp?: Timestamp; endOfTrack?: boolean } = {}) {
 		this.payload = payload;
 		this.timestamp = timestamp;
+		this.endOfTrack = endOfTrack;
 	}
 
 	/**
@@ -284,7 +295,10 @@ export class Frame {
 			await w.write(extensions);
 		}
 
-		if (this.payload !== undefined) {
+		if (this.endOfTrack) {
+			await w.u53(0); // length = 0
+			await w.u53(END_OF_TRACK);
+		} else if (this.payload !== undefined) {
 			await w.u53(this.payload.byteLength);
 
 			if (this.payload.byteLength === 0) {
@@ -334,6 +348,9 @@ export class Frame {
 
 		const status = await r.u53();
 
+		// Defined on every implemented draft, whether or not the header marks the group's end.
+		if (status === END_OF_TRACK) return new Frame({ endOfTrack: true });
+
 		if (flags.hasEnd) {
 			// Empty frame
 			if (status === 0) return new Frame({ payload: new Uint8Array(0), timestamp });

js/net/src/ietf/publish.ts

@@ -213,21 +213,53 @@ export class PublishError {
 	}
 }
 
+/** PUBLISH_DONE status codes this implementation distinguishes. Stable across drafts 14 through 22. */
+export const PublishDoneStatus = {
+	INTERNAL_ERROR: 0x0,
+	TRACK_ENDED: 0x2,
+	/** Removed in draft-20, where 0x3 is unassigned. */
+	SUBSCRIPTION_ENDED: 0x3,
+} as const;
+
+/** Whether a PUBLISH_DONE status ends the track cleanly rather than aborting it. */
+export function publishDoneClean(statusCode: number, version: IetfVersion): boolean {
+	if (statusCode === PublishDoneStatus.TRACK_ENDED) return true;
+	if (statusCode !== PublishDoneStatus.SUBSCRIPTION_ENDED) return false;
+	switch (version) {
+		case Version.DRAFT_14:
+		case Version.DRAFT_15:
+		case Version.DRAFT_16:
+		case Version.DRAFT_17:
+		case Version.DRAFT_18:
+		case Version.DRAFT_19:
+			return true;
+		default:
+			return false;
+	}
+}
+
 // In draft-14, this message is renamed from SUBSCRIBE_DONE to PUBLISH_DONE
 export class PublishDone {
 	static readonly id = 0x0b;
 
 	requestId: bigint | undefined;
 	statusCode: number;
+	/**
+	 * How many data streams the publisher opened for the subscription, fill streams included.
+	 * A hint: a peer may send 0 or the "unknown" sentinel regardless.
+	 */
+	streamCount: bigint;
 	reasonPhrase: string;
 
 	constructor({
 		requestId,
 		statusCode,
+		streamCount = 0n,
 		reasonPhrase,
-	}: { requestId?: bigint; statusCode: number; reasonPhrase: string }) {
+	}: { requestId?: bigint; statusCode: number; streamCount?: bigint; reasonPhrase: string }) {
 		this.requestId = requestId;
 		this.statusCode = statusCode;
+		this.streamCount = streamCount;
 		this.reasonPhrase = reasonPhrase;
 	}
 
@@ -237,7 +269,7 @@ export class PublishDone {
 			await w.u62(this.requestId);
 		}
 		await w.u62(BigInt(this.statusCode));
-		await w.u62(BigInt(0)); // stream_count = 0 (unsupported)
+		await w.u62(this.streamCount);
 		await w.string(this.reasonPhrase);
 	}
 
@@ -255,9 +287,9 @@ export class PublishDone {
 				? await r.u62()
 				: undefined;
 		const statusCode = Number(await r.u62());
-		await r.u62(); // ignore stream_count
+		const streamCount = await r.u62();
 		const reasonPhrase = await r.string();
 
-		return new PublishDone({ requestId, statusCode, reasonPhrase });
+		return new PublishDone({ requestId, statusCode, streamCount, reasonPhrase });
 	}
 }

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

@@ -14,7 +14,7 @@ import { wireOf } from "../wire.ts";
 import { NativeSession, type Session } from "./adapter.ts";
 import type * as Cluster from "./cluster.ts";
 import { FetchHeader } from "./fetch.ts";
-import { Group as GroupMessage } from "./object.ts";
+import { Frame, Group as GroupMessage } from "./object.ts";
 import { PublishDone } from "./publish.ts";
 import { PublishNamespace } from "./publish_namespace.ts";
 import { Publisher } from "./publisher.ts";
@@ -908,6 +908,16 @@ async function readGroup(stream: ReadableStream<Uint8Array>): Promise<ServedGrou
 	return { sequence: header.groupId, firstObject: header.flags.firstObject, objects };
 }
 
+/** Read a stream carrying only an END_OF_TRACK object, returning the group it names. */
+async function readEndOfTrack(stream: ReadableStream<Uint8Array>): Promise<number> {
+	const reader = new Reader(stream, undefined, V20);
+	const header = await GroupMessage.decode(reader, V20);
+	const frame = await Frame.decode(reader, header.flags, undefined, V20);
+	expect(frame.endOfTrack).toBe(true);
+	expect(await reader.done()).toBe(true);
+	return header.groupId;
+}
+
 /**
  * Read a fill's fetch stream to its end, reporting a reset rather than throwing.
  *
@@ -1339,8 +1349,12 @@ test("draft-20: a clean close past a bounded filter's end still sends PUBLISH_DO
 		expect(await client.reader.u53()).toBe(PublishDone.id);
 		const done = await PublishDone.decode(client.reader, V20);
 		expect(done.statusCode).toBe(TRACK_ENDED_STATUS);
+		expect(done.streamCount).toBe(2n);
 
-		// Only the in-range group was ever opened.
+		// Only the in-range group was ever served; the other stream marks the track's end.
+		const end = await nextUni(fx.uni);
+		if (!end) throw new Error("the track's end was never marked");
+		expect(await readEndOfTrack(end)).toBe(2);
 		expect(await nextUni(fx.uni)).toBeUndefined();
 	} finally {
 		fx.close();
@@ -1454,3 +1468,61 @@ test("draft-20: a fill works on a dynamically requested track", async () => {
 		client.close();
 	}
 });
+
+// PUBLISH_DONE MUST wait until every stream the subscription will open is closed, so its
+// Stream Count is final. A group still queued for a stream slot when the track ends is one.
+test("draft-20: PUBLISH_DONE waits for a queued group and counts every stream", async () => {
+	const fx = fixture();
+	const track = fx.broadcast.createTrack("video");
+
+	// Park the first stream open, the way a transport at its stream cap does.
+	const slot = Promise.withResolvers<void>();
+	const create = fx.pair.server.createUnidirectionalStream.bind(fx.pair.server);
+	let parked = false;
+	fx.pair.server.createUnidirectionalStream = async (options?: WebTransportSendStreamOptions) => {
+		if (!parked) {
+			parked = true;
+			await slot.promise;
+		}
+		return create(options);
+	};
+
+	const { client } = await runSubscribe(
+		fx,
+		new Subscribe({
+			requestId: 7n,
+			trackNamespace: Path.from("test"),
+			trackName: "video",
+			subscriberPriority: 0,
+			filter: { kind: "absolute", startGroup: 0n, startObject: 0n },
+		}),
+	);
+
+	try {
+		writeGroup(track, 1);
+		track.close();
+
+		// Nothing ends the subscription while the group waits for its slot.
+		const response = client.reader.u53();
+		const idle = new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 20));
+		expect(await Promise.race([response, idle])).toBe("pending");
+
+		slot.resolve();
+		const served = await nextUni(fx.uni);
+		if (!served) throw new Error("the queued group was never served");
+		expect((await readGroup(served)).sequence).toBe(0);
+
+		expect(await response).toBe(PublishDone.id);
+		const done = await PublishDone.decode(client.reader, V20);
+		expect(done.statusCode).toBe(TRACK_ENDED_STATUS);
+		// The group's stream and the END_OF_TRACK marker's.
+		expect(done.streamCount).toBe(2n);
+
+		const end = await nextUni(fx.uni);
+		if (!end) throw new Error("the track's end was never marked");
+		expect(await readEndOfTrack(end)).toBe(1);
+	} finally {
+		fx.close();
+		client.close();
+	}
+});