diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 7adff50256..ef22aa606a 100644 --- a/doc/lib/js/net.md +++ b/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>` 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. diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index f93a2e639f..1dfbf0ec9e 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/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. diff --git a/js/net/src/goaway-requests.test.ts b/js/net/src/goaway-requests.test.ts new file mode 100644 index 0000000000..e58fc5e2e3 --- /dev/null +++ b/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 { + 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 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 { + 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> | 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"); + + // 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(); + 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(); + } +}); 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 4d10d0e8f4..7db2009683 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. @@ -150,6 +156,9 @@ export class Subscriber { // Whether the peer understands the HIDDEN parameter (MoQ Hidden). #hidden: boolean; + // Settles when the peer sends GOAWAY, repricing this session's routes to the drain cost. + #goaway?: GetPromise; + /** Marks this subscriber's deliberate local session close. @internal */ close() { this.#localClose = true; @@ -165,6 +174,7 @@ export class Subscriber { quic, cluster, hidden = false, + goaway, }: { /** The session abstraction for bidi streams and request IDs. */ session: Session; @@ -174,11 +184,38 @@ 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()); + } + + // Whether the peer has sent GOAWAY. Requests keep opening here until a replacement + // 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; + } + + // 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); + } } /** @@ -241,9 +278,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 }); @@ -263,6 +303,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; 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 8a4bf31361..209874f7e4 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 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"; 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. export 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) { @@ -155,6 +161,9 @@ export class Subscriber { // stream on the peer having advertised Probe >= Report. #peerSetup?: Signal; + // Settles when the peer sends GOAWAY, repricing this session's routes to the drain cost. + #goaway?: GetPromise; + // Distinguishes failures from streams torn down by Subscriber.close(). #closed = new AbortController(); /** @@ -164,6 +173,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 */ @@ -173,12 +183,26 @@ export class Subscriber { hop: Hop, probe?: Signal, peerSetup?: Signal, + goaway?: GetPromise, ) { this.#quic = quic; this.version = version; this.hop = hop; this.#probe = probe; this.#peerSetup = peerSetup; + this.#goaway = goaway; + } + + // Whether the peer has sent GOAWAY. Requests keep opening here until a replacement + // session's route outranks this one. + #goingAway(): boolean { + return this.#goaway?.peek() !== undefined; + } + + // 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); } /** @@ -224,6 +248,7 @@ export class Subscriber { return; } + let stopDrain: Dispose | undefined; try { // Send the announce interest. await stream.writer.u53(StreamId.Announce); @@ -272,7 +297,7 @@ export class Subscriber { if (advertised.has(path)) { throw new ProtocolViolation(`duplicate announce for ${path}`); } - const route = { hops: [UNKNOWN_HOP], cost: Cost.zero }; + const route = { hops: [UNKNOWN_HOP], cost: this.#cost() }; const live = visible(path); const captures = scopeCaptures(scope, path); advertised.set(path, { live, route, captures }); @@ -287,6 +312,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: 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) { + 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 }); + } + }; + 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 // from it. Tracked even for announces we skip as reflected, since the sender @@ -425,7 +466,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 }); @@ -469,6 +510,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?.(); } } @@ -1218,6 +1262,9 @@ export class Subscriber { if (probe < ProbeLevel.Report) return; } + // A session that is going away has no use for a new estimate. + 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 99c369dcf0..880151c4f8 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -119,7 +119,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/js-request-deadline.md b/quest/m1/js-request-deadline.md index 9a7401300a..d7e98d72ec 100644 --- a/quest/m1/js-request-deadline.md +++ b/quest/m1/js-request-deadline.md @@ -71,4 +71,3 @@ Public API: one new error. Wire: none. - [Rust request credit](/quest/m1/rs-request-credit.md) - the same rule in moq-net - [qmux no-wait opens](/quest/m1/qmux-no-wait.md) - the WebSocket fallback stops queueing over-limit creates -- [JS GOAWAY requests](/quest/m1/js-goaway-requests.md) - adds a check at every open site this touches diff --git a/quest/m1/transport-upgrade/README.md b/quest/m1/transport-upgrade/README.md index 0a9581ed02..b46b5c8f65 100644 --- a/quest/m1/transport-upgrade/README.md +++ b/quest/m1/transport-upgrade/README.md @@ -57,8 +57,9 @@ Shared decisions, which Rust implements and [JavaScript](/quest/m1/transport-upg network, so the head start comes back. Otherwise a network where WebSocket narrowly beats QUIC would open two connections on every reconnect. - The old session gets `Goaway::new()` with the configured handover cap before - it enters draining. The relay refuses new requests on it from then on; the - front cancels its subscriptions once the new session feeds them. + it enters draining. The relay prices its routes at the drain cost and keeps + opening requests on it only until the new session's route wins; the front + cancels its subscriptions once the new session feeds them. - One-shot `connect()` returns one session and never upgrades; every `Connection` upgrades, reconnecting or not. - Publishing over the old session is announced again over the new one; the diff --git a/quest/m1/transport-upgrade/js.md b/quest/m1/transport-upgrade/js.md index 1b4a1df1f5..91f1380755 100644 --- a/quest/m1/transport-upgrade/js.md +++ b/quest/m1/transport-upgrade/js.md @@ -37,9 +37,9 @@ session to close on its own or at the handover cap. See the connect deadline. A refusal, a stall, or moq-lite-03/-04 (no server SETUP) 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 self-sent GOAWAY must reprice the old session's routes to the drain cost + too, not only a received one, so new requests move to the QUIC session once + its route answers; 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 @@ -53,4 +53,3 @@ session to close on its own or at the handover cap. See the ## Required - [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 diff --git a/rs/moq-net/src/goaway.rs b/rs/moq-net/src/goaway.rs index 0ac895217e..652b9f7eeb 100644 --- a/rs/moq-net/src/goaway.rs +++ b/rs/moq-net/src/goaway.rs @@ -175,9 +175,9 @@ impl Consumer { /// The receive-side GOAWAY signal handed to a subscriber. /// /// Carries the flag and the channel together because a subscriber needs both: an -/// atomic load before opening a request stream (the hot path, which is why the -/// flag is not merely a read of the channel), and a wakeup for the per-source -/// tasks, which have to react to a peer that drains and then goes quiet. +/// atomic load when pricing each received announcement (the hot path, which is +/// why the flag is not merely a read of the channel), and a wakeup for the +/// per-source tasks, which have to react to a peer that drains and then goes quiet. #[derive(Clone)] pub(crate) struct GoingAway { flag: Arc, @@ -239,7 +239,7 @@ pub(crate) struct Protocol { trigger: kio::Consumer>, /// Written by the driver's receive path when a GOAWAY is decoded. received: kio::Producer>, - /// Set alongside `received`; checked before opening new request streams. + /// Set alongside `received`; reprices this session's routes to the drain cost. pub going_away: GoingAway, } diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 693bf38b91..7ce35c659d 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -576,8 +576,10 @@ pub(super) struct Subscriber { state: Lock, tasks: Tasks, version: Version, - // Set once the peer sends a GOAWAY; new SUBSCRIBEs are then rejected with - // Error::GoingAway (the peer told us to stop opening streams). + // Set once the peer sends a GOAWAY; this session's routes then cost + // 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, @@ -836,12 +838,6 @@ where mut stream: Stream, prefix: PathOwned, ) -> Result<(), Error> { - // A peer that sent GOAWAY told us to stop opening requests on this session, - // announce-interest included (draft-19 sect 10.4). - if self.going_away.is_set() { - return Err(Error::GoingAway); - } - // Hidden namespaces are requested too, as on moq-lite: the session mirrors the // peer into the origin and each local reader opts in on its own. The parameter // fails decoding at a peer that doesn't know it, so it waits on the peer's SETUP @@ -1731,11 +1727,6 @@ where ) { // Data streams wait on the alias bound by SUBSCRIBE_OK, so leave the model request // pending until its immutable track metadata is known. - if self.going_away.is_set() { - request.reject(Error::GoingAway); - return; - } - let track_name = request.name().to_owned(); // Group FETCHes for cache misses: standalone, so they outlive each subscription. let mut group_fetches = TaskSet::owned(); @@ -3427,10 +3418,6 @@ where ) { let sequence = request.sequence(); let start = request.frame_start(); - if self.going_away.is_set() { - request.reject(Error::GoingAway); - return; - } // Our FETCH still encodes the Fetch Type field that draft-20 removed. if Filter::is_draft20(self.version) { request.reject(Error::Unsupported); diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index bd9ae5fe5d..d2210380b7 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -38,8 +38,8 @@ pub(super) struct SubscriberConfig { /// Local policy for what pulling from this peer costs, overriding whatever it /// declared in its SETUP. `None` charges the peer's declared price. pub cost: Option, - /// Set once the peer sends a GOAWAY; new request streams are then rejected - /// with [`Error::GoingAway`] (the peer told us to stop asking). + /// Set once the peer sends a GOAWAY; this session's routes then cost + /// [`crate::origin::Cost::DRAIN`], so a replacement session outranks it. pub going_away: crate::goaway::GoingAway, } @@ -135,15 +135,6 @@ impl Subscriber { self.ended.lock().clone().unwrap_or(Error::Dropped) } - /// Reject a new request once the peer has sent a GOAWAY: it told us to stop - /// opening streams on this session (existing subscriptions keep flowing). - fn check_going_away(&self) -> Result<(), Error> { - if self.going_away.is_set() { - return Err(Error::GoingAway); - } - Ok(()) - } - /// What pulling content across this session's link costs, added to the route cost /// of every announcement received over it. /// @@ -1166,8 +1157,8 @@ impl ProbeStream { loop { match &mut self.state { ProbeState::Open => { - // After a GOAWAY the peer must not see new streams. Probe is - // best-effort; skip it rather than erroring. + // Probe is best-effort telemetry; a session that is going away + // has no use for a new estimate, so skip it rather than erroring. if self.subscriber.going_away.is_set() { return Poll::Ready(Ok(())); } @@ -1252,8 +1243,6 @@ impl AnnouncePrefix { loop { match &mut self.state { PrefixState::Open => { - // A peer that sent GOAWAY told us to stop opening streams on this session. - self.subscriber.check_going_away()?; let mut stream = ready!(Stream::poll_open( &mut self.subscriber.session, self.subscriber.version, @@ -3760,8 +3749,6 @@ impl Establish { loop { match &mut self.state { EstablishState::Open => { - // A peer that sent GOAWAY told us to stop opening streams. - self.serve.subscriber.check_going_away()?; let mut stream = ready!(Stream::poll_open( &mut self.session, self.serve.subscriber.version, @@ -4006,7 +3993,6 @@ impl TrackInfoFetch { loop { match &mut self.state { TrackInfoState::Open => { - serve.subscriber.check_going_away()?; let mut stream = ready!(Stream::poll_open(&mut self.session, serve.subscriber.version, &mut cx))?; stream.writer.buffer(&lite::ControlType::Track)?; stream.writer.buffer(&lite::Track { @@ -4468,13 +4454,6 @@ impl kio::Task for FetchServeRun { match &mut self.state { FetchRunState::Open { request } => { - // 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); - self.state = FetchRunState::Done; - return Poll::Ready(()); - } - let mut stream = match ready!(Stream::poll_open( &mut self.session, self.serve.subscriber.version, diff --git a/rs/moq-net/src/session.rs b/rs/moq-net/src/session.rs index 181ce38f60..816a63deed 100644 --- a/rs/moq-net/src/session.rs +++ b/rs/moq-net/src/session.rs @@ -293,10 +293,10 @@ impl Session { /// Observe a GOAWAY from the peer, telling us to migrate elsewhere. /// /// [`peek`](goaway::Consumer::peek) is the cheap synchronous check; - /// [`recv`](goaway::Consumer::recv) waits for one. Once a GOAWAY arrives, new - /// subscribe and announce-interest requests on this session are refused (both - /// drafts forbid opening new streams afterward); existing subscriptions keep - /// flowing until the session closes. + /// [`recv`](goaway::Consumer::recv) waits for one. Once a GOAWAY arrives, this + /// session's routes cost [`Cost::DRAIN`](crate::origin::Cost::DRAIN): requests + /// keep opening on it until a replacement session's route outranks it, and + /// existing subscriptions keep flowing until the session closes. pub fn draining(&self) -> goaway::Consumer { self.goaway.consumer() } diff --git a/rs/moq-net/tests/goaway.rs b/rs/moq-net/tests/goaway.rs index 862a9c027b..573886bf2e 100644 --- a/rs/moq-net/tests/goaway.rs +++ b/rs/moq-net/tests/goaway.rs @@ -310,20 +310,16 @@ async fn duplicate_goaway_keeps_first_payload_moq_lite_04() { .expect("test timed out (likely a mock deadlock)"); } -/// After a received GOAWAY, new subscriptions are rejected with `GoingAway` -/// while an existing subscription keeps delivering groups. -#[moq_net_sim::test] -async fn goaway_gates_new_subscribes_moq_lite_04() { +/// After a received GOAWAY, a new subscription still opens on the draining +/// session, which is the only route, while an existing one keeps delivering. +async fn goaway_keeps_new_subscribes(version: Version) { moq_net_sim::timeout(TEST_TIMEOUT, async { - let version: Version = "moq-lite-04".parse().unwrap(); - // Server publishes a broadcast with one live track. let pub_origin = produce_origin(Hop::random()); let broadcast = pub_origin.create_broadcast("test").expect("create broadcast"); broadcast.announce(moq_net::origin::Route::default()).expect("announce"); let track = broadcast.create_track("video", None).expect("create track"); - // A second track with content ready, so the gated subscribe below would - // deliver immediately if it reached the wire. + // A second track with content ready for the subscribe opened after the GOAWAY. let audio = broadcast.create_track("audio", None).expect("create track"); let mut audio_group = audio.append_group().expect("append group"); audio_group @@ -367,22 +363,21 @@ async fn goaway_gates_new_subscribes_moq_lite_04() { client.draining().recv().await.expect("goaway"); assert!(client.draining().peek().is_some()); - // A NEW subscription must not reach the wire: the upstream open is gated - // with GoingAway. The rejection can surface three ways: at subscribe (a - // gated request), through the track (lite-04 has no TRACK_INFO, so the - // request is accepted and the copy aborts at the gated SUBSCRIBE open, - // which the origin treats as a refusal and aborts the logical track), or - // as a stall (nothing within a generous bound). Any OTHER error, and any - // delivered group, fails the test. - match bc.track("audio").unwrap().subscribe(None).await { - Err(moq_net::Error::GoingAway) => {} - Err(other) => panic!("unexpected error gating a post-GOAWAY subscribe: {other}"), - Ok(mut gated) => match moq_net_sim::timeout(Duration::from_millis(500), gated.recv_group()).await { - Err(_) | Ok(Err(moq_net::Error::GoingAway)) => {} - Ok(Err(other)) => panic!("unexpected error gating a post-GOAWAY subscribe: {other}"), - Ok(Ok(_)) => panic!("new subscribe after GOAWAY must not deliver"), - }, - } + // A NEW subscription opens on the draining session rather than failing: + // with no replacement route yet, it is still the best path. + let mut late = bc + .track("audio") + .unwrap() + .subscribe(None) + .await + .expect("subscribe after GOAWAY"); + let mut group_sub = late + .recv_group() + .await + .expect("recv_group") + .expect("track closed prematurely"); + let frame = group_sub.read_frame().await.expect("read frame").expect("frame"); + assert_eq!(&frame.payload[..], b"audio"); // The EXISTING subscription keeps flowing. let mut group = track.append_group().expect("append group"); @@ -407,6 +402,21 @@ async fn goaway_gates_new_subscribes_moq_lite_04() { .expect("test timed out (likely a mock deadlock)"); } +#[moq_net_sim::test] +async fn goaway_keeps_new_subscribes_moq_lite_04() { + goaway_keeps_new_subscribes("moq-lite-04".parse().unwrap()).await; +} + +#[moq_net_sim::test] +async fn goaway_keeps_new_subscribes_moq_lite_05() { + goaway_keeps_new_subscribes("moq-lite-05".parse().unwrap()).await; +} + +#[moq_net_sim::test] +async fn goaway_keeps_new_subscribes_moq_transport_19() { + goaway_keeps_new_subscribes("moq-transport-19".parse().unwrap()).await; +} + /// A GOAWAY costs the peer's routes at [`Cost::DRAIN`] so the origin stops /// preferring a connection that is about to close. /// diff --git a/test/drill/mutations/fetch-never-sent.patch b/test/drill/mutations/fetch-never-sent.patch index 22a5a13f4a..20d7e5b94a 100644 --- a/test/drill/mutations/fetch-never-sent.patch +++ b/test/drill/mutations/fetch-never-sent.patch @@ -10,8 +10,8 @@ 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 { - +@@ -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. @@ -19,6 +19,6 @@ index 55006c5f7..93f2f9677 100644 + 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,