feat(js/net): share one session per relay URL

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:

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<void>(() => {}),
+		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<number> {
+	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);
+});

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<Established>();
+// 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<string>();
+
+/**
+ * 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<void> {
+		return this.#signals.closed;
+	}
+
 	/** Closes the handle and the broadcast it currently holds. Idempotent. */
 	close() {
 		this.#signals.close();

js/net/src/connection/connect.test.ts

@@ -1,30 +1,24 @@
-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");
 
 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<void>(() => {});
-		closed = new Promise<void>(() => {});
-
-		constructor() {
-			connects++;
-		}
-
-		close() {}
-	}
-
-	globalThis.WebTransport = CountingWebTransport as unknown as typeof WebTransport;
+	const pending = createPendingTransports();
+	globalThis.WebTransport = pending.transport;
 
 	try {
 		const controller = new AbortController();
@@ -36,26 +30,16 @@ 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;
 	}
 });
 
 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<void>(() => {});
-		closed = new Promise<void>(() => {});
-
-		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;
 	}