feat(net): restore broadcast demand and document the demand chain

doc/lib/rs/moq-net.md

@@ -22,7 +22,7 @@ What you use it for, beyond what the concept page already describes:
 - **Discovery follows what you watch.** A client session asks the relay for announcements only under the prefixes its subscribe origin's readers watch, with `announced()`, `routed_broadcast`, or `announced_broadcast`, stops asking about a prefix 2s after nothing at or beneath it is watched, and asks again on reconnect. A broadcast resolved through a discovered route keeps its path watched until dropped. `request_broadcast` watches nothing, so a path nobody watches is unroutable. A server session, or a client fed into an origin handle marked `with_link(origin::Link::Peer)` or `Link::Upstream`, asks for everything it may, so a relay discovers everything. An HTTP gateway (the moq-hls `Server`, the moq-rtc WHEP router) watches its whole scope while it lives, so the paths its callers name never open a request each. moq-lite-03/04 sessions, whose announcements mark no initial set, still ask for everything.
 - **One track per name.** Concurrent subscriptions share one request and producer. The application owns sequence numbers: the broadcast keeps nothing of an ended track, so a producer that replaces it starts at 0. A name always means the same content, so the replacement continues past its predecessor's groups with `create_group` at an explicit sequence. Readers of an aborted track resume onto the replacement only when the broadcast announces an epoch; otherwise their subscription ends with the error.
 - **End content by finishing tracks.** A broadcast has no finished state. When the input ends, finish each track; the broadcast stays announced and keeps serving them until you close it or drop its last producer. How long to keep it up for late readers is yours to decide: `moq_mux::Linger` holds a hang broadcast for a fixed window.
-- **Publish only while someone is watching.** `demand()` on a track, group, or broadcast says whether a subscriber is attached, which is how capture and transcode skip work nobody asked for. Holding a broadcast or track consumer is not demand: a track consumer requests nothing until it queries, subscribes, or fetches, and those count until dropped. A shared fetch stays up until its last reader leaves.
+- **Publish only while someone is watching.** `demand()` on a track, group, or broadcast says whether a subscriber is attached, which is how capture and transcode skip work nobody asked for. Holding a broadcast or track consumer is not demand: a track consumer requests nothing until it queries, subscribes, or fetches, and those count until dropped. A shared fetch stays up until its last reader leaves. Demand rolls up: a frame being read keeps its group wanted, a group a subscription or fetch handed out keeps its track wanted until read to its end, and any wanted track keeps its broadcast wanted.
 - **Use the model without a runtime.** Disable default features on `moq-net` to use its `poll_*` APIs and origin driver with caller-supplied time. The default `async` feature adds async helpers, `time::run`, and both wire sessions.
 - **You drive the session, or `moq-tokio` does.** `connect` and `accept` return a session plus a driver that never reads the clock itself. `moq_net::time::run` polls it on tokio or in the browser. `moq-tokio` and `moq-wasm` hide that. A custom transport implements `moq_net::transport::poll`, and `moq_net::transport::memory::pair` links two sessions in one process with no sockets, for tests or in-process links.
 - **A session bounds what its peer can hold.** Each request holds a stream, so the transport's stream credit bounds subscriptions and FETCHes. `session::Limits` (via `Client::with_limits` and `Server::with_limits`) bounds the announcements per session. The default suits a relay mesh, so lower it for untrusted peers. Past it, the session closes with `TOO_MANY_REQUESTS`. Peer-declared lengths are capped before they are buffered.

quest/m1/api-audit/wanted.md

@@ -15,11 +15,12 @@ Decided 2026-10-25 (maintainer), extending the `wanted` shape that
 
 - Yes/no demand returns `bool`; `track::Demand::wanted() -> Option<u64>`
   keeps its meaning.
