fix(net): a minted source enforces its own request validity

js/net/src/origin.test.ts

@@ -285,21 +285,6 @@ test("closing the origin closes every routed broadcast", async () => {
 	expect(origin.closed.peek()).toBe(abort);
 });
 
-test("the table is reactive", async () => {
-	const origin = new Producer();
-	const consumer = origin.consume();
-	const path = Path.from("room");
-
-	const changed = wireOf(consumer).broadcasts.changed();
-	const broadcast = publish(origin, path);
-
-	const table = await changed;
-	expect(table?.has(path)).toBe(true);
-
-	broadcast.close();
-	origin.close();
-});
-
 test("announced streams the table under a scope with origin-relative paths", async () => {
 	const origin = new Producer();
 	const consumer = origin.consume();
@@ -450,7 +435,7 @@ test("the publisher-facing table excludes remote entries", async () => {
 	serve(origin, Path.from("theirs"), provider(upstream));
 
 	// What a session announces to a peer: local only, so a shared origin cannot echo.
-	const table = wireOf(consumer).broadcasts.peek();
+	const table = wireOf(consumer).advertised.peek();
 	expect(table?.has(Path.from("mine"))).toBe(true);
 	expect(table?.has(Path.from("theirs"))).toBe(false);
 
@@ -623,7 +608,6 @@ test("requests never appear in announced or the table", async () => {
 	// stays out of the table, and out of the announcements the table drives.
 	expect(request.active.peek()).toBeDefined();
 	expect(wireOf(origin).routes(path)).toBe(false);
-	expect(wireOf(consumer).broadcasts.peek()?.has(path)).toBe(false);
 
 	const announced = consumer.announced();
 	publish(origin, Path.from("real"));
@@ -969,7 +953,6 @@ test("createBroadcast is invisible to everyone until announce", async () => {
 
 	const broadcast = origin.createBroadcast(path);
 	expect(wireOf(consumer).routes(path)).toBe(false);
-	expect(wireOf(consumer).broadcasts.peek()?.has(path)).toBe(false);
 	expect(wireOf(consumer).advertised.peek()?.has(path)).toBe(false);
 	const request = consumer.request(path);
 	expect(request.active.peek()).toBeUndefined();

js/net/src/origin.ts

@@ -1637,8 +1637,6 @@ export class Consumer {
 		);
 		registerWire(this, {
 			routes: (path) => this.#routes(scope.path(path)),
-			broadcasts:
-				scope === Scope.all ? state.local : new Derived([state.local], (local) => scope.projectPaths(local)),
 			advertised:
 				scope === Scope.all
 					? state.originated

js/net/src/wire.ts

@@ -75,7 +75,6 @@ export interface OriginProducer {
 /** The protocol-facing operations behind an origin consumer. */
 export interface OriginConsumer {
 	routes(path: Path.Valid): boolean;
-	readonly broadcasts: Getter<ReadonlyMap<Path.Valid, broadcast.Consumer> | undefined>;
 	readonly advertised: Getter<Advertisements | undefined>;
 	/** The announced local broadcast at `path`, when it is the route peers are offered there.
 	 * A request naming an `epoch` only matches a route serving it. */

quest/m1/api-audit/README.md

@@ -69,7 +69,6 @@ Rejected in planning:
 - [Bindings read gaps and ends](/quest/m1/api-audit/decode-read-bindings.md) - moq-ffi and every wrapper surface `Gap` and `End` instead of discarding them
 - [Pinned consumers adopt the CLI flag names](/quest/m1/api-audit/cli-flags-consumers.md) - pinned downstream deployments move past the flag rename and use its spellings
 - [Session-only items stay private](/quest/m1/api-audit/model-private.md) - model items only the built-in sessions use are private
-- [A request's source enforces its validity](/quest/m1/api-audit/answer.md) - a stale source refuses new track requests by itself, with no public additions
 - [moq-transport stays private](/quest/m1/api-audit/ietf-private.md) - no public moq-net item exists only for moq-transport, and a client names its request target once as a URL
 - [One lifecycle vocabulary](/quest/m1/api-audit/lifecycle.md) - finish, abort, closed, and close mean one thing on every handle, and Rust and JS observers fire alike
 - [moq-net names say one thing](/quest/m1/api-audit/net-names.md) - demand says `wanted`, colliding stats and transport names are renamed, and Rust and JS info lookups mirror

quest/m2/audit-dead-code.md

@@ -34,8 +34,9 @@ Start by comparing the live tree against these baselines:
 - Rust had 423 dead-code/unused suppression lines: 77 unconditional, 346
   through `cfg_attr`. The unconditional ones are drop guards, compile-time
   Send assertions, generated settings registries, shared test and benchmark
-  harnesses, deserialization fixtures, FFI tables, and two `GroupConsumer`
-  helpers waiting on the [answering API](/quest/m1/api-audit/answer.md).
+  harnesses, deserialization fixtures, FFI tables, and
+  `group::Producer::abort_unused`, which only tests call since the answering
+  API that wanted it was dropped.
   Once the lint audit lands, every unconditional one is an `expect`, so a
   stale one fails the build; check what the conditional ones still guard.
   `cargo shear` found no unused dependencies.

rs/moq-net/src/ietf/subscriber.rs

@@ -2650,12 +2650,9 @@ impl Subscriber {
 				Some(requested) => requested.to_owned(),
 				None => continue,
 			};
-			let source = crate::model::broadcast::SourceGuard::new(self.origin.create_source(&requested));
-			// The handler exists before the requester sees the source, so a track it asks
-			// for before `run_broadcast` first polls queues instead of failing `NotFound`.
-			let dynamic = source.dynamic();
 			// The source serves the instance the route names now: every request for it names
-			// that epoch, so the peer refuses one that would reach another instance.
+			// that epoch when the peer can read one, and the source takes new tracks only
+			// while the peer still selects that instance.
 			let (validity, epoch) = {
 				let state = self.state.lock();
 				let entry = state
@@ -2670,6 +2667,10 @@ impl Subscriber {
 					});
 				(validity, entry.and_then(|entry| entry.route.epoch.clone()))
 			};
+			let source = crate::model::broadcast::SourceGuard::new(self.origin.create_source(&requested, validity));
+			// The handler exists before the requester sees the source, so a track it asks
+			// for before `run_broadcast` first polls queues instead of failing `NotFound`.
+			let dynamic = source.dynamic();
 			request.accept(&*source);
 
 			// The namespace's token closes the source as it is retracted. If it was
@@ -2687,7 +2688,7 @@ impl Subscriber {
 			}
 
 			let mut this = self.clone();
-			this.validity = validity;
+			this.validity = dynamic.validity().clone();
 			this.epoch = epoch;
 			broadcasts.push(async move {
 				if let Err(err) = this.run_broadcast(requested.borrow(), source, dynamic, route).await {
@@ -2718,8 +2719,8 @@ impl Subscriber {
 	}
 
 	/// Serve one minted source's track requests, taken through its `broadcast` handler,
-	/// until it closes: its namespace is retracted (`route` closes), nothing holds it any
-	/// more, or the session dies.
+	/// until it closes: its namespace is retracted (`route` closes), the peer stops
+	/// selecting its instance, nothing holds it any more, or the session dies.
 	async fn run_broadcast(
 		&self,
 		path: Path<'_>,
@@ -9644,7 +9645,7 @@ mod tests {
 	async fn a_minted_source_closes_unheld_or_retracted() {
 		let (subscriber, origin) = cluster_subscriber(crate::Hop::new(1).unwrap());
 		let run = |route: &kio::Producer<()>| {
-			let source = crate::model::broadcast::SourceGuard::new(origin.create_source("pool/p"));
+			let source = crate::model::broadcast::SourceGuard::new(origin.create_source("pool/p", Default::default()));
 			let (holder, dynamic) = (source.consume(), source.dynamic());
 			let (this, route) = (subscriber.clone(), route.consume());
 			let task = moq_clock::sim::spawn(async move {

rs/moq-net/src/lite/subscriber.rs

@@ -1831,7 +1831,6 @@ impl AnnouncePrefix {
 
 /// A source the announce half minted for one requested path, for the driver to serve.
 struct MintedSource {
-	validity: crate::model::broadcast::RequestValidity,
 	path: PathOwned,
 	/// The publisher instance the route announced, asked for on every request.
 	epoch: Option<crate::Epoch>,
@@ -1843,12 +1842,12 @@ struct MintedSource {
 }
 
 /// Serves the origin's track requests for one minted source until it closes: its
-/// route goes, nothing holds it any more, or the session dies. A closed source
+/// route goes, the peer stops selecting its instance, nothing holds it any more, or
+/// the session dies. A closed source
 /// takes no new tracks but lets those in flight run to their own end (moq-lite:
 /// retraction does not disturb subscriptions already in flight); the session
 /// dying drops them.
 struct SourceServe {
-	validity: crate::model::broadcast::RequestValidity,
 	subscriber: Subscriber,
 	path: PathOwned,
 	epoch: Option<crate::Epoch>,
@@ -1966,7 +1965,6 @@ impl SourceServe {
 				.expect("only the source answers its broadcast's listings"),
 			listing: None,
 			subscriber,
-			validity: minted.validity,
 			path: minted.path,
 			epoch: minted.epoch,
 			source: minted.source,
@@ -2005,7 +2003,7 @@ impl SourceServe {
 							crate::auth::Direction::Subscribe,
 						),
 						admitted: false,
-						validity: self.validity.start(),
+						validity: self.dynamic.validity().start(),
 					});
 				}
 				Some(_) if self.lister.poll_unused(waiter).is_ready() => {
@@ -2057,7 +2055,7 @@ impl kio::Task for SourceServe {
 			match self.dynamic.poll_requested_track(waiter) {
 				Poll::Ready(Ok(request)) => {
 					let serve = TrackServe {
-						validity: self.validity.clone(),
+						validity: self.dynamic.validity().clone(),
 						route: Some(self.route.clone()),
 						subscriber: self.subscriber.clone(),
 						path: self.path.clone(),
@@ -3935,14 +3933,13 @@ mod tests {
 			going_away: Default::default(),
 			auth,
 		});
-		let broadcast = crate::broadcast::Info::new().produce();
+		let broadcast = crate::broadcast::Info::new().produce().with_validity(validity);
 		let consumer = broadcast.consume();
 		let dynamic = broadcast.dynamic();
 		let route = kio::Producer::new(());
 		let serve = SourceServe::new(
 			subscriber,
 			MintedSource {
-				validity,
 				path: Path::new("room").to_owned(),
 				epoch: None,
 				source: crate::model::broadcast::SourceGuard::new(broadcast),
@@ -3981,7 +3978,7 @@ mod tests {
 	}
 
 	/// An instance retired while its listing waits for a token lists nothing: a TRACKS
-	/// request without an epoch would list its replacement.
+	/// request without an epoch would list its replacement. The source ends with it.
 	#[moq_clock::test]
 	async fn a_listing_of_a_retired_instance_never_reaches_the_wire() {
 		let auth = crate::auth::Link::new(true);
@@ -3996,7 +3993,7 @@ mod tests {
 		assert!(kio::Task::poll(&mut serve, &kio::Waiter::noop()).is_pending());
 		drop(selected);
 		drop(token);
-		assert!(kio::Task::poll(&mut serve, &kio::Waiter::noop()).is_pending());
+		assert!(kio::Task::poll(&mut serve, &kio::Waiter::noop()).is_ready());
 		assert!(matches!(tracks.next().now_or_never(), Some(Err(Error::Unroutable))));
 		assert_eq!(session.log.bi_opens(), 0);
 	}
@@ -6008,20 +6005,21 @@ impl Announced {
 					continue;
 				};
 				let path = path.to_owned();
-				let source = subscriber.origin.create_source(&path);
-				// The handler exists before the requester sees the source, so a track it
-				// asks for before the serve machine first runs queues instead of failing
-				// `NotFound`. Accepted before the push, so the requester holds the source
-				// before its serve machine can see it unheld.
-				let dynamic = source.dynamic();
+				// A request naming the epoch reaches only that instance, so only a source
+				// without one must stop once the peer no longer selects what minted it.
 				let validity = if entry.route.epoch.is_some() {
 					Default::default()
 				} else {
 					crate::model::broadcast::RequestValidity::for_source(&entry.dynamic, request.path())
 				};
+				let source = subscriber.origin.create_source(&path, validity);
+				// The handler exists before the requester sees the source, so a track it
+				// asks for before the serve machine first runs queues instead of failing
+				// `NotFound`. Accepted before the push, so the requester holds the source
+				// before its serve machine can see it unheld.
+				let dynamic = source.dynamic();
 				request.accept(&source);
 				let _ = subscriber.sources.try_push(MintedSource {
-					validity,
 					path,
 					epoch: entry.route.epoch.clone(),
 					source: crate::model::broadcast::SourceGuard::new(source),

rs/moq-net/src/model/broadcast.rs

@@ -274,6 +274,15 @@ impl Producer {
 		self
 	}
 
+	/// Answer track requests only while `validity` holds, ending the broadcast once it
+	/// lapses. Set when a session mints a source, before any handle is shared.
+	pub(crate) fn with_validity(mut self, validity: RequestValidity) -> Self {
+		Arc::get_mut(&mut self.alive)
+			.expect("validity is set on a fresh broadcast")
+			.validity = validity;
+		self
+	}
+
 	/// The route this broadcast is announced with, or `None` while it is not.
 	pub fn route(&self) -> Option<Route> {
 		self.alive.announcer.lock().as_ref()?.route()
@@ -529,6 +538,10 @@ struct Alive {
 	// The advertisement of the broadcast's exact path, owned here so it retracts
 	// with the broadcast: on close or the last producer-side handle dropping. `None` for a standalone broadcast.
 	announcer: kio::Lock<Option<Announcer>>,
+	// How long a session's source still names the instance that minted it. Held on the
+	// producer side, never by a consumer, so the selection it may hold is dropped by its
+	// session rather than wherever the last consumer goes.
+	validity: RequestValidity,
 }
 
 impl Alive {
@@ -537,6 +550,7 @@ impl Alive {
 			token: kio::Producer::default(),
 			state,
 			announcer: kio::Lock::new(None),
+			validity: RequestValidity::default(),
 		})
 	}
 
@@ -634,15 +648,13 @@ impl RequestValidity {
 	}
 
 	/// Capture one selection period so a pending request never revives after a replacement.
-	#[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))]
 	pub(crate) fn start(&self) -> Self {
 		match self {
 			Self::Selected(selected) => Self::Period(selected.current(), selected.clone()),
 			other => other.clone(),
 		}
 	}
 
-	#[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))]
 	pub(crate) fn check(&self, waiter: &kio::Waiter) -> Result<(), crate::Error> {
 		match self {
 			Self::Route(route) if route.poll_closed(waiter).is_ready() => Err(crate::Error::Unroutable),
@@ -776,6 +788,12 @@ impl Dynamic {
 	/// Returns [`Error::Closed`] once the broadcast has ended, so a serving loop
 	/// knows to stop and release its handle.
 	pub fn poll_requested_track(&mut self, waiter: &kio::Waiter) -> Poll<Result<track::Request, Error>> {
+		// A source that no longer names the instance that minted it takes no new track:
+		// ending it refuses what is queued with `Unroutable`, while the tracks it handed
+		// out run on. Checked outside the broadcast lock, since the selection reads the origin.
+		if self.alive.validity.check(waiter).is_err() {
+			self.alive.close();
+		}
 		let mut state = ready!(self.state.poll(waiter, |state| {
 			if state.requests.has_queued() || state.closing {
 				Poll::Ready(())
@@ -813,6 +831,13 @@ impl Dynamic {
 		}
 	}
 
+	/// What a request to the peer for this source must still hold when it goes out, for
+	/// an operation on a track the source already handed out.
+	#[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))]
+	pub(crate) fn validity(&self) -> &RequestValidity {
+		&self.alive.validity
+	}
+
 	/// Answer the broadcast's track listings, for a handler that knows the tracks it
 	/// would serve, such as a relay listing its upstream's. See [`Lister`].
 	///
@@ -1971,6 +1996,33 @@ mod test {
 		assert!(matches!(pending.await, Err(Error::Unroutable)));
 	}
 
+	/// A source that no longer names the instance that minted it takes no new track, even
+	/// from a handler that checks nothing: its request stream ends and every later request
+	/// answers `Unroutable`, while the track it accepted earlier runs on.
+	#[moq_clock::test]
+	#[cfg(feature = "async")]
+	async fn a_lapsed_source_refuses_new_tracks() {
+		let route = kio::Producer::new(());
+		let producer = Info::new()
+			.produce()
+			.with_validity(RequestValidity::Route(route.consume()));
+		let mut dynamic = producer.dynamic();
+		let consumer = producer.consume();
+
+		let pending = subscribe_pending!(consumer, "accepted");
+		let served = dynamic.assert_request().accept(crate::track::Info::timed());
+		let mut accepted = pending.await.unwrap();
+		let queued = subscribe_pending!(consumer, "queued");
+
+		drop(route);
+		assert!(matches!(dynamic.requested_track().await, Err(Error::Closed)));
+		assert!(matches!(queued.await, Err(Error::Unroutable)));
+		assert!(matches!(consumer.track("later"), Err(Error::Unroutable)));
+
+		served.append_group().unwrap().finish().unwrap();
+		accepted.assert_group();
+	}
+
 	/// With a producer still alive, losing the last handler is not the broadcast ending:
 	/// the queued request fails as `Dropped`.
 	#[moq_clock::test]