fix(net): end a track with its session's error when the session dies

js/net/src/error.ts

@@ -424,6 +424,31 @@ export function fromClose(info: WebTransportCloseInfo): Session | null {
 	return new Session(code, { reason: info.reason });
 }
 
+/**
+ * The session's close as the error it ends everything with, carrying the peer's code. A clean
+ * close code is still an error here: whatever the close cut off did not end.
+ *
+ * @internal
+ */
+export function closeError(quic: WebTransport): Promise<Error> {
+	return quic.closed.then(
+		(info) => fromClose(info) ?? new Session(SessionCode.Cancel, { reason: info.reason }),
+		(err: unknown) => error(err),
+	);
+}
+
+/**
+ * Report a failure the session's close caused as the session's own error, which carries the
+ * peer's close code; any other failure passes through.
+ *
+ * @internal
+ */
+export async function sessionCause(quic: WebTransport | undefined, err: unknown): Promise<Error> {
+	const source = typeof err === "object" && err !== null ? (err as { source?: unknown }).source : undefined;
+	if (quic && source === "session") return closeError(quic);
+	return error(err);
+}
+
 /**
  * Coerce an unknown thrown value into an `Error`.
  *

js/net/src/ietf/adapter.ts

@@ -1,4 +1,5 @@
 import { Mutex } from "async-mutex";
+import { error, ProtocolViolation } from "../error.ts";
 import { Reader, Stream, type Writer } from "../stream.ts";
 import * as Varint from "../varint.ts";
 import * as Namespace from "./namespace.ts";
@@ -246,6 +247,8 @@ export class ControlStreamAdapter implements Session {
 	 * Must be called after construction. Runs until the control stream closes.
 	 */
 	async run(): Promise<void> {
+		// Why the virtual streams end: undefined only for a GOAWAY, which is not a failure.
+		let cause: Error | undefined;
 		try {
 			// v16: also accept real bidi streams (for SubscribeNamespace)
 			if (this.version === Version.DRAFT_16) {
@@ -254,7 +257,10 @@ export class ControlStreamAdapter implements Session {
 
 			for (;;) {
 				const done = await this.#reader.done();
-				if (done) break;
+				if (done) {
+					cause = new ProtocolViolation("control stream closed");
+					break;
+				}
 
 				const typeId = await this.#reader.u53();
 				const size = await this.#reader.u16();
@@ -293,8 +299,11 @@ export class ControlStreamAdapter implements Session {
 						break;
 				}
 			}
+		} catch (err: unknown) {
+			cause = error(err);
+			throw err;
 		} finally {
-			this.close();
+			this.close(cause);
 		}
 	}
 
@@ -719,15 +728,19 @@ export class ControlStreamAdapter implements Session {
 		}
 	}
 
-	close() {
+	/**
+	 * Ends every virtual stream: cleanly for a deliberate close, or with `err` when the
+	 * control stream died under them, since every request riding it was cut off.
+	 */
+	close(err?: Error) {
 		if (this.#closed) return;
 		this.#closed = true;
 		console.debug("adapter: close() called");
 
-		// Close all virtual streams
 		for (const entry of this.#streams.values()) {
 			try {
-				entry.controller.close();
+				if (err) entry.controller.error(err);
+				else entry.controller.close();
 			} catch {
 				// Already closed
 			}

js/net/src/ietf/connection.ts

@@ -141,7 +141,7 @@ export class Connection implements Established {
 		});
 		this.#solicit = solicit;
 		this.#cluster = cluster;
-		this.#subscriber = new Subscriber({ session: this.#session, cluster, hidden });
+		this.#subscriber = new Subscriber({ session: this.#session, quic, cluster, hidden });
 		registerWire(this, { consume: (path) => this.#subscriber.consume(path) });
 
 		void this.#run();

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

@@ -639,8 +639,9 @@ test("closing the session ends the unsolicited announce loop", async () => {
 	// The session ends. The origin is untouched: it is shared, and other sessions keep using it.
 	pair.server.close();
 
+	// Ending with the session's error is ending too: the close fails its open streams.
 	await Promise.race([
-		loop,
+		loop.catch(() => undefined),
 		new Promise((_resolve, reject) =>
 			setTimeout(() => reject(new Error("the announce loop outlived its session")), STREAM_WAIT),
 		),

js/net/src/ietf/subscriber.ts

@@ -2,7 +2,7 @@ import { race, Signal } from "@moq/signals";
 import * as announce from "../announced.ts";
 import * as broadcast from "../broadcast.ts";
 import { BroadcastCache } from "../consume.ts";
-import { controlTimeout, error, ProtocolViolation, reason } from "../error.ts";
+import { closeError, controlTimeout, error, ProtocolViolation, reason, sessionCause } from "../error.ts";
 import * as netGroup from "../group.ts";
 import { Cost, type Route, routesEqual, UNKNOWN_HOP } from "../hop.ts";
 import { hiddenBelow, hooks, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts";
@@ -97,6 +97,10 @@ function sees(filter: Filter, path: Path.Valid): boolean {
 export class Subscriber {
 	#session: Session;
 
+	// The transport, so a request cut off by the session's close ends with the session's
+	// error. Optional for tests that drive a bare session.
+	#quic?: WebTransport;
+
 	// The Hop IDs this session declared; see {@link Cluster}. What the peer declared is what
 	// says whether an advertisement carries a hop path, and ours is what a path looping back
 	// to us contains.
@@ -140,17 +144,21 @@ export class Subscriber {
 	 */
 	constructor({
 		session,
+		quic,
 		cluster,
 		hidden = false,
 	}: {
 		/** The session abstraction for bidi streams and request IDs. */
 		session: Session;
+		/** The transport the session runs on. */
+		quic?: WebTransport;
 		/** The Hop IDs the SETUP exchange settled (MoQ Cluster). */
 		cluster?: Cluster.Hops;
 		/** Whether the peer understands the HIDDEN parameter (MoQ Hidden). */
 		hidden?: boolean;
 	}) {
 		this.#session = session;
+		this.#quic = quic;
 		this.#cluster = cluster;
 		this.#hidden = hidden;
 	}
@@ -479,10 +487,22 @@ export class Subscriber {
 		return consumer;
 	}
 
+	// The adapter is gone. If the transport has already closed, that close is the
+	// error. A still-open transport, such as a GOAWAY drain, has no peer code yet,
+	// so this does not wait for it.
+	async #closedSession(): Promise<Error> {
+		const quic = this.#quic;
+		if (!quic) return new Error("session closed");
+		return Promise.race([
+			closeError(quic),
+			new Promise<Error>((resolve) => queueMicrotask(() => resolve(new Error("session closed")))),
+		]);
+	}
+
 	async #runSubscribe(broadcast: Path.Valid, request: track.Request) {
 		const requestId = await this.#session.nextRequestId();
 		if (requestId === undefined) {
-			request.reject(new Error("session closed"));
+			request.reject(await this.#closedSession());
 			return;
 		}
 
@@ -533,7 +553,7 @@ export class Subscriber {
 			console.debug(`subscribe ok: id=${requestId} broadcast=${broadcast} track=${request.name}`);
 		} catch (err) {
 			// A control request that timed out is not late content, so it carries its own code.
-			const e = err instanceof TimeoutError ? controlTimeout(err) : error(err);
+			const e = err instanceof TimeoutError ? controlTimeout(err) : await sessionCause(this.#quic, err);
 			request.reject(e);
 			console.warn(
 				`subscribe error: id=${requestId} broadcast=${broadcast} track=${request.name} error=${reason(e)}`,
@@ -617,7 +637,7 @@ export class Subscriber {
 			stream.close();
 			console.debug(`subscribe close: id=${requestId} broadcast=${broadcast} track=${request.name}`);
 		} catch (err) {
-			const e = error(err);
+			const e = await sessionCause(this.#quic, err);
 			producer.close(e);
 			stream.abort(e);
 			console.warn(

js/net/src/integration.test.ts

@@ -9,7 +9,7 @@ import {
 	connect as connectSession,
 	type Established,
 } from "./connection/index.ts";
-import { StreamCode, StreamError, TooFarBehind } from "./error.ts";
+import { SessionCode, SessionError, StreamCode, StreamError, TooFarBehind } from "./error.ts";
 import * as Ietf from "./ietf/index.ts";
 import * as Lite from "./lite/index.ts";
 import { createMockTransportPair } from "./mock.ts";
@@ -2218,3 +2218,100 @@ test("a handle serves a request under live/** over the wire", async () => {
 	server.close();
 	origin.close();
 });
+
+// The peer's close code a killed session carries.
+const DEATH = SessionCode(71);
+
+/**
+ * Serve one track with a group left open, kill the publisher's session once the subscriber has
+ * read into it, and return how the subscriber's track ended.
+ */
+async function runSessionDeath(protocol: string, version?: number): Promise<Error | null> {
+	const pair = createMockTransportPair(protocol);
+	const origin = new OriginProducer();
+
+	const [client, server] = await Promise.all([
+		connect(url, { transport: pair.client }),
+		accept(pair.server, url, { version, publish: origin.consume() }),
+	]);
+
+	const broadcast = publish(origin, Path.from("test"));
+	const serving = (async () => {
+		for (;;) {
+			const req = await wireOf(broadcast).requested();
+			if (!req) break;
+			req.accept().appendGroup().writeString("head");
+		}
+	})();
+
+	const remote = wireOf(client).consume(Path.from("test"));
+	const track = remote.track("video").subscribe();
+	const group = await track.recvGroup();
+	expect(await group?.readString()).toBe("head");
+
+	pair.server.close({ closeCode: DEATH, reason: "killed" });
+	const closed = await withTimeout(Promise.resolve(track.closed), 2000, "the track never ended");
+
+	broadcast.close();
+	await serving;
+	remote.close();
+	client.close();
+	server.close();
+	origin.close();
+	return closed;
+}
+
+for (const [name, protocol, version] of [
+	["lite draft-03", Lite.ALPN_03, undefined],
+	["lite draft-05", Lite.ALPN_05, undefined],
+	["ietf draft-14", "", Ietf.Version.DRAFT_14],
+	["ietf draft-17", Ietf.ALPN.DRAFT_17, undefined],
+] as const) {
+	test(`integration: ${name} ends a track with its session's error`, async () => {
+		const closed = await runSessionDeath(protocol, version);
+		expect(closed).toBeInstanceOf(SessionError);
+		expect((closed as SessionError).code).toBe(DEATH);
+	});
+}
+
+// On lite-05+ the subscribe stream carries responses until its FIN, so a reset of it is how the
+// publisher ends a subscription with an error, and the track ends with that error.
+test("integration: lite draft-05 ends a track with the publisher's reset", async () => {
+	const pair = createMockTransportPair(Lite.ALPN_05);
+	const origin = new OriginProducer();
+
+	const [client, server] = await Promise.all([
+		connect(url, { transport: pair.client }),
+		accept(pair.server, url, { publish: origin.consume() }),
+	]);
+
+	const broadcast = publish(origin, Path.from("test"));
+	const served: TrackProducer[] = [];
+	const serving = (async () => {
+		for (;;) {
+			const req = await wireOf(broadcast).requested();
+			if (!req) break;
+			const producer = req.accept();
+			producer.appendGroup().writeString("head");
+			served.push(producer);
+		}
+	})();
+
+	const remote = wireOf(client).consume(Path.from("test"));
+	const track = remote.track("video").subscribe();
+	const group = await track.recvGroup();
+	expect(await group?.readString()).toBe("head");
+
+	const reset = StreamCode(70);
+	for (const producer of served) producer.close(new StreamError(reset));
+	const closed = await withTimeout(Promise.resolve(track.closed), 2000, "the track never ended");
+	expect(closed).toBeInstanceOf(StreamError);
+	expect((closed as StreamError).code).toBe(reset);
+
+	broadcast.close();
+	await serving;
+	remote.close();
+	client.close();
+	server.close();
+	origin.close();
+});

js/net/src/lite/connection.ts

@@ -3,7 +3,7 @@ import type * as announce from "../announced.ts";
 import type { Established } from "../connection/established.ts";
 import { type Probe, type Stats, transportStats } from "../connection/stats.ts";
 import { type Transport, transportOf } from "../connection/transport.ts";
-import { error, fromClose, StreamCode, StreamError } from "../error.ts";
+import { closeError, error, fromClose, StreamCode, StreamError, sessionCause } from "../error.ts";
 import { type Hop, randomHop } from "../hop.ts";
 import type { Consumer as OriginConsumer } from "../origin.ts";
 import type * as Path from "../path.ts";
@@ -161,11 +161,17 @@ export class Connection implements Established {
 			tasks.push(this.#subscriber.runDatagrams());
 		}
 
+		let fatal: Error | undefined;
 		try {
 			await Promise.all(tasks);
 		} catch (err) {
 			console.error("fatal error running connection", err);
+			// A session-sourced failure is the peer's close, not the raw transport error.
+			fatal = await sessionCause(this.#quic, err);
 		} finally {
+			// The session died under every track it was receiving, so they end with its
+			// error. A deliberate close() already ended them cleanly, which makes this a no-op.
+			this.#subscriber.close(fatal ?? (await closeError(this.#quic)));
 			this.close();
 		}
 	}