feat(js/net)!: route publish and consume through Origins, share one connection per relay

demo/web/src/index.ts

@@ -79,7 +79,7 @@ const metaSignal = new Signals.Signal<unknown>(undefined);
 const relayUrl = new Signals.Signal<URL | undefined>(new URL(RELAY_URL));
 
 // Discovery connection (the tiles each open their own connection internally).
-const connection = new Net.Connection.Reload({ url: relayUrl, enabled: true });
+const connection = new Net.Connection.Shared({ url: relayUrl });
 
 // ---------------------------------------------------------------------------
 // Per-broadcast tile (a <moq-watch-ui> in the left column)
@@ -382,8 +382,8 @@ ui.run((effect) => {
 	section.hidden = false;
 
 	// Report the transport negotiated by the live connection.
-	const conn = effect.get(connection.established);
-	$("network-transport").textContent = conn ? (conn.transport === "websocket" ? "WebSocket" : "WebTransport") : "";
+	const transport = effect.get(connection.transport);
+	$("network-transport").textContent = transport ? (transport === "websocket" ? "WebSocket" : "WebTransport") : "";
 
 	const video = effect.get(watch.video.out.stats);
 	const audio = effect.get(watch.audio.out.stats);

demo/web/src/publish.ts

@@ -363,8 +363,8 @@ ui.run((effect) => {
 
 // Report the transport negotiated by the live connection.
 ui.run((effect) => {
-	const conn = effect.get(publish.connection.established);
-	$("network-transport").textContent = conn ? (conn.transport === "websocket" ? "WebSocket" : "WebTransport") : "";
+	const transport = effect.get(publish.connection.transport);
+	$("network-transport").textContent = transport ? (transport === "websocket" ? "WebSocket" : "WebTransport") : "";
 });
 
 // Audio: the resolved audio config (codec / sample rate / channels / bitrate).

demo/web/src/stats.ts

@@ -84,18 +84,18 @@ const selectedNode = new Signals.Signal<string | undefined>(undefined);
 
 // The relay URL, editable at runtime (see the input binding below).
 const relayUrl = new Signals.Signal<URL | undefined>(new URL(RELAY_URL));
-const connection = new Net.Connection.Reload({ url: relayUrl, enabled: true });
+const connection = new Net.Connection.Shared({ url: relayUrl });
 
 // ---- Discover nodes + subscribe to each -----------------------------------
 
 const discovery = new Signals.Effect();
 discovery.run((effect) => {
-	const conn = effect.get(connection.established);
+	const origin = effect.get(connection.origin);
 	nodeStats.set({});
-	if (!conn) return;
+	if (!origin) return;
 
 	const prefix = Net.Path.from(STATS_PREFIX);
-	const announced = conn.announced(prefix);
+	const announced = origin.announced(prefix);
 	effect.cleanup(() => announced.close());
 
 	// One sub-effect per node so we can tear a node's subscriptions down when it
@@ -117,7 +117,7 @@ discovery.run((effect) => {
 				if (subs.has(node)) continue;
 				const ne = new Signals.Effect();
 				subs.set(node, ne);
-				subscribeNode(ne, conn, path, node);
+				subscribeNode(ne, origin, path, node);
 			} else {
 				subs.get(node)?.close();
 				subs.delete(node);
@@ -129,7 +129,7 @@ discovery.run((effect) => {
 	});
 });
 
-function subscribeNode(effect: Signals.Effect, conn: Net.Connection.Established, path: Net.Path.Valid, node: string) {
+function subscribeNode(effect: Signals.Effect, origin: Net.Origin.Table, path: Net.Path.Valid, node: string) {
 	nodeStats.mutate((s) => {
 		s[node] = {
 			egress: {},
@@ -138,8 +138,11 @@ function subscribeNode(effect: Signals.Effect, conn: Net.Connection.Established,
 		};
 	});
 
-	const consumer = conn.consume(path);
-	effect.cleanup(() => consumer.close());
+	// The path was just announced, so the request resolves from the table immediately.
+	const request = origin.request(path);
+	effect.cleanup(() => request.close());
+	const consumer = request.active.peek();
+	if (!consumer) return;
 
 	const sub = <K extends keyof NodeStats>(trackName: string, key: K) => {
 		const track = consumer.subscribe(trackName);
@@ -256,7 +259,7 @@ sampler.run((effect) => {
 	// Only sample while connected; the interval restarts on reconnect. Drop the
 	// rolling history when disconnected so a reconnect doesn't splice new
 	// samples onto stale ones across the downtime gap.
-	if (!effect.get(connection.established)) {
+	if (effect.get(connection.status) !== "connected") {
 		history.clear();
 		clusterMembership = "";
 		clock.update((n) => n + 1);

doc/lib/js/@moq/net.md

@@ -44,6 +44,37 @@ See [`js/net/examples/discovery.ts`](https://github.com/moq-dev/moq/blob/main/js
 
 ## Core Concepts
 
+### Origins
+
+A routing table of broadcasts by path, independent of any connection. Publishing goes through an origin, not a session: create one, publish broadcasts into it, and hand it to a connection via the `publish` option. The connection announces and serves the table for as long as the session lasts, and a reconnect re-announces whatever is still published.
+
+```ts
+const origin = new Moq.Origin.Producer();
+await Moq.Connection.connect(url, { publish: origin.consume() });
+
+const broadcast = origin.publish(Moq.Path.from("my-broadcast"));
+broadcast.createTrack("chat");
+```
+
+Closing the connection unannounces the broadcasts but does not close them; they stay in the origin for the next session. Closing a broadcast's producer unpublishes just that path.
+
+The other direction works the same way: pass an origin as the `subscribe` option and everything the peer announces appears in its table, gone when the session dies.
+
+`origin.request(path)` is how you consume by path, whether or not anything announced it:
+
+```ts
+// One origin can back both directions of the same connection.
+await Moq.Connection.connect(url, { publish: origin.consume(), subscribe: origin });
+
+const request = origin.request(Moq.Path.from("some-broadcast"));
+const broadcast = request.active.peek(); // or effect.get(request.active)
+request.close(); // when done
+```
+
+`request.active` follows whatever the table routes: a local publish first, so a page that publishes and watches the same broadcast reads its own copy with no round trip, then any session's announcement, swapping when a republish takes the path. When nothing routes it (a relay without discovery, or subscribing before the publisher exists on purpose), an attached session answers it blind instead. Either way the request stands across reconnects, so hold it for as long as you want the path, and close it when you don't.
+
+Use `origin.announced(prefix)` to discover what is available rather than asking for a path you already know.
+
 ### Broadcasts
 
 A collection of related tracks.

doc/lib/js/@moq/publish.md

@@ -116,9 +116,14 @@ The overlay has no `simulcast` control; enable it via the attribute on the neste
 
 ```typescript
 import * as Publish from "@moq/publish";
+import * as Moq from "@moq/net";
+
+// A connection shared with every other component pointed at the same URL. Its `origin` is
+// where the broadcasts live, so they survive a reconnect.
+const connection = new Moq.Connection.Shared({ url: new URL("https://relay.example.com/anon") });
 
 const broadcast = new Publish.Broadcast({
-    connection,
+    origin: connection.origin,
     enabled: true,
     name: "alice.hang",
     // Publish two video renditions: video/hd plus a lower-resolution video/sd.

doc/lib/js/@moq/watch.md

@@ -94,9 +94,14 @@ from the broadcast name extension by default. `room/alice.hang` uses hang,
 
 ```typescript
 import * as Watch from "@moq/watch";
+import * as Moq from "@moq/net";
+
+// A connection shared with every other component pointed at the same URL. Its `origin` is
+// where the broadcasts live, so the handle spans reconnects.
+const connection = new Moq.Connection.Shared({ url: new URL("https://relay.example.com/anon") });
 
 const broadcast = new Watch.Broadcast({
-    connection,
+    origin: connection.origin,
     enabled: true,
     name: "alice.hang",
     catalogFormat: "msf",
@@ -109,7 +114,7 @@ broadcast.catalogFormat.set("msf");
 ### Manual catalogs
 
 Use `catalog-format="manual"` (or `catalogFormat: "manual"`) to skip the catalog
-track entirely and supply a `Catalog.Root` directly. The connection and
+track entirely and supply a `Catalog.Root` directly. The origin and
 broadcast name are still required, since they're used to subscribe to the media
 tracks named by the catalog. Update the catalog at any time by writing to
 the signal:
@@ -118,7 +123,7 @@ the signal:
 import * as Watch from "@moq/watch";
 
 const broadcast = new Watch.Broadcast({
-    connection,
+    origin: connection.origin,
     enabled: true,
     name: "alice.hang",
     catalogFormat: "manual",
@@ -268,13 +273,16 @@ The `<moq-watch-ui>` element automatically discovers the nested `<moq-watch>` an
 
 ```typescript
 import * as Watch from "@moq/watch";
+import * as Moq from "@moq/net";
 import { Signal } from "@moq/signals";
 
+const connection = new Moq.Connection.Shared({ url: new URL("https://relay.example.com/anon") });
+
 // Inputs are read-only on the component, so keep a handle to anything you want to change later.
 const reload = new Signal(true);
 
 const broadcast = new Watch.Broadcast({
-    connection,
+    origin: connection.origin,
     enabled: true,
     name: "alice.hang",
     reload,

js/clock/src/main.ts

@@ -72,12 +72,13 @@ ENVIRONMENT VARIABLES:
 }
 
 async function publish(config: Config) {
-	const connection = await Moq.Connection.connect(new URL(config.url));
+	// The origin holds what we publish; the connection announces and serves it.
+	const origin = new Moq.Origin.Producer();
+	await Moq.Connection.connect(new URL(config.url), { publish: origin.consume() });
 	console.log("✅ Connected to relay:", config.url);
 
 	// Create a new "broadcast", which is a collection of tracks.
-	const broadcast = new Moq.Broadcast.Producer();
-	connection.publish(Moq.Path.from(config.broadcast), broadcast);
+	const broadcast = origin.publish(Moq.Path.from(config.broadcast));
 
 	console.log("✅ Published broadcast:", config.broadcast);
 

js/moq-boy/src/element.tsx

@@ -25,6 +25,8 @@ export default class MoqBoy extends HTMLElement {
 	static observedAttributes = OBSERVED;
 
 	readonly connection: Moq.Connection.Reload;
+	/** The origin viewer broadcasts are published into, served across reconnects. */
+	readonly origin = new Moq.Origin.Producer();
 	readonly expanded = new Moq.Signals.Signal<string | undefined>(undefined);
 
 	/** Reactive map of active game sessions. Emits on add/remove. */
@@ -42,8 +44,15 @@ export default class MoqBoy extends HTMLElement {
 		super();
 		cleanup.register(this, this.#signals);
 
-		this.connection = new Moq.Connection.Reload({ enabled: this.#enabled });
+		// One origin, both directions: viewer broadcasts are published into it and the
+		// relay's announced games arrive in it, with no risk of echoing either back.
+		this.connection = new Moq.Connection.Reload({
+			enabled: this.#enabled,
+			publish: this.origin.consume(),
+			subscribe: this.origin,
+		});
 		this.#signals.cleanup(() => this.connection.close());
+		this.#signals.cleanup(() => this.origin.close());
 
 		// Discover game sessions via announcements.
 		this.#signals.run(this.#runDiscovery.bind(this));
@@ -113,15 +122,14 @@ export default class MoqBoy extends HTMLElement {
 	}
 
 	#runDiscovery(effect: Moq.Signals.Effect) {
-		const conn = effect.get(this.connection.established);
-		if (!conn) return;
-
 		const base = effect.get(this.#prefix);
 		const gamePrefix = effect.get(this.#gamePrefixOverride) ?? `${base}/game`;
 		const viewerPrefix = effect.get(this.#viewerPrefixOverride) ?? `${base}/viewer`;
 		const prefix = Moq.Path.from(gamePrefix);
 
-		const announced = conn.announced(prefix);
+		// The origin's stream spans reconnects: entries retract when the session dies and
+		// return when the next one re-announces them, so this loop never needs to restart.
+		const announced = this.origin.consume().announced(prefix);
 		effect.cleanup(() => announced.close());
 
 		effect.spawn(async () => {
@@ -138,6 +146,7 @@ export default class MoqBoy extends HTMLElement {
 					const config: GameConfig = {
 						sessionId: id,
 						connection: this.connection,
+						origin: this.origin,
 						expanded: this.expanded,
 						gamePrefix,
 						viewerPrefix,

js/moq-boy/src/game.ts

@@ -18,6 +18,8 @@ export interface GameConfig {
 	sessionId: string;
 	/** MoQ connection to the relay. */
 	connection: Moq.Connection.Reload;
+	/** The origin viewer broadcasts are published into; the connection serves it. */
+	origin: Moq.Origin.Producer;
 	/** Shared signal tracking which game is currently expanded. */
 	expanded: Moq.Signals.Signal<string | undefined>;
 	/** MoQ path prefix for game broadcasts (e.g. "anon/boy/game"). */
@@ -114,7 +116,7 @@ export class Game {
 
 		// Video pipeline.
 		this.broadcast = new Watch.Broadcast({
-			connection: connection.established,
+			origin: config.origin,
 			name: Moq.Path.from(`${gamePrefix}/${sessionId}`),
 			enabled: true,
 		});
@@ -124,6 +126,7 @@ export class Game {
 			broadcast: this.broadcast,
 			target: this.#target,
 			supported: Watch.Video.Decoder.supported,
+			probe: connection.probe,
 		});
 		this.#signals.cleanup(() => this.videoSource.close());
 
@@ -138,7 +141,7 @@ export class Game {
 		const videoJitter = new Moq.Signals.Signal<Moq.Time.Milli | undefined>(undefined);
 		this.sync = new Watch.Sync({
 			latency: this.latency,
-			connection: connection.established,
+			probe: connection.probe,
 			video: videoJitter,
 			audio: this.audioSource.out.jitter,
 		});
@@ -185,7 +188,7 @@ export class Game {
 		this.#signals.run(this.#runStatus.bind(this));
 
 		// Command publishing.
-		this.#signals.run(this.#runCommands.bind(this, connection));
+		this.#signals.run(this.#runCommands.bind(this, connection, config.origin));
 	}
 
 	/** Send a button state update. */
@@ -275,7 +278,9 @@ export class Game {
 		});
 	}
 
-	#runCommands(connection: Moq.Connection.Reload, effect: Moq.Signals.Effect) {
+	#runCommands(connection: Moq.Connection.Reload, origin: Moq.Origin.Producer, effect: Moq.Signals.Effect) {
+		// Publishing goes through the origin, but gate on a live connection anyway: a command
+		// broadcast for a game nobody is connected to is feedback into the void.
 		const conn = effect.get(connection.established);
 		if (!conn) return;
 
@@ -293,8 +298,7 @@ export class Game {
 		const viewerId = Math.random().toString(36).slice(2, 8);
 		this.viewerId.set(viewerId);
 
-		const viewerBroadcast = new Moq.Broadcast.Producer();
-		conn.publish(Moq.Path.from(`${this.#viewerPrefix}/${this.sessionId}/${viewerId}`), viewerBroadcast);
+		const viewerBroadcast = origin.publish(Moq.Path.from(`${this.#viewerPrefix}/${this.sessionId}/${viewerId}`));
 		effect.cleanup(() => {
 			viewerBroadcast.close();
 			this.viewerId.set(undefined);

js/net/examples/publish.ts

@@ -2,17 +2,17 @@ import * as Moq from "@moq/net";
 
 async function main() {
 	const url = new URL("https://cdn.moq.dev/anon");
-	const connection = await Moq.Connection.connect(url);
 
-	// Create a broadcast (a collection of tracks)
-	const broadcast = new Moq.Broadcast.Producer();
+	// The origin holds what we publish; the connection announces and serves it.
+	const origin = new Moq.Origin.Producer();
+	await Moq.Connection.connect(url, { publish: origin.consume() });
+
+	// Create a broadcast (a collection of tracks) at a path on the origin
+	const broadcast = origin.publish(Moq.Path.from("my-broadcast"));
 
 	// Insert the "chat" track up front. A subscriber is served directly from this
 	// track, no requested() round-trip needed. Mirrors the Rust createTrack/insertTrack.
 	void publishTrack(broadcast.createTrack("chat"));
-
-	// Publish the broadcast to the connection
-	connection.publish(Moq.Path.from("my-broadcast"), broadcast);
 	console.log("Published broadcast: my-broadcast");
 
 	// Tracks created on demand (instead of up front) are still supported: handle any