fix(net): keep opening requests on a draining session until the replacement wins

doc/lib/js/net.md

@@ -50,7 +50,7 @@ for (;;) {
 
 - **Origins** hold the broadcasts, not the connection: closing a session unannounces them but leaves them created for the next one. `origin.request(path)` resolves an announced local broadcast with no round trip, so a page that watches what it publishes reads its own copy, unless a cheaper route announces the same path. A refusal from the most specific route ends the request: `request.closed` settles with the handler's error and no broader route is asked. Create, populate, then `announce()` for an exact path; use `dynamic(prefix, route)` when the set of paths is not known: an exact-path subscribe before the tracks exist is refused, and nobody, local or remote, can see or reach a broadcast until it announces.
 - **Epoch pinning**: `consumer.request(path, { epoch })` pins an announced publisher instance. The request stays unresolved and reports `unroutable` when the winning route names a different or missing epoch, including while an asynchronous handler answers. Omit `epoch` for unpinned lookup. The resolved broadcast's `epoch` names the route it actually resolved through, pinned or not.
-- **Connections** race WebTransport against WebSocket. `new Connection({ url })` pools one connection per relay URL and reconnects with backoff, which the elements use. Supplying WebTransport/WebSocket options, discovery, delay, `goaway`, or a caller-owned origin selects a private loop; explicit `share: true` refuses those options. On a relay's GOAWAY the connection dials the replacement at once while the old session keeps serving its groups until it closes or the `goaway.handover` cap (default 10s, lowered to the relay's deadline) passes; the handle and its origin stay the same. An empty GOAWAY redials the same URL through a fresh DNS resolve. A redirect is followed onto the same host by default (`goaway.redirect: "follow"` lets the relay pick the host, `"ignore"` never moves), and moves the pooled entry to the new URL; a malformed or refused one, including a host change under a pinned certificate, ends the connection with `Error.RefusedRedirect` instead of reconnecting. `closed` settles when the handle is released (`null` on a clean close); the failure that stopped retrying the current URL is `error`, and a new URL recovers the same handle. A connection owns one send-rate sampler and one `Bandwidth.Allocator`; publishers reserve against it so their encoder targets sum to the estimate instead of each matching it.
+- **Connections** race WebTransport against WebSocket. `new Connection({ url })` pools one connection per relay URL and reconnects with backoff, which the elements use. Supplying WebTransport/WebSocket options, discovery, delay, `goaway`, or a caller-owned origin selects a private loop; explicit `share: true` refuses those options. On a relay's GOAWAY the connection dials the replacement at once while the old session keeps serving its groups until it closes or the `goaway.handover` cap (default 10s, lowered to the relay's deadline) passes; the handle and its origin stay the same. New requests keep opening on the old session until the replacement's route outranks it. An empty GOAWAY redials the same URL through a fresh DNS resolve. A redirect is followed onto the same host by default (`goaway.redirect: "follow"` lets the relay pick the host, `"ignore"` never moves), and moves the pooled entry to the new URL; a malformed or refused one, including a host change under a pinned certificate, ends the connection with `Error.RefusedRedirect` instead of reconnecting. `closed` settles when the handle is released (`null` on a clean close); the failure that stopped retrying the current URL is `error`, and a new URL recovers the same handle. A connection owns one send-rate sampler and one `Bandwidth.Allocator`; publishers reserve against it so their encoder targets sum to the estimate instead of each matching it.
 - **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). A route event's `kind` says whether the route started, updated, or ended; its `prefix` is the covered prefix relative to the origin, and `captures` reports what the scope's wildcards matched when the prefix pins them. 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.
 - **Publisher retention** is optional `Track.Info.maxAge`, measured against media timestamps: omission sets no limit, while zero serves only the live edge. It clamps every subscriber's max delay, and stalled media does not age out merely because wall time passes. Separately, cached groups nobody writes for 30 seconds are reclaimed, except a live track's newest group, which stays replayable until the track closes.

drafts/draft-lcurley-moq-lite.md

@@ -522,7 +522,9 @@ Either endpoint can open a Goaway Stream (0x5) to initiate a graceful session sh
 
 The sender sends a GOAWAY message containing an optional new session URI.
 If the URI is non-empty, the peer SHOULD establish a new session at the provided URI and migrate any active subscriptions.
