From e29083743c017a1cf209df4a1c3370958feb52fb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 23:53:45 -0700 Subject: [PATCH 1/2] chore(quest): claim quest/m1/api-audit/answer.md Quest-Claim: e1e56c8aa0a778eb From 76a5fdd4209d273aa28b067750cd213a41677465 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 02:09:54 -0700 Subject: [PATCH 2/2] fix(net): a minted source enforces its own request validity The session that mints a source for a dynamic request now hands its RequestValidity to the source itself. Once the route is replaced or the selection period retires, the source's request stream ends: queued and later track requests answer Unroutable, and tracks it already accepted run to their own end. Sessions read per-operation validity from the source instead of threading their own copy. Also drops the unread `OriginConsumer.broadcasts` wire getter in @moq/net and the uncalled group::Producer::poll_unused, and finishes the quest. Co-Authored-By: Claude Opus 5.5 --- js/net/src/origin.test.ts | 19 +---------- js/net/src/origin.ts | 2 -- js/net/src/wire.ts | 1 - quest/m1/api-audit/README.md | 1 - quest/m1/api-audit/answer.md | 27 --------------- quest/m2/audit-dead-code.md | 5 +-- rs/moq-net/src/ietf/subscriber.rs | 19 ++++++----- rs/moq-net/src/lite/subscriber.rs | 32 +++++++++--------- rs/moq-net/src/model/broadcast.rs | 56 +++++++++++++++++++++++++++++-- rs/moq-net/src/model/group.rs | 15 ++------- rs/moq-net/src/model/origin.rs | 8 ++++- 11 files changed, 93 insertions(+), 92 deletions(-) delete mode 100644 quest/m1/api-audit/answer.md diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 6717f94274b..2851a8df8e4 100644 --- a/js/net/src/origin.test.ts +++ b/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(); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index 2c54c2cb456..1d331596015 100644 --- a/js/net/src/origin.ts +++ b/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 diff --git a/js/net/src/wire.ts b/js/net/src/wire.ts index 4e668fe37e8..ced76578ad7 100644 --- a/js/net/src/wire.ts +++ b/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 | undefined>; readonly advertised: Getter; /** 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. */ diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md index 889e9fb6107..84d9d3f0cf9 100644 --- a/quest/m1/api-audit/README.md +++ b/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 diff --git a/quest/m1/api-audit/answer.md b/quest/m1/api-audit/answer.md deleted file mode 100644 index 9d6ae251a5a..00000000000 --- a/quest/m1/api-audit/answer.md +++ /dev/null @@ -1,27 +0,0 @@ -# [S] A request's source enforces its validity - -## Goal - -The source minted for a dynamic request refuses new track requests with -`Unroutable` once its route is replaced or its selection period retires, so -no session can serve a stale instance by forgetting a check. Tracks it -already accepted keep running and finish. Nothing becomes public. - -## Plan - -Decided 2026-10-23 (maintainer), reshaping -[moq-dev/moq#6411](https://github.com/moq-dev/moq/pull/6411) after the -public-model goal was dropped -([session-only items stay private](/quest/m1/api-audit/model-private.md)): - -- Keep only the internal simplification: `broadcast::RequestValidity` and - `origin::Selection` checks move into the source. The answering session - learns of it when the source's request stream ends. -- Drop every planned public addition: `answer_source`, the unused-race - family (`poll_unheld`, `close_unheld`, group `poll_unused` and - `abort_unused`, `reject_unused`), `Producer::empty`/`Consumer::empty`, - `Dynamic::withdrawn`, JS `Origin.Producer.fallback()`, and JS - `Broadcast.requested` (deferred until a JS app needs it). -- Delete `wire.ts` `OriginConsumer.broadcasts`: registered, never read. - -Public API: none. Wire: none. diff --git a/quest/m2/audit-dead-code.md b/quest/m2/audit-dead-code.md index 3958b821bb0..7bf5dcf988a 100644 --- a/quest/m2/audit-dead-code.md +++ b/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. diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index e58bd3dc828..e9822218bfe 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/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 { diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index ba0a93f6c9b..58542148e49 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/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, @@ -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, @@ -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), diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index 3e0527d629c..a870725dee5 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/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 { 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>, + // 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,7 +648,6 @@ 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()), @@ -642,7 +655,6 @@ impl RequestValidity { } } - #[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> { + // 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] diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 02cae7143af..c0aafa374c2 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -1675,11 +1675,11 @@ impl Producer { } /// Abort with `err` only while nothing consumes the group, returning whether it is - /// closed. Consumer creation and the check share a lock, so a reader arriving after - /// [`poll_unused`](Self::poll_unused) keeps the group alive instead of reading the abort. + /// closed. Consumer creation and the check share a lock, so a reader arriving first + /// keeps the group alive instead of reading the abort. #[allow( dead_code, - reason = "the answering API retains these on the writer as well as the session observer" + reason = "only tests call it; the dead-code audit decides whether it stays" )] pub(crate) fn abort_unused(&self, err: Error) -> bool { self.handle.abort_unused(err) @@ -1711,15 +1711,6 @@ impl Producer { pub fn demand(&self) -> Demand { self.handle.demand() } - - /// Poll for the group becoming unused (every consumer dropped). - #[allow( - dead_code, - reason = "the answering API retains these on the writer as well as the session observer" - )] - pub(crate) fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> { - self.handle.poll_unused(waiter) - } } impl Handle { diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 7bb068b1879..76a7732ed35 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2237,8 +2237,13 @@ impl Producer { /// is *not* entered into the route table. Sessions answer /// [`Dynamic`] requests with one of these; the requester already holds /// the request's result channel, so the table never needs to resolve it. + /// The source takes new tracks only while `validity` holds. #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] - pub(crate) fn create_source(&self, path: impl AsPath) -> broadcast::Producer { + pub(crate) fn create_source( + &self, + path: impl AsPath, + validity: crate::model::broadcast::RequestValidity, + ) -> broadcast::Producer { let path = path.as_path(); let full = self.root.join(&path).to_owned(); let ingress = self.stats.ingress(&full); @@ -2251,6 +2256,7 @@ impl Producer { } .produce() .with_stats(ingress) + .with_validity(validity) } /// Advertise a route without serving it: a claim that paths under `prefix`