diff --git a/js/net/src/connection/accept.ts b/js/net/src/connection/accept.ts index 4a6ea050ef..89eedc2812 100644 --- a/js/net/src/connection/accept.ts +++ b/js/net/src/connection/accept.ts @@ -158,7 +158,7 @@ async function acceptSetup( const encoder = new TextEncoder(); const params = new Ietf.SetupOptions(); - params.setVarint(Ietf.SetupOption.MaxRequestId, 42069n); + params.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(true)); params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); Ietf.solicitIntoSetup(params); Ietf.hiddenIntoSetup(params); @@ -166,7 +166,7 @@ async function acceptSetup( const server = new Ietf.ServerSetup({ version, parameters: params }); await server.encode(stream.writer, version); - const maxRequestId = 42069n; + const maxRequestId = client.parameters.getVarint(Ietf.SetupOption.MaxRequestId) ?? 0n; return new Ietf.Connection({ ...wiring, @@ -218,7 +218,7 @@ async function acceptNegotiated( const encoder = new TextEncoder(); const params = new Ietf.SetupOptions(); - params.setVarint(Ietf.SetupOption.MaxRequestId, 42069n); + params.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(true)); params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); Ietf.solicitIntoSetup(params); Ietf.hiddenIntoSetup(params); diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index 444e776932..980fbf40de 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -302,7 +302,7 @@ async function negotiate(url: URL, session: WebTransport, wiring: SessionProps): const encoder = new TextEncoder(); const params = new Ietf.SetupOptions(); - params.setVarint(Ietf.SetupOption.MaxRequestId, 42069n); + params.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(false)); params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); Ietf.solicitIntoSetup(params); Ietf.hiddenIntoSetup(params); diff --git a/js/net/src/ietf/adapter.test.ts b/js/net/src/ietf/adapter.test.ts index ebde36a91f..5f96a9433e 100644 --- a/js/net/src/ietf/adapter.test.ts +++ b/js/net/src/ietf/adapter.test.ts @@ -1,4 +1,5 @@ -import { expect, test } from "bun:test"; +import { expect, spyOn, test } from "bun:test"; +import { SessionCode } from "../error.ts"; import { createMockTransportPair } from "../mock.ts"; import * as Path from "../path.ts"; import { Stream } from "../stream.ts"; @@ -6,8 +7,10 @@ import { ControlStreamAdapter } from "./adapter.ts"; import { toRequestCode } from "./error.ts"; import { GoAway } from "./goaway.ts"; import { PublishNamespace, PublishNamespaceCancel, PublishNamespaceDone } from "./publish_namespace.ts"; -import { RequestError } from "./request.ts"; -import { ALPN, Version } from "./version.ts"; +import { initialMaxRequestId, MaxRequestId, RequestError } from "./request.ts"; +import { SubscribeUpdate } from "./subscribe.ts"; +import { TrackStatusRequest } from "./track.ts"; +import { ALPN, type IetfVersion, Version } from "./version.ts"; test("draft-14 TRACK_STATUS_OK cannot be routed as NAMESPACE_DONE", async () => { const pair = createMockTransportPair(ALPN.DRAFT_14); @@ -233,3 +236,240 @@ test("a second GOAWAY on the control stream closes the session", async () => { } await expect(running).rejects.toThrow("duplicate GOAWAY"); }); + +// Past the advertised maximum, or more open requests than the window, is TOO_MANY_REQUESTS. +// A parity or duplicate error is draft-14 to -16's INVALID_REQUEST_ID. +const INVALID_REQUEST_ID = 0x4; + +const WINDOW_DRAFTS = [ + [Version.DRAFT_14, ALPN.DRAFT_14], + [Version.DRAFT_15, ALPN.DRAFT_15], + [Version.DRAFT_16, ALPN.DRAFT_16], +] as const; + +/** + * Adapter admitting `window` open requests, plus the peer's control stream. + * `client` is this side, so the peer uses the other parity. + */ +async function windowed( + version: IetfVersion, + alpn: string, + window: bigint, + client: boolean, +): Promise<{ + pair: ReturnType; + adapter: ControlStreamAdapter; + peer: Stream; + running: Promise; +}> { + const pair = createMockTransportPair(alpn); + const control = await Stream.open(pair.server, { version }); + const adapter = new ControlStreamAdapter(pair.server, control, version, 100n, client, window); + const running = adapter.run(); + const peer = await Stream.accept(pair.client, version); + if (!peer) throw new Error("no control stream"); + return { pair, adapter, peer, running }; +} + +/** A new request of the peer's parity. TRACK_STATUS spends an id and holds no namespace. */ +async function trackStatus(peer: Stream, version: IetfVersion, requestId: bigint): Promise { + await peer.writer.u53(TrackStatusRequest.id); + await new TrackStatusRequest({ + requestId, + trackNamespace: Path.from("t"), + trackName: "a", + }).encode(peer.writer, version); +} + +/** A REQUEST_UPDATE spending `requestId`. */ +async function update(peer: Stream, version: IetfVersion, requestId: bigint): Promise { + await peer.writer.u53(SubscribeUpdate.id); + await new SubscribeUpdate({ requestId }).encode(peer.writer, version); +} + +/** The next MAX_REQUEST_ID on the peer's control stream, skipping anything else. */ +async function nextGrant(peer: Stream, version: IetfVersion): Promise { + for (;;) { + const type = await peer.reader.u53(); + if (type === MaxRequestId.id) return (await MaxRequestId.decode(peer.reader, version)).requestId; + const size = await peer.reader.u16(); + await peer.reader.read(size); + } +} + +/** Resolves with the run loop's error, or undefined while it is still running. */ +function settled(running: Promise): () => Promise { + const failed = running.then( + () => undefined, + (err: unknown) => err, + ); + return () => Promise.race([failed, Promise.resolve(undefined)]); +} + +test("an id at the advertised maximum closes the session", async () => { + for (const [version, alpn] of WINDOW_DRAFTS) { + for (const client of [false, true]) { + // A window of 2 admits ids below 4 from a client peer and below 5 from a server peer. + // A wrong-parity id that is also past the maximum is still too many: the maximum is checked first. + const past = client ? 5n : 4n; + const pastWrong = client ? 6n : 5n; + for (const requestId of [past, pastWrong]) { + const { pair, peer, running } = await windowed(version, alpn, 2n, client); + await trackStatus(peer, version, requestId); + await expect(running).rejects.toThrow("request id exceeds max"); + expect((await pair.client.closed).closeCode).toBe(SessionCode.TooManyRequests); + } + } + } +}); + +test("a request id with the wrong parity closes the session", async () => { + for (const [version, alpn] of WINDOW_DRAFTS) { + for (const client of [false, true]) { + const wrong = client ? 0n : 1n; + const { pair, peer, running } = await windowed(version, alpn, 2n, client); + await trackStatus(peer, version, wrong); + await expect(running).rejects.toThrow("wrong parity"); + expect((await pair.client.closed).closeCode).toBe(INVALID_REQUEST_ID); + } + } +}); + +test("a second use of an open request id closes the session", async () => { + for (const [version, alpn] of WINDOW_DRAFTS) { + const { pair, peer, running } = await windowed(version, alpn, 2n, false); + await trackStatus(peer, version, 0n); + await trackStatus(peer, version, 0n); + await expect(running).rejects.toThrow("duplicate request id"); + expect((await pair.client.closed).closeCode).toBe(INVALID_REQUEST_ID); + } +}); + +test("once half the window closes, one MAX_REQUEST_ID grants the ids it spent", async () => { + for (const [version, alpn] of WINDOW_DRAFTS) { + for (const client of [false, true]) { + const ids = client ? [1n, 3n] : [0n, 2n]; + const { adapter, peer, running } = await windowed(version, alpn, 4n, client); + const state = settled(running); + for (const requestId of ids) { + await trackStatus(peer, version, requestId); + const stream = await adapter.acceptBi(); + if (!stream) throw new Error(`no stream for ${requestId}`); + stream.close(); + } + // One close is below half the window, so the first grant covers both. + expect(await nextGrant(peer, version)).toBe(initialMaxRequestId(!client, 4n) + 4n); + expect(await state()).toBeUndefined(); + } + } +}); + +test("a request the peer closes is granted back", async () => { + const { adapter, peer, running } = await windowed(VERSION, ALPN.DRAFT_15, 1n, true); + const state = settled(running); + await announce(peer, 1n, Path.from("room")); + const stream = await adapter.acceptBi(); + if (!stream) throw new Error("no stream"); + await withdraw(peer, Path.from("room")); + expect(await nextGrant(peer, VERSION)).toBe(initialMaxRequestId(false, 1n) + 2n); + expect(await state()).toBeUndefined(); +}); + +test("a request update spends an id and retires it immediately", async () => { + const warn = spyOn(console, "warn").mockImplementation(() => undefined); + try { + for (const [version, alpn] of WINDOW_DRAFTS) { + for (const client of [false, true]) { + const updateId = client ? 1n : 0n; + const nextId = client ? 3n : 2n; + const { adapter, peer, running } = await windowed(version, alpn, 1n, client); + const state = settled(running); + await update(peer, version, updateId); + // The update's own id is granted back, so the next id is admitted. + await trackStatus(peer, version, nextId); + const stream = await adapter.acceptBi(); + if (!stream) throw new Error("update held the only slot"); + expect(await nextGrant(peer, version)).toBe(initialMaxRequestId(!client, 1n) + 2n); + expect(await state()).toBeUndefined(); + } + } + } finally { + warn.mockRestore(); + } +}); + +test("an update naming an open request does not grow the window", async () => { + const warn = spyOn(console, "warn").mockImplementation(() => undefined); + try { + for (const [version, alpn] of WINDOW_DRAFTS) { + for (const client of [false, true]) { + const heldId = client ? 1n : 0n; + const nextId = client ? 3n : 2n; + const { pair, adapter, peer, running } = await windowed(version, alpn, 1n, client); + await trackStatus(peer, version, heldId); + const held = await adapter.acceptBi(); + if (!held) throw new Error("no stream"); + await update(peer, version, heldId); + await trackStatus(peer, version, nextId); + await expect(running).rejects.toThrow("request id exceeds max"); + expect((await pair.client.closed).closeCode).toBe(SessionCode.TooManyRequests); + } + } + } finally { + warn.mockRestore(); + } +}); + +/** Repeated updates raise the maximum, but the peer still cannot hold more than the window open. */ +test("reused update ids cannot hold more than the window", async () => { + const warn = spyOn(console, "warn").mockImplementation(() => undefined); + try { + for (const [version, alpn] of WINDOW_DRAFTS) { + const { pair, adapter, peer, running } = await windowed(version, alpn, 2n, false); + await trackStatus(peer, version, 0n); + if (!(await adapter.acceptBi())) throw new Error("no stream"); + for (let i = 0; i < 8; i++) await update(peer, version, 2n); + await trackStatus(peer, version, 4n); + if (!(await adapter.acceptBi())) throw new Error("no stream"); + await trackStatus(peer, version, 6n); + await expect(running).rejects.toThrow("too many open requests"); + expect((await pair.client.closed).closeCode).toBe(SessionCode.TooManyRequests); + } + } finally { + warn.mockRestore(); + } +}); + +/** Three windows of requests, one at a time, only complete when grants keep raising the maximum. */ +test("more requests than the setup window complete on one session", async () => { + const version = Version.DRAFT_15; + const window = 64n; + const initial = initialMaxRequestId(true, window); + const { pair, adapter, peer, running } = await windowed(version, ALPN.DRAFT_15, window, false); + const state = settled(running); + + let limit = initial; + const reading = (async () => { + try { + for (;;) limit = await nextGrant(peer, version); + } catch { + // The session closed under the reader. + } + })(); + + try { + for (let id = 0n; id < initial * 3n; id += 2n) { + await trackStatus(peer, version, id); + const stream = await adapter.acceptBi(); + if (!stream) throw new Error(`session closed at request ${id}`); + stream.close(); + } + expect(await state()).toBeUndefined(); + while (limit < initial * 3n) await new Promise((resolve) => setTimeout(resolve, 0)); + expect(limit).toBeGreaterThanOrEqual(initial * 3n); + } finally { + pair.server.close(); + await running.catch(() => undefined); + await reading; + } +}); diff --git a/js/net/src/ietf/adapter.ts b/js/net/src/ietf/adapter.ts index fa9671cfca..f30e529d24 100644 --- a/js/net/src/ietf/adapter.ts +++ b/js/net/src/ietf/adapter.ts @@ -1,13 +1,28 @@ import { Once } from "@moq/signals"; import { Mutex } from "async-mutex"; import type { Drain } from "../connection/goaway.ts"; -import { error, ProtocolViolation } from "../error.ts"; +import { error, ProtocolViolation, SessionCode } from "../error.ts"; import { Reader, Stream, type Writer } from "../stream.ts"; import * as Varint from "../varint.ts"; import { GoAway } from "./goaway.ts"; import * as Namespace from "./namespace.ts"; +import { initialMaxRequestId, MaxRequestId, REQUEST_WINDOW } from "./request.ts"; import { type IetfVersion, Version } from "./version.ts"; +/** Draft-14 to -16: the request ID is not valid for this peer (wrong parity or a duplicate). */ +const INVALID_REQUEST_ID = 0x4; + +/** The peer used a request ID the draft closes the session for. */ +export class RequestWindowError extends Error { + readonly code: number; + + constructor(code: number, message: string) { + super(message); + this.name = "RequestWindowError"; + this.code = code; + } +} + /** * Interface for opening outgoing bidi streams and allocating request IDs. * Implemented by both ControlStreamAdapter (v14-v16) and NativeSession (v17). @@ -128,6 +143,16 @@ export class ControlStreamAdapter implements Session { #maxRequestId: bigint; #maxRequestIdResolves: (() => void)[] = []; + // The peer's requests, flow controlled as moq-net does: at most `#peerWindow` open at + // once, every ID below `#peerMax`, and once half the window has closed the bound rises + // by the IDs they spent and goes out as one MAX_REQUEST_ID. + #peerWindow: bigint; + #peerMax: bigint; + #open = new Set(); + #retired = 0n; + #granting = false; + #grantDirty = false; + #closed = false; // Whether this side opened the session. Only a server may name a redirect. @@ -139,6 +164,7 @@ export class ControlStreamAdapter implements Session { version: IetfVersion, maxRequestId: bigint, client: boolean, + window: bigint = REQUEST_WINDOW, ) { this.#quic = quic; this.#reader = controlStream.reader; @@ -147,6 +173,8 @@ export class ControlStreamAdapter implements Session { this.#writer.version = version; this.version = version; this.#maxRequestId = maxRequestId; + this.#peerWindow = window; + this.#peerMax = initialMaxRequestId(!client, window); this.#requestId = client ? 0n : 1n; this.#client = client; } @@ -233,6 +261,19 @@ export class ControlStreamAdapter implements Session { return Stream.open(this.#quic, { version: this.version }); } + /** + * Take a slot for a request that arrived on its own stream (draft-16 SUBSCRIBE_NAMESPACE). + * Control-stream requests are admitted while they are classified. + */ + admitRequest(requestId: bigint): void { + this.#admitHeld(requestId); + } + + /** Give back a slot taken by {@link admitRequest}. */ + releaseRequest(requestId: bigint): void { + if (this.#open.delete(requestId)) this.#retire(); + } + /** * Allocate the next request ID, blocking if flow control limit reached. */ @@ -292,6 +333,7 @@ export class ControlStreamAdapter implements Session { switch (route) { case Route.NewRequest: + this.#admitHeld(requestId); this.#newRequest(typeId, size, body, requestId); break; case Route.Response: @@ -305,6 +347,9 @@ export class ControlStreamAdapter implements Session { this.#closeStream(requestId); break; case Route.FollowUp: + // REQUEST_UPDATE consumes its own request ID. One that names an open + // request is not a new ID; the routing quest owns where the bytes go. + if (typeId === 0x02) this.#admitUpdate(requestId); this.#pushMessage(requestId, typeId, size, body); break; case Route.MaxRequestId: @@ -402,18 +447,109 @@ export class ControlStreamAdapter implements Session { * locally still gives its namespace back. */ #forget(requestId: bigint) { + const held = this.#open.delete(requestId); this.#streams.delete(requestId); this.#subscribeNamespaces.delete(requestId); const registered = this.#namespacesByRequestId.get(requestId); - if (registered === undefined) return; - this.#namespacesByRequestId.delete(requestId); - const namespaces = this.#namespaceRequests(registered.direction); - - // Only the request holding the live announcement may retract it. A refused duplicate - // carries the same name, and letting its close withdraw the name would strand the - // original: its own DONE would then resolve to nothing. - if (namespaces.get(registered.namespace) === requestId) namespaces.delete(registered.namespace); + if (registered !== undefined) { + this.#namespacesByRequestId.delete(requestId); + const namespaces = this.#namespaceRequests(registered.direction); + + // Only the request holding the live announcement may retract it. A refused duplicate + // carries the same name, and letting its close withdraw the name would strand the + // original: its own DONE would then resolve to nothing. + if (namespaces.get(registered.namespace) === requestId) namespaces.delete(registered.namespace); + } + + if (held) this.#retire(); + } + + /** Reject a request ID the peer is not allowed to use. */ + #checkPeer(requestId: bigint): void { + if (requestId >= this.#peerMax) { + // Section 9.1 also names INVALID_REQUEST_ID here, but the session registry and + // section 9.5 assign an ID past the maximum to TOO_MANY_REQUESTS. + this.#refuse(SessionCode.TooManyRequests, "request id exceeds max"); + } + // Even IDs are the client's. `#client` is this side, so the peer's IDs are the other parity. + const even = (requestId & 1n) === 0n; + if (even === this.#client) { + this.#refuse(INVALID_REQUEST_ID, "request id has the wrong parity"); + } + } + + #admitHeld(requestId: bigint): void { + this.#checkPeer(requestId); + if (this.#open.has(requestId)) { + this.#refuse(INVALID_REQUEST_ID, "duplicate request id"); + } + this.#checkLive(); + this.#open.add(requestId); + } + + /** + * An update spends a request ID and does not keep a stream, so the slot it took + * is retired immediately. Naming an open request does not spend one. + */ + #admitUpdate(requestId: bigint): void { + if (this.#open.has(requestId)) return; + this.#checkPeer(requestId); + this.#checkLive(); + this.#retire(); + } + + /** A peer reusing IDs below the bound still cannot hold more than the window open. */ + #checkLive(): void { + if (BigInt(this.#open.size) >= this.#peerWindow) { + this.#refuse(SessionCode.TooManyRequests, "too many open requests"); + } + } + + #refuse(code: number, message: string): never { + try { + this.#quic.close({ closeCode: code, reason: message }); + } catch { + // The session is already closed. + } + throw new RequestWindowError(code, message); + } + + /** + * Count a request that ended. Once half the window has, raise the bound by the IDs + * they spent and send it, so grants cost a fraction of the requests they admit. + */ + #retire(): void { + this.#retired += 1n; + if (this.#retired * 2n < this.#peerWindow) return; + this.#peerMax += this.#retired * 2n; + this.#retired = 0n; + this.#grantDirty = true; + if (this.#granting || this.#closed) return; + this.#granting = true; + void this.#flushGrant(); + } + + async #flushGrant(): Promise { + try { + while (this.#grantDirty && !this.#closed) { + this.#grantDirty = false; + const max = this.#peerMax; + await this.#writeMutex.runExclusive(async () => { + if (this.#closed) return; + await this.#writer.u53(MaxRequestId.id); + await new MaxRequestId({ requestId: max }).encode(this.#writer, this.version); + }); + } + } catch { + // The control stream closed under the grant. + } finally { + this.#granting = false; + } + if (this.#grantDirty && !this.#closed) { + this.#granting = true; + void this.#flushGrant(); + } } /** diff --git a/js/net/src/ietf/connection.test.ts b/js/net/src/ietf/connection.test.ts index c36a1600d5..306b9f7b2a 100644 --- a/js/net/src/ietf/connection.test.ts +++ b/js/net/src/ietf/connection.test.ts @@ -10,7 +10,7 @@ import * as Message from "./message.ts"; import * as Namespace from "./namespace.ts"; import { Group } from "./object.ts"; import { Parameters, SetupOption, SetupOptions } from "./parameters.ts"; -import { RequestError, RequestOk } from "./request.ts"; +import { initialMaxRequestId, MaxRequestId, RequestError, RequestOk } from "./request.ts"; import { Setup } from "./setup.ts"; import { SUBSCRIBE_TRACKS_ID, @@ -193,12 +193,12 @@ for (const [version, alpn, options, namespaces] of [ const stream = await Stream.open(pair.client, { version }); if (options === undefined) { await stream.writer.u53(SubscribeNamespace.id); - await new SubscribeNamespace({ namespace: Path.empty(), requestId: 1n }).encode(stream.writer, version); + await new SubscribeNamespace({ namespace: Path.empty(), requestId: 0n }).encode(stream.writer, version); } else { await stream.writer.u53(SubscribeNamespaceLegacy.id); await new SubscribeNamespaceLegacy({ namespace: Path.empty(), - requestId: 1n, + requestId: 0n, subscribeOptions: options, }).encode(stream.writer, version); } @@ -212,7 +212,7 @@ for (const [version, alpn, options, namespaces] of [ expect(await stream.reader.u53()).toBe(RequestError.id); const refused = await RequestError.decode(stream.reader, version); expect(refused.errorCode).toBe(0x3); - expect(refused.requestId).toBe(version === Version.DRAFT_16 ? 1n : undefined); + expect(refused.requestId).toBe(version === Version.DRAFT_16 ? 0n : undefined); expect(await stream.reader.done()).toBe(true); } } finally { @@ -371,6 +371,91 @@ test("truncated uni stream before SETUP is skipped", async () => { expect(early.length).toBe(0); }); +// A PUBLISH-only request is refused, but it still spends an id and must give it back. +for (const [name, options] of [ + ["namespaces", SubscribeOptions.NAMESPACE], + ["publish only", SubscribeOptions.PUBLISH], +] as const) { + /** Draft-16 carries SUBSCRIBE_NAMESPACE on its own stream, so the control adapter never sees the id. */ + test(`draft-16 subscribe namespace for ${name} past the request window closes the session`, async () => { + const logged = spyOn(console, "error").mockImplementation(() => undefined); + const version = Version.DRAFT_16; + const pair = createMockTransportPair(ALPN.DRAFT_16); + const control = await Stream.open(pair.server, { version }); + const connection = new Connection({ + url: new URL("https://example.com"), + quic: pair.server, + control, + maxRequestId: 100n, + version, + client: false, + }); + + try { + const stream = await Stream.open(pair.client, { version }); + await stream.writer.u53(SubscribeNamespaceLegacy.id); + await new SubscribeNamespaceLegacy({ + requestId: initialMaxRequestId(true), + namespace: Path.from("room"), + subscribeOptions: options, + }).encode(stream.writer, version); + + const info = await pair.client.closed; + expect(info.closeCode).toBe(SessionCode.TooManyRequests); + expect(logged).not.toHaveBeenCalled(); + } finally { + logged.mockRestore(); + connection.abort(); + } + }); + + /** Ending that stream has to grant another id, or the next namespace request stalls at the same ceiling. */ + test(`draft-16 subscribe namespace for ${name} grants another request id when it ends`, async () => { + const logged = spyOn(console, "error").mockImplementation(() => undefined); + const version = Version.DRAFT_16; + const pair = createMockTransportPair(ALPN.DRAFT_16); + const control = await Stream.open(pair.server, { version }); + const connection = new Connection({ + url: new URL("https://example.com"), + quic: pair.server, + control, + maxRequestId: 100n, + version, + client: false, + requestWindow: 1n, + }); + const peerControl = await Stream.accept(pair.client, version); + if (!peerControl) throw new Error("no control stream"); + + try { + const stream = await Stream.open(pair.client, { version }); + await stream.writer.u53(SubscribeNamespaceLegacy.id); + await new SubscribeNamespaceLegacy({ + requestId: 0n, + namespace: Path.from("room"), + subscribeOptions: options, + }).encode(stream.writer, version); + + if (options === SubscribeOptions.PUBLISH) { + expect(await stream.reader.u53()).toBe(RequestError.id); + expect((await RequestError.decode(stream.reader, version)).requestId).toBe(0n); + } else { + expect(await stream.reader.u53()).toBe(RequestOk.id); + expect((await RequestOk.decode(stream.reader, version)).requestId).toBe(0n); + } + stream.close(); + + expect(await peerControl.reader.u53()).toBe(MaxRequestId.id); + const grant = await MaxRequestId.decode(peerControl.reader, version); + expect(grant.requestId).toBe(initialMaxRequestId(true, 1n) + 2n); + expect(logged).not.toHaveBeenCalled(); + } finally { + logged.mockRestore(); + connection.abort(); + } + }); +} + for (const [version, alpn] of [ [Version.DRAFT_18, ALPN.DRAFT_18], [Version.DRAFT_21, ALPN.DRAFT_21], diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 9a751a9262..1956b1aaa5 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -10,7 +10,7 @@ import type * as Path from "../path.ts"; import { type Reader, Readers, type Stream } from "../stream.ts"; import { withTimeout } from "../util/timeout.ts"; import { registerWire } from "../wire.ts"; -import { ControlStreamAdapter, NativeSession, type Session } from "./adapter.ts"; +import { ControlStreamAdapter, NativeSession, RequestWindowError, type Session } from "./adapter.ts"; import * as Cluster from "./cluster.ts"; import { Fetch, FetchHeader } from "./fetch.ts"; import { GoAway } from "./goaway.ts"; @@ -107,6 +107,7 @@ export class Connection implements Established { hidden = false, cluster, early = [], + requestWindow, }: { url: URL; quic: WebTransport; @@ -132,6 +133,8 @@ export class Connection implements Established { cluster?: Cluster.Hops; /** Uni streams that arrived before the peer's SETUP, type unread (v17+). */ early?: Reader[]; + /** Requests the peer may hold open on drafts 14 to 16, the window our SETUP advertised (default `REQUEST_WINDOW`). */ + requestWindow?: bigint; }) { this.url = url; this.discovery = discovery; @@ -147,12 +150,17 @@ export class Connection implements Established { // v17+: control/setup stream only carries GoAway void this.#runGoAway(control, version); } else { - const adapter = new ControlStreamAdapter(quic, control, version, maxRequestId, client); + const adapter = new ControlStreamAdapter(quic, control, version, maxRequestId, client, requestWindow); this.#session = adapter; this.#goaway = adapter.goaway; // Start the adapter read loop (routes control messages to virtual streams) void adapter.run().catch((err: unknown) => { - if (!this.#closed) console.error("adapter error", err); + if (this.#closed) return; + if (err instanceof RequestWindowError) { + this.#close({ closeCode: err.code, reason: err.message }); + return; + } + console.error("adapter error", err); this.#close(); }); } @@ -249,6 +257,10 @@ export class Connection implements Established { if (!stream) break; void this.#runBidi(stream).catch((err: unknown) => { + if (err instanceof RequestWindowError) { + this.#close({ closeCode: err.code, reason: err.message }); + return; + } console.error("error processing bidi stream", err); stream.abort(new Error("bidi stream error")); if (err instanceof ProtocolViolation) this.#violated(err); @@ -275,16 +287,28 @@ export class Connection implements Established { } case BigInt(SubscribeNamespaceLegacy.id): { const legacy = await SubscribeNamespaceLegacy.decode(stream.reader, this.#session.version); - if (legacy.subscribeOptions === SubscribeOptions.PUBLISH) { - await this.#publisher.refuseSubscribeNamespace(legacy.requestId, stream); - break; + // Draft-16 carries this on its own stream, so the control adapter never sees the ID. + // Draft-14 and 15 already admitted it off the control stream. A refused request + // still spends its ID, so it is admitted and released like any other. + const adapter = + this.#session instanceof ControlStreamAdapter && this.#session.version === Version.DRAFT_16 + ? this.#session + : undefined; + if (adapter) adapter.admitRequest(legacy.requestId); + try { + if (legacy.subscribeOptions === SubscribeOptions.PUBLISH) { + await this.#publisher.refuseSubscribeNamespace(legacy.requestId, stream); + } else { + const msg = new SubscribeNamespace({ + requestId: legacy.requestId, + namespace: legacy.namespace, + hidden: legacy.hidden, + }); + await this.#publisher.runSubscribeNamespace(msg, stream); + } + } finally { + adapter?.releaseRequest(legacy.requestId); } - const msg = new SubscribeNamespace({ - requestId: legacy.requestId, - namespace: legacy.namespace, - hidden: legacy.hidden, - }); - await this.#publisher.runSubscribeNamespace(msg, stream); break; } case BigInt(SubscribeUpdate.id): { diff --git a/js/net/src/ietf/request.ts b/js/net/src/ietf/request.ts index 926721c969..9b9f86da84 100644 --- a/js/net/src/ietf/request.ts +++ b/js/net/src/ietf/request.ts @@ -4,6 +4,25 @@ import { Parameters } from "./parameters.ts"; import * as Properties from "./properties.ts"; import { type IetfVersion, Version } from "./version.ts"; +/** + * Requests a draft-14 to -16 peer may hold open at once. + * + * Sized like moq-net's window: twice its default 100,000 announce and 10,000 + * subscription caps. Draft-17 dropped MAX_REQUEST_ID. Grants are batched per + * half window, so only about half of it is guaranteed to a peer holding requests open. + */ +export const REQUEST_WINDOW = 220_000n; + +/** + * The SETUP MAX_REQUEST_ID that first admits `window` requests from the peer. + * + * It is exclusive (the draft's "maximum request ID plus 1"), and the peer uses + * only its own parity: even IDs are the client's, odd the server's. + */ +export function initialMaxRequestId(peerClient: boolean, window: bigint = REQUEST_WINDOW): bigint { + return window * 2n + (peerClient ? 0n : 1n); +} + export class MaxRequestId { static id = 0x15; diff --git a/quest/m1/README.md b/quest/m1/README.md index 5dacd47d8b..c5edf997db 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -88,7 +88,6 @@ blocks. The quests that gated m0 lines moved under them. - [IETF object gaps](/quest/m1/ietf-object-gaps.md) - a gapped object ID is refused loudly like a subgroup in Rust and JS, and an overflowing one closes the session in Rust - [JS IETF datagrams](/quest/m1/js-ietf-datagram.md) - `@moq/net` sends and receives datagram groups over moq-transport, like Rust - [JavaScript FETCH](/quest/m1/js-fetch.md) - browser publishers answer IETF FETCH through the JS ranges request surface -- [JS request window](/quest/m1/js-request-window.md) - @moq/net grants MAX_REQUEST_ID as requests close, so a peer never stalls after ~21k requests - [Relay session limits](/quest/m1/relay-session-limits.md) - moq-relay sets per-session request limits, tighter for clients than peers, and the bindings name a refused request - [Churn with held subscriptions](/quest/m1/session-churn-held.md) - opening and closing a request costs the same with 1 or 1,024 held subscriptions - [Export linger](/quest/m1/export-ts-linger.md) - `moq export ts --linger` fails on an export error while the broadcast is live, and lingers once it closes or is replaced diff --git a/quest/m1/js-request-window.md b/quest/m1/js-request-window.md deleted file mode 100644 index e873acdf24..0000000000 --- a/quest/m1/js-request-window.md +++ /dev/null @@ -1,20 +0,0 @@ -# [S] @moq/net grants request IDs as requests close - -## Goal - -A Rust peer can make any number of requests over one JS session on -moq-transport drafts 14 to 16. Today `js/net` advertises -`MaxRequestId = 42069` in SETUP, as a server (`js/net/src/connection/accept.ts`) -and as a client (`js/net/src/connection/connect.ts`), and never grants more, -so the peer stalls after about 21k requests. - -## Plan - -Advertise a finite window and send MAX_REQUEST_ID to grant more as requests -close. Refuse an incoming ID past the window with the draft's error. This -can start before request caps (#4820), since the JS bug exists on main; use -the drafts as the contract and align with Rust's landed window when -available. A mocked-time test drives more than 21k requests through one -session. - -Public API: none. Wire: behaviour within the drafts; no format change. diff --git a/quest/m1/js-session-parity.md b/quest/m1/js-session-parity.md index 2d29a038a6..8fd91701d3 100644 --- a/quest/m1/js-session-parity.md +++ b/quest/m1/js-session-parity.md @@ -9,8 +9,8 @@ relay behaves like a Rust one. - Per-session caps (#4820, landed): add Rust's live announce and subscription caps to JS with the same defaults (100,000 and 10,000) and - close the session with TOO_MANY_REQUESTS (0x7) past them, as Rust does. JS's own request-ID window is - [JS request window](/quest/m1/js-request-window.md). + close the session with TOO_MANY_REQUESTS (0x7) past them, as Rust does. JS already grants the + request-ID window (#4966) and refuses past it with the same code. - Pending tail (#4225, still open): once Rust readers of a received track wait for missing groups below a declared end until the tail settles, mirror the hold. JS readers still end at the end with groups missing @@ -21,7 +21,3 @@ relay behaves like a Rust one. value. Test each against the Rust behavior, with mocked time. - -## Related - -- [JS request window](/quest/m1/js-request-window.md) - JS grants request IDs as requests close, the other half of JS request limits