From 400491e6977f5cfff6224385c28e0c711f1d7ba0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:04:36 -0700 Subject: [PATCH 1/5] claim quest/m1/js-goaway-requests Co-Authored-By: Grok 4.7 From 2d5624beaa1d72de46161150b12ac6e71c87ac30 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:42:26 -0700 Subject: [PATCH 2/5] fix(js): stop opening requests after GOAWAY Co-Authored-By: Grok 4.7 --- js/net/src/goaway-requests.test.ts | 260 +++++++++++++++++++++++++++++ js/net/src/ietf/connection.ts | 2 +- js/net/src/ietf/subscriber.ts | 63 ++++++- js/net/src/lite/connection.ts | 9 +- js/net/src/lite/subscriber.ts | 72 +++++++- quest/m1/README.md | 1 - quest/m1/js-goaway-requests.md | 35 ---- quest/m1/transport-upgrade/js.md | 5 +- 8 files changed, 402 insertions(+), 45 deletions(-) create mode 100644 js/net/src/goaway-requests.test.ts delete mode 100644 quest/m1/js-goaway-requests.md diff --git a/js/net/src/goaway-requests.test.ts b/js/net/src/goaway-requests.test.ts new file mode 100644 index 0000000000..0c56eb4803 --- /dev/null +++ b/js/net/src/goaway-requests.test.ts @@ -0,0 +1,260 @@ +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, StreamError } 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 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 { + 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 { + const chunks: Uint8Array[] = []; + const writable = new WritableStream({ + 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 } { + const original = transport.createUnidirectionalStream.bind(transport); + let write: ((bytes: Uint8Array) => Promise) | 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({ + 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 { + 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 opens no subscribe, fetch, or announce-interest stream. + * An existing subscription still delivers, and a new request resolves on the replacement. + */ +async function handover(kind: "lite" | "ietf"): Promise { + 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(); + if (req.name === "video" && armVideo) track.writeString("one"); + 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> | undefined; + let replacementServer: Awaited> | 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"); + + const later = wireOf(client).consume(Path.from("room")).track("later").subscribe(); + expect(await later.closed).toMatchObject({ code: StreamCode.GoingAway }); + + const fetched = wireOf(client).consume(Path.from("room")).track("video").fetchGroup(0); + if (kind === "lite") await expect(fetched).rejects.toMatchObject({ code: StreamCode.GoingAway }); + else await expect(fetched).rejects.toThrow(/not supported/); + + const interest = client.announced(); + await settle(); + interest.close(); + + expect(pair.client.sendStreams.bidi.length).toBe(opened); + expect(server).toBeDefined(); + + 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"); + expect(pair.client.sendStreams.bidi.length).toBe(opened); + 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: a GOAWAY opens no new request and the replacement serves the next one", async () => { + await handover("lite"); +}); + +test("ietf: a GOAWAY opens no new request and the replacement serves the next one", async () => { + await handover("ietf"); +}); + +test("announcing and subscribing after GOAWAY opens nothing, on the adapter too", async () => { + for (const version of [IetfVersion.DRAFT_14, IetfVersion.DRAFT_16, IetfVersion.DRAFT_19]) { + const goaway = new Once(); + 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 announced = subscriber.announced(); + const track = subscriber.consume(Path.from("room")).track("video").subscribe(); + await settle(); + expect(opened).toBe(false); + const cause = await track.closed; + if (!(cause instanceof StreamError)) throw new Error(`expected GoingAway, got ${String(cause)}`); + expect(cause.code).toBe(StreamCode.GoingAway); + announced.close(); + subscriber.close(); + } +}); diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 9bf0084651..9d0e40840d 100644 --- a/js/net/src/ietf/connection.ts +++ b/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); diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 1c8ff68367..c0b9f8992b 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -1,6 +1,7 @@ -import { race, Signal } from "@moq/signals"; +import { type GetPromise, race, Signal } from "@moq/signals"; import * as announce from "../announced.ts"; import * as broadcast from "../broadcast.ts"; +import type { Drain } from "../connection/goaway.ts"; import { BroadcastCache } from "../consume.ts"; import { closeError, @@ -55,6 +56,11 @@ import { Version } from "./version.ts"; // blocks. The timeout turns that into a clear error. const SUBSCRIBE_OK_TIMEOUT_MS = 10_000; +// Wire ceiling (2^62-1). A draining session stamps it on every live route so any other +// candidate outranks it, while the route stays selectable as the last path. Matches Rust +// Cost::DRAIN: cost is the whole mechanism, not a separate state. +const DRAIN_COST = { warm: 2n ** 62n - 1n, cold: 2n ** 62n - 1n }; + // A live subscription, as the track alias its data streams name resolves to. type Subscription = { // The write side incoming group streams are routed into. @@ -154,6 +160,10 @@ export class Subscriber { // Whether the peer understands the HIDDEN parameter (MoQ Hidden). #hidden: boolean; + // Settles when the peer sends GOAWAY. New opens stop; subscriptions already running + // keep flowing until the session closes. + #goaway?: GetPromise; + /** Marks this subscriber's deliberate local session close. @internal */ close() { this.#localClose = true; @@ -169,6 +179,7 @@ export class Subscriber { quic, cluster, hidden = false, + goaway, }: { /** The session abstraction for bidi streams and request IDs. */ session: Session; @@ -178,11 +189,40 @@ export class Subscriber { cluster?: Cluster.Hops; /** Whether the peer understands the HIDDEN parameter (MoQ Hidden). */ hidden?: boolean; + /** Settles when the peer sends GOAWAY. */ + goaway?: GetPromise; }) { this.#session = session; this.#quic = quic; this.#cluster = cluster; this.#hidden = hidden; + this.#goaway = goaway; + // A draining peer usually stops publishing namespaces, so reprice from the signal + // itself. Waiting for another message would leave the route primary until close. + if (goaway) void goaway.then(() => this.#drainAnnounced()); + } + + // The peer told us to stop opening streams on this session. + #goingAway(): boolean { + return this.#goaway?.peek() !== undefined; + } + + #goingAwayError(): StreamError { + return new StreamError(StreamCode.GoingAway, { message: "going away" }); + } + + // What a route costs once the peer has asked us to leave. + #priced(route: Route): Route { + if (!this.#goingAway()) return route; + if (route.cost.warm === DRAIN_COST.warm && route.cost.cold === DRAIN_COST.cold) return route; + return { ...route, cost: DRAIN_COST }; + } + + // Reprice every live advertisement. Idempotent, since the signal stays set. + #drainAnnounced(): void { + for (const [path, info] of this.#announced) { + this.#updateAnnounce(path, info.route); + } } /** @@ -251,9 +291,12 @@ export class Subscriber { * first. A second one is the same namespace said twice, not news. */ #attachAnnounce(path: Path.Valid, route: Route) { + route = this.#priced(route); const existing = this.#announced.get(path); if (existing) { existing.count += 1; + // A second advertisement after GOAWAY still must not win selection. + if (this.#goingAway()) this.#updateAnnounce(path, existing.route); return; } this.#announced.set(path, { count: 1, route }); @@ -273,6 +316,7 @@ export class Subscriber { * the first hop now says, so the shared consume stays. */ #updateAnnounce(path: Path.Valid, route: Route) { + route = this.#priced(route); const existing = this.#announced.get(path); if (existing === undefined || routesEqual(existing.route, route)) return; existing.route = route; @@ -319,6 +363,15 @@ export class Subscriber { } async #runAnnounced(announced: announce.Producer, prefix: Path.Valid, hidden: boolean) { + // The peer asked us to stop opening streams, including through the draft-14 to -16 + // adapter (virtual openBi and native openNativeBi alike). Leaving the consumer + // pending, rather than closing it, keeps a forwarder from treating discovery as + // failed and answering new requests on this session. + if (this.#goingAway()) { + await announced.closed; + return; + } + const version = this.#session.version; // Suffixes live on this stream, so a repeat is recognized as an update to the @@ -529,6 +582,13 @@ export class Subscriber { } async #runSubscribe(broadcast: Path.Valid, request: track.Request) { + // A peer that sent GOAWAY told us to stop opening streams. Reject before taking a + // request id, so nothing is registered and the caller sees the refusal. + if (this.#goingAway()) { + request.reject(this.#goingAwayError()); + return; + } + const requestId = await this.#session.nextRequestId(); if (requestId === undefined) { request.reject(await this.#closedSession()); @@ -742,6 +802,7 @@ export class Subscriber { ): Promise<{ stream: Stream; alias: bigint }> { const version = this.#session.version; + if (this.#goingAway()) throw this.#goingAwayError(); state.stream = await this.#session.openBi(); // The timeout can fire while the open is still in flight, in which case cleanup ran diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index 2dda1492d7..bcd6e39cc5 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -137,7 +137,14 @@ export class Connection implements Established { this.hop = randomHop(); this.#publisher = new Publisher(this.#quic, this.#version, this.hop, publish); - this.#subscriber = new Subscriber(this.#quic, this.#version, this.hop, this.#probe, this.#peerSetup); + this.#subscriber = new Subscriber( + this.#quic, + this.#version, + this.hop, + this.#probe, + this.#peerSetup, + this.#goaway, + ); registerWire(this, { consume: (path) => this.#subscriber.consume(path), goaway: this.#goaway }); void this.#run(); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index eb2bae2d21..2de262a54c 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -1,6 +1,7 @@ -import { race, Signal } from "@moq/signals"; +import { type GetPromise, race, Signal } from "@moq/signals"; import * as announce from "../announced.ts"; import * as broadcast from "../broadcast.ts"; +import type { Drain } from "../connection/goaway.ts"; import type { Probe as ProbeStats } from "../connection/stats.ts"; import { BroadcastCache } from "../consume.ts"; import type * as Epoch from "../epoch.ts"; @@ -59,6 +60,11 @@ import { // until the peer frees a slot. The timeout turns a stall into a clear error. const SUBSCRIBE_SETUP_TIMEOUT_MS = 10_000; +// Wire ceiling (2^62-1). A draining session stamps it on every live route so any other +// candidate outranks it, while the route stays selectable as the last path. Matches Rust +// Cost::DRAIN: cost is the whole mechanism, not a separate state. +const DRAIN_COST = { warm: 2n ** 62n - 1n, cold: 2n ** 62n - 1n }; + // The TRACK stream and implicit SUBSCRIBE acceptance are lite-05+. function supportsTrackStream(version: Version): boolean { switch (version) { @@ -159,6 +165,10 @@ export class Subscriber { // stream on the peer having advertised Probe >= Report. #peerSetup?: Signal; + // Settles when the peer sends GOAWAY. New opens stop; subscriptions already running + // keep flowing until the session closes. + #goaway?: GetPromise; + // Distinguishes failures from streams torn down by Subscriber.close(). #closed = new AbortController(); /** @@ -168,6 +178,7 @@ export class Subscriber { * @param origin - Hop id shared with the Publisher * @param probe - Optional sink for the peer's PROBE estimates * @param peerSetup - Optional peer SETUP slot for capability gating (lite-05+) + * @param goaway - Settles when the peer sends GOAWAY * * @internal */ @@ -177,6 +188,7 @@ export class Subscriber { hop: Hop, probe?: Signal, peerSetup?: Signal, + goaway?: GetPromise, ) { this.#quic = quic; this.version = version; @@ -184,6 +196,22 @@ export class Subscriber { this.#stamp = randomHop(); this.#probe = probe; this.#peerSetup = peerSetup; + this.#goaway = goaway; + } + + // The peer told us to stop opening streams on this session. + #goingAway(): boolean { + return this.#goaway?.peek() !== undefined; + } + + #goingAwayError(): StreamError { + return new StreamError(StreamCode.GoingAway, { message: "going away" }); + } + + // What a route costs once the peer has asked us to leave. A later announce on a + // draining session must not win selection, however cheap the path it advertises. + #cost(cost?: Cost): Cost { + return this.#goingAway() ? DRAIN_COST : (cost ?? Cost.zero); } /** @@ -212,6 +240,13 @@ export class Subscriber { hidden: boolean, ): Promise { console.debug(`announced: prefix=${prefix}`); + // The peer asked us to stop opening streams. Leaving the consumer pending, rather + // than closing it, keeps a forwarder from treating discovery as failed and answering + // new requests on this session. + if (this.#goingAway()) { + await announced.closed; + return; + } // Lite04/05: send our own session-level Hop ID so the peer can skip announces // whose hop chain already passed through us. Encoding drops it on every other // version, where we drop the reflected announce on receipt instead. Matches the @@ -277,7 +312,7 @@ export class Subscriber { if (advertised.has(path)) { throw new ProtocolViolation(`duplicate announce for ${path}`); } - const route = { hops: [this.#stamp, UNKNOWN_HOP], cost: Cost.zero }; + const route = { hops: [this.#stamp, UNKNOWN_HOP], cost: this.#cost() }; const live = visible(path); const captures = scopeCaptures(scope, path); advertised.set(path, { live, route, captures }); @@ -292,6 +327,22 @@ export class Subscriber { break; } + // A draining peer usually stops announcing, so reprice from the GOAWAY itself. + // Waiting for another message would leave the route primary until the session + // closed. Idempotent: the signal stays set, and an unchanged cost emits nothing. + const drainAdvertised = () => { + if (announced.closed.peek() !== undefined) return; + for (const [path, ad] of advertised) { + if (!ad.live) continue; + const route = { ...ad.route, cost: DRAIN_COST }; + if (routesEqual(ad.route, route)) continue; + advertised.set(path, { ...ad, route }); + announced.append({ prefix: path, captures: ad.captures, kind: "update", route }); + } + }; + if (this.#goaway) void this.#goaway.then(() => drainAdvertised()); + if (this.#goingAway()) drainAdvertised(); + // Lite06+: announce ids. Each received `active` implicitly assigns the next // per-stream ordinal; `endedId`/`restart` reference it, and lite-07 bases copy // from it. Tracked even for announces we skip as reflected, since the sender @@ -433,7 +484,7 @@ export class Subscriber { }); continue; } - const route: Route = { epoch, hops: fullHops, cost: cost ?? Cost.zero }; + const route: Route = { epoch, hops: fullHops, cost: this.#cost(cost) }; const captures = scopeCaptures(scope, path); if (!visible(path)) { advertised.set(path, { live: false, route, captures }); @@ -513,6 +564,13 @@ export class Subscriber { } async #runSubscribe(broadcast: Path.Valid, epoch: Epoch.Valid | undefined, request: track.Request) { + // A peer that sent GOAWAY told us to stop opening streams. Reject before accepting, + // so the caller sees the refusal and nothing is registered. + if (this.#goingAway()) { + request.reject(this.#goingAwayError()); + return; + } + const id = this.#subscribeNext++; const subscription = request.subscription; const initialBounds = groupBounds(subscription.groups); @@ -678,6 +736,7 @@ export class Subscriber { }; this.#subscribes.set(id, entry); + if (this.#goingAway()) throw this.#goingAwayError(); state.stream = await Stream.open(this.#quic, { version: this.version }); await state.stream.writer.u53(StreamId.Subscribe); await msg.encode(state.stream.writer, this.version); @@ -709,6 +768,7 @@ export class Subscriber { // fails. Subscriber.close() also resets it while `run` is pending, so a peer that never // answers cannot hold it open, and a stream that opens after the close is reset at once. async #exchange(options: OpenOptions, run: (stream: Stream) => Promise): Promise { + if (this.#goingAway()) throw this.#goingAwayError(); const closed = this.#closed.signal; closed.throwIfAborted(); const stream = await Stream.open(this.#quic, options); @@ -770,6 +830,8 @@ export class Subscriber { if (entry && !entry.group.isClosed) { consumer = entry.group.mirror(); } else { + // A fetch already in flight keeps its stream. A new one does not open. + if (this.#goingAway()) throw this.#goingAwayError(); const group = new netGroup.Producer(sequence); consumer = group.mirror(); entry = { @@ -1208,6 +1270,10 @@ export class Subscriber { if (probe < ProbeLevel.Report) return; } + // Probe is best-effort telemetry. After GOAWAY there is nothing to open and nothing + // to fail: the session keeps serving what it already has. + if (this.#goingAway()) return; + // Probe is best-effort: any failure (stream reset by peer, missing peer support, // transport hiccup) MUST NOT tear down the connection. On error, drop the // estimates so consumers know they're stale. diff --git a/quest/m1/README.md b/quest/m1/README.md index 21d2307fea..a6152c729c 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -130,7 +130,6 @@ blocks. The quests that gated m0 lines moved under them. - [Media stats](/quest/m1/stats/README.md) - publishers announce a stats track in the catalog, viewers answer a soliciting catalog through a per-catalog `.echo` broadcast, and one model turns both into a health verdict and a preflight report - [JS track handover](/quest/m1/js-group-handover.md) - a JS track subscription resumes across a route swap from the first frame it lacks, so `test/drain` passes at zero latency budget - [JS startup hole](/quest/m1/js-startup-hole.md) - a `@moq/hang` consumer delivers its first live group at once, even when an empty group died before it -- [JS GOAWAY requests](/quest/m1/js-goaway-requests.md) - after GOAWAY the JS client opens no new request on the old session, like Rust - [Drain handshakes](/quest/m1/drain-handshakes.md) - a drain GOAWAYs and waits for sessions still in their handshake instead of exiting under them - [Transport upgrade](/quest/m1/transport-upgrade/README.md) - a session that came up over WebSocket moves to QUIC once the QUIC dial lands, handing over without dropping a group - [Scope track priority](/quest/m1/track-priority-scope.md) - priority orders one owner's streams, and a shared cluster session is fair across tenants diff --git a/quest/m1/js-goaway-requests.md b/quest/m1/js-goaway-requests.md deleted file mode 100644 index bf8de798b2..0000000000 --- a/quest/m1/js-goaway-requests.md +++ /dev/null @@ -1,35 +0,0 @@ -# [S] JS refuses new requests after GOAWAY - -## Goal - -Once a `@moq/net` session has received GOAWAY, it opens no new subscribe, -fetch, or announce-interest stream on that session, on lite and IETF, as -moq-net already refuses them with `Error::GoingAway`. Existing subscriptions -keep flowing until the handover ends, and a refused request is served by the -replacement session rather than failing the caller. - -## Plan - -The drain line's JS migration (https://github.com/moq-dev/moq/pull/4143) -keeps the old session serving after GOAWAY but left its subscribers unaware -of it, so during the handover an origin request can still open a stream on a -session the peer asked us to leave. The lite draft says the recipient must -not open new streams after GOAWAY, and a compliant draining relay rejects -them. Rust checks before every new open in both wires (`check_going_away` in -the lite subscriber, the `going_away` checks in the IETF one). - -Guidance: - -- The drain signal already reaches the connection wire view - (`wireOf(session).goaway`). Hand it to both subscribers and check it at - every open site, including the draft-14 to -16 adapter route. -- The refusal matters as much as the check: the origin should see it as this - route declining, so the request moves to the replacement's route (the - origin already keeps the outranked route until the new one answers). Match - what the Rust origin does with `GoingAway` rather than inventing a policy. -- Test on lite and IETF with the existing mock transports: after GOAWAY a new - request opens no stream on the old session and resolves on the new one, - while an existing subscription keeps receiving groups. - -No public API change is expected; the error is internal unless a caller can -observe it today. diff --git a/quest/m1/transport-upgrade/js.md b/quest/m1/transport-upgrade/js.md index 0038f64c67..186dde4b17 100644 --- a/quest/m1/transport-upgrade/js.md +++ b/quest/m1/transport-upgrade/js.md @@ -38,8 +38,8 @@ session to close on its own or at the handover cap. See the keeps WebSocket with no GOAWAY sent. This crate's servers send SETUP after admission; other servers may still refuse afterward. - A self-sent GOAWAY must gate new requests on the old session too, not only - a received one: [JS GOAWAY requests](/quest/m1/js-goaway-requests.md) - covers the received case, so check it also covers this path. + a received one. Receiving GOAWAY already stops new subscribe, fetch, and + announce-interest opens; sending one still has to set that same signal. - On a successful upgrade delete the URL from `websocketWon`. - Tests in the browser harness against the in-tree relay: with the WebTransport dial delayed past the head start, a watched track keeps every @@ -54,4 +54,3 @@ session to close on its own or at the handover cap. See the - [JS qmux finish](/quest/m1/transport-upgrade/js-qmux-finish.md) - the GOAWAY this sends over the WebSocket session must not read as a failure - [JS track handover](/quest/m1/js-group-handover.md) - tracks carry across the handover this upgrade reuses without a dropped group -- [JS GOAWAY requests](/quest/m1/js-goaway-requests.md) - no new request opens on a session that is going away From 327153be95b6e267999dd206c18a868733e81fea Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 13:54:46 -0700 Subject: [PATCH 3/5] drafts(lite): a GOAWAY sender SHOULD keep answering requests, MAY reset with GOING_AWAY Co-Authored-By: Claude Opus 5.5 --- drafts/draft-lcurley-moq-lite.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index e9c39fde67..4dcb2c6298 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -521,7 +521,8 @@ 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. -After receiving a GOAWAY, the peer MAY keep opening requests on the current session until it has moved to a replacement session, and the sender MUST keep answering them until the session ends. +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, which the peer retries on the replacement session rather than treating as fatal. 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. @@ -1382,7 +1383,7 @@ The `Message Length` describes the payload size on the wire. - 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, which the sender MUST keep answering; previously it MUST NOT open new streams. +- 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 retries on the replacement. - 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. From 8f672f9dca0f8c65c74a578f73cc852128a9fbc9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 16:48:17 -0700 Subject: [PATCH 4/5] fix(net): retarget the fetch drill mutation and release the lite drain listener - test/drill/mutations/fetch-never-sent.patch anchors on the FETCH open arm now that the GOAWAY gate is gone; `just test drill-sensitivity fetch-never-sent` passes. - The lite announce interest disposes its GOAWAY listener when it ends, so a session that never drains doesn't retain every closed interest. - The lite draft makes the GOING_AWAY retry a recipient SHOULD on another route. - Note at the IETF sites that requests keep opening past draft-19's SHOULD NOT. Co-Authored-By: Claude Opus 5.5 --- drafts/draft-lcurley-moq-lite.md | 5 +++-- js/net/src/ietf/subscriber.ts | 3 ++- js/net/src/lite/subscriber.ts | 11 ++++++++--- rs/moq-net/src/ietf/subscriber.rs | 4 +++- test/drill/mutations/fetch-never-sent.patch | 12 ++++++------ 5 files changed, 22 insertions(+), 13 deletions(-) diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 4dcb2c6298..8b05ff2794 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -522,7 +522,8 @@ 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. 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, which the peer retries on the replacement session rather than treating as fatal. +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. @@ -1383,7 +1384,7 @@ The `Message Length` describes the payload size on the wire. - 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 retries on the replacement. +- 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. diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 6a1c56095d..dbb93d28c6 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -202,7 +202,8 @@ export class Subscriber { } // Whether the peer has sent GOAWAY. Requests keep opening here until a replacement - // session's route outranks this one. + // session's route outranks this one, deliberately past draft-19 section 10.4's SHOULD + // NOT: refusing them would fail requests that land before the replacement is up. #goingAway(): boolean { return this.#goaway?.peek() !== undefined; } diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index bcb1caeffb..297c75560e 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -1,4 +1,4 @@ -import { type GetPromise, race, Signal } from "@moq/signals"; +import { type Dispose, type GetPromise, race, Signal } from "@moq/signals"; import * as announce from "../announced.ts"; import * as broadcast from "../broadcast.ts"; import type { Drain } from "../connection/goaway.ts"; @@ -253,6 +253,7 @@ export class Subscriber { return; } + let stopDrain: Dispose | undefined; try { // Send the announce interest. await stream.writer.u53(StreamId.Announce); @@ -318,7 +319,8 @@ export class Subscriber { // A draining peer usually stops announcing, so reprice from the GOAWAY itself. // Waiting for another message would leave the route primary until the session - // closed. Idempotent: the signal stays set, and an unchanged cost emits nothing. + // closed. Idempotent: an unchanged cost emits nothing. A GOAWAY that already + // arrived needs no listener, since `#cost()` priced every route above. const drainAdvertised = () => { if (announced.closed.peek() !== undefined) return; for (const [path, ad] of advertised) { @@ -329,7 +331,7 @@ export class Subscriber { announced.append({ prefix: path, captures: ad.captures, kind: "update", route }); } }; - if (this.#goaway) void this.#goaway.then(() => drainAdvertised()); + stopDrain = this.#goaway?.changed(() => drainAdvertised()); // Lite06+: announce ids. Each received `active` implicitly assigns the next // per-stream ordinal; `endedId`/`restart` reference it, and lite-07 bases copy @@ -516,6 +518,9 @@ export class Subscriber { if (e instanceof ProtocolViolation) { this.#quic.close({ closeCode: PROTOCOL_VIOLATION_CODE, reason: closeReason(reason(e)) }); } + } finally { + // Releases this interest's routes on a session that never drains. + stopDrain?.(); } } diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index d11f6afd2d..3681a0e4d7 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -581,7 +581,9 @@ pub(super) struct Subscriber { tasks: Tasks, version: Version, // Set once the peer sends a GOAWAY; this session's routes then cost - // Cost::DRAIN, so a replacement session outranks it. + // Cost::DRAIN, so a replacement session outranks it. Requests keep opening, + // deliberately past draft-19 section 10.4's SHOULD NOT: refusing them would + // fail requests that land before the replacement is up. going_away: crate::goaway::GoingAway, // What this session may allocate up front for objects still arriving. frames: frame::Budget, diff --git a/test/drill/mutations/fetch-never-sent.patch b/test/drill/mutations/fetch-never-sent.patch index 22a5a13f4a..3d48447a7c 100644 --- a/test/drill/mutations/fetch-never-sent.patch +++ b/test/drill/mutations/fetch-never-sent.patch @@ -10,15 +10,15 @@ diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber. index 55006c5f7..93f2f9677 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs -@@ -4445,6 +4445,11 @@ impl kio::Task for FetchServeRun { +@@ -4466,6 +4466,11 @@ impl kio::Task for FetchServeRun { - match &mut self.state { - FetchRunState::Open { request } => { + match &mut self.state { + FetchRunState::Open { request } => { + // MUTATION: never send the FETCH, so nothing ever answers it. + if request.is_some() { + return Poll::Pending; + } + - // A peer that sent GOAWAY told us to stop opening streams on this session. - if self.serve.subscriber.going_away.is_set() { - request.take().expect("request pending").reject(Error::GoingAway); + let mut stream = match ready!(Stream::poll_open( + &mut self.session, + self.serve.subscriber.version, From e3f82cea8301c5ced82a81dbe462f15363ae069e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 20:11:38 -0700 Subject: [PATCH 5/5] test(drill): restore context prefixes in the fetch-never-sent mutation Co-Authored-By: Claude Opus 5.5 --- test/drill/mutations/fetch-never-sent.patch | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/test/drill/mutations/fetch-never-sent.patch b/test/drill/mutations/fetch-never-sent.patch index 3d48447a7c..20d7e5b94a 100644 --- a/test/drill/mutations/fetch-never-sent.patch +++ b/test/drill/mutations/fetch-never-sent.patch @@ -10,15 +10,15 @@ diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber. index 55006c5f7..93f2f9677 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs -@@ -4466,6 +4466,11 @@ impl kio::Task for FetchServeRun { - - match &mut self.state { - FetchRunState::Open { request } => { +@@ -4456,6 +4456,11 @@ impl kio::Task for FetchServeRun { + + match &mut self.state { + FetchRunState::Open { request } => { + // MUTATION: never send the FETCH, so nothing ever answers it. + if request.is_some() { + return Poll::Pending; + } + - let mut stream = match ready!(Stream::poll_open( - &mut self.session, - self.serve.subscriber.version, + let mut stream = match ready!(Stream::poll_open( + &mut self.session, + self.serve.subscriber.version,