diff --git a/quest/m1/dropped-sources.md b/quest/m1/dropped-sources.md index 9fc495174f..2934835fed 100644 --- a/quest/m1/dropped-sources.md +++ b/quest/m1/dropped-sources.md @@ -2,26 +2,29 @@ ## 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. -#4179 fixed one path (a revoked upstream subscription now reads -`Unauthorized`); the rest still surface `Dropped`. +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 -- 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. +- 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. +## 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/src/ietf/publish.rs b/rs/moq-net/src/ietf/publish.rs index 787496a600..cf93727fd1 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, }, } @@ -164,6 +167,7 @@ impl PublishDone<'_> { pub(crate) fn end(&self, version: Version) -> Result<(), crate::Error> { match self.status_code { 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!( @@ -731,6 +735,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/front.rs b/rs/moq-net/src/model/front.rs index df6a582a65..b56136b02a 100644 --- a/rs/moq-net/src/model/front.rs +++ b/rs/moq-net/src/model/front.rs @@ -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 } @@ -429,12 +430,16 @@ impl Front { } fn source_closed(&mut self, source: u64, actions: &mut Vec) { - let Some((serving, _)) = self.serving else { + let Some((serving, route)) = self.serving else { return; }; if serving != source { return; } + // 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; actions.push(Action::Detach { source }); @@ -813,6 +818,7 @@ mod tests { front.step(Event::SourceClosed { source: 100 }), &[Action::Detach { source: 100 }, Action::Reselect], ); + assert!(front.refused_routes().contains(&1)); assert_actions( front.step(Event::Selected { best: Some(remote(3, 10)), diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 52c06cc5f6..2b90758fd7 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -6228,6 +6228,27 @@ 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 a_closed_source_is_not_requested_again_from_its_standing_route() { + 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(); + + 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, /// so the publisher's `unused()` resolves far below `TRACK_IDLE_LINGER`. /// Cached groups stay on the front for the linger; a returning reader