From c431a1f63ee453803f3d506480475a68dd754532 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 18:42:31 -0700 Subject: [PATCH 1/4] 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/4] 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 5f6cb60673c..49a00e0c5ce 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 9fc495174fd..d8ce8cf59b2 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 ae998dd9048..c5e32041052 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 787496a600f..75ce5053575 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 7cdf0618eb6..ff6e536e0d2 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 df6a582a656..9185c4d38a1 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 52c06cc5f6c..29b9dfc3de6 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 From 43720d28c5d0c91118ee969af9343657bd7cf925 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 08:01:29 -0700 Subject: [PATCH 3/4] fix(net): trim to PUBLISH_DONE Unauthorized and standing-route exclusion Broadcast ends carry no cause after the broadcast-close questline, so drop the abort-cause propagation through origin close events, front termination, and late track lookups, along with its bench and docs. Keep the IETF PUBLISH_DONE Unauthorized decode and the dead source's route exclusion, each with a regression test. Co-Authored-By: Claude Opus 5.5 --- doc/lib/rs/moq-net.md | 1 - quest/m1/dropped-sources.md | 29 +++++++------- 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 | 50 ++++++++---------------- rs/moq-net/src/model/origin.rs | 64 ++++++++++--------------------- 7 files changed, 56 insertions(+), 151 deletions(-) diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 49a00e0c5ce..5f6cb60673c 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -18,7 +18,6 @@ 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 d8ce8cf59b2..2934835fed2 100644 --- a/quest/m1/dropped-sources.md +++ b/quest/m1/dropped-sources.md @@ -2,25 +2,22 @@ ## Goal -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. -The open #4179 fixes revoked upstream subscriptions. The merged #4120 preserves -session death and resumed-track errors; remaining paths still need verification. +A track 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. A broadcast +end carries no cause ([Broadcast close](/quest/m1/broadcast-close/README.md)), +so only track errors are in scope. ## Plan -- 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. +- Rust already maps IETF `PUBLISH_DONE` Unauthorized to `Error::Unauthorized`, + and a closed source's standing route no longer re-requests it. +- #4179 fixes revoked upstream subscriptions and #4120 preserves session death + and resumed-track errors. After #4179 reaches `main`, verify the remaining + track paths (source close, route removal, broadcast withdrawal) locally and + over a mock session, and map JS `PUBLISH_DONE` Unauthorized to #4179's shared + error. Preserve causes at the source rather than remapping `Dropped` at + consumers. Public API: none expected; error values consumers observe change. Wire: none. diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index 440efc77f4f..b2cef774cb8 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -374,54 +374,6 @@ 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, @@ -431,7 +383,6 @@ criterion_group!( bench_serve_idle, bench_subscribe, bench_request, - bench_handoff, - bench_source_close + bench_handoff ); criterion_main!(benches); diff --git a/rs/moq-net/src/ietf/publish.rs b/rs/moq-net/src/ietf/publish.rs index 75ce5053575..cf93727fd1a 100644 --- a/rs/moq-net/src/ietf/publish.rs +++ b/rs/moq-net/src/ietf/publish.rs @@ -124,6 +124,8 @@ use super::Version; pub(crate) enum PublishDoneStatus { /// An implementation-specific failure ended the subscription. InternalError, + /// The subscriber is no longer authorized for the track. + Unauthorized, /// The track is no longer being published. TrackEnded, } @@ -144,6 +146,7 @@ impl PublishDoneStatus { | Version::Draft21 | Version::Draft22 => match self { Self::InternalError => 0x0, + Self::Unauthorized => 0x1, Self::TrackEnded => 0x2, }, } @@ -163,8 +166,8 @@ 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(()), + code if code == PublishDoneStatus::Unauthorized.code(version) => Err(crate::Error::Unauthorized), // SUBSCRIPTION_ENDED: the subscription reached the end its filter asked for. // Draft-20 removed it and left 0x3 unassigned. 0x3 if matches!( diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index ff6e536e0d2..7cdf0618eb6 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(self.error()); + return Err(Error::Dropped); } let mut state = self.state.lock(); @@ -927,11 +927,6 @@ 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 9185c4d38a1..b56136b02a8 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, err: Error }, + SourceClosed { source: u64 }, /// 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. @@ -202,7 +202,8 @@ pub(super) struct Front { serving_closing: bool, /// The route an upstream request is in flight through. upstream: Option, - /// Routes that refused the path while another source was serving. + /// Routes excluded from selection: they refused the path while another + /// source was serving, or their source ended while still advertised. refused: HashSet, /// Why the last candidate fell through, reported if the front ends unresolved. last_err: Option, @@ -244,7 +245,7 @@ impl Front { self.identity.pin() } - /// The routes that refused the path; the driver skips them when selecting. + /// The routes excluded from selection; the driver skips them. pub(super) fn refused_routes(&self) -> &HashSet { &self.refused } @@ -274,7 +275,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, err } => self.source_closed(source, err, &mut actions), + Event::SourceClosed { source } => self.source_closed(source, &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,15 +429,16 @@ impl Front { }; } - fn source_closed(&mut self, source: u64, err: Error, actions: &mut Vec) { + fn source_closed(&mut self, source: u64, 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. + // A standing route can outlive the source it produced. Asking it again + // would re-request the broadcast that just ended; another route to the + // same publisher may still resume it. self.refused.insert(route); self.serving = None; self.serving_closing = false; @@ -447,11 +449,11 @@ impl Front { track.state = TrackState::Idle; } } - self.last_err = Some(err.clone()); + self.last_err = Some(Error::Dropped); 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(err, actions), + Identity::Local | Identity::Anonymous { .. } | Identity::Undetermined => self.end(Error::Dropped, actions), Identity::Publisher(_) => actions.push(Action::Reselect), } } @@ -813,10 +815,7 @@ 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, - err: Error::Dropped, - }), + front.step(Event::SourceClosed { source: 100 }), &[Action::Detach { source: 100 }, Action::Reselect], ); assert!(front.refused_routes().contains(&1)); @@ -882,10 +881,7 @@ mod tests { #[test] fn dead_source_with_no_replacement_ends() { let mut front = serving(remote(1, 10), 100); - front.step(Event::SourceClosed { - source: 100, - err: Error::Dropped, - }); + front.step(Event::SourceClosed { source: 100 }); assert_actions( front.step(Event::Selected { best: None, @@ -905,10 +901,7 @@ mod tests { let mut front = serving(candidate, 100); assert_eq!(front.pin(), Pin::Route(1)); assert_actions( - front.step(Event::SourceClosed { - source: 100, - err: Error::Dropped, - }), + front.step(Event::SourceClosed { source: 100 }), &[Action::Detach { source: 100 }, Action::End { err: Error::Dropped }], ); } @@ -1010,10 +1003,7 @@ mod tests { ); assert!(front.tracks[&name("audio")].refused.is_empty()); // The replacement is asked afresh. - front.step(Event::SourceClosed { - source: 100, - err: Error::Dropped, - }); + front.step(Event::SourceClosed { source: 100 }); front.step(Event::Selected { best: Some(remote(2, 10)), serving_closing: false, @@ -1202,10 +1192,7 @@ mod tests { fn local_incumbent_ending_ends_the_front() { let mut front = serving(local(1), 100); assert_actions( - front.step(Event::SourceClosed { - source: 100, - err: Error::Dropped, - }), + front.step(Event::SourceClosed { source: 100 }), &[Action::Detach { source: 100 }, Action::End { err: Error::Dropped }], ); } @@ -1249,10 +1236,7 @@ mod tests { standing: true, }), }, - Event::SourceClosed { - source: 100, - err: Error::Dropped, - }, + Event::SourceClosed { source: 100 }, 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 29b9dfc3de6..2b90758fd71 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, Error), + SourceClosed(u64), Info(Arc, u64, Result), Ended(Arc, u64, Result<(), Error>), Demand(Arc), @@ -2051,17 +2051,6 @@ 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)); @@ -2297,11 +2286,7 @@ 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. - if matches!(err, Error::Dropped) { - broadcast.finish(); - } else { - let _ = broadcast.clone().abort(err.clone()); - } + broadcast.finish(); broadcast.release_spliced(err.clone()); for (_, mut io) in tracks.drain() { // A reader still waiting on its source's answer is in flight @@ -2351,7 +2336,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, source.error())); + return Poll::Ready(Step::SourceClosed(id)); } for (name, io) in &tracks { if let Some((source, _, query)) = &io.query @@ -2419,7 +2404,7 @@ async fn run_front(task: FrontTask) { } } } - Step::SourceClosed(source, err) => Event::SourceClosed { source, err }, + Step::SourceClosed(source) => Event::SourceClosed { source }, 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 }; @@ -6243,34 +6228,25 @@ mod tests { assert!(end.is_none(), "a group followed the final one"); } + /// A standing route outlives the source it produced: the front ends instead of + /// asking that route for the broadcast that just closed. #[tokio::test] - async fn an_aborted_source_preserves_its_broadcast_error() { + async fn a_closed_source_is_not_requested_again_from_its_standing_route() { 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))); - } + 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(); - #[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))); - } + drop(source); + settle(|| resolved.is_closed()).await; + assert!( + server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(), + "the closed source was requested again" + ); } /// An origin front drops the source track as soon as its last reader leaves, From 8225bb9da44694d252af84ee3805ea877e2d00b3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 09:46:54 -0700 Subject: [PATCH 4/4] fix(net): align dropped-sources with broadcast close Close the source with close() in the standing-route regression and point the quest at #4031 now that the broadcast-close questline is gone. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 2 +- quest/m1/dropped-sources.md | 3 +-- rs/moq-net/src/model/origin.rs | 2 +- 3 files changed, 3 insertions(+), 4 deletions(-) diff --git a/quest/m1/README.md b/quest/m1/README.md index bf58cf9600c..5b229013274 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -20,7 +20,7 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [BBR classic ECN](/quest/m1/bbr-classic-ecn.md) - Startup and bandwidth probing respond to CE marks before the bottleneck drops packets - [libmoq hidden opt-in](/quest/m1/libmoq-hidden.md) - `moq_origin_announced` takes a `hidden` flag so C callers can list `.`-named broadcasts - [lite-07 count settle](/quest/m1/lite-count-settle.md) - moq-lite-07 subscribers stop waiting for a subscription's tail once SUBSCRIBE_END's stream count is reached -- [Dropped sources](/quest/m1/dropped-sources.md) - consumers see the producer's real error on every end path, never `Dropped` +- [Dropped sources](/quest/m1/dropped-sources.md) - track consumers see the producer's real error on every end path, never `Dropped` - [JS group guard](/quest/m1/js-group-guard.md) - a `@moq/net` publisher abandons a group past its max age without an unhandled rejection - [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop` - [Interop flakes](/quest/m1/interop-flakes.md) - the interop harness passes with other runs sharing the machine diff --git a/quest/m1/dropped-sources.md b/quest/m1/dropped-sources.md index 2934835fed2..9e29b468966 100644 --- a/quest/m1/dropped-sources.md +++ b/quest/m1/dropped-sources.md @@ -5,8 +5,7 @@ A track 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. A broadcast -end carries no cause ([Broadcast close](/quest/m1/broadcast-close/README.md)), -so only track errors are in scope. +end carries no cause since #4031, so only track errors are in scope. ## Plan diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index adcd49dd105..dfa1089b612 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -6238,7 +6238,7 @@ mod tests { queued(&server).await.accept(&source); let resolved = pending.await.unwrap(); - drop(source); + source.close(); settle(|| resolved.is_closed()).await; assert!( server.poll_requested_broadcast(&kio::Waiter::noop()).is_pending(),