From 3cbaab88498d7fa1eab30add09664522666170cf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:22:23 -0700 Subject: [PATCH 1/6] claim quest/m1/js-request-window Co-authored-by: Grok 4.7 From db8a9539d8a35ff50276cb76cb60c1159eadaea3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:41:15 -0700 Subject: [PATCH 2/6] fix(net): grant MAX_REQUEST_ID as draft 14-16 requests close Closing a request raises the advertised window by one id of the peer's parity and sends MAX_REQUEST_ID, so a session no longer stalls after about 21k requests. An id past the window closes the session. Co-authored-by: Grok 4.7 --- js/net/src/connection/accept.ts | 6 +- js/net/src/connection/connect.ts | 2 +- js/net/src/ietf/adapter.test.ts | 247 ++++++++++++++++++++++++++++- js/net/src/ietf/adapter.ts | 139 +++++++++++++++- js/net/src/ietf/connection.test.ts | 77 ++++++++- js/net/src/ietf/connection.ts | 26 ++- js/net/src/ietf/request.ts | 10 ++ quest/m1/README.md | 1 - quest/m1/js-request-window.md | 20 --- 9 files changed, 487 insertions(+), 41 deletions(-) delete mode 100644 quest/m1/js-request-window.md diff --git a/js/net/src/connection/accept.ts b/js/net/src/connection/accept.ts index 4a6ea050ef..00ee632b10 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.REQUEST_LIMIT); 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 = Ietf.REQUEST_LIMIT; 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.REQUEST_LIMIT); 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..9508e6ecdb 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.REQUEST_LIMIT); 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..fca1e5840f 100644 --- a/js/net/src/ietf/adapter.test.ts +++ b/js/net/src/ietf/adapter.test.ts @@ -1,4 +1,4 @@ -import { expect, test } from "bun:test"; +import { expect, spyOn, test } from "bun:test"; import { createMockTransportPair } from "../mock.ts"; import * as Path from "../path.ts"; import { Stream } from "../stream.ts"; @@ -6,8 +6,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 { MaxRequestId, REQUEST_LIMIT, 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 +235,242 @@ test("a second GOAWAY on the control stream closes the session", async () => { } await expect(running).rejects.toThrow("duplicate GOAWAY"); }); + +// Draft-14 to -16 session codes. Past the advertised maximum is TOO_MANY_REQUESTS; +// a parity or duplicate error is INVALID_REQUEST_ID. +const TOO_MANY_REQUESTS = 0x7; +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 plus the peer's control stream. `client` is this side, so the peer uses the other parity. */ +async function windowed( + version: IetfVersion, + alpn: string, + peerLimit: 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, peerLimit); + 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); +} + +/** Read grant updates until one is at least `target`. Hangs when no grant is sent, which is the bug. */ +async function granted(peer: Stream, version: IetfVersion, target: bigint): Promise { + let max = 0n; + while (max < target) { + const type = await peer.reader.u53(); + if (type !== MaxRequestId.id) { + const size = await peer.reader.u16(); + await peer.reader.read(size); + continue; + } + const msg = await MaxRequestId.decode(peer.reader, version); + if (msg.requestId > max) max = msg.requestId; + } + return max; +} + +test("an id at the advertised maximum closes the session", async () => { + for (const [version, alpn] of WINDOW_DRAFTS) { + for (const client of [false, true]) { + // Correct parity, one past the window. A wrong-parity id that is also past the + // window is still too many: the maximum is checked first. + const past = client ? 5n : 4n; + const pastWrong = client ? 4n : 5n; + for (const requestId of [past, pastWrong]) { + const { pair, peer, running } = await windowed(version, alpn, 4n, client); + await trackStatus(peer, version, requestId); + await expect(running).rejects.toThrow("request id exceeds max"); + expect((await pair.client.closed).closeCode).toBe(TOO_MANY_REQUESTS); + } + } + } +}); + +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, 4n, 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, 4n, 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("closing a request grants one more id of the peer's parity", async () => { + for (const [version, alpn] of WINDOW_DRAFTS) { + for (const client of [false, true]) { + const ids = client ? [1n, 3n, 5n] : [0n, 2n, 4n]; + const { adapter, peer, running } = await windowed(version, alpn, 4n, client); + const failed = running.then( + () => undefined, + (err: unknown) => err, + ); + 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(); + } + // Three closes, each raising the exclusive max by 2. + expect(await granted(peer, version, 10n)).toBeGreaterThanOrEqual(10n); + expect(await Promise.race([failed, Promise.resolve(undefined)])).toBeUndefined(); + } + } +}); + +test("a request update spends an id and frees 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, 2n, client); + const failed = running.then( + () => undefined, + (err: unknown) => err, + ); + await peer.writer.u53(SubscribeUpdate.id); + await new SubscribeUpdate({ requestId: updateId }).encode(peer.writer, version); + await trackStatus(peer, version, nextId); + const stream = await adapter.acceptBi(); + if (!stream) throw new Error("update held the only slot"); + stream.close(); + expect(await granted(peer, version, 4n)).toBeGreaterThanOrEqual(4n); + expect(await Promise.race([failed, Promise.resolve(undefined)])).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, 2n, client); + await trackStatus(peer, version, heldId); + const held = await adapter.acceptBi(); + if (!held) throw new Error("no stream"); + await peer.writer.u53(SubscribeUpdate.id); + await new SubscribeUpdate({ requestId: heldId }).encode(peer.writer, version); + await trackStatus(peer, version, nextId); + await expect(running).rejects.toThrow("request id exceeds max"); + expect((await pair.client.closed).closeCode).toBe(TOO_MANY_REQUESTS); + } + } + } finally { + warn.mockRestore(); + } +}); + +/** + * SETUP advertises 42069 and used to stop there. One more even id than that window + * (0, 2, ..., 42070) only completes when each close raises the maximum. + */ +test( + "more requests than the setup window complete on one session", + async () => { + const version = Version.DRAFT_15; + const { pair, adapter, peer, running } = await windowed(version, ALPN.DRAFT_15, REQUEST_LIMIT, false); + const failed = running.then( + () => undefined, + (err: unknown) => err, + ); + + let limit = 0n; + let wake: (() => void) | undefined; + const reading = (async () => { + try { + for (;;) { + if (await peer.reader.done()) return; + const type = await peer.reader.u53(); + if (type !== MaxRequestId.id) { + const size = await peer.reader.u16(); + await peer.reader.read(size); + continue; + } + const msg = await MaxRequestId.decode(peer.reader, version); + if (msg.requestId > limit) { + limit = msg.requestId; + wake?.(); + } + } + } catch { + // The session closed under the reader. + } + })(); + + try { + for (let id = 0n; id <= REQUEST_LIMIT + 1n; id += 2n) { + await trackStatus(peer, version, id); + const stream = await adapter.acceptBi(); + if (!stream) throw new Error(`session closed at request ${id}`); + stream.close(); + } + + if (limit <= REQUEST_LIMIT) { + await new Promise((resolve, reject) => { + const timer = setTimeout( + () => reject(new Error("peer was not granted a larger request window")), + 1000, + ); + wake = () => { + if (limit <= REQUEST_LIMIT) return; + clearTimeout(timer); + resolve(); + }; + if (limit > REQUEST_LIMIT) wake(); + }); + } + expect(limit).toBeGreaterThan(REQUEST_LIMIT); + expect(await Promise.race([failed, Promise.resolve(undefined)])).toBeUndefined(); + } finally { + pair.server.close(); + await running.catch(() => undefined); + await reading; + } + }, + { timeout: 60_000 }, +); diff --git a/js/net/src/ietf/adapter.ts b/js/net/src/ietf/adapter.ts index fa9671cfca..125d0bc2c7 100644 --- a/js/net/src/ietf/adapter.ts +++ b/js/net/src/ietf/adapter.ts @@ -6,8 +6,31 @@ 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 { MaxRequestId, REQUEST_LIMIT } 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; + +/** + * Draft-14 to -16: the request ID is at or past the MAX_REQUEST_ID we advertised. + * + * Section 9.1 also names INVALID_REQUEST_ID for an ID past the maximum. The session + * registry and section 9.5 assign that case to this code, so this is the one we send. + */ +const TOO_MANY_REQUESTS = 0x7; + +/** 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 +151,13 @@ export class ControlStreamAdapter implements Session { #maxRequestId: bigint; #maxRequestIdResolves: (() => void)[] = []; + // The exclusive bound advertised to the peer, and the requests still holding a slot. + // Closing one raises the bound by 2 (one ID of the peer's parity) and sends MAX_REQUEST_ID. + #peerMax: bigint; + #open = new Set(); + #granting = false; + #grantDirty = false; + #closed = false; // Whether this side opened the session. Only a server may name a redirect. @@ -139,6 +169,7 @@ export class ControlStreamAdapter implements Session { version: IetfVersion, maxRequestId: bigint, client: boolean, + peerLimit: bigint = REQUEST_LIMIT, ) { this.#quic = quic; this.#reader = controlStream.reader; @@ -147,6 +178,7 @@ export class ControlStreamAdapter implements Session { this.#writer.version = version; this.version = version; this.#maxRequestId = maxRequestId; + this.#peerMax = peerLimit; this.#requestId = client ? 0n : 1n; this.#client = client; } @@ -233,6 +265,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.#grant(); + } + /** * Allocate the next request ID, blocking if flow control limit reached. */ @@ -292,6 +337,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 +351,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 +451,92 @@ 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.#grant(); + } + + /** Reject a request ID the peer is not allowed to use. */ + #checkPeer(requestId: bigint): void { + if (requestId >= this.#peerMax) { + this.#refuse(TOO_MANY_REQUESTS, "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.#open.add(requestId); + } + + /** + * An update spends a request ID and does not keep a stream, so the slot it took + * is free again immediately. Naming an open request does not spend one. + */ + #admitUpdate(requestId: bigint): void { + if (this.#open.has(requestId)) return; + this.#checkPeer(requestId); + this.#grant(); + } + + #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); + } + + /** Raise the peer's exclusive bound by one ID and send it, coalescing a burst into one write. */ + #grant(): void { + this.#peerMax += 2n; + 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 15e8879b4a..d12891078b 100644 --- a/js/net/src/ietf/connection.test.ts +++ b/js/net/src/ietf/connection.test.ts @@ -2,13 +2,14 @@ import { expect, spyOn, test } from "bun:test"; import { exchangeSetup } from "../connection/handshake.ts"; import { SessionCode } from "../error.ts"; import { createMockTransportPair } from "../mock.ts"; +import * as Path from "../path.ts"; import { Stream, Writer } from "../stream.ts"; import { Connection } from "./connection.ts"; import { Group } from "./object.ts"; import { SetupOption, SetupOptions } from "./parameters.ts"; -import { RequestError } from "./request.ts"; +import { MaxRequestId, REQUEST_LIMIT, RequestError, RequestOk } from "./request.ts"; import { Setup } from "./setup.ts"; -import { SUBSCRIBE_TRACKS_ID } from "./subscribe_namespace.ts"; +import { SUBSCRIBE_TRACKS_ID, SubscribeNamespaceLegacy } from "./subscribe_namespace.ts"; import { ALPN, type IetfVersion, Version } from "./version.ts"; const PADDING = 0x132b3e28n; @@ -257,3 +258,75 @@ test("truncated uni stream before SETUP is skipped", async () => { const { early } = await exchangeSetup(pair.server, version, "test"); expect(early.length).toBe(0); }); + +/** Draft-16 carries SUBSCRIBE_NAMESPACE on its own stream, so the control adapter never sees the id. */ +test("draft-16 subscribe namespace 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: REQUEST_LIMIT + 1n, + namespace: Path.from("room"), + }).encode(stream.writer, version); + + const info = await pair.client.closed; + expect(info.closeCode).toBe(0x7); + 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 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, + }); + 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"), + }).encode(stream.writer, version); + + expect(await stream.reader.u53()).toBe(RequestOk.id); + const ok = await RequestOk.decode(stream.reader, version); + expect(ok.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(REQUEST_LIMIT + 2n); + expect(logged).not.toHaveBeenCalled(); + } finally { + logged.mockRestore(); + connection.abort(); + } +}); diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 9bf0084651..db50604e8c 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"; @@ -147,7 +147,12 @@ export class Connection implements Established { 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(); }); } @@ -240,6 +245,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); @@ -270,7 +279,18 @@ export class Connection implements Established { namespace: legacy.namespace, hidden: legacy.hidden, }); - await this.#publisher.runSubscribeNamespace(msg, stream); + // 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. + const adapter = + this.#session instanceof ControlStreamAdapter && this.#session.version === Version.DRAFT_16 + ? this.#session + : undefined; + if (adapter) adapter.admitRequest(legacy.requestId); + try { + await this.#publisher.runSubscribeNamespace(msg, stream); + } finally { + adapter?.releaseRequest(legacy.requestId); + } break; } case BigInt(SubscribeUpdate.id): { diff --git a/js/net/src/ietf/request.ts b/js/net/src/ietf/request.ts index 926721c969..3aea0b88a6 100644 --- a/js/net/src/ietf/request.ts +++ b/js/net/src/ietf/request.ts @@ -4,6 +4,16 @@ import { Parameters } from "./parameters.ts"; import * as Properties from "./properties.ts"; import { type IetfVersion, Version } from "./version.ts"; +/** + * Request IDs a draft-14 to -16 peer may have in flight. + * + * SETUP's MAX_REQUEST_ID is this exclusive bound (the draft's "maximum request + * ID plus 1"). Even IDs are the client's and odd IDs are the server's, so each + * peer can use about half of them. Closing a request raises the bound by 2, + * one more ID of that parity. Draft-17 dropped the field. + */ +export const REQUEST_LIMIT = 42069n; + export class MaxRequestId { static id = 0x15; diff --git a/quest/m1/README.md b/quest/m1/README.md index 98f4f6cb5a..150e882cc2 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -92,7 +92,6 @@ blocks. The quests that gated m0 lines moved under them. - [JavaScript FETCH](/quest/m1/js-fetch.md) - browser publishers answer IETF FETCH through the JS ranges request surface - [Bounded HLS playlists](/quest/m1/hls-bounded.md) - `moq-hls` renders a capped sliding window for a fresh viewer, with a stable media sequence, gap slots, and no video before a sync point - [Request update routing](/quest/m1/ietf-subscribe-update-route.md) - drafts 14 and 15 SUBSCRIBE_UPDATE and draft 16 REQUEST_UPDATE reach the request named by their second field -- [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 - [Wide varint tests](/quest/m1/varint-test-gaps.md) - wide sequences and both varint forms are tested where they differ 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. From 24a31a043860741f01da02991f9c4ce7b62f9732 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 13:51:32 -0700 Subject: [PATCH 3/6] fix(net): bound the JS request window by open requests, as moq-net does Mirror #4820's Rust window: advertise twice a 220,000-request window (plus the peer's parity), refuse TOO_MANY_REQUESTS once that many are open, and grant the spent IDs back in one MAX_REQUEST_ID once half the window has closed. A peer repeating an update ID now raises the bound without holding more than the window open. acceptSetup takes our outgoing IDs from the client's MAX_REQUEST_ID instead of our own advertisement. Co-Authored-By: Claude Opus 5.5 --- js/net/src/connection/accept.ts | 6 +- js/net/src/connection/connect.ts | 2 +- js/net/src/ietf/adapter.test.ts | 223 ++++++++++++++--------------- js/net/src/ietf/adapter.ts | 43 ++++-- js/net/src/ietf/connection.test.ts | 7 +- js/net/src/ietf/connection.ts | 5 +- js/net/src/ietf/request.ts | 20 ++- 7 files changed, 168 insertions(+), 138 deletions(-) diff --git a/js/net/src/connection/accept.ts b/js/net/src/connection/accept.ts index 00ee632b10..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, Ietf.REQUEST_LIMIT); + 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 = Ietf.REQUEST_LIMIT; + 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, Ietf.REQUEST_LIMIT); + 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 9508e6ecdb..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, Ietf.REQUEST_LIMIT); + 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 fca1e5840f..9b998658ee 100644 --- a/js/net/src/ietf/adapter.test.ts +++ b/js/net/src/ietf/adapter.test.ts @@ -6,7 +6,7 @@ import { ControlStreamAdapter } from "./adapter.ts"; import { toRequestCode } from "./error.ts"; import { GoAway } from "./goaway.ts"; import { PublishNamespace, PublishNamespaceCancel, PublishNamespaceDone } from "./publish_namespace.ts"; -import { MaxRequestId, REQUEST_LIMIT, RequestError } from "./request.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"; @@ -236,8 +236,8 @@ test("a second GOAWAY on the control stream closes the session", async () => { await expect(running).rejects.toThrow("duplicate GOAWAY"); }); -// Draft-14 to -16 session codes. Past the advertised maximum is TOO_MANY_REQUESTS; -// a parity or duplicate error is INVALID_REQUEST_ID. +// Draft-14 to -16 session codes. Past the advertised maximum, or more open requests +// than the window, is TOO_MANY_REQUESTS; a parity or duplicate error is INVALID_REQUEST_ID. const TOO_MANY_REQUESTS = 0x7; const INVALID_REQUEST_ID = 0x4; @@ -247,11 +247,14 @@ const WINDOW_DRAFTS = [ [Version.DRAFT_16, ALPN.DRAFT_16], ] as const; -/** Adapter plus the peer's control stream. `client` is this side, so the peer uses the other parity. */ +/** + * 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, - peerLimit: bigint, + window: bigint, client: boolean, ): Promise<{ pair: ReturnType; @@ -261,7 +264,7 @@ async function windowed( }> { const pair = createMockTransportPair(alpn); const control = await Stream.open(pair.server, { version }); - const adapter = new ControlStreamAdapter(pair.server, control, version, 100n, client, peerLimit); + 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"); @@ -278,31 +281,40 @@ async function trackStatus(peer: Stream, version: IetfVersion, requestId: bigint }).encode(peer.writer, version); } -/** Read grant updates until one is at least `target`. Hangs when no grant is sent, which is the bug. */ -async function granted(peer: Stream, version: IetfVersion, target: bigint): Promise { - let max = 0n; - while (max < target) { +/** 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) { - const size = await peer.reader.u16(); - await peer.reader.read(size); - continue; - } - const msg = await MaxRequestId.decode(peer.reader, version); - if (msg.requestId > max) max = msg.requestId; + if (type === MaxRequestId.id) return (await MaxRequestId.decode(peer.reader, version)).requestId; + const size = await peer.reader.u16(); + await peer.reader.read(size); } - return max; +} + +/** 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]) { - // Correct parity, one past the window. A wrong-parity id that is also past the - // window is still too many: the maximum is checked first. + // 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 ? 4n : 5n; + const pastWrong = client ? 6n : 5n; for (const requestId of [past, pastWrong]) { - const { pair, peer, running } = await windowed(version, alpn, 4n, client); + 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(TOO_MANY_REQUESTS); @@ -315,7 +327,7 @@ 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, 4n, client); + 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); @@ -325,7 +337,7 @@ test("a request id with the wrong parity closes the session", async () => { 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, 4n, false); + 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"); @@ -333,48 +345,52 @@ test("a second use of an open request id closes the session", async () => { } }); -test("closing a request grants one more id of the peer's parity", async () => { +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, 5n] : [0n, 2n, 4n]; + const ids = client ? [1n, 3n] : [0n, 2n]; const { adapter, peer, running } = await windowed(version, alpn, 4n, client); - const failed = running.then( - () => undefined, - (err: unknown) => err, - ); + 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(); } - // Three closes, each raising the exclusive max by 2. - expect(await granted(peer, version, 10n)).toBeGreaterThanOrEqual(10n); - expect(await Promise.race([failed, Promise.resolve(undefined)])).toBeUndefined(); + // 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 update spends an id and frees it immediately", async () => { +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, 2n, client); - const failed = running.then( - () => undefined, - (err: unknown) => err, - ); - await peer.writer.u53(SubscribeUpdate.id); - await new SubscribeUpdate({ requestId: updateId }).encode(peer.writer, version); + 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"); - stream.close(); - expect(await granted(peer, version, 4n)).toBeGreaterThanOrEqual(4n); - expect(await Promise.race([failed, Promise.resolve(undefined)])).toBeUndefined(); + expect(await nextGrant(peer, version)).toBe(initialMaxRequestId(!client, 1n) + 2n); + expect(await state()).toBeUndefined(); } } } finally { @@ -389,12 +405,11 @@ test("an update naming an open request does not grow the window", async () => { 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, 2n, client); + 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 peer.writer.u53(SubscribeUpdate.id); - await new SubscribeUpdate({ requestId: heldId }).encode(peer.writer, version); + 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(TOO_MANY_REQUESTS); @@ -405,72 +420,56 @@ test("an update naming an open request does not grow the window", async () => { } }); -/** - * SETUP advertises 42069 and used to stop there. One more even id than that window - * (0, 2, ..., 42070) only completes when each close raises the maximum. - */ -test( - "more requests than the setup window complete on one session", - async () => { - const version = Version.DRAFT_15; - const { pair, adapter, peer, running } = await windowed(version, ALPN.DRAFT_15, REQUEST_LIMIT, false); - const failed = running.then( - () => undefined, - (err: unknown) => err, - ); - - let limit = 0n; - let wake: (() => void) | undefined; - const reading = (async () => { - try { - for (;;) { - if (await peer.reader.done()) return; - const type = await peer.reader.u53(); - if (type !== MaxRequestId.id) { - const size = await peer.reader.u16(); - await peer.reader.read(size); - continue; - } - const msg = await MaxRequestId.decode(peer.reader, version); - if (msg.requestId > limit) { - limit = msg.requestId; - wake?.(); - } - } - } catch { - // The session closed under the reader. - } - })(); +/** 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(TOO_MANY_REQUESTS); + } + } 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 (let id = 0n; id <= REQUEST_LIMIT + 1n; id += 2n) { - await trackStatus(peer, version, id); - const stream = await adapter.acceptBi(); - if (!stream) throw new Error(`session closed at request ${id}`); - stream.close(); - } + for (;;) limit = await nextGrant(peer, version); + } catch { + // The session closed under the reader. + } + })(); - if (limit <= REQUEST_LIMIT) { - await new Promise((resolve, reject) => { - const timer = setTimeout( - () => reject(new Error("peer was not granted a larger request window")), - 1000, - ); - wake = () => { - if (limit <= REQUEST_LIMIT) return; - clearTimeout(timer); - resolve(); - }; - if (limit > REQUEST_LIMIT) wake(); - }); - } - expect(limit).toBeGreaterThan(REQUEST_LIMIT); - expect(await Promise.race([failed, Promise.resolve(undefined)])).toBeUndefined(); - } finally { - pair.server.close(); - await running.catch(() => undefined); - await reading; + 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(); } - }, - { timeout: 60_000 }, -); + 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 125d0bc2c7..cd96701803 100644 --- a/js/net/src/ietf/adapter.ts +++ b/js/net/src/ietf/adapter.ts @@ -6,7 +6,7 @@ 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 { MaxRequestId, REQUEST_LIMIT } from "./request.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). */ @@ -151,10 +151,13 @@ export class ControlStreamAdapter implements Session { #maxRequestId: bigint; #maxRequestIdResolves: (() => void)[] = []; - // The exclusive bound advertised to the peer, and the requests still holding a slot. - // Closing one raises the bound by 2 (one ID of the peer's parity) and sends MAX_REQUEST_ID. + // 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; @@ -169,7 +172,7 @@ export class ControlStreamAdapter implements Session { version: IetfVersion, maxRequestId: bigint, client: boolean, - peerLimit: bigint = REQUEST_LIMIT, + window: bigint = REQUEST_WINDOW, ) { this.#quic = quic; this.#reader = controlStream.reader; @@ -178,7 +181,8 @@ export class ControlStreamAdapter implements Session { this.#writer.version = version; this.version = version; this.#maxRequestId = maxRequestId; - this.#peerMax = peerLimit; + this.#peerWindow = window; + this.#peerMax = initialMaxRequestId(!client, window); this.#requestId = client ? 0n : 1n; this.#client = client; } @@ -275,7 +279,7 @@ export class ControlStreamAdapter implements Session { /** Give back a slot taken by {@link admitRequest}. */ releaseRequest(requestId: bigint): void { - if (this.#open.delete(requestId)) this.#grant(); + if (this.#open.delete(requestId)) this.#retire(); } /** @@ -466,7 +470,7 @@ export class ControlStreamAdapter implements Session { if (namespaces.get(registered.namespace) === requestId) namespaces.delete(registered.namespace); } - if (held) this.#grant(); + if (held) this.#retire(); } /** Reject a request ID the peer is not allowed to use. */ @@ -486,17 +490,26 @@ export class ControlStreamAdapter implements Session { 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 free again immediately. Naming an open request does not spend one. + * 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.#grant(); + 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(TOO_MANY_REQUESTS, "too many open requests"); + } } #refuse(code: number, message: string): never { @@ -508,9 +521,15 @@ export class ControlStreamAdapter implements Session { throw new RequestWindowError(code, message); } - /** Raise the peer's exclusive bound by one ID and send it, coalescing a burst into one write. */ - #grant(): void { - this.#peerMax += 2n; + /** + * 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; diff --git a/js/net/src/ietf/connection.test.ts b/js/net/src/ietf/connection.test.ts index d12891078b..64ae88aab1 100644 --- a/js/net/src/ietf/connection.test.ts +++ b/js/net/src/ietf/connection.test.ts @@ -7,7 +7,7 @@ import { Stream, Writer } from "../stream.ts"; import { Connection } from "./connection.ts"; import { Group } from "./object.ts"; import { SetupOption, SetupOptions } from "./parameters.ts"; -import { MaxRequestId, REQUEST_LIMIT, RequestError, RequestOk } from "./request.ts"; +import { initialMaxRequestId, MaxRequestId, RequestError, RequestOk } from "./request.ts"; import { Setup } from "./setup.ts"; import { SUBSCRIBE_TRACKS_ID, SubscribeNamespaceLegacy } from "./subscribe_namespace.ts"; import { ALPN, type IetfVersion, Version } from "./version.ts"; @@ -278,7 +278,7 @@ test("draft-16 subscribe namespace past the request window closes the session", const stream = await Stream.open(pair.client, { version }); await stream.writer.u53(SubscribeNamespaceLegacy.id); await new SubscribeNamespaceLegacy({ - requestId: REQUEST_LIMIT + 1n, + requestId: initialMaxRequestId(true), namespace: Path.from("room"), }).encode(stream.writer, version); @@ -304,6 +304,7 @@ test("draft-16 subscribe namespace grants another request id when it ends", asyn maxRequestId: 100n, version, client: false, + requestWindow: 1n, }); const peerControl = await Stream.accept(pair.client, version); if (!peerControl) throw new Error("no control stream"); @@ -323,7 +324,7 @@ test("draft-16 subscribe namespace grants another request id when it ends", asyn expect(await peerControl.reader.u53()).toBe(MaxRequestId.id); const grant = await MaxRequestId.decode(peerControl.reader, version); - expect(grant.requestId).toBe(REQUEST_LIMIT + 2n); + expect(grant.requestId).toBe(initialMaxRequestId(true, 1n) + 2n); expect(logged).not.toHaveBeenCalled(); } finally { logged.mockRestore(); diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index db50604e8c..72f2a72df8 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -102,6 +102,7 @@ export class Connection implements Established { hidden = false, cluster, early = [], + requestWindow, }: { url: URL; quic: WebTransport; @@ -127,6 +128,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; @@ -142,7 +145,7 @@ 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) diff --git a/js/net/src/ietf/request.ts b/js/net/src/ietf/request.ts index 3aea0b88a6..1674550bc5 100644 --- a/js/net/src/ietf/request.ts +++ b/js/net/src/ietf/request.ts @@ -5,14 +5,22 @@ import * as Properties from "./properties.ts"; import { type IetfVersion, Version } from "./version.ts"; /** - * Request IDs a draft-14 to -16 peer may have in flight. + * Requests a draft-14 to -16 peer may hold open at once. * - * SETUP's MAX_REQUEST_ID is this exclusive bound (the draft's "maximum request - * ID plus 1"). Even IDs are the client's and odd IDs are the server's, so each - * peer can use about half of them. Closing a request raises the bound by 2, - * one more ID of that parity. Draft-17 dropped the field. + * Sized like moq-net's window: twice its default 100,000 announce and 10,000 + * subscription caps. Draft-17 dropped MAX_REQUEST_ID. */ -export const REQUEST_LIMIT = 42069n; +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; From f0a6530a5aced99012364ac309bf09c2303c2113 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 23:13:18 -0700 Subject: [PATCH 4/6] quest: drop links to the finished js-request-window quest Co-Authored-By: Claude Opus 5.5 --- quest/m1/js-session-parity.md | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) 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 From 85779498513144c709f79a150f2ce7cf04eed7fd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 23:20:52 -0700 Subject: [PATCH 5/6] fix(net): admit draft-16 PUBLISH-only namespace requests before refusing them A refused SUBSCRIBE_NAMESPACE still spends its request id, so it takes a window slot and gives it back like any other, and an id past the bound or of the wrong parity closes the session. Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/connection.test.ts | 148 ++++++++++++++++------------- js/net/src/ietf/connection.ts | 23 ++--- 2 files changed, 92 insertions(+), 79 deletions(-) diff --git a/js/net/src/ietf/connection.test.ts b/js/net/src/ietf/connection.test.ts index 30a254aee2..306b9f7b2a 100644 --- a/js/net/src/ietf/connection.test.ts +++ b/js/net/src/ietf/connection.test.ts @@ -371,78 +371,90 @@ test("truncated uni stream before SETUP is skipped", async () => { expect(early.length).toBe(0); }); -/** Draft-16 carries SUBSCRIBE_NAMESPACE on its own stream, so the control adapter never sees the id. */ -test("draft-16 subscribe namespace 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, - }); +// 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"), - }).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(); - } -}); + 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); -/** Ending that stream has to grant another id, or the next namespace request stalls at the same ceiling. */ -test("draft-16 subscribe namespace 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 info = await pair.client.closed; + expect(info.closeCode).toBe(SessionCode.TooManyRequests); + expect(logged).not.toHaveBeenCalled(); + } finally { + logged.mockRestore(); + connection.abort(); + } }); - 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"), - }).encode(stream.writer, version); - - expect(await stream.reader.u53()).toBe(RequestOk.id); - const ok = await RequestOk.decode(stream.reader, version); - expect(ok.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(); - } -}); + /** 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], diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index e813ba38fd..1956b1aaa5 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -287,24 +287,25 @@ 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; - } - const msg = new SubscribeNamespace({ - requestId: legacy.requestId, - namespace: legacy.namespace, - hidden: legacy.hidden, - }); // 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. + // 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 { - await this.#publisher.runSubscribeNamespace(msg, stream); + 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); } From 5990fb3a196a2cd611dafaf954b8f652d4e7c23c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 23:57:46 -0700 Subject: [PATCH 6/6] docs(net): note that batched grants guarantee about half the request window Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/request.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/js/net/src/ietf/request.ts b/js/net/src/ietf/request.ts index 1674550bc5..9b9f86da84 100644 --- a/js/net/src/ietf/request.ts +++ b/js/net/src/ietf/request.ts @@ -8,7 +8,8 @@ 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. + * 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;