-The peer MUST NOT open new streams on the current session after receiving a GOAWAY.
+After receiving a GOAWAY, the peer MAY keep opening requests on the current session until it has moved to a replacement session.
+The sender SHOULD keep answering them until the session ends, and MAY reset one with GOING_AWAY.
+The recipient SHOULD treat that reset as a route failure and retry the request on the replacement session or another route.
 
 The sender closes the stream (FIN) when it is ready to terminate the session.
 The peer SHOULD close all streams and the session after migrating or when it no longer needs the session.
@@ -1388,6 +1390,7 @@ The `Message Length` describes the payload size on the wire.
 - Assigned 0x3A NOT_FETCHABLE in the stream error table: a FETCH for a group delivered only as a datagram.
 - The subscriber FINs its Subscribe Stream after settling its tail; graceful session close waits for that FIN or reset.
 - A refusal is not retried at another route of the same prefix either.
+- A GOAWAY recipient MAY keep opening requests on the current session until it has moved to a replacement session; previously it MUST NOT open new streams. The sender SHOULD keep answering them and MAY reset one with GOING_AWAY, which the recipient SHOULD retry on another route.
 - Made TRACK_INFO Publisher Max Age optional, encoded as milliseconds plus one with zero meaning no limit.
 - Added `Largest Group` and `Largest Frame` to SUBSCRIBE_OK: the publisher's largest position when it answers, which a subscriber takes as where the live feed is. A publisher MUST answer at once when the requested start is past it. Earlier versions carry no such position, so a subscriber takes the first frame instead.
 - Added `Epoch` to ANNOUNCE_START, SUBSCRIBE, TRACK, and FETCH: a UUIDv7 naming the publisher instance, or empty. A path and an Epoch name one Broadcast. A relay MAY move a subscription between routes with the same Epoch, continuing from the first frame the subscriber lacks instead of at a group boundary, and never between routes with different Epochs or none. The newest Epoch wins a path and ends subscriptions to the older one. Replaces the first-hop identity.

js/net/src/goaway-requests.test.ts

