diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index d3a4f8d9cb..b0c833901e 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -38,14 +38,14 @@ broadcast.announce(); - **Rank several CDNs.** Connect to each with its own `consume: origin.withPreference(Origin.Preference(rank))` (higher wins; 0 is neutral and a negative rank demotes), all on one origin. Requests use the preferred link that has a route and fail over to the next; see [multiple CDNs](/concept/moq-lite#multiple-cdns). - **Follow restarts.** `origin.request(path, { announced: true })` follows whichever publisher instance serves the path: its `active` changes to a fresh broadcast on a restart, clears on an end, and resolves again on the next start, so read it reactively and close the request when done. Instances are never stitched, so subscribe again to each new broadcast. Another route of the same epoch taking over (epochs travel only on `moq-lite-07-wip`) keeps `active` as is, and each track subscription resumes there from the first group it lacks. A refusal ends the request. Add `epoch` to pin one instance; a plain `request(path)` keeps its accepted instance, can resolve blindly, and ends once that instance stops serving. To list what is live, iterate `announced(scope)`. - **Discovery follows what you watch.** A `consume` origin asks the relay for announcements only under the prefixes its readers watch with `announced(scope)` or `request(path, { announced: true })`, stops asking about a prefix 2s after nothing at or beneath it is watched, and asks again on reconnect. A plain `request(path)` asks nothing and subscribes blind, but once it resolves through a discovered route it watches its path until closed, so it outlives the listing that found it. A relay that refuses an announce request turns that session's announced requests blind too. -- **One connection per URL, unless you opt out.** `new Connection({ url })` pools and reconnects with backoff, which the elements use. Supplying your own transport options or origin selects a private loop. Receive estimates are on demand: observe `connection.probe` reactively or subscribe while you need them; `peek()` alone opens no PROBE stream. +- **Connections share by URL and versions.** `new Connection({ url })` pools and reconnects with backoff, which the elements use. An ordered `versions` list, spelled like `moq-lite-06` as in Rust, applies to both WebTransport and WebSocket; equivalent lists share, while a different order or an omitted list gets its own pool entry. Supplying your own transport options or origin selects a private loop. Receive estimates are on demand: observe `connection.probe` reactively or subscribe while you need them; `peek()` alone opens no PROBE stream. - **GOAWAY moves the session.** The connection dials the replacement at once while the old session keeps serving its groups, up to `goaway.handover` (default 10s, or the relay's deadline when that is sooner). New requests keep opening on the old session until the replacement's route outranks it, which is a restart without an epoch: an unpinned announced request follows the replacement, while a blind or epoch-pinned request stays on its accepted instance. With an epoch, which travels only when `moq-lite-07-wip` is offered, the replacement serves the same broadcast, so open subscriptions move across without missing a group. An empty GOAWAY redials the same URL. A redirect stays on the same host unless `goaway.redirect` is `"follow"`. - **One send estimate per connection.** `Bandwidth.Allocator` divides it by track priority, max-min fair within a tier. An idle track claims nothing. Publishers reserve against it so their targets sum to the estimate. - **One track per name.** Concurrent subscriptions share one request and producer. `createTrack` on a name with a request still queued answers it. The application owns sequence numbers: a track that replaces an ended one, created or requested, throws on an append that would reuse a sequence the name already wrote, so it writes its first group with an explicit sequence. Only a new broadcast starts over. - **Timing is per track, with no default.** A track that declares a `timescale` carries a `timestamp` on every frame; one without is [untimed](/concept/moq-lite#subscriptions) and its frames carry none. Omitting it means untimed, as in Rust. Nothing stamps a frame for you: the publisher passes the timestamp, and the `Timed` type (`{ value, at }`) carries one into the `@moq/json` and `@moq/flate` producers. - **Subscriber staleness is `maxDelay`.** It counts from a group's successor, by media time and by wall clock, whichever reaches it first. Passing the old `maxAge` key throws a `TypeError` naming `maxDelay`. Publisher retention is the separate `Track.Info.maxAge`. - **Hidden paths** stay out of discovery unless the announce request opts in. See [hidden broadcasts](/concept/moq-lite#hidden-broadcasts). -- **Authorization is in band.** On moq-lite 07 (offered only when `moq-lite-07-wip` is listed in `webtransport.protocols` or `websocket.protocols`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `auth.grant` watches what the relay lets this side publish and subscribe to, and `auth.add(token)` presents another token without reconnecting. Publishing outside the grant closes the connection with `SessionCode.Unauthorized`, naming the path. On moq-lite, a subscription the grant stops covering resets with `StreamCode.Unauthorized` and the session stays up. On moq-transport, `auth.add(token)` keeps the token even without the extension, and a broadcast announced with the `Auth.Auth` it returns (`broadcast.announce({ auth })`) carries it on that announcement's `PUBLISH_NAMESPACE`; announcing again with another renews the request. An `Auth.Auth` belongs to the session that added it, so after a reconnect, add the token again and re-announce. +- **Authorization is in band.** On moq-lite 07 (opt in with `versions: ["moq-lite-07-wip", "moq-lite-06"]`, listing every fallback you still want) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `auth.grant` watches what the relay lets this side publish and subscribe to, and `auth.add(token)` presents another token without reconnecting. Publishing outside the grant closes the connection with `SessionCode.Unauthorized`, naming the path. On moq-lite, a subscription the grant stops covering resets with `StreamCode.Unauthorized` and the session stays up. On moq-transport, `auth.add(token)` keeps the token even without the extension, and a broadcast announced with the `Auth.Auth` it returns (`broadcast.announce({ auth })`) carries it on that announcement's `PUBLISH_NAMESPACE`; announcing again with another renews the request. An `Auth.Auth` belongs to the session that added it, so after a reconnect, add the token again and re-announce. - **A graceful close waits.** `await connection.close()` withdraws announcements and drains for up to one second. Application delivery requires explicit `moq-lite-07-wip`; older Lite versions wait only for transport ACKs. IETF media streams are not drained. `abort()` ends immediately. Examples: diff --git a/js/net/MIGRATE.md b/js/net/MIGRATE.md index 3c2bc0e0b6..5b81d137fa 100644 --- a/js/net/MIGRATE.md +++ b/js/net/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `webtransport.protocols` is `versions` (#5979) + +Move the list to the top-level `versions` of `connect` or `Connection`, spelled as version names: `moq-transport-22`, not the ALPN `moqt-22`; lite names are unchanged. +It now applies to the WebSocket fallback too, so list every fallback you still want, such as `versions: ["moq-lite-07-wip", "moq-lite-06"]`. +Passing `webtransport.protocols` or `websocket.protocols` throws a `TypeError`; move either to `versions`. + ### Track demand reports the wanted group (#5930) `Track.Demand.used` and `unused()` are gone. diff --git a/js/net/src/connection/connect.test.ts b/js/net/src/connection/connect.test.ts index ed99c92415..d69e39883f 100644 --- a/js/net/src/connection/connect.test.ts +++ b/js/net/src/connection/connect.test.ts @@ -1,7 +1,9 @@ import { expect, jest, test } from "bun:test"; import { SessionCode, SessionError } from "../error.ts"; -import { ALPN_05, ALPN_06, ALPN_07_WIP } from "../lite/version.ts"; +import * as Ietf from "../ietf/index.ts"; +import { ALPN, ALPN_05, ALPN_06, ALPN_07_WIP } from "../lite/version.ts"; import { createMockTransportPair } from "../mock.ts"; +import { accept } from "./accept.ts"; import { type ConnectProps, connect as connectSession } from "./connect.ts"; const url = new URL("https://example.com/test"); @@ -84,7 +86,7 @@ test("WebTransport offers lite-06 first by default", async () => { }); test("WebTransport negotiates lite-07-wip only when explicitly offered", async () => { - const { protocols, version } = await offered(ALPN_07_WIP, { webtransport: { protocols: [ALPN_07_WIP] } }); + const { protocols, version } = await offered(ALPN_07_WIP, { versions: ["moq-lite-07-wip"] }); expect(protocols).toEqual(["moq-lite-07-wip"]); expect(version).toBe("moq-lite-07-wip"); }); @@ -293,6 +295,84 @@ test("SETUP preserves a delayed session close code", async () => { } }); +test("versions map to WebTransport ALPNs in order", async () => { + const versions = ["moq-lite-07-wip", "moq-lite-06", "moq-transport-22", "moq-lite-02", "moq-transport-14"]; + expect((await offered(ALPN_07_WIP, { versions })).protocols).toEqual([ + ALPN_07_WIP, + ALPN_06, + Ietf.ALPN.DRAFT_22, + ALPN, + ]); +}); + +test("versions reject empty, duplicate, unsupported and ALPN spellings", async () => { + for (const versions of [[], ["moq-lite-06", "moq-lite-06"], ["moq-lite-99"], ["toString"], [Ietf.ALPN.DRAFT_22]]) { + expect(await connect(url, { versions }).catch((err: unknown) => err)).toBeInstanceOf(TypeError); + } +}); + +test("the removed per-transport protocols fail loudly", async () => { + for (const props of [{ webtransport: { protocols: [ALPN_06] } }, { websocket: { protocols: [ALPN_06] } }]) { + await expect(connect(url, props as Omit)).rejects.toThrow("protocols was removed"); + } +}); + +test("a peer cannot select outside the explicit versions", async () => { + const pair = createMockTransportPair(ALPN_05); + const restore = stubWebTransport(pair.client); + try { + await expect(connect(url, { versions: ["moq-lite-06"], websocket: { enabled: false } })).rejects.toThrow( + `server selected an unoffered protocol: ${ALPN_05}`, + ); + } finally { + restore(); + } +}); + +test("an IETF version name negotiates its ALPN", async () => { + const pair = createMockTransportPair(Ietf.ALPN.DRAFT_22); + const restore = stubWebTransport(pair.client); + try { + const [client, server] = await Promise.all([ + connect(url, { versions: ["moq-transport-22"], websocket: { enabled: false } }), + accept({ url, transport: pair.server }), + ]); + expect(client.version).toBe("moq-transport-22"); + client.abort(); + server.abort(); + } finally { + restore(); + } +}); + +test("SETUP offers only the named versions to a peer without an ALPN", async () => { + const pair = createMockTransportPair(""); + const restore = stubWebTransport(pair.client); + try { + const [client, server] = await Promise.all([ + connect(url, { versions: ["moq-lite-06", "moq-transport-14"], websocket: { enabled: false } }), + accept({ url, transport: pair.server }), + ]); + expect(client.version).toBe("moq-transport-14"); + client.abort(); + server.abort(); + } finally { + restore(); + } +}); + +test("a peer without an ALPN is refused when no named version negotiates in SETUP", async () => { + const pair = createMockTransportPair(""); + const restore = stubWebTransport(pair.client); + try { + await expect(connect(url, { versions: ["moq-lite-06"], websocket: { enabled: false } })).rejects.toThrow( + "server negotiated no protocol", + ); + } finally { + restore(); + } +}); + // The WebSocket subprotocols one connect() attempt asks a server for. async function websocketOffer(props: Omit): Promise { const original = globalThis.WebTransport; @@ -322,26 +402,12 @@ async function websocketOffer(props: Omit): Promise { +test("the WebSocket fallback offers the same versions", async () => { const defaults = await websocketOffer({}); expect(defaults.some((protocol) => protocol.endsWith(ALPN_06))).toBe(true); expect(defaults.some((protocol) => protocol.endsWith(ALPN_07_WIP))).toBe(false); - const asked = await websocketOffer({ websocket: { protocols: [ALPN_07_WIP, ALPN_06] } }); + const asked = await websocketOffer({ versions: ["moq-lite-07-wip", "moq-lite-06"] }); expect(asked[0]?.endsWith(ALPN_07_WIP)).toBe(true); expect(asked.some((protocol) => protocol.endsWith(ALPN_05))).toBe(false); }); - -test("the WebSocket fallback refuses an unsupported protocol", async () => { - const original = globalThis.WebTransport; - globalThis.WebTransport = undefined as unknown as typeof WebTransport; - try { - for (const protocol of ["moq-lite-99", "toString"]) { - await expect(connect(url, { websocket: { protocols: [protocol] } })).rejects.toThrow( - "unsupported WebSocket protocol", - ); - } - } finally { - globalThis.WebTransport = original; - } -}); diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index b3924d532b..afb9eca3a8 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -33,13 +33,6 @@ export interface WebSocketProps { * If WebSocket won the previous race for a given URL, this is 0. */ delay?: Milli; - - /** - * The protocols to offer, most preferred first, as `webtransport.protocols` does for - * WebTransport. Defaults to every supported protocol except the work-in-progress - * `moq-lite-07-wip`; an unsupported one throws. - */ - protocols?: string[]; } /** @@ -54,8 +47,11 @@ export interface CertificateHash { value: BufferSource | string; } -/** WebTransport options extended with friendlier certificate pinning (hex hashes or a raw certificate). */ -export interface WebTransportProps extends Omit { +/** + * WebTransport options extended with friendlier certificate pinning (hex hashes or a raw certificate). + * The subprotocols come from {@link ConnectProps.versions}, shared with the WebSocket fallback. + */ +export interface WebTransportProps extends Omit { /** * Pin the server to one or more certificate hashes. Each `value` may be raw * bytes or a hex string; the algorithm defaults to `sha-256`. @@ -70,11 +66,100 @@ export interface WebTransportProps extends Omit = { + "moq-lite-07-wip": { alpn: Lite.ALPN_07_WIP }, + "moq-lite-06": { alpn: Lite.ALPN_06 }, + "moq-lite-05": { alpn: Lite.ALPN_05 }, + "moq-lite-04": { alpn: Lite.ALPN_04 }, + "moq-lite-03": { alpn: Lite.ALPN_03 }, + "moq-lite-02": { alpn: Lite.ALPN, setup: Lite.Version.DRAFT_02 }, + "moq-lite-01": { alpn: Lite.ALPN, setup: Lite.Version.DRAFT_01 }, + "moq-transport-22": { alpn: Ietf.ALPN.DRAFT_22 }, + "moq-transport-21": { alpn: Ietf.ALPN.DRAFT_21 }, + "moq-transport-20": { alpn: Ietf.ALPN.DRAFT_20 }, + "moq-transport-19": { alpn: Ietf.ALPN.DRAFT_19 }, + "moq-transport-18": { alpn: Ietf.ALPN.DRAFT_18 }, + "moq-transport-17": { alpn: Ietf.ALPN.DRAFT_17 }, + "moq-transport-16": { alpn: Ietf.ALPN.DRAFT_16 }, + "moq-transport-15": { alpn: Ietf.ALPN.DRAFT_15 }, + "moq-transport-14": { alpn: Lite.ALPN, setup: Ietf.Version.DRAFT_14 }, +}; + +// Experimental lite-07 is offered only when named. +const DEFAULT_VERSIONS = Object.keys(VERSIONS).filter((name) => name !== "moq-lite-07-wip"); + +// moq-transport-18 and newer are pinned to qmux-01. Every other ALPN is currently +// negotiated as `qmux-00.{alpn}`, but `null` lets the polyfill advertise every QMux draft +// it knows and the server pick one. +const QMUX: Record = { + [Ietf.ALPN.DRAFT_22]: "qmux-01", + [Ietf.ALPN.DRAFT_21]: "qmux-01", + [Ietf.ALPN.DRAFT_20]: "qmux-01", + [Ietf.ALPN.DRAFT_19]: "qmux-01", + [Ietf.ALPN.DRAFT_18]: "qmux-01", +}; + +/** What one version list offers: the ALPNs in preference order, and the SETUP versions behind `moql`. */ +interface Offer { + alpns: string[]; + setup: SetupVersion[]; + /** Whether the caller named the versions, so the peer must pick one of them. */ + explicit: boolean; +} + +function offer(versions?: readonly string[]): Offer { + const names = versions ?? DEFAULT_VERSIONS; + const alpns = new Set(); + const setup: SetupVersion[] = []; + for (const name of names) { + const version = VERSIONS[name]; + if (!version) throw new TypeError(`unsupported version: ${name}`); + alpns.add(version.alpn); + if (version.setup !== undefined) setup.push(version.setup); + } + return { alpns: [...alpns], setup, explicit: versions !== undefined }; +} + +/** Validate and snapshot an ordered version list before acquiring a connection. @internal */ +export function normalizeVersions(props?: { + versions?: readonly string[]; + webtransport?: WebTransportProps; + websocket?: WebSocketProps; +}): string[] | undefined { + if (props?.webtransport && "protocols" in props.webtransport) + throw new TypeError("webtransport.protocols was removed; set versions instead"); + if (props?.websocket && "protocols" in props.websocket) + throw new TypeError("websocket.protocols was removed; set versions instead"); + const versions = props?.versions; + if (versions === undefined) return undefined; + if (versions.length === 0) throw new TypeError("versions must not be empty"); + const seen = new Set(); + for (const name of versions) { + if (!Object.hasOwn(VERSIONS, name)) throw new TypeError(`unsupported version: ${name}`); + if (seen.has(name)) throw new TypeError(`duplicate version: ${name}`); + seen.add(name); + } + return [...versions]; +} + /** Options for {@link connect}. */ export interface ConnectProps { /** The relay URL. */ url: URL; + /** + * Protocol versions to offer on every transport, most preferred first, spelled like + * `moq-lite-06`. Defaults to every supported version except the work-in-progress + * `moq-lite-07-wip`; list every fallback you still want. A peer that negotiates no + * subprotocol (such as WebTransport in Firefox before 155) is refused unless the list + * includes `moq-lite-02`, `moq-lite-01` or `moq-transport-14`, which negotiate in SETUP. + */ + versions?: readonly string[]; + /** WebTransport options. */ webtransport?: WebTransportProps; @@ -149,6 +234,8 @@ const NEVER_ABORTED = new AbortController().signal; */ export async function connect({ url, ...props }: ConnectProps): Promise { if ("discovery" in props) throw new TypeError(DISCOVERY_REMOVED); + props.versions = normalizeVersions(props); + if (props.transport && props.versions) throw new TypeError("versions cannot reconfigure a supplied transport"); const signal = props.signal ?? NEVER_ABORTED; signal.throwIfAborted(); @@ -181,26 +268,30 @@ async function connectInner(url: URL, props: Omit, abort: P extensions: Extensions.offered(props.extensions), }; + const offered = offer(props.versions); + if (props.transport) { const transport = props.transport; void abort.then(() => transport.close()); - return connectTransport(url, transport, wiring); + return connectTransport(url, transport, wiring, offered.setup); } - // Resolved before racing, so a misconfigured protocol fails the call rather than one transport. - const versions = props.websocket?.enabled !== false ? websocketVersions(props.websocket?.protocols) : undefined; - // Stop transports after one connects or the caller aborts. const { promise: raced, resolve: done } = Promise.withResolvers(); const cancel = Promise.race([raced, abort]); - const webtransport = isWebTransportSupported() ? connectWebTransport(url, cancel, props.webtransport) : undefined; + const webtransport = isWebTransportSupported() + ? connectWebTransport(url, cancel, offered.alpns, props.webtransport) + : undefined; // Give QUIC a head start to connect before trying WebSocket, unless WebSocket has won in the past. // NOTE that QUIC should be faster because it involves 1/2 fewer RTTs. const headstart = !webtransport || websocketWon.has(url.toString()) ? 0 : (props.websocket?.delay ?? DEFAULT_WEBSOCKET_DELAY_MS); - const websocket = versions ? connectWebSocket(props.websocket?.url ?? url, headstart, cancel, versions) : undefined; + const websocket = + props.websocket?.enabled !== false + ? connectWebSocket(props.websocket?.url ?? url, headstart, cancel, offered.alpns) + : undefined; if (!websocket && !webtransport) { throw new Error("no transport available; WebTransport not supported and WebSocket is disabled"); @@ -232,17 +323,31 @@ async function connectInner(url: URL, props: Omit, abort: P } // The remaining setup is identical whether the transport was raced or supplied. - return await connectTransport(url, session as WebTransport, wiring); + const protocol = (session as { protocol?: string }).protocol || undefined; + if (offered.explicit && protocol && !offered.alpns.includes(protocol)) { + session.close(); + throw new Error(`server selected an unoffered protocol: ${protocol}`); + } + if (offered.explicit && !protocol && offered.setup.length === 0) { + session.close(); + throw new Error("server negotiated no protocol, and no offered version negotiates in SETUP"); + } + return await connectTransport(url, session as WebTransport, wiring, offered.setup); } -async function connectTransport(url: URL, session: WebTransport, wiring: SessionProps): Promise { +async function connectTransport( + url: URL, + session: WebTransport, + wiring: SessionProps, + setup: SetupVersion[], +): Promise { const closed = session.closed.then(fromClose, error); const terminal = closed.then((cause) => { throw cause ?? new Error("session closed during SETUP"); }); try { - return await Promise.race([terminal, negotiate(url, session, wiring)]); + return await Promise.race([terminal, negotiate(url, session, wiring, setup)]); } catch (cause) { // We refused the peer's SETUP and already closed with the code that says why. if (cause instanceof ProtocolViolation) throw cause; @@ -266,7 +371,13 @@ async function connectTransport(url: URL, session: WebTransport, wiring: Session // Negotiate the MoQ protocol over an established transport. The caller races this against // the session closing so a close code is not lost behind a failed or stalled SETUP stream. -async function negotiate(url: URL, session: WebTransport, wiring: SessionProps): Promise { +// `setup` lists the versions a `moql` or ALPN-less session offers in SETUP. +async function negotiate( + url: URL, + session: WebTransport, + wiring: SessionProps, + setup: SetupVersion[], +): Promise { // The DOM lib has no `protocol` property yet. It is "" when none was negotiated, and // undefined in a browser that predates subprotocols (Firefox before 155). const protocol: string | undefined = (session as { protocol?: string }).protocol || undefined; @@ -328,7 +439,7 @@ async function negotiate(url: URL, session: WebTransport, wiring: SessionProps): ? [Ietf.Version.DRAFT_16] : setupVersion === Ietf.Version.DRAFT_15 ? [Ietf.Version.DRAFT_15] - : [Lite.Version.DRAFT_02, Lite.Version.DRAFT_01, Ietf.Version.DRAFT_14], + : setup, parameters: params, }); await client.encode(stream.writer, setupVersion); @@ -451,6 +562,7 @@ async function resolveCertificateHashes(options?: WebTransportProps): Promise, + alpns: string[], options?: WebTransportProps, ): Promise { let finalUrl = url; @@ -461,24 +573,8 @@ async function connectWebTransport( const finalOptions: WebTransportOptions = { allowPooling: false, congestionControl: "low-latency", - protocols: [ - // Lite.ALPN_07_WIP is intentionally omitted: lite-07 is work-in-progress and - // not advertised by default (negotiate still accepts it if a server selects it). - Lite.ALPN_06, - Lite.ALPN_05, - Lite.ALPN_04, - Lite.ALPN_03, - Lite.ALPN, - Ietf.ALPN.DRAFT_22, - Ietf.ALPN.DRAFT_21, - Ietf.ALPN.DRAFT_20, - Ietf.ALPN.DRAFT_19, - Ietf.ALPN.DRAFT_18, - Ietf.ALPN.DRAFT_17, - Ietf.ALPN.DRAFT_16, - Ietf.ALPN.DRAFT_15, - ], ...webtransport, + protocols: alpns, }; // Accumulate caller-provided pins first, then append anything we fetch below, @@ -532,49 +628,12 @@ async function connectWebTransport( return quic; } -// moq-transport-18 and newer are pinned to qmux-01. Every other ALPN we -// support is currently negotiated as `qmux-00.{alpn}` on the wire, but we -// don't want to lock that in: set the value to `null` so the polyfill -// advertises every QMux draft it knows about and the server picks one. -// Insertion order is the negotiation preference on the wire. -const WEBSOCKET_VERSIONS = { - [Lite.ALPN_06]: null, - [Lite.ALPN_05]: null, - [Lite.ALPN_04]: null, - [Lite.ALPN_03]: null, - [Lite.ALPN]: null, - [Ietf.ALPN.DRAFT_22]: "qmux-01", - [Ietf.ALPN.DRAFT_21]: "qmux-01", - [Ietf.ALPN.DRAFT_20]: "qmux-01", - [Ietf.ALPN.DRAFT_19]: "qmux-01", - [Ietf.ALPN.DRAFT_18]: "qmux-01", - [Ietf.ALPN.DRAFT_17]: null, - [Ietf.ALPN.DRAFT_16]: null, - [Ietf.ALPN.DRAFT_15]: null, -} as const satisfies Record; - -// The QMux versions to offer per ALPN: `protocols` in their order, or the defaults. Lite-07 is -// work-in-progress, so it is offered only when asked for. -function websocketVersions(protocols?: string[]): Record { - if (!protocols) return WEBSOCKET_VERSIONS; - const supported: Record = { - [Lite.ALPN_07_WIP]: null, - ...WEBSOCKET_VERSIONS, - }; - const versions: Record = {}; - for (const protocol of protocols) { - if (!Object.hasOwn(supported, protocol)) throw new Error(`unsupported WebSocket protocol: ${protocol}`); - versions[protocol] = supported[protocol]; - } - return versions; -} - // TODO accept arguments to control the port/path used. async function connectWebSocket( url: URL, delay: number, cancel: Promise, - versions: Record, + alpns: string[], ): Promise { const timer = sleep(Nano.fromMilli(Milli(delay))); @@ -586,8 +645,8 @@ async function connectWebSocket( // know a wire-format version (today's moq-relay only accepts bare // `webtransport`). const quic = new Session(url, { - protocols: Object.keys(versions), - versions, + protocols: alpns, + versions: Object.fromEntries(alpns.map((alpn) => [alpn, QMUX[alpn] ?? null])), }); // Wait for the WebSocket to connect, or for the cancel promise to resolve. diff --git a/js/net/src/connection/pool.test.ts b/js/net/src/connection/pool.test.ts index 3a8a4451c7..fe1c3f20bc 100644 --- a/js/net/src/connection/pool.test.ts +++ b/js/net/src/connection/pool.test.ts @@ -6,6 +6,7 @@ import * as Lite from "../lite/index.ts"; import { createMockTransportPair } from "../mock.ts"; import { Producer as OriginProducer } from "../origin.ts"; import * as Path from "../path.ts"; +import { Stream } from "../stream.ts"; import * as Time from "../time.ts"; import { connect } from "./connect.ts"; import { type AcceptProps, accept as acceptSession } from "./index.ts"; @@ -624,3 +625,85 @@ test("a handle's announcements sync with the shared origin", async () => { watch.close(); handle.close(); }); + +test("shared connections distinguish ordered versions", async () => { + const dials = stubTransports(); + const versions = ["moq-lite-07-wip", "moq-lite-06", "moq-lite-05"]; + const first = new Connection({ url, versions, share: true, linger }); + const same = new Connection({ url, versions: [...versions], share: true, linger }); + const reversed = new Connection({ url, versions: [...versions].reverse(), share: true, linger }); + const defaults = new Connection({ url, linger }); + try { + await advance(); + expect(first.status.peek()).toBe("connected"); + expect(same.origin.peek()).toBe(first.origin.peek()); + expect(reversed.origin.peek()).not.toBe(first.origin.peek()); + expect(defaults.origin.peek()).not.toBe(first.origin.peek()); + expect(dials.count()).toBe(3); + } finally { + first.close(); + same.close(); + reversed.close(); + defaults.close(); + } +}); + +test("a caller cannot mutate acquired versions", async () => { + const dials = stubTransports(); + const versions = ["moq-lite-07-wip", "moq-lite-05"]; + const first = new Connection({ url, versions, linger }); + versions.reverse(); + const original = new Connection({ url, versions: ["moq-lite-07-wip", "moq-lite-05"], linger }); + const changed = new Connection({ url, versions, linger }); + try { + await advance(); + expect(first.origin.peek()).toBe(original.origin.peek()); + expect(changed.origin.peek()).not.toBe(first.origin.peek()); + expect(dials.count()).toBe(2); + } finally { + first.close(); + original.close(); + changed.close(); + } +}); + +test("GOAWAY keeps version pools distinct at the redirected URL", async () => { + const moved = new URL("https://example.com:5443/pool"); + const peers: WebTransport[] = []; + globalThis.WebTransport = function RedirectTransport(address: string | URL) { + const pair = createMockTransportPair(Lite.ALPN_06); + peers.push(pair.server); + void accept(pair.server, new URL(address)); + return pair.client; + } as unknown as typeof WebTransport; + const versions = ["moq-lite-07-wip", "moq-lite-06"]; + const first = new Connection({ url, versions, linger }); + const resident = new Connection({ url: moved, linger }); + let joined: Connection | undefined; + let defaults: Connection | undefined; + try { + await advance(); + const origin = first.origin.peek(); + // Exercise a healthy migration after the immediate-redirect backoff guard. + await advance(2_000); + const peer = peers[0]; + if (!peer) throw new Error("missing initial peer"); + const stream = await Stream.open(peer, { version: Lite.Version.DRAFT_06 }); + await stream.writer.u53(Lite.StreamId.Goaway); + await new Lite.Goaway(moved.href).encode(stream.writer, Lite.Version.DRAFT_06); + await advance(); + joined = new Connection({ url: moved, versions, linger }); + defaults = new Connection({ url: moved, linger }); + await advance(); + expect(joined.status.peek()).toBe("connected"); + expect(joined.origin.peek()).toBe(origin); + expect(defaults.origin.peek()).toBe(resident.origin.peek()); + expect(defaults.origin.peek()).not.toBe(origin); + expect(peers).toHaveLength(3); + } finally { + first.close(); + resident.close(); + joined?.close(); + defaults?.close(); + } +}); diff --git a/js/net/src/connection/pool.ts b/js/net/src/connection/pool.ts index ac37be1183..66b7dc6ec1 100644 --- a/js/net/src/connection/pool.ts +++ b/js/net/src/connection/pool.ts @@ -1,5 +1,5 @@ /** - * A reconnecting, shareable connection: one origin and one reconnect loop per relay URL. + * A reconnecting, shareable connection: one origin and reconnect loop per URL and version list. * * @module */ @@ -19,6 +19,7 @@ import { type ConnectProps as ConnectPropsType, certificateHash, connect, + normalizeVersions, type WebSocketProps as WebSocketPropsType, type WebTransportProps as WebTransportPropsType, } from "./connect.ts"; @@ -51,12 +52,15 @@ export interface ConnectionProps { linger?: Milli; /** - * Share a pooled transport keyed on the URL. By default, options specific to this - * handle use a private reconnect loop; otherwise connections share by URL. + * Share a pooled transport keyed on the URL and ordered versions. By default, options specific to this + * handle use a private reconnect loop; otherwise connections share by URL and versions. * Explicit `true` refuses options a pool cannot honor. */ share?: boolean; + /** Protocol versions to offer, most preferred first, spelled like `moq-lite-06`; connections pool by URL and this list. */ + versions?: readonly string[]; + /** WebTransport options applied to each connection attempt (not reactive). */ webtransport?: WebTransportPropsType; @@ -88,7 +92,7 @@ export interface ConnectionProps { /** * A cloneable handle on a MoQ connection: `new Connection({ url })` reconnects and, by - * default, shares one origin and one reconnect loop with every other handle on that URL. + * default, shares one origin and reconnect loop with handles using the same URL and versions. * * The shared {@link origin} is wired to both directions. Everything the relay announces * lands in it, a publish into it is announced to the relay, and a page that publishes and @@ -202,6 +206,7 @@ export class Connection { * (plus the linger window) wants them. */ constructor(props?: ConnectionProps) { + const versions = normalizeVersions(props); refuse(props); this.#signals = new Effect(); @@ -221,7 +226,7 @@ export class Connection { const href = this.#signals.computed((effect) => effect.get(this.url)?.href); if (props && (props.share === false || requiresPrivate(props))) { - this.#runPrivate(props, href); + this.#runPrivate({ ...props, versions }, href); return; } @@ -230,7 +235,7 @@ export class Connection { const key = effect.get(href); if (!key) return; - const lease = acquire(key, linger); + const lease = acquire(key, versions, linger); effect.cleanup(lease.release); effect.set(this.#origin, lease.origin, undefined); @@ -254,6 +259,7 @@ export class Connection { enabled: this.enabled, publish: props.publish ?? (owned ? origin.consume() : undefined), consume: origin, + versions: props.versions, webtransport: props.webtransport, websocket: props.websocket, delay: props.delay, @@ -400,7 +406,7 @@ function requiresPrivate(props: ConnectionProps): boolean { /** One shared connection and the handles keeping it alive. */ interface Entry { - /** The pool key: the dialed URL, which a GOAWAY redirect moves. */ + /** The pool key: the dialed URL and versions, moved together on GOAWAY. */ key: string; origin: Origin.Producer; connection: Reload; @@ -412,13 +418,20 @@ interface Entry { /** The process-wide pool backing {@link Connection}. */ const pool = new Map(); -/** Take a reference on the shared entry for `key`, creating it on first use. */ -function acquire(key: string, linger?: Milli): Entry & { release: () => void } { +// Preserve the ordered versions when a redirect moves the URL. +function poolKey(url: string, versions?: readonly string[]): string { + return JSON.stringify([url, versions]); +} + +/** Take a reference on the shared entry, creating it on first use. */ +function acquire(url: string, versions?: readonly string[], linger?: Milli): Entry & { release: () => void } { + const key = poolKey(url, versions); let entry = pool.get(key); if (!entry) { const origin = new Origin.Producer(); const connection = new Reload({ - url: new URL(key), + url: new URL(url), + versions, enabled: true, publish: origin.consume(), consume: origin, @@ -439,9 +452,11 @@ function acquire(key: string, linger?: Milli): Entry & { release: () => void } { // A live entry already at the target wins: this one leaves the pool and serves only // the handles it has until they release it. connection.redirect.subscribe((redirect) => { - if (!redirect || redirect.href === created.key) return; + if (!redirect) return; + const key = poolKey(redirect.href, versions); + if (key === created.key) return; if (pool.get(created.key) === created) pool.delete(created.key); - created.key = redirect.href; + created.key = key; if (!pool.has(created.key)) pool.set(created.key, created); }); diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index 9570ddb592..8d5bab18ca 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -7,7 +7,13 @@ import type { Consumer as OriginConsumer, Producer as OriginProducer } from "../ import * as Path from "../path.ts"; import { Follow } from "../util/observed.ts"; import { wireOf } from "../wire.ts"; -import { type ConnectProps, connect, type WebSocketProps, type WebTransportProps } from "./connect.ts"; +import { + type ConnectProps, + connect, + normalizeVersions, + type WebSocketProps, + type WebTransportProps, +} from "./connect.ts"; import type { Established } from "./established.ts"; import { DISCOVERY_REMOVED } from "./forward.ts"; import { DEFAULT_HANDOVER, type Drain, dialed, type GoawayProps, handover, pinnedTransport, target } from "./goaway.ts"; @@ -136,6 +142,9 @@ export class Reload { */ readonly bandwidth: Allocator; + /** Protocol versions to offer, most preferred first, kept across retries and redirects. */ + readonly versions?: readonly string[]; + /** WebTransport options applied to each connection attempt (not reactive). */ webtransport?: WebTransportProps; @@ -232,6 +241,7 @@ export class Reload { this.delay = props?.delay ?? {}; this.goaway = props?.goaway ?? {}; this.redirect = this.#redirect; + this.versions = normalizeVersions(props); this.webtransport = props?.webtransport; this.websocket = props?.websocket; this.publish = props?.publish; @@ -389,6 +399,7 @@ export class Reload { url: dialing, // A redirect names the relay; a fallback URL pinned for the old one does not follow. websocket: this.#redirect.peek() ? { ...this.websocket, url: undefined } : this.websocket, + versions: this.versions, webtransport: this.webtransport, publish: this.publish, consume: this.consume, diff --git a/test/drain/drain.ts b/test/drain/drain.ts index eb4f5d225a..834381e909 100644 --- a/test/drain/drain.ts +++ b/test/drain/drain.ts @@ -55,7 +55,7 @@ const HANDOVER = Milli(2000); const path = Moq.Path.from("drain"); // Publisher epochs travel on the work-in-progress moq-lite-07 only, which nothing offers by default. -const websocket = { protocols: ["moq-lite-07-wip"] }; +const versions = ["moq-lite-07-wip"]; const trackName = "seq"; function log(message: string) { @@ -116,7 +116,7 @@ broadcast.announce({ epoch: Moq.Epoch.mint() }); const publisher = new Moq.Connection({ url: new URL(`http://127.0.0.1:${bPort}/`), publish: published.consume(), - websocket, + versions, }); let lastPublished = -1; @@ -143,7 +143,7 @@ const viewer = new Moq.Connection({ url: new URL(`http://127.0.0.1:${proxyPort}/`), consume: watched, goaway: { handover: HANDOVER }, - websocket, + versions, }); // One request and one subscription, at the default zero budget, read for the whole run: the diff --git a/test/interop/clients/js-native/subscribe.ts b/test/interop/clients/js-native/subscribe.ts index f118dc7644..8132bbdfa2 100644 --- a/test/interop/clients/js-native/subscribe.ts +++ b/test/interop/clients/js-native/subscribe.ts @@ -66,11 +66,11 @@ async function video(bc: Moq.Broadcast.Consumer): Promise<[string, Catalog.Video async function run(): Promise { const origin = new Moq.Origin.Producer(); // The grant arrives over AUTH, which only the work-in-progress moq-lite-07 carries and no - // client offers by default. It is offered over WebTransport alone, so the fallback stays off. + // client offers by default. const connection = await Moq.Connection.connect({ url: new URL(url as string), consume: origin, - webtransport: { protocols: ["moq-lite-07-wip"] }, + versions: ["moq-lite-07-wip"], websocket: { enabled: false }, }); // The grant the relay sent, in the Rust client's `auth granted` shape, so the harness can diff --git a/test/interop/clients/js-native/tail.ts b/test/interop/clients/js-native/tail.ts index 10377e6eda..3339f86695 100644 --- a/test/interop/clients/js-native/tail.ts +++ b/test/interop/clients/js-native/tail.ts @@ -18,7 +18,7 @@ const origin = new Moq.Origin.Producer(); const connection = await Moq.Connection.connect({ url: new URL(url), websocket: { enabled: false }, - ...(version ? { webtransport: { protocols: [version] } } : {}), + ...(version ? { versions: [version] } : {}), ...(role === "publish" ? { publish: origin.consume() } : { consume: origin }), }); try {