fix(moq-net): resolve reordered track aliases (moq-dev#2262)

Co-authored-by: OpenAI Codex <codex@openai.com>

Author: kixelated

js/net/src/ietf/adapter.ts

@@ -13,7 +13,7 @@ export interface Session {
 	openNativeBi?(): Promise<Stream>;
 	acceptBi(): Promise<Stream | undefined>;
 	nextRequestId(): Promise<bigint | undefined>;
-	close?(): void;
+	close(): void;
 	readonly version: IetfVersion;
 }
 
@@ -44,6 +44,11 @@ export class NativeSession implements Session {
 		this.#requestId += 2n;
 		return id;
 	}
+
+	/** Closes the underlying WebTransport session. */
+	close() {
+		this.#quic.close();
+	}
 }
 
 // Route classification for control stream messages.

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

@@ -0,0 +1,39 @@
+import { expect, test } from "bun:test";
+import { TrackAliases } from "./aliases.ts";
+
+test("waits for the control message that establishes an alias", async () => {
+	const aliases = new TrackAliases<object>();
+	const track = {};
+	const pending = aliases.get(7n);
+
+	aliases.set(7n, track);
+
+	expect(await pending).toBe(track);
+});
+
+test("resolves every subgroup waiting for the same alias", async () => {
+	const aliases = new TrackAliases<object>();
+	const track = {};
+	const pending = [aliases.get(7n), aliases.get(7n)];
+
+	aliases.set(7n, track);
+
+	expect(await Promise.all(pending)).toEqual([track, track]);
+});
+
+test("rejects an alias used by two active tracks", () => {
+	const aliases = new TrackAliases<object>();
+	aliases.set(7n, {});
+
+	expect(() => aliases.set(7n, {})).toThrow("duplicate track alias");
+});
+
+test("does not let stale cleanup remove a reused alias", async () => {
+	const aliases = new TrackAliases<object>();
+	const active = {};
+	aliases.set(7n, active);
+
+	aliases.delete(7n, {});
+
+	expect(await aliases.get(7n)).toBe(active);
+});

js/net/src/ietf/aliases.ts