-- Broadcast demand is deleted: Rust's `broadcast::Demand` went in
-  [#6744](https://github.com/moq-dev/moq/pull/6744), and the `@internal` JS
-  `Broadcast.Demand` (read only by tests and a bench) goes too. The origin
-  closes a superseded producer explicitly instead of a publisher watching
-  demand.
+- Rust's `broadcast::Demand`, deleted in
+  [#6744](https://github.com/moq-dev/moq/pull/6744), is back in the `wanted`
+  shape for publishers that steer on whether anyone reads a broadcast. The
+  `@internal` JS `Broadcast.Demand` (read only by tests and a bench) goes. The
+  origin closes a superseded producer explicitly instead of a publisher
+  watching demand.
 - `group::Demand` already says `wanted`. The JS group demand stays internal.
 - `bandwidth::Producer` gets a `demand()` handle with `wanted() -> bool` and
   `wanted_changed()`, like track.

rs/moq-e2ee/src/tests.rs

@@ -836,10 +836,6 @@ fn grouped_failure_wakes_pending_reads_and_releases_demand() {
 
 	assert!(frame.woken(), "parked frame read was not woken");
 	assert!(track.woken(), "parked group and datagram reads were not woken");
-	assert!(
-		!demand.wanted().is_some(),
-		"track demand stayed live after authentication failure"
-	);
 	assert!(!bad_demand.wanted(), "failed group demand stayed live");
 	assert!(!pair.net.is_closed(), "publisher must stay open");
 
@@ -853,6 +849,11 @@ fn grouped_failure_wakes_pending_reads_and_releases_demand() {
 		!open_demand.wanted(),
 		"sibling group demand stayed live after its woken read"
 	);
+	// The sibling's reader held the track wanted until that read released it.
+	assert!(
+		!demand.wanted().is_some(),
+		"track demand stayed live after authentication failure"
+	);
 	assert_eq!(
 		track
 			.poll(|waiter| pair.consumer.poll_recv_group(waiter))
@@ -873,20 +874,23 @@ fn grouped_failure_wakes_pending_reads_and_releases_demand() {
 }
 
 #[test]
-fn dropping_track_releases_demand_while_group_is_held() {
+fn a_held_group_keeps_demand_after_the_track_drops() {
 	let mut pair = pair("video");
 	let demand = pair.net.demand();
 	let mut open = pair.producer.append_group().unwrap();
 	open.write_frame(ms(1), b"ok").unwrap();
 
 	let mut group = recv_group(&mut pair.consumer);
 	drop(pair.consumer);
-	assert!(!demand.wanted().is_some(), "a held group kept the track subscription");
+	assert!(demand.wanted().is_some(), "a held group keeps its track wanted");
 
 	// The held group is not failed by the unsubscribe.
 	assert_eq!(read_frame(&mut group).plaintext, &b"ok"[..]);
 	open.write_frame(ms(2), b"next").unwrap();
 	assert_eq!(read_frame(&mut group).plaintext, &b"next"[..]);
+
+	drop(group);
+	assert!(demand.wanted().is_none(), "the last reader leaving releases the track");
 }
 
 #[test]

rs/moq-net/Cargo.toml

@@ -143,6 +143,10 @@ required-features = ["async"]
 name = "tracks"
 harness = false
 
+[[bench]]
+name = "demand"
+harness = false
+
 [[bench]]
 name = "tail"
 harness = false

rs/moq-net/MIGRATE.md

@@ -4,6 +4,22 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### `broadcast::Demand` reads like `track::Demand`
+
+Replace `is_used()` with `wanted()`, and `used()`, `unused()`, `poll_used`, and `poll_unused` with a loop over `wanted_changed()` (or `poll_wanted_changed`), which returns the new value each time it changes and errors once the broadcast closes.
+`broadcast::Consumer::demand()` is gone: call `demand()` on the broadcast's `Producer`.
+
+```rust
+// Before
+demand.used().await?;
+demand.unused().await?;
+// After
+while !demand.wanted_changed().await? {}
+while demand.wanted_changed().await? {}
+```
+
+A group a subscription or fetch handed out now keeps its track wanted until the group finishes and a reader reads its end (a range cap is not the end): until then `track::Demand::wanted` stays `Some` and `track::Producer::abort_unused` declines. Drop the group, or read it to its end, to let the demand go.
+
 ### Paths fit moq-transport's 4096-byte namespace ([#6927](https://github.com/moq-dev/moq/pull/6927))
 
 `Path::MAX_BYTES` is 4096, moq-transport's namespace limit, so every path fits every wire and building a longer one fails with `path::Invalid::TooLong`. Keep broadcast paths within 4096 bytes.
@@ -78,7 +94,6 @@ A sender that relied on `None` to hold its rate while unwatched should check its
 
 These had no caller outside tests; replace them as follows.
 
-- `broadcast::Demand` and both `broadcast::{Producer, Consumer}::demand()`: watch each track's `track::Demand` instead.
 - `goaway::Consumer::{peek, recv}`: poll it, `goaway.poll(&kio::Waiter::noop())` to peek or `kio::wait(|waiter| goaway.poll(waiter)).await` to wait.
 - `bandwidth::Producer::{abort, closed}`: drop the last producer to close it.
 - `Version::{names, from_alpn, is_ietf}`: use `Versions::all().iter()` or `Version::from_code`, match `alpn()` yourself, and `!is_lite()`.

rs/moq-net/benches/demand.rs

@@ -0,0 +1,78 @@
+//! What a broadcast's demand costs as it fans out.
+//!
+//! - edge: a reader subscribes to one track through the broadcast and leaves, and every
+//!   [`broadcast::Demand`] handle reads both edges. Swept over the tracks the broadcast
+//!   holds and the handles watching it, so a cost that grows with the table rather than
+//!   the touched track shows up as a slope.
+//! - churn: a short-lived track comes and goes while handles that never poll watch the
+//!   broadcast, the insert each handle's inbox absorbs. Swept the same way, with the live
+//!   tracks one short of a power of two so they fill an inbox's capacity but one slot.
+//!
+//! Run with `cargo bench -p moq-net --bench demand`.
+
+use std::task::Poll;
+
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use moq_net::broadcast;
+
+/// `(tracks, handles)` shapes: the broadcast's tracks and the demand handles watching it.
+const SHAPES: [(usize, usize); 5] = [(1, 1), (100, 1), (10_000, 1), (1, 100), (10_000, 100)];
+
+/// `(tracks, handles)` shapes for churn: each `tracks` is one under the inbox capacity it grows to.
+const CHURN_SHAPES: [(usize, usize); 5] = [(3, 1), (127, 1), (16_383, 1), (3, 100), (16_383, 100)];
+
+fn changed(demand: &mut broadcast::Demand, waiter: &kio::Waiter) -> Option<bool> {
+	match demand.poll_wanted_changed(waiter) {
+		Poll::Ready(wanted) => Some(wanted.expect("broadcast closed")),
+		Poll::Pending => None,
+	}
+}
+
+fn bench_edge(c: &mut Criterion) {
+	let mut group = c.benchmark_group("demand/edge");
+	for (tracks, handles) in SHAPES {
+		let id = BenchmarkId::from_parameter(format!("{tracks}t_{handles}h"));
+		group.bench_function(id, |b| {
+			let producer = broadcast::Info::new().produce();
+			let _tracks: Vec<_> = (0..tracks)
+				.map(|i| producer.create_track(format!("t{i}"), None).unwrap())
+				.collect();
+			let consumer = producer.consume();
+			let waiter = kio::Waiter::noop();
+			let mut demands: Vec<_> = (0..handles).map(|_| producer.demand()).collect();
+			for demand in &mut demands {
+				assert_eq!(changed(demand, &waiter), Some(false));
+			}
+			b.iter(|| {
+				let reader = consumer.track("t0").unwrap().subscribe(None);
+				for demand in &mut demands {
+					assert_eq!(changed(demand, &waiter), Some(true));
+				}
+				drop(reader);
+				for demand in &mut demands {
+					assert_eq!(changed(demand, &waiter), Some(false));
+				}
+			});
+		});
+	}
+	group.finish();
+}
+
+fn bench_churn(c: &mut Criterion) {
+	let mut group = c.benchmark_group("demand/churn");
+	for (tracks, handles) in CHURN_SHAPES {
+		let id = BenchmarkId::from_parameter(format!("{tracks}t_{handles}h"));
+		group.bench_function(id, |b| {
+			let producer = broadcast::Info::new().produce();
+			let _demands: Vec<_> = (0..handles).map(|_| producer.demand()).collect();
+			let _tracks: Vec<_> = (0..tracks)
+				.map(|i| producer.create_track(format!("t{i}"), None).unwrap())
+				.collect();
+			b.iter(|| drop(producer.create_track("churn", None).unwrap()));
+		});
+	}
+	group.finish();
+}
+
+criterion_group!(benches, bench_edge, bench_churn);
+criterion_main!(benches);