From 66e88c7424bb227ceacb4309347a7a43dbe91c5b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 4 Aug 2026 19:22:16 -0700 Subject: [PATCH 1/6] feat(js/net): share one session per relay URL Every connection to the same URL dialed its own WebTransport session, so a page showing N broadcasts from one relay opened N sessions. connect() now leases a shared session keyed by URL and options, lingering briefly after the last handle so a component torn down and rebuilt costs no handshake. Folds the pool and reconnect knobs into one ConnectProps used by both connect() and Reload, with pool, reload, and enabled all defaulting on. Co-Authored-By: Claude Opus 5 --- doc/lib/js/@moq/net.md | 12 + js/net/src/announced.ts | 11 +- js/net/src/connection/connect.test.ts | 8 +- js/net/src/connection/connect.ts | 100 ++++++- js/net/src/connection/index.ts | 3 +- js/net/src/connection/pool.test.ts | 349 ++++++++++++++++++++++++ js/net/src/connection/pool.ts | 367 ++++++++++++++++++++++++++ js/net/src/connection/reload.test.ts | 141 +++++++++- js/net/src/connection/reload.ts | 83 +++--- js/net/src/ietf/publisher.ts | 5 +- js/net/src/lite/publisher.test.ts | 36 +++ js/net/src/lite/publisher.ts | 5 +- 12 files changed, 1064 insertions(+), 56 deletions(-) create mode 100644 js/net/src/connection/pool.test.ts create mode 100644 js/net/src/connection/pool.ts diff --git a/doc/lib/js/@moq/net.md b/doc/lib/js/@moq/net.md index c840d3d64f..7f72d95747 100644 --- a/doc/lib/js/@moq/net.md +++ b/doc/lib/js/@moq/net.md @@ -64,6 +64,18 @@ See the [publishing example](https://github.com/moq-dev/moq/blob/main/js/net/exa ## Advanced Usage +### Shared sessions + +`connect()` shares one session with every other connection to the same URL and options, so a page showing a dozen broadcasts from one relay dials it once. What you get back is a reference-counted handle: `close()` releases yours, and the connection goes away once the last one does. It lingers for a couple of seconds after that, so tearing a component down and rebuilding it costs no handshake. + +```ts +const a = await Moq.Connection.connect(url); // dials +const b = await Moq.Connection.connect(url); // same session +a.close(); // b keeps working +``` + +Pass `pool: false` when the session has to be yours alone. One case needs it: a session never sees its own announcements, so publishing and consuming the same broadcast over one shared session leaves the consumer waiting forever. Sharing is also skipped automatically when it can't be done safely, with a supplied `transport` or a pinned server certificate. + ### Remote errors When a peer resets a stream it sends a numeric code, and a read or write in progress rejects with `Moq.RemoteError` carrying it: diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index b9eaf6d4b3..2f6d614b47 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -129,9 +129,10 @@ export class Consumer { } } -// Connections already warned about missing broadcast discovery, so the fallback logs at most -// once per connection instead of once per watched path. -const warnedNoDiscovery = new WeakSet(); +// Relays already warned about missing broadcast discovery, so the fallback logs at most once +// per relay instead of once per watched path. Keyed by URL rather than by session, since +// several handles can share one connection. +const warnedNoDiscovery = new Set(); /** * What to watch, for {@link Broadcast}. @@ -218,8 +219,8 @@ export class Broadcast { // Without discovery no announcement ever arrives, so waiting would hang forever. if (!conn.discovery) { - if (!warnedNoDiscovery.has(conn)) { - warnedNoDiscovery.add(conn); + if (!warnedNoDiscovery.has(conn.url.href)) { + warnedNoDiscovery.add(conn.url.href); console.warn("relay does not support broadcast discovery; consuming without waiting."); } diff --git a/js/net/src/connection/connect.test.ts b/js/net/src/connection/connect.test.ts index 525570da98..4fcd0f4ecb 100644 --- a/js/net/src/connection/connect.test.ts +++ b/js/net/src/connection/connect.test.ts @@ -1,7 +1,8 @@ -import { expect, test } from "bun:test"; +import { beforeEach, expect, test } from "bun:test"; import { ALPN_05 } from "../lite/version.ts"; import { createMockTransportPair } from "../mock.ts"; import { connect } from "./connect.ts"; +import { resetPool } from "./pool.ts"; const url = new URL("https://example.com/test"); @@ -9,6 +10,11 @@ async function settle() { await new Promise((resolve) => setTimeout(resolve, 0)); } +// Sessions are shared by URL, so a leftover one would answer the next case. +beforeEach(() => { + resetPool(); +}); + test("already-aborted signal rejects without connecting", async () => { const original = globalThis.WebTransport; let connects = 0; diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index ca7efdcb3b..e7c9cfeb43 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -1,4 +1,5 @@ import Session, { type Version as QmuxVersion } from "@moq/qmux"; +import type { Signal } from "@moq/signals"; import * as Ietf from "../ietf/index.ts"; import * as Lite from "../lite/index.ts"; import { Stream } from "../stream.ts"; @@ -6,10 +7,30 @@ import * as Hex from "../util/hex.ts"; import { isWebTransportSupported } from "./browser.ts"; import type { Established } from "./established.ts"; import { exchangeSetup } from "./handshake.ts"; +import { type PoolProps, poolKey, sessionPool } from "./pool.ts"; // Default head start for WebTransport before attempting the WebSocket fallback. const DEFAULT_WEBSOCKET_DELAY_MS = 500; +/** Exponential backoff settings for the reconnect loop; see `reload` on the connect options. */ +export type ReloadDelay = { + /** The delay in milliseconds before reconnecting (default: 1000). */ + initial: DOMHighResTimeStamp; + + /** The multiplier for the delay (default: 2). */ + multiplier: number; + + /** The maximum delay in milliseconds (default: 30000). */ + max: DOMHighResTimeStamp; + + /** + * Maximum total time in milliseconds to spend retrying before giving up (default: + * 300000, 5 minutes). Resets after each successful connection. Set to 0 for + * unlimited retries. + */ + timeout?: DOMHighResTimeStamp; +}; + /** Tuning for the WebSocket fallback used when WebTransport is unavailable or loses the connect race. */ export interface WebSocketOptions { /** Enable the WebSocket fallback. Defaults to `true`. */ @@ -53,7 +74,12 @@ export interface WebTransportProps extends Omit; + + /** Whether to connect at all (default: true); pass a {@link Signal} to disconnect and reconnect live. */ + enabled?: boolean | Signal; + + /** + * Reconnect with exponential backoff when the session drops (default: true). + * + * `false` connects once and gives up when that session ends. An object tunes the backoff. + */ + reload?: boolean | ReloadDelay; } // Relays that don't implement broadcast discovery (SUBSCRIBE_NAMESPACE), so `announced()` would @@ -103,12 +160,40 @@ const NEVER_ABORTED = new AbortController().signal; /** * Establishes a connection to a MOQ server. * + * Shared with every other connection to the same URL by default, so `close()` releases your + * handle rather than terminating the session. See {@link ConnectProps.pool}. + * * @param url - The URL of the server to connect to * @param props - Connection options * @returns A promise that resolves to a Connection instance */ export async function connect(url: URL, props?: ConnectProps): Promise { - const signal = props?.signal ?? NEVER_ABORTED; + // Before anything opens, so an already-aborted caller can't leave a session behind. + props?.signal?.throwIfAborted(); + + const discovery = props?.discovery ?? defaultDiscovery(url); + + const key = poolKey(url, props, discovery); + if (key === undefined) return await dial(url, props, discovery, props?.signal); + + const pool = sessionPool(); + + const cached = pool.get(key); + if (cached) return await cached.acquire(props?.signal); + + // The entry owns the attempt, so one caller walking away doesn't abandon the others. + const entry = pool.insert(key, typeof props?.pool === "object" ? props.pool.grace : undefined); + entry.settle(dial(url, props, discovery, entry.signal)); + return await entry.acquire(props?.signal); +} + +/** Connect without sharing, aborting the attempt when `signal` fires. */ +async function dial( + url: URL, + props: ConnectProps | undefined, + discovery: boolean, + signal: AbortSignal = NEVER_ABORTED, +): Promise { signal.throwIfAborted(); // Resolves on abort so every in-flight transport tears itself down. @@ -116,7 +201,7 @@ export async function connect(url: URL, props?: ConnectProps): Promise resolve(); signal.addEventListener("abort", onAbort, { once: true }); - const pending = connectInner(url, props, abort); + const pending = connectInner(url, props, discovery, abort); try { // A `pending` rejection propagates unless the abort beat it to the finish line. const connection = await Promise.race([pending, abort.then(() => undefined)]); @@ -130,9 +215,12 @@ export async function connect(url: URL, props?: ConnectProps): Promise): Promise { - const discovery = props?.discovery ?? defaultDiscovery(url); - +async function connectInner( + url: URL, + props: ConnectProps | undefined, + discovery: boolean, + abort: Promise, +): Promise { if (props?.transport) { const transport = props.transport; void abort.then(() => transport.close()); diff --git a/js/net/src/connection/index.ts b/js/net/src/connection/index.ts index ed0d2facd7..55db234b22 100644 --- a/js/net/src/connection/index.ts +++ b/js/net/src/connection/index.ts @@ -1,5 +1,5 @@ /** - * Connection helpers: connect to or accept a MoQ session and reconnect on failure. + * Connection helpers: connect to or accept a MoQ session, share it by URL, and reconnect on failure. * * @module */ @@ -7,6 +7,7 @@ export * from "./accept.ts"; export { isWebTransportSupported } from "./browser.ts"; export * from "./connect.ts"; export * from "./established.ts"; +export type { PoolProps } from "./pool.ts"; export * from "./reload.ts"; export type { Probe, Stats } from "./stats.ts"; export type { Transport } from "./transport.ts"; diff --git a/js/net/src/connection/pool.test.ts b/js/net/src/connection/pool.test.ts new file mode 100644 index 0000000000..a5bb400929 --- /dev/null +++ b/js/net/src/connection/pool.test.ts @@ -0,0 +1,349 @@ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { ALPN_05 } from "../lite/version.ts"; +import { createMockTransportPair, type MockTransport } from "../mock.ts"; +import * as Path from "../path.ts"; +import { connect } from "./connect.ts"; +import { resetPool } from "./pool.ts"; + +const url = new URL("https://example.com/pool"); + +async function settle() { + await new Promise((resolve) => setTimeout(resolve, 0)); +} + +// A tiny window keeps the linger tests quick without mocking timers. +const grace = 20; + +async function expired() { + await new Promise((resolve) => setTimeout(resolve, grace * 3)); +} + +const original = globalThis.WebTransport; + +beforeEach(() => { + resetPool(); +}); + +afterEach(() => { + resetPool(); + globalThis.WebTransport = original; +}); + +/** Hand out a live mock transport per dial, counting how many were opened. */ +function stubTransports(): { transports: MockTransport[]; servers: MockTransport[] } { + const transports: MockTransport[] = []; + const servers: MockTransport[] = []; + + // Biome forbids returning a value from a class constructor. + function StubWebTransport(this: unknown) { + const pair = createMockTransportPair(ALPN_05); + transports.push(pair.client); + servers.push(pair.server); + return pair.client; + } + globalThis.WebTransport = StubWebTransport as unknown as typeof WebTransport; + + return { transports, servers }; +} + +/** Hand out a transport that never finishes connecting, counting dials and closes. */ +function stubPending(): { connects: () => number; closes: () => number } { + let connects = 0; + let closes = 0; + + class PendingWebTransport { + ready = new Promise(() => {}); + closed = new Promise(() => {}); + + constructor() { + connects++; + } + + close() { + closes++; + } + } + + globalThis.WebTransport = PendingWebTransport as unknown as typeof WebTransport; + return { connects: () => connects, closes: () => closes }; +} + +test("two connections to one URL share a session", async () => { + const { transports } = stubTransports(); + + const first = await connect(url, { websocket: { enabled: false } }); + const second = await connect(url, { websocket: { enabled: false } }); + + expect(transports.length).toBe(1); + expect(second).not.toBe(first); + expect(second.url.href).toBe(url.href); + + first.close(); + second.close(); +}); + +test("the session survives until the last handle lets go", async () => { + const { transports } = stubTransports(); + + const first = await connect(url, { websocket: { enabled: false }, pool: { grace } }); + const second = await connect(url, { websocket: { enabled: false } }); + + let closed = false; + void transports[0]?.closed.then(() => { + closed = true; + }); + + first.close(); + await expired(); + expect(closed).toBe(false); + + second.close(); + await expired(); + expect(closed).toBe(true); +}); + +test("a session reacquired within the grace period is the same one", async () => { + const { transports } = stubTransports(); + + const first = await connect(url, { websocket: { enabled: false }, pool: { grace } }); + first.close(); + + const second = await connect(url, { websocket: { enabled: false } }); + expect(transports.length).toBe(1); + + let closed = false; + void transports[0]?.closed.then(() => { + closed = true; + }); + + await expired(); + expect(closed).toBe(false); + + second.close(); +}); + +test("a session dropped after the grace period is dialed again", async () => { + const { transports } = stubTransports(); + + const first = await connect(url, { websocket: { enabled: false }, pool: { grace } }); + first.close(); + await expired(); + + const second = await connect(url, { websocket: { enabled: false } }); + expect(transports.length).toBe(2); + + second.close(); +}); + +test("a session that dies is evicted, even when closed rejects", async () => { + class RejectingWebTransport { + ready = Promise.resolve(undefined); + // A real WebTransport rejects `closed` on an abnormal termination. + closed = Promise.reject(new Error("connection reset")); + protocol = ALPN_05; + + close() {} + + createBidirectionalStream() { + return new Promise(() => {}); + } + + createUnidirectionalStream() { + return new Promise(() => {}); + } + + incomingBidirectionalStreams = new ReadableStream(); + incomingUnidirectionalStreams = new ReadableStream(); + datagrams = { maxDatagramSize: 0, readable: new ReadableStream(), writable: new WritableStream() }; + } + + globalThis.WebTransport = RejectingWebTransport as unknown as typeof WebTransport; + + const first = await connect(url, { websocket: { enabled: false }, pool: { grace } }); + await settle(); + + // The dead session is gone from the pool, so the next caller dials rather than + // leasing a corpse. + const second = await connect(url, { websocket: { enabled: false }, pool: { grace } }); + expect(second).not.toBe(first); + + first.close(); + second.close(); +}); + +test("concurrent callers share one connection attempt", async () => { + const pending = stubPending(); + + const first = connect(url, { websocket: { enabled: false } }).catch(() => undefined); + const second = connect(url, { websocket: { enabled: false } }).catch(() => undefined); + + await settle(); + expect(pending.connects()).toBe(1); + + resetPool(); + await Promise.all([first, second]); +}); + +test("one caller aborting mid-connect leaves the attempt alone", async () => { + const pending = stubPending(); + const controller = new AbortController(); + + const abandoned = connect(url, { websocket: { enabled: false }, signal: controller.signal }).then( + () => undefined, + (err: unknown) => err, + ); + const waiting = connect(url, { websocket: { enabled: false } }).then( + () => undefined, + (err: unknown) => err, + ); + + await settle(); + + const reason = new Error("gave up"); + controller.abort(reason); + expect(await abandoned).toBe(reason); + + await settle(); + // Somebody is still waiting on it, so the attempt is untouched. + expect(pending.closes()).toBe(0); + + resetPool(); + await waiting; +}); + +test("the last caller aborting mid-connect drops the attempt immediately", async () => { + const pending = stubPending(); + const controller = new AbortController(); + + const abandoned = connect(url, { websocket: { enabled: false }, signal: controller.signal, pool: { grace } }).then( + () => undefined, + (err: unknown) => err, + ); + + await settle(); + controller.abort(); + await abandoned; + await settle(); + + // No linger: a half-open dial has nothing warm worth keeping. + expect(pending.closes()).toBe(1); +}); + +test("a caller arriving after an abort starts a fresh attempt", async () => { + const pending = stubPending(); + const controller = new AbortController(); + + const abandoned = connect(url, { websocket: { enabled: false }, signal: controller.signal }).then( + () => undefined, + (err: unknown) => err, + ); + + await settle(); + controller.abort(); + await abandoned; + + const next = connect(url, { websocket: { enabled: false } }).catch(() => undefined); + await settle(); + expect(pending.connects()).toBe(2); + + resetPool(); + await next; +}); + +test("pool: false gets a session of its own", async () => { + const { transports } = stubTransports(); + + const shared = await connect(url, { websocket: { enabled: false } }); + const alone = await connect(url, { websocket: { enabled: false }, pool: false }); + + expect(transports.length).toBe(2); + + // Closing the dedicated session doesn't touch the shared one. + let sharedClosed = false; + void transports[0]?.closed.then(() => { + sharedClosed = true; + }); + alone.close(); + await settle(); + expect(sharedClosed).toBe(false); + + shared.close(); +}); + +test("options that change the session are not shared", async () => { + const { transports } = stubTransports(); + + const first = await connect(url, { websocket: { enabled: false } }); + const second = await connect(url, { websocket: { enabled: false }, discovery: false }); + const third = await connect(url, { + websocket: { enabled: false }, + webtransport: { serverCertificateHashes: [{ value: "ab" }] }, + }); + + expect(transports.length).toBe(3); + expect(second.discovery).toBe(false); + + first.close(); + second.close(); + third.close(); +}); + +test("a supplied transport is never shared", async () => { + const first = createMockTransportPair(ALPN_05); + const second = createMockTransportPair(ALPN_05); + + const one = await connect(url, { transport: first.client }); + const two = await connect(url, { transport: second.client }); + expect(two).not.toBe(one); + + let closed = false; + void second.client.closed.then(() => { + closed = true; + }); + + one.close(); + await settle(); + expect(closed).toBe(false); + + two.close(); +}); + +test("releasing a handle closes what it opened, not the session", async () => { + const { transports } = stubTransports(); + + const first = await connect(url, { websocket: { enabled: false } }); + const second = await connect(url, { websocket: { enabled: false } }); + + const mine = first.consume(Path.from("mine")); + const yours = second.consume(Path.from("yours")); + const announced = first.announced(); + + first.close(); + + expect(mine.closed.peek()).not.toBeUndefined(); + expect(announced.closed.peek()).not.toBeUndefined(); + expect(yours.closed.peek()).toBeUndefined(); + + let closed = false; + void transports[0]?.closed.then(() => { + closed = true; + }); + await settle(); + expect(closed).toBe(false); + + yours.close(); + second.close(); +}); + +test("a released handle refuses to hand out more", async () => { + stubTransports(); + + const connection = await connect(url, { websocket: { enabled: false } }); + connection.close(); + + expect(() => connection.consume(Path.from("late"))).toThrow(); + expect(() => connection.announced()).toThrow(); + + // Idempotent, so a second close doesn't double-release the session. + connection.close(); +}); diff --git a/js/net/src/connection/pool.ts b/js/net/src/connection/pool.ts new file mode 100644 index 0000000000..f2c31771a9 --- /dev/null +++ b/js/net/src/connection/pool.ts @@ -0,0 +1,367 @@ +import type { Getter } from "@moq/signals"; +import * as announce from "../announced.ts"; +import type * as broadcast from "../broadcast.ts"; +import type * as Path from "../path.ts"; +import type { ConnectProps } from "./connect.ts"; +import type { Established } from "./established.ts"; +import type { Probe, Stats } from "./stats.ts"; +import type { Transport } from "./transport.ts"; + +/** How long an unreferenced session lingers before it actually closes. */ +const DEFAULT_GRACE_MS = 2000; + +/** Tuning for the shared session pool; see `pool` on the connect options. */ +export interface PoolProps { + /** + * How long an unreferenced session lingers before closing, in milliseconds (default: 2000). + * + * The window is what makes moving an element around the DOM free: the session is still + * warm when the new owner asks for it. Applied by whoever opens the session, so a later + * caller sharing it inherits the original window. + */ + grace?: DOMHighResTimeStamp; +} + +/** + * The pool key for a connection, or `undefined` when it must not be shared. + * + * Keyed on the full URL because auth tokens ride in the query string, plus everything else + * that shapes the session. Options that can't be compared safely opt out entirely rather + * than risk sharing a session that doesn't match what the caller asked for. + * + * @internal + */ +export function poolKey(url: URL, props: ConnectProps | undefined, discovery: boolean): string | undefined { + if (props?.pool === false) return undefined; + + // The caller owns this session; we can't take it over and hand it to someone else. + if (props?.transport) return undefined; + + // Certificate pins are `BufferSource | string`, which no fingerprint compares honestly. + // Sharing a session pinned to the wrong certificate is not a failure worth risking. + const webtransport = props?.webtransport; + if (webtransport?.serverCertificate !== undefined) return undefined; + if (webtransport?.serverCertificateHashes !== undefined) return undefined; + + const websocket = props?.websocket; + const fingerprint = JSON.stringify([ + discovery, + webtransport ?? null, + websocket?.enabled ?? null, + websocket?.url?.href ?? null, + websocket?.delay ?? null, + ]); + + // NUL can't appear in a URL, so the two halves can't run together. + return `${url.href}\0${fingerprint}`; +} + +/** + * One shared session, plus the handles that keep it alive. + * + * @internal + */ +export class Entry { + /** Aborts the shared connection attempt once nobody is waiting on it. */ + readonly signal: AbortSignal; + + /** True once the session died, so the pool stops handing it out. */ + dead = false; + + #pool: Pool; + #key: string; + #grace: DOMHighResTimeStamp; + #controller = new AbortController(); + + // The dial, then the session it produced. `session` is what makes an entry reusable + // without awaiting, and what tells a release whether there is anything warm to keep. + #pending?: Promise; + #session?: Established; + + #refs = 0; + #timer?: ReturnType; + + constructor(pool: Pool, key: string, grace: DOMHighResTimeStamp) { + this.#pool = pool; + this.#key = key; + this.#grace = grace; + this.signal = this.#controller.signal; + } + + /** Hand the entry the connection attempt made with its {@link signal}. */ + settle(pending: Promise): void { + this.#pending = pending; + + void pending.then( + (session) => { + this.#session = session; + + // Read `closed` once: it derives a fresh promise per access, and it rejects on an + // abnormal close, so a one-argument `then` would both leak an unhandled rejection + // and leave the dead session in the map for the next caller to lease. + const died = () => this.#died(); + void session.closed.then(died, died); + + // The last holder left while this was still in flight, so nothing wants it now. + if (this.#refs === 0) session.close(); + }, + () => this.#evict(), + ); + } + + /** + * Take a handle on the session, waiting if the attempt is still in flight. + * + * `signal` aborts this caller's wait only. The shared attempt survives as long as anyone + * else is still waiting on it. + */ + async acquire(signal?: AbortSignal): Promise { + // Claim the reference before awaiting, so a release in between doesn't tear down the + // entry underneath us. + this.#refs += 1; + this.#cancelGrace(); + + try { + signal?.throwIfAborted(); + + const session = this.#session ?? (await abortable(this.#pending, signal)); + if (this.dead) throw new Error("connection closed"); + + return new Lease(this, session); + } catch (err) { + this.release(); + throw err; + } + } + + /** Drop a handle, closing the session once the grace period expires with none left. */ + release(): void { + this.#refs -= 1; + // Only the handle that takes the count to zero decides what happens next. + if (this.#refs !== 0) return; + + if (this.dead) { + this.#evict(); + return; + } + + const session = this.#session; + if (!session) { + // Nothing warm to preserve yet, so don't hold a half-open dial. Evict first: the + // aborted attempt is doomed, and a caller arriving in this same turn has to start a + // fresh one rather than join it. + this.#evict(); + this.#controller.abort(); + return; + } + + this.#timer = setTimeout(() => { + this.#evict(); + session.close(); + }, this.#grace); + + // Don't hold a Node process open for a session nobody is using. + (this.#timer as { unref?: () => void }).unref?.(); + } + + /** Close the session and drop the entry, whoever still holds a handle. */ + close(): void { + // Dead, so a handle released afterwards doesn't schedule a second teardown. + this.dead = true; + this.#cancelGrace(); + this.#evict(); + this.#controller.abort(); + this.#session?.close(); + } + + #died(): void { + this.dead = true; + this.#cancelGrace(); + this.#evict(); + } + + #cancelGrace(): void { + if (this.#timer === undefined) return; + clearTimeout(this.#timer); + this.#timer = undefined; + } + + #evict(): void { + this.#pool.evict(this.#key, this); + } +} + +/** Reject as soon as `signal` aborts, without disturbing `pending`. */ +async function abortable(pending: Promise | undefined, signal?: AbortSignal): Promise { + if (!pending) throw new Error("connection attempt never started"); + if (!signal) return await pending; + + const { promise: aborted, reject } = Promise.withResolvers(); + const onAbort = () => reject(signal.reason); + signal.addEventListener("abort", onAbort, { once: true }); + + try { + return await Promise.race([pending, aborted]); + } finally { + signal.removeEventListener("abort", onAbort); + } +} + +/** + * A reference-counted handle on a shared session. + * + * Everything is the underlying session's, except {@link close}, which releases this handle + * rather than terminating the session: the session goes away once every holder has let go. + * The same shape as {@link broadcast.Consumer.clone}. + */ +class Lease implements Established { + readonly url: URL; + readonly version: string; + readonly transport: Transport; + readonly probe: Getter; + readonly discovery: boolean; + + #entry: Entry; + #session: Established; + #closed = false; + + // What this handle handed out, so releasing it doesn't leave subscriptions running on a + // session that outlives it. + #announced = new Set(); + #consumed = new Set(); + #broadcasts = new Set(); + #published = new Set(); + + constructor(entry: Entry, session: Established) { + this.#entry = entry; + this.#session = session; + + this.url = session.url; + this.version = session.version; + this.transport = session.transport; + this.probe = session.probe; + this.discovery = session.discovery; + } + + announced(prefix?: Path.Valid): announce.Consumer { + this.#check(); + + const consumer = this.#session.announced(prefix); + this.#announced.add(consumer); + void consumer.closed.then(() => this.#announced.delete(consumer)); + return consumer; + } + + publish(path: Path.Valid, producer: broadcast.Producer): void { + this.#check(); + + this.#session.publish(path, producer); + this.#published.add(producer); + void producer.closed.then(() => this.#published.delete(producer)); + } + + consume(path: Path.Valid): broadcast.Consumer { + this.#check(); + + const consumer = this.#session.consume(path); + this.#consumed.add(consumer); + void consumer.closed.then(() => this.#consumed.delete(consumer)); + return consumer; + } + + announcedBroadcast(path: Path.Valid): announce.Broadcast { + this.#check(); + + // Built against this handle, not the session, so it consumes through us and goes away + // with us. + const watch = new announce.Broadcast({ connection: this, path }); + this.#broadcasts.add(watch); + return watch; + } + + async stats(): Promise { + return await this.#session.stats(); + } + + get closed(): Promise { + return this.#session.closed; + } + + close(): void { + if (this.#closed) return; + this.#closed = true; + + for (const watch of this.#broadcasts) watch.close(); + for (const consumer of this.#consumed) consumer.close(); + for (const consumer of this.#announced) consumer.close(); + for (const producer of this.#published) producer.close(); + + this.#broadcasts.clear(); + this.#consumed.clear(); + this.#announced.clear(); + this.#published.clear(); + + this.#entry.release(); + } + + #check(): void { + if (this.#closed) throw new Error("connection released"); + } +} + +/** + * Sessions shared by URL, so N components watching one relay dial it once. + * + * @internal + */ +export class Pool { + #entries = new Map(); + + /** The live entry for `key`, or `undefined` when nothing usable is cached. */ + get(key: string): Entry | undefined { + const entry = this.#entries.get(key); + if (entry && !entry.dead) return entry; + return undefined; + } + + /** + * Cache a new entry for `key`, to be handed the connection attempt made with its + * {@link Entry.signal}. Call on a {@link get} miss. + */ + insert(key: string, grace: DOMHighResTimeStamp = DEFAULT_GRACE_MS): Entry { + const entry = new Entry(this, key, grace); + this.#entries.set(key, entry); + return entry; + } + + /** Drop `entry` from the cache, unless a newer one already replaced it. */ + evict(key: string, entry: Entry): void { + if (this.#entries.get(key) === entry) this.#entries.delete(key); + } + + /** Close every session, whoever still holds a handle. */ + close(): void { + const entries = [...this.#entries.values()]; + this.#entries.clear(); + for (const entry of entries) entry.close(); + } +} + +/** The process-wide pool backing the `pool` connect option. */ +const pool = new Pool(); + +/** @internal */ +export function sessionPool(): Pool { + return pool; +} + +/** + * Close every pooled session, so the next connection dials fresh. + * + * Exists for tests, which otherwise share sessions across cases. + * + * @internal + */ +export function resetPool(): void { + pool.close(); +} diff --git a/js/net/src/connection/reload.test.ts b/js/net/src/connection/reload.test.ts index 18d7ae7d18..b0bc8429b1 100644 --- a/js/net/src/connection/reload.test.ts +++ b/js/net/src/connection/reload.test.ts @@ -1,15 +1,21 @@ -import { expect, test } from "bun:test"; +import { beforeEach, expect, test } from "bun:test"; import { Producer as BroadcastProducer } from "../broadcast.ts"; import * as Lite from "../lite/index.ts"; import { createMockTransportPair } from "../mock.ts"; import * as Path from "../path.ts"; import { accept } from "./index.ts"; -import { Reload, type ReloadProps } from "./reload.ts"; +import { resetPool } from "./pool.ts"; +import { Reload } from "./reload.ts"; async function settle() { await new Promise((resolve) => setTimeout(resolve, 0)); } +// Sessions are shared by URL, so a leftover one would answer the next case. +beforeEach(() => { + resetPool(); +}); + test("equivalent URL instances do not restart a pending connection", async () => { const original = globalThis.WebTransport; let connects = 0; @@ -49,10 +55,133 @@ test("equivalent URL instances do not restart a pending connection", async () => } }); -test("ReloadProps excludes signal", () => { - // @ts-expect-error signal is not part of ReloadProps - const props: ReloadProps = { signal: new AbortController().signal }; - expect(props.enabled).toBeUndefined(); +test("aborting the signal stops the loop", async () => { + const original = globalThis.WebTransport; + let closes = 0; + + class PendingWebTransport { + ready = new Promise(() => {}); + closed = new Promise(() => {}); + + close() { + closes++; + } + } + + globalThis.WebTransport = PendingWebTransport as unknown as typeof WebTransport; + const controller = new AbortController(); + const reload = new Reload({ + url: new URL("https://example.com/signal"), + websocket: { enabled: false }, + signal: controller.signal, + }); + + try { + await settle(); + expect(reload.status.peek()).toBe("connecting"); + + controller.abort(); + await settle(); + expect(closes).toBe(1); + await reload.closed; + } finally { + reload.close(); + globalThis.WebTransport = original; + } +}); + +test("connecting is the default", async () => { + const original = globalThis.WebTransport; + let connects = 0; + + class PendingWebTransport { + ready = new Promise(() => {}); + closed = new Promise(() => {}); + + constructor() { + connects++; + } + + close() {} + } + + globalThis.WebTransport = PendingWebTransport as unknown as typeof WebTransport; + const reload = new Reload({ url: new URL("https://example.com/default"), websocket: { enabled: false } }); + + try { + await settle(); + expect(connects).toBe(1); + } finally { + reload.close(); + globalThis.WebTransport = original; + } +}); + +test("two connections to one URL share a session", async () => { + const original = globalThis.WebTransport; + const url = new URL("https://example.com/shared"); + let connects = 0; + + const stub = function StubWebTransport() { + connects++; + const pair = createMockTransportPair(Lite.ALPN_06_WIP); + void accept(pair.server, url); + return pair.client; + }; + globalThis.WebTransport = stub as unknown as typeof WebTransport; + + const first = new Reload({ url, websocket: { enabled: false } }); + const second = new Reload({ url, websocket: { enabled: false } }); + + try { + await waitUntil(() => first.established.peek() !== undefined && second.established.peek() !== undefined); + expect(connects).toBe(1); + + // One going away leaves the other connected: the session belongs to both. + first.close(); + await settle(); + expect(second.established.peek()).not.toBeUndefined(); + expect(second.status.peek()).toBe("connected"); + } finally { + first.close(); + second.close(); + globalThis.WebTransport = original; + } +}); + +test("reload: false gives up after one session", async () => { + const original = globalThis.WebTransport; + const url = new URL("https://example.com/once"); + let connects = 0; + + const sessions: { close: () => void }[] = []; + const stub = function StubWebTransport() { + connects++; + const pair = createMockTransportPair(Lite.ALPN_06_WIP); + void accept(pair.server, url).then((server) => sessions.push(server)); + return pair.client; + }; + globalThis.WebTransport = stub as unknown as typeof WebTransport; + + const reload = new Reload({ url, websocket: { enabled: false }, reload: false }); + + try { + await waitUntil(() => reload.established.peek() !== undefined && sessions.length > 0); + + // A clean drop is the end of the line rather than the start of a backoff. + sessions[0]?.close(); + await reload.closed; + expect(connects).toBe(1); + expect(reload.established.peek()).toBeUndefined(); + expect(reload.status.peek()).toBe("disconnected"); + + // Nothing is scheduled, so it stays down. + await settle(); + expect(connects).toBe(1); + } finally { + reload.close(); + globalThis.WebTransport = original; + } }); test("closing mid-connect aborts the pending attempt", async () => { diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index bf2360b29f..3459fd5925 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -2,50 +2,44 @@ import { Effect, type Getter, Signal } from "@moq/signals"; import * as Announce from "../announced.ts"; import type * as Path from "../path.ts"; import { empty as emptyPath } from "../path.ts"; -import { type ConnectProps, connect, type WebSocketOptions, type WebTransportProps } from "./connect.ts"; +import { + type ConnectProps, + connect, + type ReloadDelay, + type WebSocketOptions, + type WebTransportProps, +} from "./connect.ts"; import type { Established } from "./established.ts"; +import type { PoolProps } from "./pool.ts"; import type { Probe, Stats } from "./stats.ts"; -/** Exponential backoff settings for {@link Reload}'s reconnect loop. */ -export type ReloadDelay = { - /** The delay in milliseconds before reconnecting (default: 1000). */ - initial: DOMHighResTimeStamp; - - /** The multiplier for the delay (default: 2). */ - multiplier: number; - - /** The maximum delay in milliseconds (default: 30000). */ - max: DOMHighResTimeStamp; - - /** - * Maximum total time in milliseconds to spend retrying before giving up (default: - * 300000, 5 minutes). Resets after each successful connection. Set to 0 for - * unlimited retries. - */ - timeout?: DOMHighResTimeStamp; -}; - -/** Connection and retry options for {@link Reload}. */ -export type ReloadProps = Omit & { - /** Whether to reload the connection when it disconnects (default: true). */ - enabled?: boolean | Signal; - - /** The URL of the relay server. */ - url?: URL | Signal; - - /** Backoff settings for the reconnect loop. */ +/** + * Connection and retry options for {@link Reload}. + * + * @internal + */ +export type ReloadProps = ConnectProps & { + /** @internal Superseded by `reload`, which also disables the loop. */ delay?: ReloadDelay; }; +/** The backoff used when nothing else is asked for. */ +const DEFAULT_DELAY: ReloadDelay = { initial: 1000, multiplier: 2, max: 30000 }; + /** Current state of a {@link Reload} connection. */ export type ReloadStatus = "connecting" | "connected" | "disconnected"; -/** Maintains a MoQ connection, reconnecting with exponential backoff when it drops. */ +/** + * Maintains a MoQ connection, reconnecting with exponential backoff when it drops. + * + * Takes the same {@link ConnectProps} as {@link connect}, including the shared session pool, so + * several of these on one relay URL cost one connection. + */ export class Reload { /** Relay URL to connect to; updating it triggers a reconnect. */ url: Signal; - /** Whether reconnecting is active. */ + /** Whether to connect at all; clearing it disconnects and setting it reconnects. */ enabled: Signal; /** Current connection status. */ @@ -77,6 +71,12 @@ export class Reload { /** Backoff settings for the reconnect loop. */ delay: ReloadDelay; + /** Whether a dropped session is reconnected at all, applied to each drop (not reactive). */ + reload: boolean; + + /** Whether sessions are shared with other connections to the same URL (not reactive). */ + pool: boolean | PoolProps; + /** The reactive effect scope driving the connect loop; closed by {@link Reload.close}. */ #signals = new Effect(); @@ -101,8 +101,10 @@ export class Reload { #url: Getter; constructor(props?: ReloadProps) { this.url = Signal.from(props?.url); - this.enabled = Signal.from(props?.enabled ?? false); - this.delay = props?.delay ?? { initial: 1000, multiplier: 2, max: 30000 }; + this.enabled = Signal.from(props?.enabled ?? true); + this.reload = props?.reload !== false; + this.delay = (typeof props?.reload === "object" ? props.reload : props?.delay) ?? DEFAULT_DELAY; + this.pool = props?.pool ?? true; this.webtransport = props?.webtransport; this.websocket = props?.websocket; this.discovery = props?.discovery; @@ -131,6 +133,13 @@ export class Reload { this.#url = this.#signals.computed((effect) => effect.get(this.url)?.href); // Create a reactive root so cleanup is easier. this.#signals.run(this.#connect.bind(this)); + + // The whole managed connection goes away with the caller's signal, since there is no + // single attempt for it to cancel. + if (props?.signal) { + if (props.signal.aborted) this.close(); + else this.#signals.event(props.signal, "abort", () => this.close()); + } } #connect(effect: Effect): void { @@ -160,6 +169,7 @@ export class Reload { websocket: this.websocket, webtransport: this.webtransport, discovery: this.discovery, + pool: this.pool, signal, }); @@ -200,6 +210,13 @@ export class Reload { this.established.set(undefined); this.status.set("disconnected"); + // One attempt was all that was asked for, so this is the end of the line. + if (!this.reload) { + if (cause === undefined) this.#closedResolve(); + else this.#closedReject(cause instanceof Error ? cause : new Error(String(cause))); + return; + } + // A session that outlived the initial delay was healthy, so clear the backoff and // start a fresh retry window: a one-off drop should reconnect promptly. Anything // shorter is a peer that accepts and immediately severs, which has to keep diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 3cccde4a9b..cf211ed1ab 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -57,10 +57,11 @@ export class Publisher { broadcasts.set(path, broadcast); }); - // Remove the broadcast from the lookup when it's closed. + // Remove the broadcast from the lookup when it's closed, unless a republish already + // replaced it: a stale producer closing must not unpublish the live one. void broadcast.closed.then(() => { this.#broadcasts.mutate((broadcasts) => { - broadcasts?.delete(path); + if (broadcasts?.get(path) === broadcast) broadcasts.delete(path); }); }); } diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 04017c4a1f..bc9c6340dd 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -68,3 +68,39 @@ test("lite draft-05: subscribe end clears the max sequence when groups arrive ou test("lite draft-05: subscribe end is 0 when no groups were produced", async () => { expect(await subscribeEnd([])).toBe(0); }); + +// A shared session lets two components publish the same path, so the registration has to +// belong to whoever holds it now. Without the identity check the stale producer's close +// deletes the live one's entry, and the path stops answering subscribes. +test("a stale producer closing does not unpublish a republished path", async () => { + const pair = createMockTransportPair(ALPN_05); + const publisher = new Publisher(pair.server, Version.DRAFT_05, randomOrigin()); + const path = Path.from("test"); + + const stale = new BroadcastProducer(); + publisher.publish(path, stale); + + const live = new BroadcastProducer(); + publisher.publish(path, live); + + stale.close(); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const client = await Stream.open(pair.client); + const server = await Stream.accept(pair.server); + if (!server) throw new Error("publisher never accepted the subscribe stream"); + + const msg = new Subscribe({ id: 0n, broadcast: path, track: "video", priority: 0 }); + void publisher.runSubscribe(msg, server); + + try { + // An unpublished path is reset rather than served, so the request never arrives; + // race a deadline so the regression fails instead of hanging. + const deadline = new Promise((resolve) => setTimeout(() => resolve("not served"), 100)); + const request = await Promise.race([live.requested(), deadline]); + expect(request).toHaveProperty("name", "video"); + } finally { + live.close(); + client.close(); + } +}); diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 8b074b5ad3..7f056e9807 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -110,10 +110,11 @@ export class Publisher { broadcasts.set(path, broadcast); }); - // Remove the broadcast from the lookup when it's closed. + // Remove the broadcast from the lookup when it's closed, unless a republish already + // replaced it: a stale producer closing must not unpublish the live one. void broadcast.closed.then(() => { this.#broadcasts.mutate((broadcasts) => { - broadcasts?.delete(path); + if (broadcasts?.get(path) === broadcast) broadcasts.delete(path); }); }); } From 7898cc56f93cddb47b7abb4f4592105957a37968 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 4 Aug 2026 19:42:32 -0700 Subject: [PATCH 2/6] fix(js/net): address review on the session pool Reload settled `closed` on both terminal paths without closing its effect scope, leaving the page listeners, the probe computed, and the announce pumps behind. Route both through one helper that tears down first. The no-discovery warning cache keyed on the full href, which pins every auth token an app ever used; key on origin and path instead. Also prune a lease's announcedBroadcast handles when they close, document that a pooled handle's `closed` follows the shared session rather than the handle, publish ReloadProps' option shape via ConnectProps, and share one pending-transport stub across the connection tests. Co-Authored-By: Claude Opus 5 --- js/net/src/announced.ts | 18 +++++- js/net/src/connection/connect.test.ts | 38 +++--------- js/net/src/connection/connect.ts | 10 +++- js/net/src/connection/pool.test.ts | 35 +++++------ js/net/src/connection/pool.ts | 7 +++ js/net/src/connection/reload.test.ts | 83 ++++++++------------------- js/net/src/connection/reload.ts | 34 ++++++----- js/net/src/lite/publisher.test.ts | 14 ++++- js/net/src/mock.ts | 39 +++++++++++++ 9 files changed, 143 insertions(+), 135 deletions(-) diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index 2f6d614b47..c9ad534d54 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -130,10 +130,16 @@ export class Consumer { } // Relays already warned about missing broadcast discovery, so the fallback logs at most once -// per relay instead of once per watched path. Keyed by URL rather than by session, since +// per relay instead of once per watched path. Keyed by relay rather than by session, since // several handles can share one connection. const warnedNoDiscovery = new Set(); +// Never the full href: the query carries the auth token, so keying on it would pin every token +// an app ever used in memory and mint an entry per rotation. Origin plus path is the relay. +function relayKey(url: URL): string { + return `${url.origin}${url.pathname}`; +} + /** * What to watch, for {@link Broadcast}. * @@ -219,8 +225,9 @@ export class Broadcast { // Without discovery no announcement ever arrives, so waiting would hang forever. if (!conn.discovery) { - if (!warnedNoDiscovery.has(conn.url.href)) { - warnedNoDiscovery.add(conn.url.href); + const key = relayKey(conn.url); + if (!warnedNoDiscovery.has(key)) { + warnedNoDiscovery.add(key); console.warn("relay does not support broadcast discovery; consuming without waiting."); } @@ -287,6 +294,11 @@ export class Broadcast { }); } + /** Resolves once the handle is closed, so an owner can drop its reference. */ + get closed(): Promise { + return this.#signals.closed; + } + /** Closes the handle and the broadcast it currently holds. Idempotent. */ close() { this.#signals.close(); diff --git a/js/net/src/connection/connect.test.ts b/js/net/src/connection/connect.test.ts index 4fcd0f4ecb..2e3e4aaef9 100644 --- a/js/net/src/connection/connect.test.ts +++ b/js/net/src/connection/connect.test.ts @@ -1,6 +1,6 @@ import { beforeEach, expect, test } from "bun:test"; import { ALPN_05 } from "../lite/version.ts"; -import { createMockTransportPair } from "../mock.ts"; +import { createMockTransportPair, createPendingTransports } from "../mock.ts"; import { connect } from "./connect.ts"; import { resetPool } from "./pool.ts"; @@ -17,20 +17,8 @@ beforeEach(() => { test("already-aborted signal rejects without connecting", async () => { const original = globalThis.WebTransport; - let connects = 0; - - class CountingWebTransport { - ready = new Promise(() => {}); - closed = new Promise(() => {}); - - constructor() { - connects++; - } - - close() {} - } - - globalThis.WebTransport = CountingWebTransport as unknown as typeof WebTransport; + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; try { const controller = new AbortController(); @@ -42,7 +30,7 @@ test("already-aborted signal rejects without connecting", async () => { ); expect(err).toBeInstanceOf(DOMException); expect((err as DOMException).name).toBe("AbortError"); - expect(connects).toBe(0); + expect(pending.connects()).toBe(0); } finally { globalThis.WebTransport = original; } @@ -50,18 +38,8 @@ test("already-aborted signal rejects without connecting", async () => { test("abort mid-connect rejects with the reason and closes the transport", async () => { const original = globalThis.WebTransport; - let closes = 0; - - class PendingWebTransport { - ready = new Promise(() => {}); - closed = new Promise(() => {}); - - close() { - closes++; - } - } - - globalThis.WebTransport = PendingWebTransport as unknown as typeof WebTransport; + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; try { const controller = new AbortController(); @@ -73,13 +51,13 @@ test("abort mid-connect rejects with the reason and closes the transport", async ); await settle(); - expect(closes).toBe(0); + expect(pending.closes()).toBe(0); controller.abort(reason); expect(await result).toBe(reason); await settle(); - expect(closes).toBe(1); + expect(pending.closes()).toBe(1); } finally { globalThis.WebTransport = original; } diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index e7c9cfeb43..e60f3f2f65 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -75,10 +75,11 @@ export interface WebTransportProps extends Omit setTimeout(resolve, 0)); } -// A tiny window keeps the linger tests quick without mocking timers. +// A tiny window keeps the linger tests quick without mocking timers. The wait is a wide +// multiple of it so a loaded runner's timer drift can't be mistaken for the session lingering. const grace = 20; async function expired() { - await new Promise((resolve) => setTimeout(resolve, grace * 3)); + await new Promise((resolve) => setTimeout(resolve, grace * 15)); } const original = globalThis.WebTransport; @@ -47,25 +53,10 @@ function stubTransports(): { transports: MockTransport[]; servers: MockTransport } /** Hand out a transport that never finishes connecting, counting dials and closes. */ -function stubPending(): { connects: () => number; closes: () => number } { - let connects = 0; - let closes = 0; - - class PendingWebTransport { - ready = new Promise(() => {}); - closed = new Promise(() => {}); - - constructor() { - connects++; - } - - close() { - closes++; - } - } - - globalThis.WebTransport = PendingWebTransport as unknown as typeof WebTransport; - return { connects: () => connects, closes: () => closes }; +function stubPending(): PendingTransports { + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; + return pending; } test("two connections to one URL share a session", async () => { diff --git a/js/net/src/connection/pool.ts b/js/net/src/connection/pool.ts index f2c31771a9..cf99dade71 100644 --- a/js/net/src/connection/pool.ts +++ b/js/net/src/connection/pool.ts @@ -276,6 +276,7 @@ class Lease implements Established { // with us. const watch = new announce.Broadcast({ connection: this, path }); this.#broadcasts.add(watch); + void watch.closed.then(() => this.#broadcasts.delete(watch)); return watch; } @@ -283,6 +284,12 @@ class Lease implements Established { return await this.#session.stats(); } + /** + * Resolves when the shared session closes, which is not when you release this handle. + * + * Somebody else may still be using the connection after your {@link close}, so don't await + * this to find out that your own teardown finished; it says the connection is gone. + */ get closed(): Promise { return this.#session.closed; } diff --git a/js/net/src/connection/reload.test.ts b/js/net/src/connection/reload.test.ts index b0bc8429b1..2fbf04d593 100644 --- a/js/net/src/connection/reload.test.ts +++ b/js/net/src/connection/reload.test.ts @@ -1,7 +1,7 @@ import { beforeEach, expect, test } from "bun:test"; import { Producer as BroadcastProducer } from "../broadcast.ts"; import * as Lite from "../lite/index.ts"; -import { createMockTransportPair } from "../mock.ts"; +import { createMockTransportPair, createPendingTransports } from "../mock.ts"; import * as Path from "../path.ts"; import { accept } from "./index.ts"; import { resetPool } from "./pool.ts"; @@ -18,20 +18,8 @@ beforeEach(() => { test("equivalent URL instances do not restart a pending connection", async () => { const original = globalThis.WebTransport; - let connects = 0; - - class PendingWebTransport { - ready = new Promise(() => {}); - closed = new Promise(() => {}); - - constructor() { - connects++; - } - - close() {} - } - - globalThis.WebTransport = PendingWebTransport as unknown as typeof WebTransport; + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; const reload = new Reload({ enabled: true, url: new URL("https://example.com/broadcast"), @@ -40,15 +28,15 @@ test("equivalent URL instances do not restart a pending connection", async () => try { await settle(); - expect(connects).toBe(1); + expect(pending.connects()).toBe(1); reload.url.set(new URL("https://example.com/broadcast")); await settle(); - expect(connects).toBe(1); + expect(pending.connects()).toBe(1); reload.url.set(new URL("https://example.com/other")); await settle(); - expect(connects).toBe(2); + expect(pending.connects()).toBe(2); } finally { reload.close(); globalThis.WebTransport = original; @@ -57,18 +45,8 @@ test("equivalent URL instances do not restart a pending connection", async () => test("aborting the signal stops the loop", async () => { const original = globalThis.WebTransport; - let closes = 0; - - class PendingWebTransport { - ready = new Promise(() => {}); - closed = new Promise(() => {}); - - close() { - closes++; - } - } - - globalThis.WebTransport = PendingWebTransport as unknown as typeof WebTransport; + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; const controller = new AbortController(); const reload = new Reload({ url: new URL("https://example.com/signal"), @@ -82,7 +60,7 @@ test("aborting the signal stops the loop", async () => { controller.abort(); await settle(); - expect(closes).toBe(1); + expect(pending.closes()).toBe(1); await reload.closed; } finally { reload.close(); @@ -92,25 +70,13 @@ test("aborting the signal stops the loop", async () => { test("connecting is the default", async () => { const original = globalThis.WebTransport; - let connects = 0; - - class PendingWebTransport { - ready = new Promise(() => {}); - closed = new Promise(() => {}); - - constructor() { - connects++; - } - - close() {} - } - - globalThis.WebTransport = PendingWebTransport as unknown as typeof WebTransport; + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; const reload = new Reload({ url: new URL("https://example.com/default"), websocket: { enabled: false } }); try { await settle(); - expect(connects).toBe(1); + expect(pending.connects()).toBe(1); } finally { reload.close(); globalThis.WebTransport = original; @@ -178,6 +144,13 @@ test("reload: false gives up after one session", async () => { // Nothing is scheduled, so it stays down. await settle(); expect(connects).toBe(1); + + // Settling `closed` also tears the effect scope down, so the page listeners, the probe + // computed, and the announce pumps go with it. A URL change proves it: a live scope + // would rerun the connect effect and dial again. + reload.url.set(new URL("https://example.com/once-again")); + await settle(); + expect(connects).toBe(1); } finally { reload.close(); globalThis.WebTransport = original; @@ -186,18 +159,8 @@ test("reload: false gives up after one session", async () => { test("closing mid-connect aborts the pending attempt", async () => { const original = globalThis.WebTransport; - let closes = 0; - - class PendingWebTransport { - ready = new Promise(() => {}); - closed = new Promise(() => {}); - - close() { - closes++; - } - } - - globalThis.WebTransport = PendingWebTransport as unknown as typeof WebTransport; + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; const reload = new Reload({ enabled: true, url: new URL("https://example.com/broadcast"), @@ -206,11 +169,11 @@ test("closing mid-connect aborts the pending attempt", async () => { try { await settle(); - expect(closes).toBe(0); + expect(pending.closes()).toBe(0); reload.close(); await settle(); - expect(closes).toBe(1); + expect(pending.closes()).toBe(1); } finally { globalThis.WebTransport = original; } diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index 3459fd5925..0dd3c8486f 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -13,15 +13,8 @@ import type { Established } from "./established.ts"; import type { PoolProps } from "./pool.ts"; import type { Probe, Stats } from "./stats.ts"; -/** - * Connection and retry options for {@link Reload}. - * - * @internal - */ -export type ReloadProps = ConnectProps & { - /** @internal Superseded by `reload`, which also disables the loop. */ - delay?: ReloadDelay; -}; +/** @internal Superseded by {@link ConnectProps}, which {@link Reload} takes directly. */ +export type ReloadProps = ConnectProps; /** The backoff used when nothing else is asked for. */ const DEFAULT_DELAY: ReloadDelay = { initial: 1000, multiplier: 2, max: 30000 }; @@ -99,7 +92,7 @@ export class Reload { // Use the serialized URL as the reactive connection key. URL objects use identity // equality, but replacing one with an equivalent instance should not reconnect. #url: Getter; - constructor(props?: ReloadProps) { + constructor(props?: ConnectProps) { this.url = Signal.from(props?.url); this.enabled = Signal.from(props?.enabled ?? true); this.reload = props?.reload !== false; @@ -212,8 +205,7 @@ export class Reload { // One attempt was all that was asked for, so this is the end of the line. if (!this.reload) { - if (cause === undefined) this.#closedResolve(); - else this.#closedReject(cause instanceof Error ? cause : new Error(String(cause))); + this.#finish(cause); return; } @@ -235,8 +227,7 @@ export class Reload { if (elapsed >= timeout) { console.warn("reconnect timed out"); // A graceful close has no error, so report the timeout itself. - if (cause === undefined) this.#closedReject(new Error("reconnect timed out")); - else this.#closedReject(cause instanceof Error ? cause : new Error(String(cause))); + this.#finish(cause ?? new Error("reconnect timed out")); return; } } @@ -335,7 +326,20 @@ export class Reload { /** Stop reconnecting, close the current connection, and resolve {@link Reload.closed}. */ close() { + this.#finish(); + } + + /** + * Stop for good: tear down the effect scope, then settle {@link Reload.closed}. + * + * Both terminal paths run through here so neither leaves the page listeners, the `probe` + * computed, and the `announced()` pumps behind. Settling twice is a no-op, so the public + * {@link Reload.close} after a timeout keeps the original cause. + */ + #finish(cause?: unknown): void { this.#signals.close(); - this.#closedResolve(); + + if (cause === undefined) this.#closedResolve(); + else this.#closedReject(cause instanceof Error ? cause : new Error(String(cause))); } } diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index bc9c6340dd..5c0d024f25 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -4,6 +4,7 @@ import { Producer as GroupProducer } from "../group.ts"; import { createMockTransportPair } from "../mock.ts"; import * as Path from "../path.ts"; import { Stream } from "../stream.ts"; +import { Request as TrackRequest } from "../track.ts"; import { randomOrigin } from "./origin.ts"; import { Publisher } from "./publisher.ts"; import { decodeSubscribeResponse, Subscribe } from "./subscribe.ts"; @@ -69,6 +70,9 @@ test("lite draft-05: subscribe end is 0 when no groups were produced", async () expect(await subscribeEnd([])).toBe(0); }); +// How long a served path gets to answer before the test calls it unpublished. +const REQUEST_DEADLINE_MS = 100; + // A shared session lets two components publish the same path, so the registration has to // belong to whoever holds it now. Without the identity check the stale producer's close // deletes the live one's entry, and the path stops answering subscribes. @@ -91,16 +95,22 @@ test("a stale producer closing does not unpublish a republished path", async () if (!server) throw new Error("publisher never accepted the subscribe stream"); const msg = new Subscribe({ id: 0n, broadcast: path, track: "video", priority: 0 }); - void publisher.runSubscribe(msg, server); + const serving = publisher.runSubscribe(msg, server).catch(() => undefined); try { // An unpublished path is reset rather than served, so the request never arrives; // race a deadline so the regression fails instead of hanging. - const deadline = new Promise((resolve) => setTimeout(() => resolve("not served"), 100)); + const deadline = new Promise((resolve) => setTimeout(() => resolve("not served"), REQUEST_DEADLINE_MS)); const request = await Promise.race([live.requested(), deadline]); expect(request).toHaveProperty("name", "video"); + + // The handler parks on track.info() until the request is answered, so reject it rather + // than leaving the handler running past the test. + if (request instanceof TrackRequest) request.reject(new Error("done")); } finally { live.close(); + publisher.close(); client.close(); + await serving; } }); diff --git a/js/net/src/mock.ts b/js/net/src/mock.ts index b9ba43345a..20bd1c62f5 100644 --- a/js/net/src/mock.ts +++ b/js/net/src/mock.ts @@ -238,6 +238,45 @@ export class MockTransport implements WebTransport { } } +/** A {@link WebTransport} stub that never connects, plus counters for what it was asked to do. */ +export interface PendingTransports { + /** Assign to `globalThis.WebTransport` for the duration of a test. */ + transport: typeof WebTransport; + /** How many were constructed, i.e. how many dials were attempted. */ + connects(): number; + /** How many were closed, i.e. how many attempts were abandoned. */ + closes(): number; +} + +/** + * A {@link WebTransport} whose `ready` and `closed` never settle, so a connection attempt stays + * in flight until something aborts it. Counts constructions and closes for tests asserting on + * how many dials happened and whether they were cleaned up. + */ +export function createPendingTransports(): PendingTransports { + let connects = 0; + let closes = 0; + + class PendingWebTransport { + ready = new Promise(() => {}); + closed = new Promise(() => {}); + + constructor() { + connects++; + } + + close() { + closes++; + } + } + + return { + transport: PendingWebTransport as unknown as typeof WebTransport, + connects: () => connects, + closes: () => closes, + }; +} + type DatagramWritableApi = "writable" | "createWritable" | "none"; /** Options for {@link createMockTransportPair}. */ From 608255790c7665066b340ad35bb71e987dd16703 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 4 Aug 2026 19:48:03 -0700 Subject: [PATCH 3/6] fix(js/net): bound the no-discovery warning cache Keying it by relay stopped auth tokens piling up, but the set still had no eviction, so an app rotating through relays grew it for the process lifetime. Cap it and drop the oldest entry, giving up exactly-once for a hard bound: the cache exists to keep the log readable, not to guarantee a single line forever. Co-Authored-By: Claude Opus 5 --- js/net/src/announced.test.ts | 63 +++++++++++++++++++++++++++++++++++- js/net/src/announced.ts | 35 ++++++++++++++------ 2 files changed, 88 insertions(+), 10 deletions(-) diff --git a/js/net/src/announced.test.ts b/js/net/src/announced.test.ts index 1cd3dc31fc..5576df214c 100644 --- a/js/net/src/announced.test.ts +++ b/js/net/src/announced.test.ts @@ -1,5 +1,6 @@ -import { expect, test } from "bun:test"; +import { expect, spyOn, test } from "bun:test"; import * as Announce from "./announced.ts"; +import { resetNoDiscoveryWarnings } from "./announced.ts"; import * as Path from "./path.ts"; const p = (s: string) => Path.from(s); @@ -44,3 +45,63 @@ test("aborting rejects next", async () => { producer.close(new Error("boom")); await expect(consumer.next()).rejects.toThrow("boom"); }); + +// A stub session with no discovery, so Broadcast takes the warn-and-consume-blind path. +function noDiscovery(url: string): Announce.BroadcastProps["connection"] { + return { + url: new URL(url), + discovery: false, + closed: new Promise(() => {}), + consume: () => ({ close() {}, closed: { peek: () => undefined } }), + } as unknown as Announce.BroadcastProps["connection"]; +} + +// Count what the handles log on their first run, so the warning is measured rather than +// eyeballed. The handles are closed only after that run, since closing one in the same job +// tears its effect down before it ever warns. +async function countWarnings(fn: () => Announce.Broadcast[]): Promise { + const warn = spyOn(console, "warn").mockImplementation(() => {}); + const handles = fn(); + + try { + await new Promise((resolve) => setTimeout(resolve, 0)); + return warn.mock.calls.length; + } finally { + for (const handle of handles) handle.close(); + warn.mockRestore(); + } +} + +test("the no-discovery warning is once per relay, ignoring the auth token", async () => { + resetNoDiscoveryWarnings(); + + // Same relay, different tokens and different watched paths: still one relay. + const warnings = await countWarnings(() => [ + new Announce.Broadcast({ connection: noDiscovery("https://relay.example/anon?jwt=a"), path: p("one") }), + new Announce.Broadcast({ connection: noDiscovery("https://relay.example/anon?jwt=b"), path: p("two") }), + new Announce.Broadcast({ connection: noDiscovery("https://other.example/anon"), path: p("one") }), + ]); + + expect(warnings).toBe(2); +}); + +test("the no-discovery warning cache is bounded", async () => { + resetNoDiscoveryWarnings(); + + // Fill past the cap, then come back to the very first relay. An unbounded cache would + // still remember it and stay silent; a bounded one has evicted it and warns again. + const first = "https://relay0.example/anon"; + await countWarnings(() => + Array.from( + { length: 200 }, + (_, i) => + new Announce.Broadcast({ connection: noDiscovery(`https://relay${i}.example/anon`), path: p("x") }), + ), + ); + + const warnings = await countWarnings(() => [ + new Announce.Broadcast({ connection: noDiscovery(first), path: p("x") }), + ]); + + expect(warnings).toBe(1); +}); diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index c9ad534d54..14b9be6918 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -134,10 +134,31 @@ export class Consumer { // several handles can share one connection. const warnedNoDiscovery = new Set(); -// Never the full href: the query carries the auth token, so keying on it would pin every token -// an app ever used in memory and mint an entry per rotation. Origin plus path is the relay. -function relayKey(url: URL): string { - return `${url.origin}${url.pathname}`; +// Enough that a real deployment never evicts, small enough that the cache can't grow into a +// leak. Past it the oldest relay is forgotten and may warn a second time, which is the right +// thing to give up: this exists to keep the log readable, not to guarantee exactly-once. +const WARNED_MAX = 64; + +/** Warn that `url`'s relay lacks discovery, at most once per relay. */ +function warnNoDiscovery(url: URL): void { + // Never the full href: the query carries the auth token, so keying on it would pin every + // token an app ever used and mint an entry per rotation. Origin plus path is the relay. + const key = `${url.origin}${url.pathname}`; + if (warnedNoDiscovery.has(key)) return; + + // A Set iterates in insertion order, so the first entry is the oldest. + if (warnedNoDiscovery.size >= WARNED_MAX) { + const oldest = warnedNoDiscovery.values().next().value; + if (oldest !== undefined) warnedNoDiscovery.delete(oldest); + } + + warnedNoDiscovery.add(key); + console.warn("relay does not support broadcast discovery; consuming without waiting."); +} + +/** @internal Forget every warned relay, so a test starts from a clean cache. */ +export function resetNoDiscoveryWarnings(): void { + warnedNoDiscovery.clear(); } /** @@ -225,11 +246,7 @@ export class Broadcast { // Without discovery no announcement ever arrives, so waiting would hang forever. if (!conn.discovery) { - const key = relayKey(conn.url); - if (!warnedNoDiscovery.has(key)) { - warnedNoDiscovery.add(key); - console.warn("relay does not support broadcast discovery; consuming without waiting."); - } + warnNoDiscovery(conn.url); const blind = conn.consume(path); effect.cleanup(() => blind.close()); From 49d4a9c6c8e061130e8434f2edddc8b46e1eb16e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 4 Aug 2026 19:52:06 -0700 Subject: [PATCH 4/6] fix(js/net): keep transport out of the Reload contract Reload accepts ConnectProps but never forwards `transport`, so a caller supplying a pre-existing WebTransport silently got a fresh dial instead. A supplied session is good for one connection and a reconnect loop has nothing to reuse after the first drop, so drop the option from the type rather than pretending to honor it: ReloadProps is that view of ConnectProps, published again so the constructor's shape is visible. Also cover the state a closed Reload reports, and drop two doc comments that described how the options got here rather than what they do. Co-Authored-By: Claude Opus 5 --- js/net/src/connection/connect.ts | 2 +- js/net/src/connection/reload.test.ts | 27 +++++++++++++++++++++++++++ js/net/src/connection/reload.ts | 13 ++++++++++--- 3 files changed, 38 insertions(+), 4 deletions(-) diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index e60f3f2f65..1b242e1b9c 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -141,7 +141,7 @@ export interface ConnectProps { */ reload?: boolean | ReloadDelay; - /** @internal Superseded by {@link ConnectProps.reload}, which also disables the loop. */ + /** @internal Backoff settings, ignored when {@link ConnectProps.reload} carries its own. */ delay?: ReloadDelay; } diff --git a/js/net/src/connection/reload.test.ts b/js/net/src/connection/reload.test.ts index 2fbf04d593..6c30fb0922 100644 --- a/js/net/src/connection/reload.test.ts +++ b/js/net/src/connection/reload.test.ts @@ -266,3 +266,30 @@ test("announcedBroadcast follows the reconnect loop", async () => { globalThis.WebTransport = original; } }); + +test("closing a live connection reports disconnected", async () => { + const original = globalThis.WebTransport; + const url = new URL("https://example.com/live"); + + const stub = function StubWebTransport() { + const pair = createMockTransportPair(Lite.ALPN_06_WIP); + void accept(pair.server, url); + return pair.client; + }; + globalThis.WebTransport = stub as unknown as typeof WebTransport; + + const reload = new Reload({ url, websocket: { enabled: false } }); + + try { + await waitUntil(() => reload.established.peek() !== undefined); + expect(reload.status.peek()).toBe("connected"); + + // Whoever is watching these gets the state they're left holding, not the dead session. + reload.close(); + expect(reload.established.peek()).toBeUndefined(); + expect(reload.status.peek()).toBe("disconnected"); + await reload.closed; + } finally { + globalThis.WebTransport = original; + } +}); diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index 0dd3c8486f..9e8e20d2ba 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -13,8 +13,13 @@ import type { Established } from "./established.ts"; import type { PoolProps } from "./pool.ts"; import type { Probe, Stats } from "./stats.ts"; -/** @internal Superseded by {@link ConnectProps}, which {@link Reload} takes directly. */ -export type ReloadProps = ConnectProps; +/** + * The {@link ConnectProps} a {@link Reload} accepts. + * + * Everything {@link connect} takes except `transport`: a supplied session is good for one + * connection, and a loop that reconnects has nothing to reuse after the first drop. + */ +export type ReloadProps = Omit; /** The backoff used when nothing else is asked for. */ const DEFAULT_DELAY: ReloadDelay = { initial: 1000, multiplier: 2, max: 30000 }; @@ -92,7 +97,7 @@ export class Reload { // Use the serialized URL as the reactive connection key. URL objects use identity // equality, but replacing one with an equivalent instance should not reconnect. #url: Getter; - constructor(props?: ConnectProps) { + constructor(props?: ReloadProps) { this.url = Signal.from(props?.url); this.enabled = Signal.from(props?.enabled ?? true); this.reload = props?.reload !== false; @@ -337,6 +342,8 @@ export class Reload { * {@link Reload.close} after a timeout keeps the original cause. */ #finish(cause?: unknown): void { + // Closing the scope also restores `established` and `status`, since the connect run set + // them through `effect.set`. this.#signals.close(); if (cause === undefined) this.#closedResolve(); From 707da6a35ea67bb1bc7c0afb91bdcd822963766d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 4 Aug 2026 20:31:44 -0700 Subject: [PATCH 5/6] fix(js/net): warn when Reload is handed a transport Reload has never forwarded ConnectProps.transport, so a caller supplying a pre-existing WebTransport silently got a fresh dial. A supplied session is good for one connection and a reconnect loop has nothing to reuse after the first drop, so it stays unsupported. Warn on use instead of narrowing the type, which would break a published API and send this PR to dev for a hole that predates it. Co-Authored-By: Claude Opus 5 --- js/net/src/connection/reload.test.ts | 18 +++++++++++++++++- js/net/src/connection/reload.ts | 16 +++++++++------- 2 files changed, 26 insertions(+), 8 deletions(-) diff --git a/js/net/src/connection/reload.test.ts b/js/net/src/connection/reload.test.ts index 6c30fb0922..8191105104 100644 --- a/js/net/src/connection/reload.test.ts +++ b/js/net/src/connection/reload.test.ts @@ -1,4 +1,4 @@ -import { beforeEach, expect, test } from "bun:test"; +import { beforeEach, expect, spyOn, test } from "bun:test"; import { Producer as BroadcastProducer } from "../broadcast.ts"; import * as Lite from "../lite/index.ts"; import { createMockTransportPair, createPendingTransports } from "../mock.ts"; @@ -293,3 +293,19 @@ test("closing a live connection reports disconnected", async () => { globalThis.WebTransport = original; } }); + +test("a supplied transport is refused out loud", () => { + const warn = spyOn(console, "warn").mockImplementation(() => {}); + const pair = createMockTransportPair(Lite.ALPN_06_WIP); + + // Silently dialing past it would hand back a session the caller never asked for. + const reload = new Reload({ enabled: false, transport: pair.client }); + + try { + expect(warn.mock.calls.length).toBe(1); + } finally { + reload.close(); + warn.mockRestore(); + pair.client.close(); + } +}); diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index 9e8e20d2ba..eac26000f0 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -13,13 +13,8 @@ import type { Established } from "./established.ts"; import type { PoolProps } from "./pool.ts"; import type { Probe, Stats } from "./stats.ts"; -/** - * The {@link ConnectProps} a {@link Reload} accepts. - * - * Everything {@link connect} takes except `transport`: a supplied session is good for one - * connection, and a loop that reconnects has nothing to reuse after the first drop. - */ -export type ReloadProps = Omit; +/** The {@link ConnectProps} a {@link Reload} accepts. */ +export type ReloadProps = ConnectProps; /** The backoff used when nothing else is asked for. */ const DEFAULT_DELAY: ReloadDelay = { initial: 1000, multiplier: 2, max: 30000 }; @@ -109,6 +104,13 @@ export class Reload { this.#delay = this.delay.initial; + // A supplied session is good for one connection, so a loop that reconnects has nothing + // to reuse after the first drop and dials its own instead. Say so out loud rather than + // silently handing back a session the caller didn't ask for. + if (props?.transport) { + console.warn("transport is ignored when reconnecting; use connect() for a supplied session"); + } + this.closed = new Promise((resolve, reject) => { this.#closedResolve = resolve; this.#closedReject = reject; From 1e0843ad6d052fe991d9133b6b51b62b31fd58d6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 4 Aug 2026 20:38:15 -0700 Subject: [PATCH 6/6] test(js/net): close the gaps in the pool review coverage The warning dedup test varied tokens and watched paths but never the relay pathname, so keying on origin alone would have passed it. The supplied-transport test asserted a warning fired without observing that a fresh dial happened, so "warned and used it anyway" would have passed too. Both now fail on those regressions. Also derive the eviction count from the cache limit rather than a literal, and close a Reload in `finally` so a failed assertion can't leave a session leased for the next case. Co-Authored-By: Claude Opus 5 --- js/net/src/announced.test.ts | 11 +++++++---- js/net/src/announced.ts | 14 ++++++++++---- js/net/src/connection/reload.test.ts | 28 ++++++++++++++++++++++------ 3 files changed, 39 insertions(+), 14 deletions(-) diff --git a/js/net/src/announced.test.ts b/js/net/src/announced.test.ts index 5576df214c..dbe3bd1f14 100644 --- a/js/net/src/announced.test.ts +++ b/js/net/src/announced.test.ts @@ -1,6 +1,6 @@ import { expect, spyOn, test } from "bun:test"; import * as Announce from "./announced.ts"; -import { resetNoDiscoveryWarnings } from "./announced.ts"; +import { resetNoDiscoveryWarnings, WARNED_MAX } from "./announced.ts"; import * as Path from "./path.ts"; const p = (s: string) => Path.from(s); @@ -75,14 +75,16 @@ async function countWarnings(fn: () => Announce.Broadcast[]): Promise { test("the no-discovery warning is once per relay, ignoring the auth token", async () => { resetNoDiscoveryWarnings(); - // Same relay, different tokens and different watched paths: still one relay. + // One relay is an origin and a path. The first two differ only by token and by what they + // watch, so they share a warning; the last two are each a relay of their own. const warnings = await countWarnings(() => [ new Announce.Broadcast({ connection: noDiscovery("https://relay.example/anon?jwt=a"), path: p("one") }), new Announce.Broadcast({ connection: noDiscovery("https://relay.example/anon?jwt=b"), path: p("two") }), + new Announce.Broadcast({ connection: noDiscovery("https://relay.example/other"), path: p("one") }), new Announce.Broadcast({ connection: noDiscovery("https://other.example/anon"), path: p("one") }), ]); - expect(warnings).toBe(2); + expect(warnings).toBe(3); }); test("the no-discovery warning cache is bounded", async () => { @@ -93,7 +95,8 @@ test("the no-discovery warning cache is bounded", async () => { const first = "https://relay0.example/anon"; await countWarnings(() => Array.from( - { length: 200 }, + // One past the cap is all it takes to push the first entry out. + { length: WARNED_MAX + 1 }, (_, i) => new Announce.Broadcast({ connection: noDiscovery(`https://relay${i}.example/anon`), path: p("x") }), ), diff --git a/js/net/src/announced.ts b/js/net/src/announced.ts index 14b9be6918..f85b89781f 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -134,10 +134,16 @@ export class Consumer { // several handles can share one connection. const warnedNoDiscovery = new Set(); -// Enough that a real deployment never evicts, small enough that the cache can't grow into a -// leak. Past it the oldest relay is forgotten and may warn a second time, which is the right -// thing to give up: this exists to keep the log readable, not to guarantee exactly-once. -const WARNED_MAX = 64; +/** + * How many relays the no-discovery warning remembers. + * + * Enough that a real deployment never evicts, small enough that the cache can't grow into a + * leak. Past it the oldest relay is forgotten and may warn a second time, which is the right + * thing to give up: this exists to keep the log readable, not to guarantee exactly-once. + * + * @internal + */ +export const WARNED_MAX = 64; /** Warn that `url`'s relay lacks discovery, at most once per relay. */ function warnNoDiscovery(url: URL): void { diff --git a/js/net/src/connection/reload.test.ts b/js/net/src/connection/reload.test.ts index 8191105104..262954d462 100644 --- a/js/net/src/connection/reload.test.ts +++ b/js/net/src/connection/reload.test.ts @@ -290,22 +290,38 @@ test("closing a live connection reports disconnected", async () => { expect(reload.status.peek()).toBe("disconnected"); await reload.closed; } finally { + // Also in `finally`, so a failed assertion doesn't leave the session leased. + reload.close(); globalThis.WebTransport = original; } }); -test("a supplied transport is refused out loud", () => { +test("a supplied transport is refused out loud, and not used", async () => { + const original = globalThis.WebTransport; + const url = new URL("https://example.com/supplied"); const warn = spyOn(console, "warn").mockImplementation(() => {}); - const pair = createMockTransportPair(Lite.ALPN_06_WIP); - // Silently dialing past it would hand back a session the caller never asked for. - const reload = new Reload({ enabled: false, transport: pair.client }); + // Counts the dials, so "warned but used it anyway" and "warned and dialed nothing" are + // both distinguishable from the documented behavior. + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; + + const supplied = createMockTransportPair(Lite.ALPN_06_WIP); + let reload: Reload | undefined; try { + reload = new Reload({ url, websocket: { enabled: false }, transport: supplied.client }); + expect(warn.mock.calls.length).toBe(1); + expect(String(warn.mock.calls[0]?.[0])).toContain("transport is ignored"); + + // It dials its own rather than silently handing back the supplied session. + await settle(); + expect(pending.connects()).toBe(1); } finally { - reload.close(); + reload?.close(); warn.mockRestore(); - pair.client.close(); + supplied.client.close(); + globalThis.WebTransport = original; } });