@@ -0,0 +1,54 @@
+const TRACK_ALIAS_TIMEOUT_MS = 1000;
+
+type Resolver<T> = PromiseWithResolvers<T>["resolve"];
+
+/** Resolves publisher-chosen track aliases after control/data stream reordering. @internal */
+export class TrackAliases<T> {
+	#active = new Map<bigint, T>();
+	#pending = new Map<bigint, Set<Resolver<T>>>();
+
+	/** Waits briefly for an alias to be established by SUBSCRIBE_OK or PUBLISH. */
+	async get(alias: bigint): Promise<T> {
+		if (this.#active.has(alias)) return this.#active.get(alias) as T;
+
+		const { promise, resolve } = Promise.withResolvers<T>();
+		let resolvers = this.#pending.get(alias);
+		if (!resolvers) {
+			resolvers = new Set();
+			this.#pending.set(alias, resolvers);
+		}
+		resolvers.add(resolve);
+
+		let timer: ReturnType<typeof setTimeout> | undefined;
+		const timeout = new Promise<never>((_, reject) => {
+			timer = setTimeout(() => reject(new Error(`unknown track alias: ${alias}`)), TRACK_ALIAS_TIMEOUT_MS);
+		});
+
+		try {
+			return await Promise.race([promise, timeout]);
+		} finally {
+			clearTimeout(timer);
+			resolvers.delete(resolve);
+			if (this.#pending.get(alias) === resolvers && resolvers.size === 0) this.#pending.delete(alias);
+		}
+	}
+
+	/** Establishes an alias and releases any data streams waiting for it. */
+	set(alias: bigint, value: T) {
+		const active = this.#active.get(alias);
+		if (this.#active.has(alias)) {
+			if (active !== value) throw new Error(`duplicate track alias: ${alias}`);
+			return;
+		}
+
+		this.#active.set(alias, value);
+		const resolvers = this.#pending.get(alias);
+		this.#pending.delete(alias);
+		for (const resolve of resolvers ?? []) resolve(value);
+	}
+
+	/** Removes an alias only if it still belongs to the supplied value. */
+	delete(alias: bigint, value: T) {
+		if (this.#active.get(alias) === value) this.#active.delete(alias);
+	}
+}

js/net/src/ietf/connection.ts

@@ -98,7 +98,7 @@ export class Connection implements Established {
 
 		this.#closed = true;
 
-		this.#session.close?.();
+		this.#session.close();
 
 		try {
 			this.#quic.close();

js/net/src/ietf/subscriber.ts

@@ -6,6 +6,7 @@ import type { Reader, Stream } from "../stream.ts";
 import type { Track } from "../track.ts";
 import { error } from "../util/error.ts";
 import type { Session } from "./adapter.ts";
+import { TrackAliases } from "./aliases.ts";
 import { Frame, type Group as GroupMessage } from "./object.ts";
 import { type Publish, PublishError } from "./publish.ts";
 import { type PublishNamespace, PublishNamespaceError, PublishNamespaceOk } from "./publish_namespace.ts";
@@ -31,8 +32,8 @@ import { Version } from "./version.ts";
 export class Subscriber {
 	#session: Session;
 
-	// Our subscribed tracks — keyed by trackAlias for group routing
-	#subscribes = new Map<bigint, Track>();
+	// Publisher-chosen aliases used by incoming group streams.
+	#aliases = new TrackAliases<Track>();
 
 	// Any currently active announcements.
 	#announced = new Set<Path.Valid>();
@@ -219,20 +220,19 @@ export class Subscriber {
 				await msg.encode(stream.writer, version);
 				console.debug(`subscribe written: id=${requestId} broadcast=${broadcast} track=${request.track.name}`);
 
-				// Pre-register with requestId so early group uni streams aren't dropped.
-				// The publisher typically uses requestId as the trackAlias.
-				this.#subscribes.set(requestId, request.track);
-
 				// Read response (SubscribeOk or error)
 				const respTypeId = await stream.reader.u53();
 				if (respTypeId === SubscribeOk.id) {
 					const ok = await SubscribeOk.decode(stream.reader, version);
-					// Update registration to use the actual trackAlias from SubscribeOk
-					if (ok.trackAlias !== requestId) {
-						this.#subscribes.delete(requestId);
-						this.#subscribes.set(ok.trackAlias, request.track);
+					try {
+						this.#aliases.set(ok.trackAlias, request.track);
+					} catch (err) {
+						this.#session.close();
+						throw err;
 					}
-					console.debug(`subscribe ok: id=${requestId} broadcast=${broadcast} track=${request.track.name}`);
+					console.debug(
+						`subscribe ok: id=${requestId} broadcast=${broadcast} track=${request.track.name} alias=${ok.trackAlias}`,
+					);
 
 					try {
 						// Wait for stream close (= PublishDone) or track close (= local unsubscribe)
@@ -259,12 +259,9 @@ export class Subscriber {
 							`subscribe close: id=${requestId} broadcast=${broadcast} track=${request.track.name}`,
 						);
 					} finally {
-						this.#subscribes.delete(ok.trackAlias);
+						this.#aliases.delete(ok.trackAlias, request.track);
 					}
 				} else {
-					// Clean up pre-registered entry on error
-					this.#subscribes.delete(requestId);
-
 					// Error response
 					let reasonPhrase = "unknown error";
 					try {
@@ -282,7 +279,6 @@ export class Subscriber {
 					throw new Error(`SUBSCRIBE error: ${reasonPhrase}`);
 				}
 			} catch (err) {
-				this.#subscribes.delete(requestId);
 				stream.abort(error(err));
 				throw err;
 			}
@@ -417,12 +413,8 @@ export class Subscriber {
 		}
 
 		try {
-			// Look up by trackAlias directly
-			const track = this.#subscribes.get(group.trackAlias);
-			if (!track) {
-				// Fallback: try treating trackAlias as requestId (for compat)
-				throw new Error(`unknown track: trackAlias=${group.trackAlias}`);
-			}
+			// The control message establishing this alias can arrive after the data stream.
+			const track = await this.#aliases.get(group.trackAlias);
 
 			track.writeGroup(producer);