From c431a1f63ee453803f3d506480475a68dd754532 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 18:42:31 -0700 Subject: [PATCH 1/2] chore(quest): claim dropped-sources Co-Authored-By: GPT-6 From 2798b3bf652e474004d009703014001a5da49785 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 19:35:01 -0700 Subject: [PATCH 2/2] fix(net): preserve aborted source errors through origins Co-Authored-By: GPT-6 --- doc/lib/rs/moq-net.md | 1 + quest/m1/dropped-sources.md | 30 ++++++++++------- rs/moq-net/benches/origin.rs | 51 ++++++++++++++++++++++++++++- rs/moq-net/src/ietf/publish.rs | 5 +++ rs/moq-net/src/model/broadcast.rs | 7 +++- rs/moq-net/src/model/front.rs | 46 ++++++++++++++++++++------- rs/moq-net/src/model/origin.rs | 53 ++++++++++++++++++++++++++++--- 7 files changed, 163 insertions(+), 30 deletions(-) diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 5f6cb60673..49a00e0c5c 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -18,6 +18,7 @@ above ([hang](/lib/rs/hang)); relays and CDNs implement only this. - **Origins** scope what a session can see, and merge duplicate subscriptions so a broadcast is pulled upstream once no matter how many local readers. - **Broadcasts** are created unannounced and invisible to everyone, then announced as an exact route, or served below a prefix with `dynamic`. A consumer of the same origin sees exactly what a peer sees. Discovery accepts pattern unions; events carry the advertised prefix and captures for a complete match. - **Patterns** (`Pattern`, `Patterns`) are re-exported from [`moq-pattern`](https://docs.rs/moq-pattern). Literal `Path` stays a coordinate. +- **Broadcast errors** preserve the source abort cause through origin routing, including requests made after closure. Existing tracks keep their independent ends. IETF `PUBLISH_DONE` Unauthorized is reported as `Error::Unauthorized`. - **Tracks** carry groups with a priority, a retention window, and a timescale. Subscribers set their own priority and max age and can change them live. - **Groups** are written frame by frame and delivered on independent streams. Old groups are cached for fetch-by-sequence; stale groups are skipped per the subscriber's budget. - **Track ends**: `finish()` ends a track at its live edge, while `finish_at(n)` declares the exclusive end ahead of it and still accepts the groups below. A subscriber awaits it with `finished()`. A remote track ends only once every group below its end has arrived or was dropped; one reset before its header arrived is skipped after the subscription's max age on moq-lite (one second without one), or after one second on IETF. diff --git a/quest/m1/dropped-sources.md b/quest/m1/dropped-sources.md index 9fc495174f..d8ce8cf59b 100644 --- a/quest/m1/dropped-sources.md +++ b/quest/m1/dropped-sources.md @@ -5,23 +5,29 @@ A track or broadcast that ends because its source ended reports the source's own error to every consumer, locally and across a relay. `Dropped` means only that a handle was dropped without an end, which a correct producer never does. -#4179 fixed one path (a revoked upstream subscription now reads -`Unauthorized`); the rest still surface `Dropped`. +The open #4179 fixes revoked upstream subscriptions. The merged #4120 preserves +session death and resumed-track errors; remaining paths still need verification. ## Plan -- Known sources, from #4179: a source closing, a route leaving the origin's - table, and a withdrawn source broadcast. Find each place a consumer can - observe `Dropped` and make the ending side carry its real error (an explicit - `abort` or a preserved cause), at the source rather than by remapping at the - consumer. -- moq-transport: a `PUBLISH_DONE` carrying Unauthorized arrives as - `Error::Remote(1)`. Map it to the same error lite reports. -- Regression tests per path, each failing on `Dropped` today, in-process and - over a mock session. +- Origin broadcasts now preserve an aborted source's cause through local and + routed fronts, including a concurrent route withdrawal and later track lookup. + A closed source's standing route is excluded from that front's failover. +- Rust maps IETF `PUBLISH_DONE` Unauthorized to `Error::Unauthorized`. +- After #4179 lands, verify the combined track paths locally and over + mock sessions, and map JS `PUBLISH_DONE` Unauthorized to the shared error from + #4179. Do not duplicate the resume changes those PRs own. + +- Complete the remaining source-close, route-removal, and broadcast-withdrawal + track regressions locally and over a mock session. Preserve causes at the + source rather than remapping `Dropped` at consumers. Public API: none expected; error values consumers observe change. Wire: none. +## Required + +- [Unauthorized](/quest/m1/auth/unauthorized.md) - #4179 supplies shared Unauthorized errors and revoked-stream handling + ## Related -- [#4179](https://github.com/moq-dev/moq/pull/4179) - fixed the revoked-upstream path +- [#4179](https://github.com/moq-dev/moq/pull/4179) - owns the revoked-upstream path diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index ae998dd904..c5e3204105 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -377,6 +377,54 @@ fn bench_handoff(c: &mut Criterion) { group.finish(); } +/// Closing one routed source with a standing route, swept over unrelated routes +/// and consumers of that source. Only the touched front should reselect. +fn bench_source_close(c: &mut Criterion) { + let mut group = c.benchmark_group("origin/source_close"); + for (routes, subscribers) in [(1, 1), (1_000, 1), (1, 100), (1_000, 100)] { + group.bench_function(BenchmarkId::from_parameter(format!("{routes}r_{subscribers}s")), |b| { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let (producer, driver) = origin::Producer::new(origin::Config::default()); + runtime.spawn(moq_net::time::run(driver)); + let consumer = producer.consume(); + let _others: Vec<_> = (0..routes) + .map(|i| { + producer + .dynamic(format!("other/{i}"), origin::Route::default()) + .unwrap() + }) + .collect(); + b.iter_custom(|iterations| { + runtime.block_on(async { + let mut total = Duration::ZERO; + for _ in 0..iterations { + let server = producer.dynamic("live", peer_route(1, 0)).unwrap(); + let source = broadcast::Info::new().produce(); + let pending = consumer.request_broadcast("live"); + server.requested_broadcast().await.unwrap().accept(&source); + let first = pending.await.unwrap(); + let mut readers = vec![first]; + for _ in 1..subscribers { + readers.push(consumer.request_broadcast("live").await.unwrap()); + } + let started = std::time::Instant::now(); + source.abort(moq_net::Error::Unauthorized).unwrap(); + for reader in readers { + assert!(matches!(reader.closed().await, moq_net::Error::Unauthorized)); + } + total += started.elapsed(); + } + total + }) + }); + }); + } + group.finish(); +} + criterion_group!( benches, bench_announce, @@ -386,6 +434,7 @@ criterion_group!( bench_serve_idle, bench_subscribe, bench_request, - bench_handoff + bench_handoff, + bench_source_close ); criterion_main!(benches); diff --git a/rs/moq-net/src/ietf/publish.rs b/rs/moq-net/src/ietf/publish.rs index 787496a600..75ce505357 100644 --- a/rs/moq-net/src/ietf/publish.rs +++ b/rs/moq-net/src/ietf/publish.rs @@ -163,6 +163,7 @@ impl PublishDone<'_> { /// How the publisher ended the subscription: cleanly, or with the error its status names. pub(crate) fn end(&self, version: Version) -> Result<(), crate::Error> { match self.status_code { + 0x1 => Err(crate::Error::Unauthorized), code if code == PublishDoneStatus::TrackEnded.code(version) => Ok(()), // SUBSCRIPTION_ENDED: the subscription reached the end its filter asked for. // Draft-20 removed it and left 0x3 unassigned. @@ -731,6 +732,10 @@ mod tests { for version in [Version::Draft14, Version::Draft19, Version::Draft20, Version::Draft22] { assert!(done(0x2).end(version).is_ok(), "{version:?}"); + assert!( + matches!(done(0x1).end(version), Err(crate::Error::Unauthorized)), + "{version:?}" + ); assert!( matches!(done(0x0).end(version), Err(crate::Error::Remote(0x0))), "{version:?}" diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index 7cdf0618eb..ff6e536e0d 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -818,7 +818,7 @@ impl Consumer { fn track_inner(&self, name: &str) -> Result { // A closed broadcast (every producer and handler gone) serves nothing. if self.is_closed() { - return Err(Error::Dropped); + return Err(self.error()); } let mut state = self.state.lock(); @@ -927,6 +927,11 @@ impl Consumer { /// tell those apart). pub async fn closed(&self) -> Error { self.alive.closed().await; + self.error() + } + + /// The recorded source error, or `Dropped` when no abort was recorded. + pub(crate) fn error(&self) -> Error { self.state.read().abort.clone().unwrap_or(Error::Dropped) } diff --git a/rs/moq-net/src/model/front.rs b/rs/moq-net/src/model/front.rs index df6a582a65..9185c4d38a 100644 --- a/rs/moq-net/src/model/front.rs +++ b/rs/moq-net/src/model/front.rs @@ -52,7 +52,7 @@ pub(super) enum Event { /// given id by the driver) or was refused. Resolved { route: u64, result: Result }, /// A source closed: it will never serve again. - SourceClosed { source: u64 }, + SourceClosed { source: u64, err: Error }, /// The spliced broadcast handed out a new logical track to serve. TrackAssigned { track: Arc }, /// A source answered a track query: its copy's metadata, or a refusal. @@ -274,7 +274,7 @@ impl Front { match event { Event::Selected { best, serving_closing } => self.selected(best, serving_closing, &mut actions), Event::Resolved { route, result } => self.resolved(route, result, &mut actions), - Event::SourceClosed { source } => self.source_closed(source, &mut actions), + Event::SourceClosed { source, err } => self.source_closed(source, err, &mut actions), // A fresh logical track, even under a name served before: an earlier // verdict belonged to that request, and a later one asks afresh. The // broadcast's track metadata is what persists. @@ -428,13 +428,16 @@ impl Front { }; } - fn source_closed(&mut self, source: u64, actions: &mut Vec) { - let Some((serving, _)) = self.serving else { + fn source_closed(&mut self, source: u64, err: Error, actions: &mut Vec) { + let Some((serving, route)) = self.serving else { return; }; if serving != source { return; } + // A standing route can still advertise a source that ended. This front + // must not ask it for new content; another route may resume the old source. + self.refused.insert(route); self.serving = None; self.serving_closing = false; actions.push(Action::Detach { source }); @@ -444,11 +447,11 @@ impl Front { track.state = TrackState::Idle; } } - self.last_err = Some(Error::Dropped); + self.last_err = Some(err.clone()); match self.identity { // A local publisher ending ends its broadcast; a newcomer at the // path gets a fresh one. An anonymous source can never be resumed. - Identity::Local | Identity::Anonymous { .. } | Identity::Undetermined => self.end(Error::Dropped, actions), + Identity::Local | Identity::Anonymous { .. } | Identity::Undetermined => self.end(err, actions), Identity::Publisher(_) => actions.push(Action::Reselect), } } @@ -810,9 +813,13 @@ mod tests { fn dead_source_reselects_through_the_same_publisher() { let mut front = serving(remote(1, 10), 100); assert_actions( - front.step(Event::SourceClosed { source: 100 }), + front.step(Event::SourceClosed { + source: 100, + err: Error::Dropped, + }), &[Action::Detach { source: 100 }, Action::Reselect], ); + assert!(front.refused_routes().contains(&1)); assert_actions( front.step(Event::Selected { best: Some(remote(3, 10)), @@ -875,7 +882,10 @@ mod tests { #[test] fn dead_source_with_no_replacement_ends() { let mut front = serving(remote(1, 10), 100); - front.step(Event::SourceClosed { source: 100 }); + front.step(Event::SourceClosed { + source: 100, + err: Error::Dropped, + }); assert_actions( front.step(Event::Selected { best: None, @@ -895,7 +905,10 @@ mod tests { let mut front = serving(candidate, 100); assert_eq!(front.pin(), Pin::Route(1)); assert_actions( - front.step(Event::SourceClosed { source: 100 }), + front.step(Event::SourceClosed { + source: 100, + err: Error::Dropped, + }), &[Action::Detach { source: 100 }, Action::End { err: Error::Dropped }], ); } @@ -997,7 +1010,10 @@ mod tests { ); assert!(front.tracks[&name("audio")].refused.is_empty()); // The replacement is asked afresh. - front.step(Event::SourceClosed { source: 100 }); + front.step(Event::SourceClosed { + source: 100, + err: Error::Dropped, + }); front.step(Event::Selected { best: Some(remote(2, 10)), serving_closing: false, @@ -1186,7 +1202,10 @@ mod tests { fn local_incumbent_ending_ends_the_front() { let mut front = serving(local(1), 100); assert_actions( - front.step(Event::SourceClosed { source: 100 }), + front.step(Event::SourceClosed { + source: 100, + err: Error::Dropped, + }), &[Action::Detach { source: 100 }, Action::End { err: Error::Dropped }], ); } @@ -1230,7 +1249,10 @@ mod tests { standing: true, }), }, - Event::SourceClosed { source: 100 }, + Event::SourceClosed { + source: 100, + err: Error::Dropped, + }, Event::TrackAssigned { track: name("v") }, Event::Used { track: name("v") }, Event::Unused { diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 52c06cc5f6..29b9dfc3de 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -2025,7 +2025,7 @@ async fn run_front(task: FrontTask) { enum Step { Assigned(Arc, super::resume::Producer), Resolved(u64, Result), - SourceClosed(u64), + SourceClosed(u64, Error), Info(Arc, u64, Result), Ended(Arc, u64, Result<(), Error>), Demand(Arc), @@ -2051,6 +2051,17 @@ async fn run_front(task: FrontTask) { if table.closed { return Event::Closed; } + // A table withdrawal can race the close wakeup. Preserve the source's cause + // before selecting a replacement or ending a front with no remaining route. + if let Some(source) = front.serving() + && let Some(broadcast) = sources.get(&source) + && broadcast.is_closed() + { + return Event::SourceClosed { + source, + err: broadcast.error(), + }; + } // Read alongside the decision, under the lock a poke takes first. *seen = watch.seen(); front.retain_routes(|route| table.routes.covers(&path.as_path(), route)); @@ -2286,7 +2297,11 @@ async fn run_front(task: FrontTask) { // subscriptions already in flight): dropping their producers // leaves each reader on the copy it was spliced from, ending // when and as that copy ends. - broadcast.finish(); + if matches!(err, Error::Dropped) { + broadcast.finish(); + } else { + let _ = broadcast.clone().abort(err.clone()); + } broadcast.release_spliced(err.clone()); for (_, mut io) in tracks.drain() { // A reader still waiting on its source's answer is in flight @@ -2336,7 +2351,7 @@ async fn run_front(task: FrontTask) { && let Some(source) = sources.get(&id) && source.poll_closed(waiter).is_ready() { - return Poll::Ready(Step::SourceClosed(id)); + return Poll::Ready(Step::SourceClosed(id, source.error())); } for (name, io) in &tracks { if let Some((source, _, query)) = &io.query @@ -2404,7 +2419,7 @@ async fn run_front(task: FrontTask) { } } } - Step::SourceClosed(source) => Event::SourceClosed { source }, + Step::SourceClosed(source, err) => Event::SourceClosed { source, err }, Step::Info(name, source, result) => { let closing = sources.get(&source).is_some_and(|s| s.is_closing()); let Some(io) = tracks.get_mut(&name) else { continue }; @@ -6228,6 +6243,36 @@ mod tests { assert!(end.is_none(), "a group followed the final one"); } + #[tokio::test] + async fn an_aborted_source_preserves_its_broadcast_error() { + let producer = origin(1).produce(); + let broadcast = producer.publish("room/alice", Route::default()).unwrap(); + let resolved = producer.consume().request_broadcast("room/alice").await.unwrap(); + broadcast.abort(Error::Unauthorized).unwrap(); + assert!(matches!(resolved.closed().await, Error::Unauthorized)); + assert!(matches!(resolved.track("video"), Err(Error::Unauthorized))); + } + + #[tokio::test] + async fn an_aborted_routed_source_preserves_its_broadcast_error() { + for withdraw in [false, true] { + let producer = origin(1).produce(); + let server = producer + .dynamic("room", Route::default().with_hops(hops(&[10]))) + .unwrap(); + let pending = producer.consume().request_broadcast("room/alice"); + let source = broadcast::Info::new().produce(); + queued(&server).await.accept(&source); + let resolved = pending.await.unwrap(); + source.abort(Error::Unauthorized).unwrap(); + if withdraw { + drop(server); + } + assert!(matches!(resolved.closed().await, Error::Unauthorized)); + assert!(matches!(resolved.track("video"), Err(Error::Unauthorized))); + } + } + /// An origin front drops the source track as soon as its last reader leaves, /// so the publisher's `unused()` resolves far below `TRACK_IDLE_LINGER`. /// Cached groups stay on the front for the linger; a returning reader