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.test.ts b/js/net/src/announced.test.ts index 1cd3dc31fc..dbe3bd1f14 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, WARNED_MAX } from "./announced.ts"; import * as Path from "./path.ts"; const p = (s: string) => Path.from(s); @@ -44,3 +45,66 @@ 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(); + + // 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(3); +}); + +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( + // 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") }), + ), + ); + + 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 b9eaf6d4b3..f85b89781f 100644 --- a/js/net/src/announced.ts +++ b/js/net/src/announced.ts @@ -129,9 +129,43 @@ 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 relay rather than by session, since +// several handles can share one connection. +const warnedNoDiscovery = new Set(); + +/** + * 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 { + // 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(); +} /** * What to watch, for {@link Broadcast}. @@ -218,10 +252,7 @@ export class Broadcast { // Without discovery no announcement ever arrives, so waiting would hang forever. if (!conn.discovery) { - if (!warnedNoDiscovery.has(conn)) { - warnedNoDiscovery.add(conn); - console.warn("relay does not support broadcast discovery; consuming without waiting."); - } + warnNoDiscovery(conn.url); const blind = conn.consume(path); effect.cleanup(() => blind.close()); @@ -286,6 +317,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 525570da98..2e3e4aaef9 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 { createMockTransportPair, createPendingTransports } from "../mock.ts"; import { connect } from "./connect.ts"; +import { resetPool } from "./pool.ts"; const url = new URL("https://example.com/test"); @@ -9,22 +10,15 @@ 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; - - 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(); @@ -36,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; } @@ -44,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(); @@ -67,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 ca7efdcb3b..45411783e7 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,34 @@ 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 {@link ConnectProps.reload}. + * + * The delays carry jitter, so a fleet of tabs knocked offline together doesn't reconnect in + * lockstep. Every failure is retried; {@link ReloadDelay.timeout} is what stops the 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: 5000). */ + max: DOMHighResTimeStamp; + + /** + * Maximum total time in milliseconds to spend retrying before giving up (default: + * 10000). 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 +78,13 @@ 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; + + /** @internal Backoff settings, ignored when {@link ConnectProps.reload} carries its own. */ + delay?: ReloadDelay; } // Relays that don't implement broadcast discovery (SUBSCRIBE_NAMESPACE), so `announced()` would @@ -103,12 +168,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 +209,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 +223,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..a44d126eb2 --- /dev/null +++ b/js/net/src/connection/pool.test.ts @@ -0,0 +1,340 @@ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { ALPN_05 } from "../lite/version.ts"; +import { + createMockTransportPair, + createPendingTransports, + type MockTransport, + type PendingTransports, +} 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. 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 * 15)); +} + +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(): PendingTransports { + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; + return pending; +} + +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..cf99dade71 --- /dev/null +++ b/js/net/src/connection/pool.ts @@ -0,0 +1,374 @@ +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); + void watch.closed.then(() => this.#broadcasts.delete(watch)); + return watch; + } + + async stats(): Promise { + 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; + } + + 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..262954d462 100644 --- a/js/net/src/connection/reload.test.ts +++ b/js/net/src/connection/reload.test.ts @@ -1,74 +1,166 @@ -import { 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 } from "../mock.ts"; +import { createMockTransportPair, createPendingTransports } 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; + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; + const reload = new Reload({ + enabled: true, + url: new URL("https://example.com/broadcast"), + websocket: { enabled: false }, + }); - class PendingWebTransport { - ready = new Promise(() => {}); - closed = new Promise(() => {}); + try { + await settle(); + expect(pending.connects()).toBe(1); - constructor() { - connects++; - } + reload.url.set(new URL("https://example.com/broadcast")); + await settle(); + expect(pending.connects()).toBe(1); - close() {} + reload.url.set(new URL("https://example.com/other")); + await settle(); + expect(pending.connects()).toBe(2); + } finally { + reload.close(); + globalThis.WebTransport = original; } +}); - globalThis.WebTransport = PendingWebTransport as unknown as typeof WebTransport; +test("aborting the signal stops the loop", async () => { + const original = globalThis.WebTransport; + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; + const controller = new AbortController(); const reload = new Reload({ - enabled: true, - url: new URL("https://example.com/broadcast"), + url: new URL("https://example.com/signal"), websocket: { enabled: false }, + signal: controller.signal, }); try { await settle(); - expect(connects).toBe(1); + expect(reload.status.peek()).toBe("connecting"); - reload.url.set(new URL("https://example.com/broadcast")); + controller.abort(); await settle(); - expect(connects).toBe(1); + expect(pending.closes()).toBe(1); + await reload.closed; + } finally { + reload.close(); + globalThis.WebTransport = original; + } +}); - reload.url.set(new URL("https://example.com/other")); +test("connecting is the default", async () => { + const original = globalThis.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(2); + expect(pending.connects()).toBe(1); } finally { reload.close(); globalThis.WebTransport = original; } }); -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("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("closing mid-connect aborts the pending attempt", async () => { +test("reload: false gives up after one session", async () => { const original = globalThis.WebTransport; - let closes = 0; + 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 }); - class PendingWebTransport { - ready = new Promise(() => {}); - closed = new Promise(() => {}); + 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); - close() { - closes++; - } + // 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; } +}); - globalThis.WebTransport = PendingWebTransport as unknown as typeof WebTransport; +test("closing mid-connect aborts the pending attempt", async () => { + const original = globalThis.WebTransport; + const pending = createPendingTransports(); + globalThis.WebTransport = pending.transport; const reload = new Reload({ enabled: true, url: new URL("https://example.com/broadcast"), @@ -77,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; } @@ -174,3 +266,62 @@ 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 { + // 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, and not used", async () => { + const original = globalThis.WebTransport; + const url = new URL("https://example.com/supplied"); + const warn = spyOn(console, "warn").mockImplementation(() => {}); + + // 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(); + warn.mockRestore(); + supplied.client.close(); + globalThis.WebTransport = original; + } +}); diff --git a/js/net/src/connection/reload.ts b/js/net/src/connection/reload.ts index e8b0bd9969..55528526ba 100644 --- a/js/net/src/connection/reload.ts +++ b/js/net/src/connection/reload.ts @@ -3,44 +3,22 @@ import * as Announce from "../announced.ts"; import { error } from "../error.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. - * - * The delays carry jitter, so a fleet of tabs knocked offline together doesn't reconnect in - * lockstep. Every failure is retried; {@link ReloadDelay.timeout} is what stops the 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: 5000). */ - max: DOMHighResTimeStamp; - - /** - * Maximum total time in milliseconds to spend retrying before giving up (default: - * 10000). Resets after each successful connection. Set to 0 for unlimited retries. - */ - timeout?: DOMHighResTimeStamp; -}; +/** The {@link ConnectProps} a {@link Reload} accepts. */ +export type ReloadProps = ConnectProps; -/** 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. */ - delay?: ReloadDelay; -}; +/** The backoff used when nothing else is asked for. */ +const DEFAULT_DELAY: ReloadDelay = { initial: 1000, multiplier: 2, max: 5000 }; /** * How long to keep retrying before giving up, when {@link ReloadDelay.timeout} is unset. @@ -53,12 +31,17 @@ const DEFAULT_TIMEOUT = 10000; /** 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. */ @@ -90,6 +73,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(); @@ -119,12 +108,21 @@ 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: 5000 }; + 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; + // 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; @@ -152,6 +150,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 { @@ -181,6 +186,7 @@ export class Reload { websocket: this.websocket, webtransport: this.webtransport, discovery: this.discovery, + pool: this.pool, signal, }); @@ -221,6 +227,12 @@ 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) { + this.#finish(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 @@ -238,7 +250,7 @@ export class Reload { if (now >= this.#deadline) { console.warn("reconnect timed out"); // A graceful close has no error, so report the timeout itself. - this.#closedReject(cause === undefined ? new Error("reconnect timed out") : error(cause)); + this.#finish(cause ?? new Error("reconnect timed out")); return; } @@ -339,7 +351,22 @@ 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 { + // Closing the scope also restores `established` and `status`, since the connect run set + // them through `effect.set`. this.#signals.close(); - this.#closedResolve(); + + if (cause === undefined) this.#closedResolve(); + else this.#closedReject(error(cause)); } } 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..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"; @@ -68,3 +69,48 @@ 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); }); + +// 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. +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 }); + 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"), 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/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); }); }); } 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}. */