refactor(net)!: watch demand through demand() and withdraw a fetch with its last caller

doc/lib/js/net.md

@@ -131,3 +131,13 @@ Examples in
 [`js/net/examples/`](https://github.com/moq-dev/moq/tree/main/js/net/examples).
 Runs in the browser and, over WebSocket, in Node, Bun, and Deno; see
 [server-side](/lib/js/#server-side).
+
+## Subscriber demand
+
+Call `demand()` on a track producer or pending track request, a group producer,
+or a broadcast producer. The returned `Track.Demand`, `Group.Demand`, or
+`Broadcast.Demand` exposes a read-only `used` signal, `unused()`, and `closed`.
+A broadcast watches subscribers to its tracks; holding a broadcast consumer
+alone is not demand. A group's demand counts its mirror readers, preserving
+fetch coalescing: the shared download becomes unused only after every reader
+leaves. Demand handles cannot write or close the producer.

doc/lib/rs/moq-net.md

@@ -178,3 +178,17 @@ subscription's upstream request.
 
 Inside a group, `set_frames(...)` applies the same range syntax to frame
 indices.
+
+## Subscriber demand
+
+Track producers, pending track requests, and dynamic track handlers return a
+`track::Demand` from `demand()`. Group producers and pending group requests
+return `group::Demand`; broadcast producers return `broadcast::Demand`.
+Watch `used()` / `unused()` (or `poll_used` / `poll_unused`) on these handles,
+or take a snapshot with `is_used()`. Demand handles cannot write or abort and
+do not keep a producer or request alive. A pending group request counts every
+caller sharing its fetch, so one caller leaving does not cancel the others.
+The last caller to leave withdraws the fetch, so a later fetch of that group
+starts a fresh request; a handler that sees the request unused just drops it.
+Use the producer's `abort_unused` to atomically end an idle track; observing
+unused demand alone does not protect against a new subscriber arriving.

doc/setup/upgrade.md

@@ -31,12 +31,22 @@ These land with the next breaking release, not the 2026-09-23 train.
   are `publish_flate_snapshot` / `publish_flate_stream`, taking
   `MoqFlateConfig` and returning `MoqFlateSnapshotProducer` /
   `MoqFlateStreamProducer`. The C `moq_publish_binary_*` calls are unchanged.
-- **Track demand is read through `demand()`.** In Rust, `track::Producer`'s
-  `is_used`, `used`, `unused`, and `poll_unused` are `producer.demand().X`.
+- **Demand is read through `demand()`.** In Rust, `track::Producer`'s
+  `is_used`, `used`, `unused`, and `poll_unused` are `producer.demand().X`,
+  and so are `group::Producer`'s `used` and `unused`.
+  `track::Request::poll_unused`, `track::Dynamic::poll_unused`, and
+  `group::Request::poll_unused` are `demand().poll_unused`, which returns
+  `Poll<Result<()>>` instead of `Poll<()>`: an `Err` means the request closed,
+  so treat it as unused too. A `group::Request` no longer needs polling to be
+  withdrawn; the last `fetch_group` caller leaving does it, so a handler that
+  sees it unused just drops it. A fetch arriving after that queues a fresh
+  request instead of joining the abandoned one, so a handler that keeps serving
+  without watching `demand()` may see its `accept` return `Error::Duplicate`.
   The moq-json snapshot and moq-flate `is_used()` is `demand().is_used()`.
   In TypeScript, `Track.Producer`'s `used` and `unused()` are
-  `producer.demand().used` and `.unused()`, and `Allocator.reserve` takes
-  `producer.demand()`, replacing the `Bandwidth.Demand` interface.
+  `producer.demand().used` and `.unused()`, as are `Group.Producer`'s, and
+  `Allocator.reserve` takes `producer.demand()`, replacing the
+  `Bandwidth.Demand` interface.
 - **The `"auto"` delay is measured, not derived from RTT** (#4162). It is sized
   from how late frames arrive (see [audio jitter](/concept/audio-jitter)) in
   `@moq/watch` and `moq play`, which now defaults `--delay` to `auto` instead of

js/flate/src/stream/stream.test.ts

@@ -121,8 +121,8 @@ test("a second group is reported while the first is still open", async () => {
 
 	// Both mirrors are released. The read that lost the race would otherwise stay registered on the
 	// first group, keeping this consumer's subscription reachable after the caller drops it.
-	expect(first.used.peek()).toBe(false);
-	expect(second.used.peek()).toBe(false);
+	expect(first.demand().used.peek()).toBe(false);
+	expect(second.demand().used.peek()).toBe(false);
 
 	// Sticky: a later read must not report the rest of the first group as a whole log.
 	first.writeFrame({ payload: payloads(3)[2], timestamp: Time.Timestamp.now() });

js/json/src/stream/stream.test.ts

@@ -89,8 +89,8 @@ test("a second group is reported while the first is still open", async () => {
 
 	// Both mirrors are released. The read that lost the race would otherwise stay registered on the
 	// first group, keeping this consumer's subscription reachable after the caller drops it.
-	expect(first.used.peek()).toBe(false);
-	expect(second.used.peek()).toBe(false);
+	expect(first.demand().used.peek()).toBe(false);
+	expect(second.demand().used.peek()).toBe(false);
 
 	// Sticky: a later read must not report the rest of the first group as a whole log.
 	first.writeFrame({ payload: encode({ n: 2 }), timestamp: Time.Timestamp.now() });

js/net/bench/broadcasts.ts

@@ -1,4 +1,6 @@
-/** Sweep route and observer counts for a single touched-path route update. */
+/** Sweep observers against routes for one route update, and against tracks for one demand edge. */
+
+import { Producer as BroadcastProducer } from "../src/broadcast.ts";
 import { Producer } from "../src/origin.ts";
 import * as Path from "../src/path.ts";
 
@@ -49,3 +51,37 @@ for (const scope of ["unscoped", "distinct-scopes", "producer-scopes", "rooted-p
 	}
 }
 if (checksum === 0) throw new Error("benchmark did no work");
+
+// Sweep both axes while only one track's demand changes.
+const demandUpdates = 1024;
+console.log("tracks,observers,demand_update_us");
+for (const trackCount of [8, 32, 128]) {
+	for (const observerCount of [1, 8, 32]) {
+		const broadcast = new BroadcastProducer();
+		const tracks = Array.from({ length: trackCount }, (_, index) => broadcast.createTrack(`track-${index}`));
+		const demand = broadcast.demand();
+		let notifications = 0;
+		const disposes = Array.from({ length: observerCount }, () =>
+			demand.used.subscribe(() => {
+				notifications++;
+			}),
+		);
+		const start = performance.now();
+		for (let index = 0; index < demandUpdates; index++) {
+			const subscriber = tracks[0].subscribe();
+			await Promise.resolve();
+			await Promise.resolve();
+			if (!demand.used.peek()) throw new Error("missing demand");
+			subscriber.close();
+			await demand.unused();
+			await Promise.resolve();
+		}
+		if (notifications !== demandUpdates * 2 * observerCount) throw new Error(`lost demand edge: ${notifications}`);
+		console.log(
+			`${trackCount},${observerCount},${(((performance.now() - start) * 1000) / demandUpdates).toFixed(1)}`,
+		);
+		for (const dispose of disposes) dispose();
+		for (const track of tracks) track.close();
+		broadcast.close();
+	}
+}

js/net/src/broadcast.test.ts

@@ -503,3 +503,45 @@ test("aborting a fetch rejects with the signal's reason", async () => {
 
 	broadcast.close();
 });
+
+test("broadcast demand watches static and pending tracks rather than broadcast handles", async () => {
+	const broadcast = new BroadcastProducer();
+	const demand = broadcast.demand();
+	const consumer = broadcast.consume();
+	expect(demand.used.peek()).toBe(false);
+	const video = broadcast.createTrack("video");
+	const first = video.subscribe();
+	await Promise.resolve();
+	expect(demand.used.peek()).toBe(true);
+	const second = consumer.track("audio").subscribe();
+	const request = await wireOf(broadcast).requested();
+	if (!request) throw new Error("expected request");
+	const requested = request.demand();
+	expect(requested.used.peek()).toBe(true);
+	first.close();
+	await Promise.resolve();
+	expect(demand.used.peek()).toBe(true);
+	second.close();
+	await requested.unused();
+	await demand.unused();
+	expect(demand.used.peek()).toBe(false);
+	request.reject();
+	video.close();
+	broadcast.close();
+	expect(await demand.closed).toBeNull();
+	consumer.close();
+});
+
+test("broadcast closure clears active demand and subscriptions", async () => {
+	const broadcast = new BroadcastProducer();
+	const track = broadcast.createTrack("video");
+	const subscriber = track.subscribe();
+	const demand = broadcast.demand();
+	await Promise.resolve();
+	expect(demand.used.peek()).toBe(true);
+	broadcast.close();
+	expect(demand.used.peek()).toBe(false);
+	await demand.unused();
+	subscriber.close();
+	track.close();
+});

js/net/src/broadcast.ts

@@ -3,7 +3,7 @@
  *
  * @module
  */
-import { type GetPromise, Once, Signal } from "@moq/signals";
+import { type Dispose, type GetPromise, type Getter, Once, Signal } from "@moq/signals";
 import { NotFound } from "./error.ts";
 import type { Consumer as GroupConsumer } from "./group.ts";
 import { Route } from "./hop.ts";
@@ -34,6 +34,42 @@ class BroadcastState {
 	// Live consumer handles sharing this state (see {@link Consumer.clone}). The broadcast
 	// closes once the last one closes, so a shared consumer can be handed to several callers.
 	consumers = 0;
+	used = new Signal(false);
+	active = 0;
+	watched = new WeakSet<track.Producer>();
+	demandCleanup = new Set<Dispose>();
+}
+
+// Each track updates the aggregate on an edge, so a demand change touches only its track.
+function watchDemand(state: BroadcastState, producer: track.Producer): void {
+	if (state.watched.has(producer)) return;
+	state.watched.add(producer);
+	const demand = producer.demand();
+	let active = false;
+	const update = () => {
+		const used = state.closed.peek() === undefined && demand.closed.peek() === undefined && demand.used.peek();
+		if (active === used) return;
+		state.active += used ? 1 : -1;
+		active = used;
+		state.used.set(state.active > 0);
+	};
+	const disposeUsed = demand.used.subscribe(update);
+	const disposeClosed = demand.closed.subscribe(() => {
+		if (demand.closed.peek() === undefined) return;
+		cleanup();
+	});
+	const cleanup = () => {
+		disposeUsed();
+		disposeClosed();
+		if (active) {
+			state.active--;
+			active = false;
+			state.used.set(state.active > 0);
+		}
+		state.demandCleanup.delete(cleanup);
+	};
+	state.demandCleanup.add(cleanup);
+	update();
 }
 
 function dequeueRequest(state: BroadcastState): track.Request | undefined {
@@ -51,6 +87,7 @@ function dequeueRequest(state: BroadcastState): track.Request | undefined {
 function closeState(state: BroadcastState) {
 	if (state.closed.peek() !== undefined) return;
 	state.closed.set(null);
+	for (const cleanup of state.demandCleanup) cleanup();
 	for (const request of state.pending) request.reject();
 	state.requested.mutate((requests) => {
 		requests.length = 0;
@@ -81,6 +118,7 @@ function subscribe(
 	}
 
 	const producer = new track.Producer(name);
+	watchDemand(state, producer);
 	const subscriber = producer.subscribe(options);
 
 	if (register) {
@@ -110,6 +148,7 @@ async function resolveTrackInfo(state: BroadcastState, name: string): Promise<tr
 	}
 
 	const producer = new track.Producer(name);
+	watchDemand(state, producer);
 	state.requested.mutate((requested) => {
 		requested.push(hooks.makeRequest({ name, producer, sequences: state.sequences, pending: state.pending }));
 	});
@@ -154,13 +193,44 @@ async function fetchGroup(
 	}
 }
 
+let makeDemand: (state: BroadcastState) => Demand;
+
+/** A watch-only view of demand for any track in a broadcast. */
+export class Demand {
+	#state: BroadcastState;
+	private constructor(state: BroadcastState) {
+		this.#state = state;
+	}
+	static {
+		makeDemand = (state) => new Demand(state);
+	}
+
+	/** Whether any track currently has subscribers. */
+	get used(): Getter<boolean> {
+		return this.#state.used;
+	}
+
+	/** Wait until every track becomes unused or the broadcast closes. */
+	async unused(): Promise<void> {
+		while (this.#state.used.peek() && this.#state.closed.peek() === undefined) {
+			await Signal.race(this.#state.used, this.#state.closed);
+		}
+	}
+
+	/** The broadcast's clean close. */
+	get closed(): GetPromise<null> {
+		return this.#state.closed;
+	}
+}
+
 /**
  * The write side of a broadcast.
  *
  * @public
  */
 export class Producer {
 	#state = new BroadcastState();
+	#demand = makeDemand(this.#state);
 	#announcer?: Announcer;
 	#path = Path.empty();
 
@@ -186,6 +256,11 @@ export class Producer {
 		return this.#state.closed;
 	}
 
+	/** Watch demand for the broadcast's tracks. */
+	demand(): Demand {
+		return this.#demand;
+	}
+
 	/** A read handle for this broadcast, named by the path the origin created it at. */
 	consume(): Consumer {
 		return makeConsumer({ state: this.#state, path: this.#path });
@@ -213,6 +288,7 @@ export class Producer {
 			throw new Error(`duplicate track: ${track.name}`);
 		}
 
+		watchDemand(this.#state, track);
 		this.#state.tracks.set(track.name, track);
 
 		void track.closed.then(() => {

js/net/src/group.test.ts

@@ -9,21 +9,21 @@ test("used reflects mirror demand and unused resolves when the last reader leave
 	const producer = new Producer(0);
 
 	// No mirror readers: no demand.
-	expect(producer.used.peek()).toBe(false);
+	expect(producer.demand().used.peek()).toBe(false);
 
 	const a = producer.mirror();
 	const b = producer.mirror();
-	expect(producer.used.peek()).toBe(true);
+	expect(producer.demand().used.peek()).toBe(true);
 
 	// Closing one of two keeps demand, so unused() stays pending.
 	a.close();
-	expect(producer.used.peek()).toBe(true);
+	expect(producer.demand().used.peek()).toBe(true);
 
 	// Closing the last reader drops demand; unused() resolves. Fetch coalescing awaits this to
 	// cancel a download that everyone has abandoned (a group may never end on its own).
 	b.close();
-	await producer.unused();
-	expect(producer.used.peek()).toBe(false);
+	await producer.demand().unused();
+	expect(producer.demand().used.peek()).toBe(false);
 });
 
 function pair(sequence: number) {