@@ -0,0 +1,268 @@
+import { expect, test } from "bun:test";
+import { Once } from "@moq/signals";
+import { accept } from "./connection/accept.ts";
+import { connect } from "./connection/connect.ts";
+import type { Drain } from "./connection/goaway.ts";
+import { StreamCode } from "./error.ts";
+import type { Session } from "./ietf/adapter.ts";
+import { GoAway } from "./ietf/goaway.ts";
+import { Subscriber } from "./ietf/subscriber.ts";
+import { ALPN, Version as IetfVersion } from "./ietf/version.ts";
+import { Goaway } from "./lite/goaway.ts";
+import { StreamId } from "./lite/stream.ts";
+import { ALPN_06, Version as LiteVersion } from "./lite/version.ts";
+import { createMockTransportPair, type MockTransport } from "./mock.ts";
+import { Producer as OriginProducer } from "./origin.ts";
+import * as Path from "./path.ts";
+import { Stream, Writer } from "./stream.ts";
+import { Timescale } from "./time.ts";
+import type { Producer as TrackProducer } from "./track.ts";
+import { wireOf } from "./wire.ts";
+
+const url = new URL("https://relay.example/room");
+
+// Same ceiling Rust stamps as Cost::DRAIN, and the lite cost varint saturates at.
+const DRAIN = 2n ** 62n - 1n;
+
+async function settle() {
+	await new Promise((resolve) => setTimeout(resolve, 0));
+}
+
+async function waitUntil(pred: () => boolean): Promise<void> {
+	const deadline = Date.now() + 2000;
+	for (;;) {
+		if (pred()) return;
+		if (Date.now() > deadline) throw new Error("timed out waiting for condition");
+		await settle();
+	}
+}
+
+/** Bytes of one GOAWAY, framed the way the draft-17 setup stream carries it. */
+async function encodeIetfGoaway(): Promise<Uint8Array> {
+	const chunks: Uint8Array[] = [];
+	const writable = new WritableStream<Uint8Array>({
+		write(chunk) {
+			chunks.push(Uint8Array.from(chunk));
+		},
+	});
+	const writer = new Writer(writable, IetfVersion.DRAFT_17);
+	await writer.u53(GoAway.id);
+	await new GoAway({ newSessionUri: "" }).encode(writer, IetfVersion.DRAFT_17);
+	const size = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0);
+	const bytes = new Uint8Array(size);
+	let offset = 0;
+	for (const chunk of chunks) {
+		bytes.set(chunk, offset);
+		offset += chunk.byteLength;
+	}
+	return bytes;
+}
+
+/**
+ * The server's setup uni is locked inside the connection. Wrap the first one so the test
+ * can append a GOAWAY after SETUP without taking the writer's lock.
+ */
+function captureFirstUni(transport: MockTransport): { inject(bytes: Uint8Array): Promise<void> } {
+	const original = transport.createUnidirectionalStream.bind(transport);
+	let write: ((bytes: Uint8Array) => Promise<void>) | undefined;
+	let captured = false;
+	transport.createUnidirectionalStream = async (options) => {
+		const writable = await original(options);
+		if (captured) return writable;
+		captured = true;
+		const writer = writable.getWriter();
+		write = (bytes) => writer.write(bytes);
+		return new WritableStream<Uint8Array>({
+			write: (chunk) => writer.write(chunk),
+			close: () => writer.close(),
+			abort: (reason) => writer.abort(reason),
+		});
+	};
+	return {
+		inject: (bytes) => {
+			if (!write) throw new Error("setup stream was not opened");
+			return write(bytes);
+		},
+	};
+}
+
+async function sendLiteGoaway(server: WebTransport): Promise<void> {
+	const stream = await Stream.open(server, { version: LiteVersion.DRAFT_06 });
+	await stream.writer.u53(StreamId.Goaway);
+	await new Goaway("").encode(stream.writer, LiteVersion.DRAFT_06);
+	stream.writer.close();
+}
+
+/**
+ * After GOAWAY the old session keeps opening subscribe, fetch, and announce-interest streams
+ * while it is the only route, at the drain cost. Once a replacement answers, it outranks the
+ * old session and new requests open there instead.
+ */
+async function handover(kind: "lite" | "ietf"): Promise<void> {
+	const protocol = kind === "lite" ? ALPN_06 : ALPN.DRAFT_17;
+	const pair = createMockTransportPair(protocol);
+	const inject = kind === "ietf" ? captureFirstUni(pair.server) : undefined;
+	const goawayBytes = kind === "ietf" ? await encodeIetfGoaway() : undefined;
+
+	const consume = new OriginProducer();
+	const publish = new OriginProducer();
+	const broadcast = publish.createBroadcast(Path.from("room"));
+	broadcast.announce();
+
+	const served: TrackProducer[] = [];
+	let armVideo = true;
+	const serving = (async () => {
+		for (;;) {
+			const req = await wireOf(broadcast).requested();
+			if (!req) break;
+			const track = req.accept({ timescale: Timescale.MILLI });
+			if (req.name === "video" && armVideo) track.writeString("one");
+			if (req.name === "later") track.writeString("from-draining");
+			if (req.name === "audio") track.writeString("from-replacement");
+			served.push(track);
+		}
+	})();
+
+	const [client, server] = await Promise.all([
+		connect({ url, transport: pair.client, consume }),
+		accept({ transport: pair.server, url, publish: publish.consume() }),
+	]);
+
+	const announced = client.announced();
+	const request = consume.request(Path.from("room"));
+	let replacement: Awaited<ReturnType<typeof connect>> | undefined;
+	let replacementServer: Awaited<ReturnType<typeof accept>> | undefined;
+
+	try {
+		const start = await announced.next();
+		expect(start?.kind).toBe("start");
+		expect(start?.prefix).toBe(Path.from("room"));
+
+		await waitUntil(() => request.active.peek() !== undefined);
+		const front = request.active.peek();
+		if (!front) throw new Error("no route");
+
+		let gaps = 0;
+		const stop = request.active.subscribe((active) => {
+			if (active === undefined) gaps++;
+		});
+
+		const video = front.track("video").subscribe().ordered();
+		expect(await video.readString()).toBe("one");
+		armVideo = false;
+
+		const opened = pair.client.sendStreams.bidi.length;
+
+		if (inject && goawayBytes) await inject.inject(goawayBytes);
+		else await sendLiteGoaway(pair.server);
+		await wireOf(client).goaway;
+
+		const update = await announced.next();
+		expect(update?.kind).toBe("update");
+		expect(update?.route.cost.warm).toBe(DRAIN);
+		expect(update?.route.cost.cold).toBe(DRAIN);
+		// Still the only route, so the subscription already open keeps its front.
+		expect(request.active.peek()).toBe(front);
+
+		for (const producer of served) {
+			if (producer.closed.peek() === undefined) producer.writeString("two");
+		}
+		expect(await video.readString()).toBe("two");
+
+		// With no replacement yet, a new subscribe opens on the draining session instead of failing.
+		const later = front.track("later").subscribe().ordered();
+		expect(await later.readString()).toBe("from-draining");
+
+		// The publisher no longer holds group 0, so its answer proves the FETCH reached the wire.
+		const fetched = wireOf(client).consume(Path.from("room")).track("video").fetchGroup(0);
+		if (kind === "lite") await expect(fetched).rejects.toMatchObject({ code: StreamCode.NotFound });
+		else await expect(fetched).rejects.toThrow(/not supported/);
+
+		// A new announce-interest opens too, and sees the route at the drain cost.
+		const interest = client.announced();
+		const drained = await interest.next();
+		expect(drained?.route.cost.warm).toBe(DRAIN);
+		interest.close();
+
+		expect(pair.client.sendStreams.bidi.length).toBeGreaterThan(opened);
+		const drainingOpened = pair.client.sendStreams.bidi.length;
+
+		const pair2 = createMockTransportPair(protocol);
+		[replacement, replacementServer] = await Promise.all([
+			connect({ url, transport: pair2.client, consume }),
+			accept({ transport: pair2.server, url, publish: publish.consume() }),
+		]);
+
+		await waitUntil(() => {
+			const active = request.active.peek();
+			return active !== undefined && active !== front;
+		});
+		const audio = request.active.peek()?.track("audio").subscribe().ordered();
+		if (!audio) throw new Error("replacement did not answer");
+		expect(await audio.readString()).toBe("from-replacement");
+		// Once the replacement wins, nothing new opens on the draining session.
+		expect(pair.client.sendStreams.bidi.length).toBe(drainingOpened);
+		expect(pair2.client.sendStreams.bidi.length).toBeGreaterThan(0);
+		expect(gaps).toBe(0);
+		stop();
+		audio.close();
+		later.close();
+		video.close();
+	} finally {
+		announced.close();
+		request.close();
+		replacement?.abort();
+		replacementServer?.abort();
+		client.abort();
+		server.abort();
+		broadcast.close();
+		await serving;
+		publish.close();
+		consume.close();
+	}
+}
+
+test("lite: requests open on the draining session until the replacement wins", async () => {
+	await handover("lite");
+});
+
+test("ietf: requests open on the draining session until the replacement wins", async () => {
+	await handover("ietf");
+});
+
+test("announcing and subscribing after GOAWAY still open streams, on the adapter too", async () => {
+	for (const version of [IetfVersion.DRAFT_14, IetfVersion.DRAFT_16, IetfVersion.DRAFT_19]) {
+		const goaway = new Once<Drain>();
+		goaway.set({ uri: "" });
+		let opened = false;
+		const session: Session = {
+			version,
+			openBi() {
+				opened = true;
+				throw new Error("openBi");
+			},
+			openNativeBi() {
+				opened = true;
+				throw new Error("openNativeBi");
+			},
+			acceptBi() {
+				return Promise.resolve(undefined);
+			},
+			nextRequestId() {
+				return Promise.resolve(0n);
+			},
+			close() {},
+		};
+
+		const subscriber = new Subscriber({ session, goaway });
+		const track = subscriber.consume(Path.from("room")).track("video").subscribe();
+		await track.closed;
+		expect(opened).toBe(true);
+
+		opened = false;
+		const announced = subscriber.announced();
+		await waitUntil(() => opened);
+		announced.close();
+		subscriber.close();
+	}
+});

js/net/src/ietf/connection.ts

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