From d59f5203a3ac2079ca70e5082c662932731e3112 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 09:51:38 -0700 Subject: [PATCH 01/10] chore(quest): claim quest/m1/broadcast-lifecycle/ts-follow.md Quest-Claim: 823ba17f811c2825 From 1e2e31f5672ac4b8669f742843157c7cc5ed4aab Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 11:01:51 -0700 Subject: [PATCH 02/10] feat(mux)!: the TS follower follows instances and classifies failures by cause Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/export_test.rs | 501 ++++++++--------- rs/moq-mux/src/container/ts/follower.rs | 591 +++++++++------------ 2 files changed, 472 insertions(+), 620 deletions(-) diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index c41b2294b40..44db94aa166 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -264,7 +264,7 @@ async fn a_follower_keeps_hangz_after_catalog_eof() { drop((broadcast, track, catalog)); let mut follower = super::Follower::new(export) .unwrap() - .with_linger(Duration::from_secs(1)); + .with_follow(Duration::from_secs(1)); let (broadcast, track, mut catalog) = publish(); // Only catalog.json.z exists, so falling back to the default encoding cannot subscribe. tokio::time::timeout( @@ -8935,43 +8935,83 @@ fn finish((broadcast, mut catalog): (moq_net::broadcast::Producer, crate::catalo drop((broadcast, catalog)); } -/// A failure while the broadcast stays announced is the export's own: it fails after the -/// grace, without waiting out the linger. +/// A failure whose cause is not the publisher leaving is the export's own: it fails at once, +/// however long the follow. moq-transport's PUBLISH_DONE has no code for a lost route, so a +/// relay's copy reads there as an unrecognized one. #[tokio::test(start_paused = true)] -async fn a_follower_fails_with_the_broadcast_up_after_the_grace() { - let origin = crate::source::produce_origin(); - let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); - let mut follower = follower(&origin).await.with_linger(Duration::from_secs(10)); - assert!( - tokio::time::timeout(Duration::from_secs(1), follower.next()) - .await - .is_err(), - "nothing to mux yet" - ); +async fn a_follower_fails_at_once_on_its_own_failure() { + for cause in [moq_net::Error::Cancel, moq_net::Error::Remote(0)] { + let origin = crate::source::produce_origin(); + let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + let track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + let mut follower = follower(&origin).await.with_follow(Duration::from_secs(10)); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "nothing to mux yet" + ); - track.abort(moq_net::Error::Cancel); - let start = tokio::time::Instant::now(); - let end = follower.next().await; - assert!(end.is_err(), "the export's own failure fails the follower: {end:?}"); - assert_eq!(start.elapsed(), Duration::from_secs(1)); + track.abort(cause.clone()); + let start = tokio::time::Instant::now(); + let end = follower.next().await; + assert!( + matches!(&end, Err(crate::Error::Moq(err)) if err.to_string() == cause.to_string()), + "{cause}: {end:?}" + ); + assert_eq!(start.elapsed(), Duration::ZERO, "{cause}"); + } } -/// A replacement followed with stitching that never serves its catalog gives up at the -/// linger, rather than waiting on the catalog past it. +/// A track that lost its route, or whose session ended, says the publisher left even with the +/// broadcast still announced: the follower waits out the follow for a return, then fails with +/// that cause. #[tokio::test(start_paused = true)] -async fn a_follower_return_without_a_catalog_expires_with_the_linger() { +async fn a_follower_waits_out_the_follow_on_a_lost_route() { + let causes = [ + moq_net::Error::Unroutable, + moq_net::Error::Stream(moq_net::StreamError::Unroutable), + moq_net::Error::Session(moq_net::SessionError::Cancel), + ]; + for cause in causes { + let origin = crate::source::produce_origin(); + let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + let track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "nothing to mux yet" + ); + + track.abort(cause.clone()); + let start = tokio::time::Instant::now(); + let end = follower.next().await; + assert!( + matches!(&end, Err(crate::Error::Moq(err)) if err.to_string() == cause.to_string()), + "{cause}: {end:?}" + ); + assert_eq!(start.elapsed(), follow, "{cause}"); + } +} + +/// A replacement followed with stitching is awaited until its catalog arrives, with no deadline +/// of its own: one whose catalog request is never answered keeps the follower waiting past the +/// follow, until it goes. +#[tokio::test(start_paused = true)] +async fn a_follower_awaits_a_replacement_catalog_past_the_follow() { let origin = crate::source::produce_origin(); let first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger).with_stitch(true); - let start = tokio::time::Instant::now(); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow).with_stitch(true); finish(first); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); // Replaced, but the replacement's catalog request is never answered. @@ -8981,58 +9021,70 @@ async fn a_follower_return_without_a_catalog_expires_with_the_linger() { moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), ) .unwrap(); - let _unanswered = second.dynamic(); - let end = follower.next().await.unwrap(); - assert!(end.is_none(), "a return that never resumes is no return"); - assert_eq!(start.elapsed(), linger); + let unanswered = second.dynamic(); + assert!( + tokio::time::timeout(follow * 3, follower.next()).await.is_err(), + "awaiting the replacement's catalog past the follow" + ); + + drop((second, unanswered)); + let end = tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .expect("the replacement going ends the wait"); + assert!(matches!(end, Ok(None)), "the original clean end: {end:?}"); } /// A return of the same instance whose catalog subscription is answered but never delivers a -/// snapshot is no return either: the linger still bounds it. +/// snapshot is awaited past the follow, and once it fails too, it was no return: the export +/// ends as it did before it. #[tokio::test(start_paused = true)] -async fn a_follower_return_whose_catalog_never_snapshots_expires_with_the_linger() { +async fn a_follower_awaits_a_return_catalog_until_it_fails() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let first = publish_bare(&origin, Some(&epoch)); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger); - let start = tokio::time::Instant::now(); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow); finish(first); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); let second = origin .publish("live", moq_net::origin::Route::default().with_epoch(epoch)) .unwrap(); - let _silent = second.create_track(hang::Catalog::DEFAULT_NAME, None).unwrap(); - let end = tokio::time::timeout(linger * 3, follower.next()) + let silent = second.create_track(hang::Catalog::DEFAULT_NAME, None).unwrap(); + assert!( + tokio::time::timeout(follow * 3, follower.next()).await.is_err(), + "awaiting the return's catalog past the follow" + ); + + silent.abort(moq_net::Error::Unroutable).unwrap(); + let end = tokio::time::timeout(Duration::from_secs(1), follower.next()) .await - .expect("the linger bounds the return") - .unwrap(); - assert!(end.is_none(), "a return that never resumes is no return"); - assert_eq!(start.elapsed(), linger); + .expect("the failed return ends the wait"); + assert!(matches!(end, Ok(None)), "the original clean end: {end:?}"); + drop(second); } /// A return that goes again before its catalog resolves is no return yet: the follower keeps -/// lingering, and with nothing else back it yields the original clean end at the deadline. +/// following, and with nothing else back it yields the original clean end at the deadline. #[tokio::test(start_paused = true)] -async fn a_follower_keeps_lingering_when_a_return_goes_before_it_resolves() { +async fn a_follower_keeps_following_when_a_return_goes_before_it_resolves() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let first = publish_bare(&origin, Some(&epoch)); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow); let start = tokio::time::Instant::now(); finish(first); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); // Back, but its catalog request is never answered before it goes again. @@ -9048,11 +9100,11 @@ async fn a_follower_keeps_lingering_when_a_return_goes_before_it_resolves() { ); drop((second, unanswered)); - let end = tokio::time::timeout(linger * 3, follower.next()) + let end = tokio::time::timeout(follow * 3, follower.next()) .await - .expect("the linger bounds the wait"); + .expect("the follow bounds the wait"); assert!(matches!(end, Ok(None)), "the original clean end: {end:?}"); - assert_eq!(start.elapsed(), linger); + assert_eq!(start.elapsed(), follow); } /// Announce `epoch` at `live`, refusing every track request with `NotFound` until the @@ -9071,20 +9123,20 @@ fn publish_refusing(origin: &moq_net::origin::Producer, epoch: &moq_net::Epoch) } /// A replacement that refuses its catalog while still announced is no return: the follower -/// keeps lingering, and follows the next instance to serve. +/// keeps following, and follows the next instance to serve. #[tokio::test(start_paused = true)] -async fn a_follower_keeps_lingering_through_a_refused_return() { +async fn a_follower_keeps_following_through_a_refused_return() { let origin = crate::source::produce_origin(); let first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger).with_stitch(true); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow).with_stitch(true); let start = tokio::time::Instant::now(); finish(first); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); // Replaced, but the replacement refuses its catalog request. @@ -9093,7 +9145,7 @@ async fn a_follower_keeps_lingering_through_a_refused_return() { tokio::time::timeout(Duration::from_secs(2), follower.next()) .await .is_err(), - "the refusal keeps it lingering" + "the refusal keeps it following" ); refusing.abort(); @@ -9109,18 +9161,18 @@ async fn a_follower_keeps_lingering_through_a_refused_return() { finish(second); let end = follower.next().await; assert!(matches!(end, Ok(None)), "the replacement's clean end: {end:?}"); - assert_eq!(start.elapsed(), Duration::from_secs(4) + linger); + assert_eq!(start.elapsed(), Duration::from_secs(4) + follow); } /// A return that stays announced but refuses its catalog fails the follower only once the -/// linger runs out, with that refusal rather than the original clean end. +/// follow runs out, with that refusal rather than the original clean end. #[tokio::test(start_paused = true)] -async fn a_follower_reports_a_refused_return_at_the_linger() { +async fn a_follower_reports_a_refused_return_at_the_follow() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let first = publish_bare(&origin, Some(&epoch)); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow); let start = tokio::time::Instant::now(); finish(first); @@ -9128,17 +9180,17 @@ async fn a_follower_reports_a_refused_return_at_the_linger() { tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); let _refusing = publish_refusing(&origin, &epoch); - let end = tokio::time::timeout(linger * 3, follower.next()) + let end = tokio::time::timeout(follow * 3, follower.next()) .await - .expect("the linger bounds the wait"); + .expect("the follow bounds the wait"); assert!( matches!(end, Err(crate::Error::Moq(moq_net::Error::NotFound))), - "the refusal surfaces at the linger: {end:?}" + "the refusal surfaces at the follow: {end:?}" ); - assert_eq!(start.elapsed(), linger); + assert_eq!(start.elapsed(), follow); } /// A stitch whose replacement goes before it resolves is no switch yet: the export stays on @@ -9151,7 +9203,7 @@ async fn a_follower_stitches_onto_the_next_replacement_when_one_goes_before_it_r let mut follower = follower(&origin) .await .with_stitch(true) - .with_linger(Duration::from_secs(10)); + .with_follow(Duration::from_secs(10)); // Replaced, but the replacement's catalog request is never answered before it goes. let second = origin.publish("live", route()).unwrap(); @@ -9183,21 +9235,22 @@ async fn a_follower_stitches_onto_the_next_replacement_when_one_goes_before_it_r } /// A stitch that fails to resolve while a return is still awaiting its catalog goes back to -/// that return, so its end starts the linger over rather than the old deadline cutting it off. +/// that return, which ending before its first snapshot never resumed: the export ends as it did +/// before it, at the original follow. #[tokio::test(start_paused = true)] async fn a_follower_keeps_a_return_awaiting_its_catalog_when_a_stitch_goes() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let first = publish_bare(&origin, Some(&epoch)); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger).with_stitch(true); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow).with_stitch(true); let start = tokio::time::Instant::now(); finish(first); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); // Back, with a catalog yet to deliver its first snapshot. @@ -9228,18 +9281,18 @@ async fn a_follower_keeps_a_return_awaiting_its_catalog_when_a_stitch_goes() { finish(second); let end = follower.next().await; assert!(matches!(end, Ok(None)), "{end:?}"); - assert_eq!(start.elapsed(), Duration::from_secs(3) + linger); + assert_eq!(start.elapsed(), follow); } -/// A failure whose broadcast went and came back before the follower looked is no failure of -/// the export's own: the return is followed instead of failing at the grace. +/// A track that lost its route is the publisher leaving, so the return that follows is carried +/// on into. #[tokio::test(start_paused = true)] -async fn a_follower_follows_a_return_that_came_back_during_the_grace() { +async fn a_follower_follows_a_return_after_a_lost_route() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&epoch)); let track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); - let mut follower = follower(&origin).await.with_linger(Duration::from_secs(10)); + let mut follower = follower(&origin).await.with_follow(Duration::from_secs(10)); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await @@ -9247,15 +9300,9 @@ async fn a_follower_follows_a_return_that_came_back_during_the_grace() { "nothing to mux yet" ); - track.abort(moq_net::Error::Cancel); - assert!( - tokio::time::timeout(Duration::from_millis(100), follower.next()) - .await - .is_err(), - "waiting out the grace" - ); - // The end and the return both arrive before the follower looks again. - drop((broadcast, catalog)); + track.abort(moq_net::Error::Unroutable); + catalog.abort(moq_net::Error::Unroutable); + drop(broadcast); let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&epoch)); let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); assert!( @@ -9264,6 +9311,7 @@ async fn a_follower_follows_a_return_that_came_back_during_the_grace() { .is_err(), "carried on into the return" ); + assert!(follower.export().cataloged(), "the return served its catalog"); track.finish().unwrap(); finish((broadcast, catalog)); @@ -9305,27 +9353,29 @@ async fn a_follower_built_after_its_route_went_follows_the_return() { finish(first); tokio::time::sleep(Duration::from_secs(1)).await; - let linger = Duration::from_secs(10); - let mut follower = super::Follower::new(export).unwrap().with_linger(linger); + let follow = Duration::from_secs(10); + let mut follower = super::Follower::new(export).unwrap().with_follow(follow); assert!( tokio::time::timeout(Duration::from_secs(2), follower.next()) .await .is_err(), - "lingering" + "following" ); - let second = publish_bare(&origin, Some(&epoch)); + let mut second = publish_bare(&origin, Some(&epoch)); + std::ops::DerefMut::deref_mut(&mut second.1.modify().unwrap()); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), "carried on" ); + assert!(follower.export().cataloged(), "the return served its catalog"); - // Carried on into the return, so its end starts the linger over. + // Carried on into the return, so its end starts the follow over. finish(second); let end = follower.next().await; assert!(matches!(end, Ok(None)), "{end:?}"); - assert_eq!(start.elapsed(), Duration::from_secs(4) + linger); + assert_eq!(start.elapsed(), Duration::from_secs(4) + follow); } /// The export's own instance winning the path back from a more specific route that went is no @@ -9375,12 +9425,12 @@ async fn a_follower_takes_its_own_instance_winning_back_for_no_replacement() { } /// Without stitching, another instance taking the path fails the follower once the export -/// ends, however long the linger. +/// ends, however long the follow. #[tokio::test(start_paused = true)] async fn a_follower_fails_on_a_replacement_without_stitch() { let origin = crate::source::produce_origin(); let first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_linger(Duration::from_secs(10)); + let mut follower = follower(&origin).await.with_follow(Duration::from_secs(10)); finish(first); let _second = publish_bare(&origin, Some(&moq_net::Epoch::mint())); @@ -9395,7 +9445,7 @@ async fn a_follower_fails_on_a_replacement_without_stitch() { async fn a_follower_takes_an_epochless_return_as_a_replacement() { let origin = crate::source::produce_origin(); let first = publish_bare(&origin, None); - let mut follower = follower(&origin).await.with_linger(Duration::from_secs(10)); + let mut follower = follower(&origin).await.with_follow(Duration::from_secs(10)); finish(first); let _second = publish_bare(&origin, None); @@ -9406,35 +9456,37 @@ async fn a_follower_takes_an_epochless_return_as_a_replacement() { assert!(matches!(err, crate::Error::Replaced(_)), "{err}"); } -/// The same instance coming back within the linger carries the export on: when it ends -/// again, the linger starts over from there. +/// The same instance coming back within the follow carries the export on: when it ends +/// again, the follow starts over from there. #[tokio::test(start_paused = true)] async fn a_follower_continues_on_the_same_instance_returning() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let first = publish_bare(&origin, Some(&epoch)); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow); let start = tokio::time::Instant::now(); finish(first); assert!( tokio::time::timeout(Duration::from_secs(2), follower.next()) .await .is_err(), - "lingering" + "following" ); - let second = publish_bare(&origin, Some(&epoch)); + let mut second = publish_bare(&origin, Some(&epoch)); + std::ops::DerefMut::deref_mut(&mut second.1.modify().unwrap()); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), "carried on" ); + assert!(follower.export().cataloged(), "the return served its catalog"); finish(second); let end = follower.next().await.unwrap(); assert!(end.is_none()); - assert_eq!(start.elapsed(), Duration::from_secs(3) + linger); + assert_eq!(start.elapsed(), Duration::from_secs(3) + follow); } /// Pull frames from `follower` until it ends, or until nothing more comes within the drain. @@ -9470,7 +9522,7 @@ async fn a_follower_continues_through_a_same_epoch_handoff_after_a_gap() { .with_replay(); let mut follower = super::Follower::new(export) .unwrap() - .with_linger(Duration::from_secs(10)); + .with_follow(Duration::from_secs(10)); let start = tokio::time::Instant::now(); for ms in (0..200).step_by(20) { write_aac(&mut track, ms); @@ -9480,7 +9532,9 @@ async fn a_follower_continues_through_a_same_epoch_handoff_after_a_gap() { // Left unpolled across the handoff, and long enough for the exact route's retraction to // end the export's request before the prefix arrives. - drop((exact, catalog, track)); + track.abort(moq_net::Error::Unroutable); + catalog.abort(moq_net::Error::Unroutable); + drop(exact); tokio::time::sleep(Duration::from_secs(1)).await; let mut info = moq_net::broadcast::Info::new(); info.epoch = Some(epoch); @@ -9546,47 +9600,50 @@ async fn a_follower_ignores_a_covering_prefix_while_the_exact_route_serves() { assert!(matches!(end, Ok(None)), "no replacement: {end:?}"); } -/// Both the catalog subscription and its first snapshot share the restart's linger bound. +/// A mid-stream stitch awaits the replacement's catalog with no deadline of its own, so even a +/// zero follow switches to a cold replacement once its first snapshot arrives. #[tokio::test(start_paused = true)] -async fn a_midstream_stitch_catalog_expires_at_the_linger() { - for snapshot in [false, true] { - for linger in [Duration::ZERO, Duration::from_secs(10)] { - let origin = crate::source::produce_origin(); - let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let second = origin - .publish( - "live", - moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), - ) - .unwrap(); - let _unanswered = second.dynamic(); - let _silent = snapshot.then(|| second.create_track(hang::Catalog::DEFAULT_NAME, None).unwrap()); - let waiter = kio::Waiter::noop(); - if !linger.is_zero() { - assert!(follower.poll_next(&waiter).is_pending()); - tokio::time::advance(linger - Duration::from_millis(1)).await; - assert!(follower.poll_next(&waiter).is_pending()); - tokio::time::advance(Duration::from_millis(1)).await; - } - let std::task::Poll::Ready(Err(err)) = follower.poll_next(&waiter) else { - panic!("replacement catalog wait must fail at the linger, snapshot={snapshot}, linger={linger:?}"); - }; - assert!(err.to_string().contains("live"), "{err}"); - assert!(err.to_string().contains("catalog"), "{err}"); - } +async fn a_midstream_stitch_awaits_the_replacement_catalog() { + for follow in [Duration::ZERO, Duration::from_secs(10)] { + let origin = crate::source::produce_origin(); + let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + let mut follower = follower(&origin).await.with_stitch(true).with_follow(follow); + let epoch = moq_net::Epoch::mint(); + let (_second, mut catalog) = publish_bare(&origin, Some(&epoch)); + assert!( + tokio::time::timeout(Duration::from_secs(60), follower.next()) + .await + .is_err(), + "awaiting the replacement's catalog, follow={follow:?}" + ); + assert!(!follower.export().cataloged(), "follow={follow:?}"); + + std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap()); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "nothing to mux, follow={follow:?}" + ); + assert!(follower.export().cataloged(), "the late snapshot switches, follow={follow:?}"); + assert_eq!(follower.export().instance(), Some(&epoch)); } } -/// Re-pricing a replacement while a stitch resolves it is the same instance: it neither -/// restarts the resolve nor extends the linger bound. +/// A mid-stream stitch keeps the old instance playing while the replacement's catalog request +/// goes unanswered, rather than stalling the program on it. #[tokio::test(start_paused = true)] -async fn a_midstream_stitch_bound_survives_route_updates() { +async fn a_midstream_stitch_keeps_the_old_instance_playing_until_it_switches() { let origin = crate::source::produce_origin(); - let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let start = tokio::time::Instant::now(); + let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + let export = open_export(crate::source::resolved(&origin, "live").await) + .await + .unwrap() + .with_delay(RECORDING_MAX_AGE) + .with_replay(); + let mut follower = super::Follower::new(export).unwrap().with_stitch(true); + let second = origin .publish( "live", @@ -9594,78 +9651,28 @@ async fn a_midstream_stitch_bound_survives_route_updates() { ) .unwrap(); let _unanswered = second.dynamic(); - let reprice = async { - let mut cost = 0u64; - loop { - tokio::time::sleep(Duration::from_secs(1)).await; - cost += 1; - second.announce(second.route().unwrap().with_cost(cost)).unwrap(); - } - }; - let end = tokio::time::timeout(linger * 3, async { - tokio::select! { - end = follower.next() => end, - () = reprice => unreachable!("re-pricing never stops"), - } - }) - .await - .expect("updates extended the stitch bound"); - let err = end.expect_err("the stitch fails at its linger"); - assert!(err.to_string().contains("catalog"), "{err}"); - assert_eq!(start.elapsed(), linger); -} - -/// A catalog ready before the bound completes the switch, including zero linger. -#[tokio::test(start_paused = true)] -async fn a_midstream_stitch_catalog_within_linger_clears_the_bound() { - for linger in [Duration::ZERO, Duration::from_secs(10)] { - let origin = crate::source::produce_origin(); - let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let epoch = moq_net::Epoch::mint(); - let (_second, mut catalog) = publish_bare(&origin, Some(&epoch)); - let waiter = kio::Waiter::noop(); - if !linger.is_zero() { - assert!(follower.poll_next(&waiter).is_pending()); - tokio::time::advance(linger - Duration::from_millis(1)).await; - } - std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap()); - // Resolve the local async serving path so zero linger sees an already available - // subscription and snapshot, rather than a request still parked on the driver. - let broadcast = origin - .consume() - .request_broadcast("live", Some(epoch.clone())) - .await - .unwrap(); - let mut available = crate::catalog::Consumer::<()>::new(&broadcast, crate::catalog::CatalogFormat::Hang) - .await - .unwrap(); - use crate::catalog::Stream; - kio::wait(|waiter| available.poll_next(waiter)).await.unwrap().unwrap(); - let next = follower.poll_next(&waiter); - assert!(next.is_pending(), "{next:?}, linger={linger:?}"); - assert!(follower.export().cataloged()); - assert_eq!(follower.export().instance(), Some(&epoch)); - tokio::time::advance(Duration::from_secs(20)).await; - assert!( - follower.poll_next(&waiter).is_pending(), - "the first snapshot clears the bound" - ); + for ms in (0..200).step_by(20) { + write_aac(&mut track, ms); } + let (frames, end) = drain_follower(&mut follower).await; + assert!(end.is_none(), "still exporting: {end:?}"); + assert_eq!(pes_count(&frames), 10, "the old instance kept playing"); } /// A stitch onto a replacement that goes before it resolves leaves the export awaiting its -/// current replacement's catalog, by that replacement's own deadline. +/// current replacement's catalog, which still completes the switch when it arrives. #[tokio::test(start_paused = true)] -async fn a_nested_stitch_keeps_the_pending_catalog_deadline() { +async fn a_nested_stitch_keeps_awaiting_the_pending_replacement() { let origin = crate::source::produce_origin(); - let linger = Duration::from_secs(10); let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let start = tokio::time::Instant::now(); + let mut follower = follower(&origin) + .await + .with_stitch(true) + .with_follow(Duration::from_secs(10)); - // Switched onto a replacement whose catalog never delivers a snapshot. - let _second = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + // Switched onto a replacement whose catalog has no snapshot yet. + let epoch = moq_net::Epoch::mint(); + let (_second, mut catalog) = publish_bare(&origin, Some(&epoch)); assert!( tokio::time::timeout(Duration::from_secs(2), follower.next()) .await @@ -9688,66 +9695,22 @@ async fn a_nested_stitch_keeps_the_pending_catalog_deadline() { "resolving the next replacement" ); drop((third, unanswered)); - - let end = tokio::time::timeout(linger * 3, follower.next()) - .await - .expect("the second replacement's deadline still bounds it"); - let err = end.expect_err("a replacement without a catalog fails"); - assert!(err.to_string().contains("catalog"), "{err}"); - assert_eq!(start.elapsed(), linger); -} - -/// A replacement restored after its own deadline, when the next one goes before it resolves, -/// fails even with its late first snapshot ready. -#[tokio::test(start_paused = true)] -async fn a_nested_stitch_restored_after_its_deadline_fails() { - let origin = crate::source::produce_origin(); - let linger = Duration::from_secs(10); - let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let start = tokio::time::Instant::now(); - - // Switched onto a replacement whose catalog has no snapshot yet, bounded at 10s. - let epoch = moq_net::Epoch::mint(); - let (_second, mut catalog) = publish_bare(&origin, Some(&epoch)); assert!( - tokio::time::timeout(Duration::from_secs(2), follower.next()) + tokio::time::timeout(Duration::from_secs(30), follower.next()) .await .is_err(), - "awaiting the replacement's catalog" + "awaiting the second replacement's catalog again" ); - // Replaced again at 2s, bounded at 12s, but that one's catalog request is never answered. - let third = origin - .publish( - "live", - moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), - ) - .unwrap(); - let unanswered = third.dynamic(); + std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap()); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "resolving the next replacement" + "nothing to mux" ); - - // At 11s, the third goes and the second's first snapshot is ready, both past its deadline. - tokio::time::advance(linger + Duration::from_secs(1) - start.elapsed()).await; - drop((third, unanswered)); - std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap()); - // Deliver the snapshot through the local serving path, so it is ready on the next poll. - let broadcast = origin.consume().request_broadcast("live", Some(epoch)).await.unwrap(); - let mut available = crate::catalog::Consumer::<()>::new(&broadcast, crate::catalog::CatalogFormat::Hang) - .await - .unwrap(); - use crate::catalog::Stream; - kio::wait(|waiter| available.poll_next(waiter)).await.unwrap().unwrap(); - - let std::task::Poll::Ready(Err(err)) = follower.poll_next(&kio::Waiter::noop()) else { - panic!("a snapshot after the restored replacement's deadline must not count"); - }; - assert!(err.to_string().contains("catalog"), "{err}"); + assert!(follower.export().cataloged(), "the second replacement served its catalog"); + assert_eq!(follower.export().instance(), Some(&epoch)); } /// A replacement whose catalog closes before its first snapshot fails the stitch, rather than @@ -9759,7 +9722,7 @@ async fn a_midstream_stitch_catalog_closed_without_a_snapshot_fails() { let mut follower = follower(&origin) .await .with_stitch(true) - .with_linger(Duration::from_secs(10)); + .with_follow(Duration::from_secs(10)); let (_second, mut catalog) = publish_bare(&origin, Some(&moq_net::Epoch::mint())); assert!( @@ -9778,38 +9741,6 @@ async fn a_midstream_stitch_catalog_closed_without_a_snapshot_fails() { assert!(err.to_string().contains("catalog"), "{err}"); } -/// A snapshot that arrives after the deadline, before the follower polls again, is too late. -#[tokio::test(start_paused = true)] -async fn a_midstream_stitch_catalog_after_the_linger_fails() { - let origin = crate::source::produce_origin(); - let linger = Duration::from_secs(10); - let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let epoch = moq_net::Epoch::mint(); - let (_second, mut catalog) = publish_bare(&origin, Some(&epoch)); - let waiter = kio::Waiter::noop(); - assert!(follower.poll_next(&waiter).is_pending()); - - tokio::time::advance(linger).await; - std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap()); - // Deliver the snapshot through the local serving path, so it is ready on the next poll. - let broadcast = origin - .consume() - .request_broadcast("live", Some(epoch.clone())) - .await - .unwrap(); - let mut available = crate::catalog::Consumer::<()>::new(&broadcast, crate::catalog::CatalogFormat::Hang) - .await - .unwrap(); - use crate::catalog::Stream; - kio::wait(|waiter| available.poll_next(waiter)).await.unwrap().unwrap(); - - let std::task::Poll::Ready(Err(err)) = follower.poll_next(&waiter) else { - panic!("a snapshot after the linger must not count"); - }; - assert!(err.to_string().contains("catalog"), "{err}"); -} - /// The audio PTS in `frames`. fn audio_pts(frames: &[Frame]) -> Vec { let bytes: Vec = frames.iter().flat_map(|f| f.payload.iter().copied()).collect(); diff --git a/rs/moq-mux/src/container/ts/follower.rs b/rs/moq-mux/src/container/ts/follower.rs index f20a4f6fb97..cb90bf456c3 100644 --- a/rs/moq-mux/src/container/ts/follower.rs +++ b/rs/moq-mux/src/container/ts/follower.rs @@ -1,4 +1,4 @@ -//! [`Follower`]: an [`Export`] driven by its path's announcements, as a player is. +//! [`Follower`]: an [`Export`] driven by the publisher instances serving its path, as a player is. use std::future::Future; use std::pin::Pin; @@ -9,103 +9,94 @@ use moq_clock::Instant; use super::{Export, catalog}; use crate::container::Frame; - -/// How long an export failure waits for its broadcast to go before it counts as the export's own. -/// -/// A killed publisher's tracks can error just before its route is withdrawn, so the two need not -/// land together. -const CLOSE_GRACE: Duration = Duration::from_secs(1); - -/// The subscriptions carrying an export on into another broadcast. -type Resolved = crate::Result<(moq_net::broadcast::Consumer, crate::catalog::Consumer)>; +use crate::follow::{Event, Instances}; #[cfg(not(target_family = "wasm"))] -type Resolving = Pin> + Send>>; +type Subscribing = Pin>> + Send>>; #[cfg(target_family = "wasm")] -type Resolving = Pin>>>; +type Subscribing = Pin>>>>; -/// What the announcements say serves the exported path. -#[derive(Clone, Debug, PartialEq, Eq)] +/// What the instances say serves the exported path. enum Serving { - /// Nothing: every route went. - Gone, - /// The publisher instance being exported. + /// The instance being exported, with no change since the export carried on into it. Ours, - /// Another instance: a newer epoch, or any return of an epochless route. - Other(Option), + /// Nothing: the instance handed out last went. + Gone, + /// The instance being exported, handed out again since: it came back after a gap. + Return(moq_net::broadcast::Consumer), + /// Another instance: another epoch, or any return of an epochless route. + Other(moq_net::broadcast::Consumer), } -/// A first catalog snapshot the export awaits, since a subscribed catalog that never answers is -/// no return or switch. -struct Awaiting { - /// When the linger runs out. - deadline: Instant, - missing: Missing, +/// Subscribing to an instance's catalog, to carry the export on into it. +struct Subscription { + target: moq_net::broadcast::Consumer, + catalog: Subscribing, } -/// What a catalog still missing at the deadline means. +/// Why the export awaits the first catalog snapshot of the instance it carried on into. enum Missing { - /// A return that never resumed: the export ends as it did before it. - Return(crate::Result<()>), - /// A replacement switched to mid-stream that never served its catalog: a failure. + /// Carried on after the export ended with `end`: an instance that ends before it serves + /// leaves the export waiting for another until `deadline`. + Ended { end: crate::Result<()>, deadline: Instant }, + /// A replacement switched to mid-stream, which ending before it serves is a failure. Switch, } enum State { - /// Exporting, until the deadline when its catalog is yet to deliver a snapshot. - Running { awaiting: Option }, - /// The export ended with `end`, waiting for the announcements to say what follows. - Settling { + /// Exporting, while switching to a replacement when stitching one, and awaiting the + /// catalog's first snapshot after carrying on into an instance. + Running { + awaiting: Option, + switching: Option>, + }, + /// The export ended with `end`, waiting until `deadline` for an instance to carry on into. + Settling { end: crate::Result<()>, deadline: Instant }, + /// The export ended with `end`, and an instance serving the path is being subscribed. + Resuming { + subscription: Subscription, end: crate::Result<()>, - /// Until when a failure with the broadcast still announced waits for it to go. - grace: Option, - /// When the linger runs out. deadline: Instant, }, - /// Resolving the instance serving the path now, to carry the export on into it. - Following { - resolving: Resolving, - /// Bounds the resolution and then the catalog's first snapshot. - awaiting: Awaiting, - /// Where the export carries on, bound and all, if the instance goes before it resolves. - fallback: Box>, - }, /// Nothing follows. Done, } -/// Carries an [`Export`] across its broadcast's returns, following the path's announcements. +/// Carries an [`Export`] across its broadcast's returns, following the publisher instances +/// serving its path through [`Instances`]. /// -/// The same publisher instance (the same epoch) returning within the [linger](Self::with_linger) +/// The same publisher instance (the same epoch) returning within the [follow](Self::with_follow) /// continues the stream. Another instance taking the path fails with /// [`Error::Replaced`](crate::Error::Replaced), unless [stitching](Self::with_stitch), which /// switches the program to it at once. Subscriptions are sticky, so without stitching a -/// replaced publisher that stays up keeps the export until it ends. An export that fails -/// while its broadcast stays announced fails here too, without lingering. +/// replaced publisher that stays up keeps the export until it ends. +/// +/// A track failure is followed only when its cause says the publisher left: the route serving +/// it was lost (`Unroutable`, which a moq-lite relay resets a copy with once its upstream +/// session ends), or the session carrying it ended. Any other failure is the export's own and +/// fails here at once. /// -/// Announcements alone decide whether the broadcast returned: a return that fails to resolve -/// waits for the next one, and the linger running out reports that failure. +/// An instance that comes back is awaited until its catalog serves a first snapshot, with no +/// deadline but its own end. A return that refuses its catalog waits for the next change, and +/// the follow running out reports that refusal. A refusal to resolve the path at all fails at +/// once, as it does for every [`Instances`] reader. /// /// An epochless route has no instance to match, so any return of one is a replacement. pub struct Follower { export: Export>, state: State, - announced: moq_net::announce::Follow, + instances: Instances, path: moq_net::PathOwned, - /// The epoch of the instance being exported. - epoch: Option, serving: Serving, - /// Every route went since the export resolved its broadcast. - ended: bool, - /// An announcement arrived since the follower was built. - started: bool, + /// Taking the instances on hand when built, which replay the route the export resolved. + priming: bool, /// The origin closed, so nothing more is announced. closed: bool, - /// Why the last return failed to resolve, reported instead of the end if the linger runs out. + /// Why the last return refused its catalog, reported instead of the end if the follow runs out. failed: Option, - /// A return failed to resolve: nothing is followed until the next announcement. + /// The instance serving now refused its catalog: nothing is followed until the next change. refused: bool, - linger: Duration, + follow: Duration, stitch: bool, timer: Option>>, } @@ -116,50 +107,41 @@ impl Follower { /// Fails when the export's origin can never cover its path. pub fn new(export: Export>) -> crate::Result { let source = export.source(); - let announced = source.origin().follow(source.path())?; - let path = source.path().clone(); - let epoch = export.instance().cloned(); + let instances = Instances::new(source.origin(), source.path(), None)?; let mut follower = Self { + path: source.path().clone(), export, - state: State::Running { awaiting: None }, - announced, - path, - epoch, + state: State::Running { + awaiting: None, + switching: None, + }, + instances, serving: Serving::Ours, - ended: false, - started: false, + priming: true, closed: false, failed: None, refused: false, - linger: Duration::ZERO, + follow: Duration::ZERO, stitch: false, timer: None, }; - // Take the start naming the route the export resolved now, before anything else can - // fold into it. The cursor replays the routes on hand at once, so none means the route - // already went, and the next start is a return. - follower.poll_announced(&kio::Waiter::noop()); - if !follower.started { - follower.started = true; - follower.serving = Serving::Gone; - follower.ended = true; - } + // Take the instance on hand now, before a gap can fold into it: one handed out at once is + // the route the export resolved, so only a later one is a return. + follower.poll_instances(&kio::Waiter::noop())?; + follower.priming = false; Ok(follower) } - /// Wait up to `linger` for the same instance to return once the broadcast ends. Defaults to - /// [`Duration::ZERO`]. - /// - /// It bounds the whole return, and a mid-stream stitch from its restart to the - /// replacement catalog's first snapshot. Zero requires an already available catalog. - pub fn with_linger(mut self, linger: Duration) -> Self { - self.linger = linger; + /// Wait up to `follow` after the broadcast ends for the same instance, or with + /// [stitching](Self::with_stitch) a replacement, to come back. Defaults to + /// [`Duration::ZERO`], which carries on only into an instance already serving the path. + pub fn with_follow(mut self, follow: Duration) -> Self { + self.follow = follow; self } /// Follow another publisher instance taking the path, as a full program switch (see - /// [`Export::follow`]). Defaults to `false`. Its first catalog must arrive within - /// [`with_linger`](Self::with_linger), including when the old instance is still live. + /// [`Export::follow`]). Defaults to `false`. pub fn with_stitch(mut self, stitch: bool) -> Self { self.stitch = stitch; self @@ -185,322 +167,244 @@ impl Follower { } fn poll_state(&mut self, waiter: &kio::Waiter) -> Poll>> { - if let Some(expired) = self.expired() { - return Poll::Ready(expired); - } - loop { match std::mem::replace(&mut self.state, State::Done) { State::Done => return Poll::Ready(Ok(None)), - State::Running { mut awaiting } => { - self.poll_announced(waiter); - // Subscriptions are sticky, so a replacement leaves the export on the old - // instance until it ends, unless asked to switch at once. - if self.stitch && matches!(self.serving, Serving::Other(_)) { - tracing::info!(path = %self.path, "broadcast replaced, switching the program to the new instance"); - self.state = State::Following { - resolving: self.resolve(), - awaiting: Awaiting { - deadline: moq_clock::now() + self.linger, - missing: Missing::Switch, - }, - fallback: Box::new(State::Running { awaiting }), - }; - continue; + State::Running { + mut awaiting, + mut switching, + } => { + self.poll_instances(waiter)?; + if self.poll_switch(waiter, &mut switching)? { + awaiting = Some(Missing::Switch); } let next = self.export.poll_next(waiter); - if self.export.cataloged() { - awaiting = None; - } + awaiting = awaiting.filter(|_| !self.export.cataloged()); let end = match next { Poll::Ready(Ok(Some(frame))) => { - self.state = State::Running { awaiting }; + self.state = State::Running { awaiting, switching }; return Poll::Ready(Ok(Some(frame))); } - // A switch whose catalog closed before its first snapshot never served one. - Poll::Ready(Ok(None)) => match awaiting { - Some(Awaiting { - missing: Missing::Switch, - .. - }) => Err(self.missing_catalog()), - _ => Ok(()), - }, - Poll::Ready(Err(err)) => Err(err), + Poll::Ready(Err(err)) if !left(&err) => { + tracing::warn!(path = %self.path, %err, "export failed, so not following"); + return Poll::Ready(Err(err)); + } + Poll::Ready(end) => end.map(|_| ()), Poll::Pending => { - if let Some(awaiting) = awaiting.take_if(|awaiting| self.elapsed(waiter, awaiting.deadline)) - { - return Poll::Ready(self.expire(awaiting)); + // An instance that goes with its catalog still open never served. + if let Some(missing) = awaiting.take_if(|_| matches!(self.serving, Serving::Gone)) { + self.state = self.withdrawn(missing); + continue; } - self.state = State::Running { awaiting }; + self.state = State::Running { awaiting, switching }; return Poll::Pending; } }; - let now = moq_clock::now(); - // A failure ends the broadcast only if the broadcast goes too. - let grace = (end.is_err() && !self.ended && self.serving == Serving::Ours) - .then(|| now + CLOSE_GRACE.min(self.linger)); - if grace.is_none() { - self.lingering(&end); + // Nor did one whose catalog ended before its first snapshot. + if let Some(missing) = awaiting { + self.state = self.withdrawn(missing); + continue; } + self.following(&end); self.state = State::Settling { end, - grace, - deadline: now + self.linger, + deadline: moq_clock::now() + self.follow, }; } - State::Settling { - end, - mut grace, - deadline, - } => { - self.poll_announced(waiter); - // A broadcast that went, even one already back, was no failure of the export's own. - if grace.is_some() && (self.serving != Serving::Ours || self.ended) { - grace = None; - self.lingering(&end); - } - if let Some(at) = grace { - // One still announced cannot return, so the failure is the export's own. - if self.closed || self.elapsed(waiter, at) { - let err = end.expect_err("only a failure waits out the grace"); - tracing::warn!(path = %self.path, %err, "export failed with the broadcast still up, so not lingering"); - return Poll::Ready(Err(err)); - } - self.state = State::Settling { end, grace, deadline }; - return Poll::Pending; - } - - let follow = match &self.serving { - _ if self.refused => false, - Serving::Ours => self.ended, - Serving::Other(_) if self.stitch => true, + State::Settling { end, deadline } => { + self.poll_instances(waiter)?; + let target = match &self.serving { + _ if self.refused => None, + Serving::Return(target) => Some(target.clone()), + Serving::Other(target) if self.stitch => Some(target.clone()), Serving::Other(_) => return Poll::Ready(Err(crate::Error::Replaced(self.path.to_string()))), - Serving::Gone => false, + Serving::Ours | Serving::Gone => None, }; - if follow { - self.state = State::Following { - resolving: self.resolve(), - fallback: Box::new(State::Settling { - end: end.clone(), - grace: None, - deadline, - }), - awaiting: Awaiting { - deadline, - missing: Missing::Return(end), - }, + if let Some(target) = target { + self.state = State::Resuming { + subscription: self.subscribe(target), + end, + deadline, }; continue; } if self.closed || self.elapsed(waiter, deadline) { return Poll::Ready(self.gone(end)); } - self.state = State::Settling { end, grace, deadline }; + self.state = State::Settling { end, deadline }; return Poll::Pending; } - State::Following { - mut resolving, - awaiting, - fallback, + State::Resuming { + mut subscription, + end, + deadline, } => { - // Whatever the request waits on, the announcements say what serves the path - // now: rather than sit out a route that went, ask afresh from there. - if self.poll_announced(waiter) { - tracing::debug!(path = %self.path, "announcement while resolving, asking again"); - // A stitch still facing a replacement asks that one, within its original bound. - if matches!(*fallback, State::Running { .. }) && matches!(self.serving, Serving::Other(_)) { - self.state = State::Following { - resolving: self.resolve(), - awaiting, - fallback, - }; - continue; - } - // Otherwise it stays on the export it has, still awaiting what it was. - self.state = *fallback; - // The fallback's wait predates this poll, so it gets no same-poll grace. - if let Some(expired) = self.expired() { - return Poll::Ready(expired); - } + // Whatever the subscription waits on, the instances say what serves the path + // now: rather than sit out one that went, decide afresh from there. + if self.poll_instances(waiter)? { + tracing::debug!(path = %self.path, "instance changed while subscribing, deciding again"); + self.state = State::Settling { end, deadline }; continue; } - if let Poll::Ready(resolved) = waiter.poll_future(resolving.as_mut()) { - let (broadcast, catalog) = match resolved { - Ok(resolved) => resolved, - // A stitch from a live export fails loud: the replacement would not serve. - Err(err) if matches!(*fallback, State::Running { .. }) => return Poll::Ready(Err(err)), - // While lingering, a return that fails to resolve is no return yet: nothing - // serves the path until the next announcement says otherwise. The error is - // reported only if the linger runs out. - Err(err) => { - tracing::warn!(path = %self.path, %err, "broadcast failed to resolve, still waiting"); + match waiter.poll_future(subscription.catalog.as_mut()) { + Poll::Ready(Ok(catalog)) => { + self.carry_on(&subscription.target, catalog)?; + self.state = State::Running { + awaiting: Some(Missing::Ended { end, deadline }), + switching: None, + }; + } + // Refused, or gone before the instances say so: nothing is followed until + // the next change. Only the follow running out reports a refusal. + Poll::Ready(Err(err)) => { + tracing::warn!(path = %self.path, %err, "catalog subscription failed, still waiting"); + if !left(&err) { self.failed = Some(err); - self.refused = true; - self.state = *fallback; - if let Some(expired) = self.expired() { - return Poll::Ready(expired); - } - continue; } - }; - self.export.followed(&broadcast, catalog)?; - self.failed = None; - self.epoch = self.export.instance().cloned(); - self.serving = Serving::Ours; - self.ended = false; - tracing::info!( - path = %self.path, - epoch = self.epoch.as_ref().map(tracing::field::display), - "exporting broadcast" - ); - self.state = State::Running { - awaiting: Some(awaiting), - }; - continue; - } - if self.elapsed(waiter, awaiting.deadline) { - return Poll::Ready(self.expire(awaiting)); + self.refused = true; + self.state = State::Settling { end, deadline }; + } + Poll::Pending => { + self.state = State::Resuming { + subscription, + end, + deadline, + }; + return Poll::Pending; + } } - self.state = State::Following { - resolving, - awaiting, - fallback, - }; - return Poll::Pending; } } } } - /// Apply every announcement on hand, returning whether any changed what serves the path. - fn poll_announced(&mut self, waiter: &kio::Waiter) -> bool { + /// Drive a switch to the replacement serving the path when stitching, returning whether the + /// export carried on into it. + /// + /// The old instance keeps playing until the replacement's catalog is subscribed, however long + /// that takes, and a replacement that goes first is no switch. + fn poll_switch(&mut self, waiter: &kio::Waiter, switching: &mut Option>) -> crate::Result { + let target = match &self.serving { + Serving::Other(target) if self.stitch && !self.refused => target, + _ => { + *switching = None; + return Ok(false); + } + }; + if !switching.as_ref().is_some_and(|switching| switching.target.is_clone(target)) { + tracing::info!(path = %self.path, "broadcast replaced, switching the program to the new instance"); + *switching = Some(self.subscribe(target.clone())); + } + let subscription = switching.as_mut().expect("just set"); + match waiter.poll_future(subscription.catalog.as_mut()) { + Poll::Ready(Ok(catalog)) => { + let subscription = switching.take().expect("just polled"); + self.carry_on(&subscription.target, catalog)?; + Ok(true) + } + // Gone before the instances say so: no switch until the next change. + Poll::Ready(Err(err)) if left(&err) => { + tracing::warn!(path = %self.path, %err, "replacement went before its catalog subscribed"); + *switching = None; + self.refused = true; + Ok(false) + } + // The replacement would not serve, so the switch fails loud. + Poll::Ready(Err(err)) => Err(err), + Poll::Pending => Ok(false), + } + } + + /// Carry the export on into `target`, whose `catalog` is subscribed. + fn carry_on( + &mut self, + target: &moq_net::broadcast::Consumer, + catalog: crate::catalog::Consumer, + ) -> crate::Result<()> { + self.export.followed(target, catalog)?; + self.serving = Serving::Ours; + self.failed = None; + tracing::info!( + path = %self.path, + epoch = self.export.instance().map(tracing::field::display), + "exporting broadcast" + ); + Ok(()) + } + + /// Apply every change on hand, returning whether any came through. + fn poll_instances(&mut self, waiter: &kio::Waiter) -> crate::Result { let mut any = false; while !self.closed { - any |= match self.announced.poll_next(waiter) { - Poll::Ready(Some(event)) => self.apply(event), - Poll::Ready(None) => { - self.closed = true; - true + match self.instances.poll_next(waiter) { + Poll::Ready(Ok(Event::Start(broadcast))) => { + self.serving = match self.export.instance() { + // An epochless route has no instance to match, so only the broadcast + // being exported is its own. + None if broadcast.is_clone(&self.export.source().broadcast()) => Serving::Ours, + Some(epoch) if broadcast.info().epoch.as_ref() == Some(epoch) => match self.priming { + true => Serving::Ours, + false => Serving::Return(broadcast), + }, + _ => Serving::Other(broadcast), + }; } + Poll::Ready(Ok(Event::End)) => self.serving = Serving::Gone, + Poll::Ready(Err(crate::Error::Moq(moq_net::Error::Closed))) => self.closed = true, + Poll::Ready(Err(err)) => return Err(err), Poll::Pending => break, - }; + } + any = true; } self.refused &= !any; - any + Ok(any) } - /// Apply an announcement, returning whether it changed what serves the path. - fn apply(&mut self, event: moq_net::announce::Event) -> bool { - let first = !std::mem::replace(&mut self.started, true); - self.serving = match event { - moq_net::announce::Event::Start(announce) => { - let epoch = announce.route.epoch; - // The first start names the route the export resolved, as near as an epochless - // route can tell. - match epoch == self.epoch && (epoch.is_some() || first) { - true => Serving::Ours, - false => Serving::Other(epoch), - } - } - // The export's own instance winning the path back, as when a more specific route of - // another instance goes, is still its own. - moq_net::announce::Event::Restart(announce) => { - let epoch = announce.route.epoch; - match epoch.is_some() && epoch == self.epoch { - true => Serving::Ours, - false => Serving::Other(epoch), - } - } - moq_net::announce::Event::End(_) => { - self.ended = true; - Serving::Gone - } - // The same instance over another route, which subscriptions ride out. A gap that - // ended the export's request comes as an end and a start instead. - moq_net::announce::Event::Update(_) => return false, - }; - true + /// Begin subscribing to `target`'s catalog. + fn subscribe(&self, target: moq_net::broadcast::Consumer) -> Subscription { + let (broadcast, format) = (target.clone(), self.export.catalog_format()); + Subscription { + target, + catalog: Box::pin(async move { crate::catalog::Consumer::new(&broadcast, format).await }), + } } - /// Begin resolving the instance serving the path now. - fn resolve(&self) -> Resolving { - let epoch = match &self.serving { - Serving::Other(epoch) => epoch.clone(), - Serving::Ours | Serving::Gone => self.epoch.clone(), - }; - let request = self.export.source().origin().request_broadcast(&self.path, epoch); - let format = self.export.catalog_format(); - Box::pin(async move { - let broadcast = request.await?; - let catalog = crate::catalog::Consumer::new(&broadcast, format).await?; - Ok((broadcast, catalog)) - }) + /// Where the export goes once the instance it carried on into ends before its first snapshot. + fn withdrawn(&mut self, missing: Missing) -> State { + tracing::warn!(path = %self.path, "broadcast ended before serving its catalog"); + match missing { + Missing::Ended { end, deadline } => State::Settling { end, deadline }, + Missing::Switch => State::Settling { + end: Err(anyhow::anyhow!("replacement broadcast `{}` ended before serving its catalog", self.path).into()), + deadline: moq_clock::now() + self.follow, + }, + } } - fn lingering(&self, end: &crate::Result<()>) { - if self.linger.is_zero() { + fn following(&self, end: &crate::Result<()>) { + if self.follow.is_zero() { return; } match end { Ok(()) => { - tracing::info!(path = %self.path, linger = ?self.linger, "broadcast finished, waiting for it to return") + tracing::info!(path = %self.path, follow = ?self.follow, "broadcast finished, waiting for it to return") } Err(err) => { - tracing::warn!(path = %self.path, %err, linger = ?self.linger, "broadcast ended, waiting for it to return") + tracing::warn!(path = %self.path, %err, follow = ?self.follow, "broadcast ended, waiting for it to return") } } } - /// End a catalog wait already past its deadline, before a late snapshot can count. - /// - /// Only waits from an earlier poll are checked: within one poll a ready catalog still makes - /// it, which zero linger relies on. - fn expired(&mut self) -> Option>> { - let awaiting = match &self.state { - State::Running { awaiting } => awaiting.as_ref(), - State::Following { awaiting, .. } => Some(awaiting), - State::Settling { .. } | State::Done => None, - }; - if !awaiting.is_some_and(|awaiting| moq_clock::now() >= awaiting.deadline) { - return None; - } - let (State::Running { - awaiting: Some(awaiting), - } - | State::Following { awaiting, .. }) = std::mem::replace(&mut self.state, State::Done) - else { - unreachable!("only a running or following state awaits a catalog"); - }; - Some(self.expire(awaiting)) - } - - /// How a wait for the catalog's first snapshot ends once the linger runs out. - fn expire(&mut self, awaiting: Awaiting) -> crate::Result> { - match awaiting.missing { - Missing::Return(end) => self.gone(end), - Missing::Switch => Err(self.missing_catalog()), - } - } - - /// How the export ends once the linger runs out with nothing back: with why the last - /// return failed to resolve, if one did, else as the broadcast ended. + /// How the export ends once the follow runs out with nothing back: with why the last + /// return refused its catalog, if one did, else as the broadcast ended. fn gone(&mut self, end: crate::Result<()>) -> crate::Result> { - tracing::info!(path = %self.path, linger = ?self.linger, "broadcast did not return"); + tracing::info!(path = %self.path, follow = ?self.follow, "broadcast did not return"); match self.failed.take() { Some(err) => Err(err), None => end.map(|()| None), } } - fn missing_catalog(&self) -> crate::Error { - anyhow::anyhow!( - "replacement broadcast `{}` did not serve its catalog within {:?}", - self.path, - self.linger - ) - .into() - } - /// Whether `at` has come, registering `waiter` to wake then if not. fn elapsed(&mut self, waiter: &kio::Waiter, at: Instant) -> bool { if moq_clock::now() >= at { @@ -513,3 +417,20 @@ impl Follower { waiter.poll_future(timer.as_mut()).is_ready() } } + +/// Whether a failure says the publisher left rather than the export failing: the route serving +/// the broadcast was lost, or the session carrying it ended. +fn left(err: &crate::Error) -> bool { + let (crate::Error::Moq(err) | crate::Error::Hang(hang::Error::Moq(err))) = err else { + return false; + }; + use moq_net::{Error, StreamError}; + matches!( + err, + Error::Unroutable + | Error::Stream(StreamError::Unroutable | StreamError::Session(_)) + | Error::Session(_) + | Error::Transport(_) + | Error::IdleTimeout + ) +} From a24306a7f095716a073d8213549817918f5f7faa Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 11:12:09 -0700 Subject: [PATCH 03/10] feat(cli)!: `export ts|srt --linger` is `--follow` Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 21 +++--- doc/bin/srt.md | 6 +- doc/setup/redundancy.md | 7 +- quest/m1/api-audit/README.md | 3 +- quest/m1/broadcast-lifecycle/README.md | 1 - quest/m1/broadcast-lifecycle/ts-follow.md | 40 ----------- quest/m1/sink-restart-follow.md | 5 +- rs/justfile | 2 +- rs/moq-cli/MIGRATE.md | 9 +-- rs/moq-cli/src/args.rs | 77 +++++++++++++--------- rs/moq-cli/src/main.rs | 2 +- rs/moq-cli/src/srt.rs | 14 ++-- rs/moq-cli/src/subscribe.rs | 7 +- rs/moq-cli/tests/integration/export.rs | 42 ++++++------ rs/moq-mux/src/container/ts/export_test.rs | 14 ++-- rs/moq-mux/src/container/ts/follower.rs | 8 ++- rs/moq-srt/src/dial.rs | 7 +- rs/moq-srt/src/server.rs | 7 +- rs/moq-srt/src/ts.rs | 4 +- 19 files changed, 127 insertions(+), 149 deletions(-) delete mode 100644 quest/m1/broadcast-lifecycle/ts-follow.md diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 235a015c52d..f1b65d57ac2 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -449,25 +449,26 @@ and writes its packets back out unchanged, from its first group on, each run between two PCRs `--delay` after the first one's time on the source clock. The output is exactly as conformant as the source, and two exporters emit the same bytes. A run that arrives after its deadline is dropped and counted, leaving a -continuity error, and the rest carry on. It takes neither `--linger`, +continuity error, and the rest carry on. It takes neither `--follow`, `--stitch` nor `--mux-rate`. Only the hang catalog names the track, so an `.msf` broadcast needs `--catalog-format hang`. -A stdout export ends with the broadcast. `export ts --linger 10s` waits that +A stdout export ends with the broadcast. `export ts --follow 10s` waits that long for the same publisher instance to come back, and carries on with the same stream. A return that cannot be read yet, such as one still routed through a -dead session, keeps it waiting for the next announcement; if the linger runs +dead session, keeps it waiting for the next announcement; if the follow runs out first, it exits 1 with that error. A replacement (a restarted publisher, under a new [epoch](/concept/moq-lite#publisher-epochs)) exits 1 unless `--stitch` follows it as a full program switch: a new PMT from its catalog, with every PID flagging -the break. `--linger` also bounds a mid-stream stitch from the replacement -announcement through its first catalog snapshot; expiry exits 1. The default -`--linger 0s` follows only a replacement whose catalog is already available. -Without `--stitch` the export stays on the old publisher while it is +the break. A stitch waits for the replacement's first catalog snapshot until it +arrives or the replacement goes, keeping the old publisher playing meanwhile, +so even the default `--follow 0s` switches to a replacement that is still +starting. Without `--stitch` the export stays on the old publisher while it is up. Only lite-07 and moq-transport draft 17+ sessions carry epochs, so on the -default version every return is a replacement. An export that fails while the broadcast stays up, such as on -a codec TS cannot carry, exits 1 without lingering. Only `ts` can mark a break, -so the other formats refuse `--linger`. +default version every return is a replacement. A failure other than the +publisher leaving, such as a codec TS cannot carry, exits 1 at once. Only `ts` +can mark a break, so the other formats refuse `--linger`, the released spelling +of `--follow`. ## Reports and feedback diff --git a/doc/bin/srt.md b/doc/bin/srt.md index 1e0069edfc4..d8f65e9b144 100644 --- a/doc/bin/srt.md +++ b/doc/bin/srt.md @@ -28,11 +28,9 @@ Each connection publishes under a fresh [epoch](/concept/moq-lite#publisher-epochs), so an encoder that reconnects while its stale connection is still open replaces it at once: viewers see the broadcast restart, and their next subscribe reaches the new feed. An -`export srt` stream ends with its broadcast; `--linger` and `--stitch` follow a +`export srt` stream ends with its broadcast; `--follow` and `--stitch` follow a return or a replacement on the same connection, as they do for -[`export ts`](/bin/cli#retention-and-latency). The linger also bounds a -mid-stream replacement's first catalog; expiry ends the export with an error. -With the default `--linger 0s`, that catalog must already be available. +[`export ts`](/bin/cli#retention-and-latency). A multi-program feed is refused unless `--program` picks one: `--program 2` imports program 2 alone, and `--program all` diff --git a/doc/setup/redundancy.md b/doc/setup/redundancy.md index 6e3c5961e8f..6e9279ae4fd 100644 --- a/doc/setup/redundancy.md +++ b/doc/setup/redundancy.md @@ -70,14 +70,13 @@ For MPEG-TS output, opt into following replacements: ```bash moq --connect https://edge.example.com/anon \ --connect-version moq-transport-17 --broadcast event.hang \ - export ts --stitch --linger 10s > event.ts + export ts --stitch --follow 10s > event.ts ``` `--stitch` follows a replacement as a program switch: its catalog supplies a new PMT and every PID flags the discontinuity. Starting a backup can cause -this switch too. `--linger 10s` bounds waiting after an end and waiting for a -replacement's first catalog; choose that window for your application. It does -not make the switch hitless. Without `--stitch`, the TS export stays on the +this switch too. `--follow 10s` bounds waiting for a publisher after an end; +choose that window for your application. It does not make the switch hitless. Without `--stitch`, the TS export stays on the old instance while it is up, then fails on a replacement. Other stdout export formats end with their broadcast; they do not follow a replacement. TS passthrough also cannot stitch. See [CLI retention and latency](/bin/cli#retention-and-latency). diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md index c68d43362f1..4e534bcf284 100644 --- a/quest/m1/api-audit/README.md +++ b/quest/m1/api-audit/README.md @@ -61,8 +61,7 @@ Rejected in planning: `close` collides with `AutoCloseable`, and no single verb fits every language's reserved names. - A follow policy enum folding `linger` and `stitch`: one name per meaning - ([TS export follows by cause](/quest/m1/broadcast-lifecycle/ts-follow.md) - renames the export wait `--follow`) fixes the confusion more cheaply. + (the export wait is `--follow`) fixes the confusion more cheaply. ## Required diff --git a/quest/m1/broadcast-lifecycle/README.md b/quest/m1/broadcast-lifecycle/README.md index 04554384a88..0bb444eb6de 100644 --- a/quest/m1/broadcast-lifecycle/README.md +++ b/quest/m1/broadcast-lifecycle/README.md @@ -47,7 +47,6 @@ timer on main at 1582fdc07d: ## Required - [Stateless broadcasts](/quest/m1/broadcast-lifecycle/stateless.md) - `Finish`, its linger, and finished-only supersession are gone from moq-net, `@moq/net`, moq-ffi, moq-c, and every binding -- [TS export follows by cause](/quest/m1/broadcast-lifecycle/ts-follow.md) - `export ts|srt --linger` becomes `--follow `, a replacement's catalog has no separate deadline, and a track failure is classified by its cause instead of a 1s grace - [SRT tells a clean end from a failure](/quest/m1/broadcast-lifecycle/srt-end.md) - an SRT ingest whose peer goes silent aborts instead of finishing as if it had shut down - [Timer cleanups](/quest/m1/broadcast-lifecycle/timers.md) - `ConnectionProps.linger` is gone and stale linger references are fixed diff --git a/quest/m1/broadcast-lifecycle/ts-follow.md b/quest/m1/broadcast-lifecycle/ts-follow.md deleted file mode 100644 index d9e2fab3ea7..00000000000 --- a/quest/m1/broadcast-lifecycle/ts-follow.md +++ /dev/null @@ -1,40 +0,0 @@ -# [M] TS export follows by cause - -## Goal - -`moq export ts --linger` and `moq export srt --linger` become `--follow -`: how long to wait for the publisher, or with `--stitch` a -replacement, to come back. The default 0 means don't follow. The Rust -`container::ts::Follower::with_linger` becomes `with_follow`, matching -moq-srt's `Subscribe::with_follow` and `Client::with_follow`. A replacement's first catalog -is awaited until it arrives or the replacement withdraws, with no deadline of -its own. A track failure is classified by its cause, not by waiting -`CLOSE_GRACE` (1s) to see whether the announcement also goes. - -## Plan - -Decided with the maintainer on 2026-10-23: - -- One knob, one meaning. Today `--linger` also bounds the stitch catalog - fetch (`rs/moq-mux/src/container/ts/follower.rs`, `Awaiting { deadline }`), - so `--stitch --linger 0` fails on a cold remote replacement. The - replacement's own lifetime is the bound instead. -- `CLOSE_GRACE` exists because the track error and the announcement removal - arrive on different streams. Find what tells "the publisher left" from "this - export failed" in the track's own end: the abort cause a relay gives a copy - whose upstream went away. If no cause distinguishes them today, add one at - the source. A wire error code change updates the drafts. -- The hidden `--linger` on the other container exports - (`rs/moq-cli/src/args.rs`) keeps refusing; point its message at `--follow` - only where following applies. -- Move the follower onto `moq_mux::follow::Instances` here (decided - 2026-10-24), as [readers follow restarts](/quest/m1/sink-restart-follow.md) - did for `moq play`, `moqsrc`, and `transcode`. `with_linger` waits from the - announcement through resolution and the first catalog, but `Instances` - yields `Start` only after resolution, so settle what `--follow` bounds - before moving the timer onto it. Coordinate the name with that quest. -- Docs: `doc/bin/cli.md`, plus a migration note for moq-cli, moq-mux, and - moq-srt in the Impact bullet. - -Public API: moq-mux rename (breaking); moq-cli flag rename. -Wire: none unless the failure cause needs a new code. diff --git a/quest/m1/sink-restart-follow.md b/quest/m1/sink-restart-follow.md index 27e1e97828d..6d4084a885a 100644 --- a/quest/m1/sink-restart-follow.md +++ b/quest/m1/sink-restart-follow.md @@ -36,9 +36,8 @@ one stream. Decided with the maintainer on 2026-10-24, when the play, moqsrc, and transcode moves landed: -- The TS follower moves onto `Instances` in - [TS export follows by cause](/quest/m1/broadcast-lifecycle/ts-follow.md), - which reworks the same wait, so its timers change once. +- The TS follower follows `Instances`, with `--follow` bounding only the wait + for a publisher after an end. - `export hls` gives each instance its own playlist URL. While the old publisher stays live, `Server::broadcaster` (`rs/moq-hls/src/server/mod.rs`) answers new requests from its cached broadcaster, so an overlapping diff --git a/rs/justfile b/rs/justfile index df314f95b21..59150fa1e46 100644 --- a/rs/justfile +++ b/rs/justfile @@ -334,7 +334,7 @@ quic-sim: send-depth: bash sh/rs/send-depth.sh -# Run moq-cli's `moq export` process tests, which feed a real relay and wait out lingers in real time. +# Run moq-cli's `moq export` process tests, which feed a real relay and wait out follows in real time. [positional-arguments] cli-export *args: just rs test -p moq-cli --test integration --run-ignored only -E 'test(/^export::/)' "$@" diff --git a/rs/moq-cli/MIGRATE.md b/rs/moq-cli/MIGRATE.md index c8153a06780..62517583371 100644 --- a/rs/moq-cli/MIGRATE.md +++ b/rs/moq-cli/MIGRATE.md @@ -79,13 +79,14 @@ Rename `--max-age` to `--max-delay` on those exports. Each frame is written that long after its decode time; the older staleness spellings are refused with `--delay` named. Rename `--max-age` to `--delay`. -### `moq export ts --linger` follows only the same publisher instance +### `moq export ts --linger` is `--follow`, and follows only the same publisher instance -A restarted publisher exits 1 unless `--stitch` follows it as a program switch. -`moq export srt` takes the same two flags. +Rename `export ts --linger` and `export srt --linger` to `--follow`; the old spelling stops the run naming it. +A restarted publisher exits 1 unless `--stitch` follows it as a program switch, and a stitched replacement's first catalog is awaited until it arrives or the replacement goes, so `--follow 0s` no longer refuses a cold one. +A failure other than the publisher leaving, such as a codec TS cannot carry, exits at once rather than after a 1 s grace. ```bash -moq export ts --linger 10s --stitch ... +moq export ts --follow 10s --stitch ... ``` ### `moq export fmp4` fixes its track set at the init segment diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index d9ee240b7aa..415f4a7f515 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -365,8 +365,8 @@ impl Invocation { } if let Some(stdout) = export.sink.stdout() { anyhow::ensure!( - stdout.linger.is_zero() || matches!(stdout.format, SubscribeFormat::Ts), - "--linger needs an output that can mark a restart, and only `export ts` can" + stdout.follow.is_zero() || matches!(stdout.format, SubscribeFormat::Ts), + "--linger is gone, and following a broadcast's return needs an output that can mark the restart: only `export ts --follow` can" ); if matches!(stdout.format, SubscribeFormat::H264 | SubscribeFormat::H265) { anyhow::ensure!( @@ -1090,7 +1090,7 @@ impl ExportSink { let container = |format, container: &Container| Stdout { format, max_delay: container.max_delay.into(), - linger: container.linger.into(), + follow: container.linger.into(), stitch: false, fragment_duration: None, mux_rate: None, @@ -1099,7 +1099,7 @@ impl ExportSink { let buffered = |format, args: &Buffered| Stdout { format, max_delay: args.delay.into(), - linger: args.linger.into(), + follow: args.linger.into(), stitch: false, fragment_duration: None, mux_rate: None, @@ -1117,7 +1117,7 @@ impl ExportSink { Self::Ts(args) => Stdout { format: SubscribeFormat::Ts, max_delay: args.delay.into(), - linger: args.linger.into(), + follow: args.follow.into(), stitch: args.stitch, fragment_duration: None, mux_rate: args.mux_rate, @@ -1137,7 +1137,9 @@ pub struct Stdout { /// The staleness budget, also the FLV/MKV release delay and how far behind the source `ts` /// runs (`--delay`). pub max_delay: Duration, - pub linger: Duration, + /// How long to wait for the broadcast to come back (`ts` only; the others refuse their + /// hidden `--linger`). + pub follow: Duration, /// Follow a replacement as a program switch (`ts` only). pub stitch: bool, pub fragment_duration: Option, @@ -1154,8 +1156,8 @@ pub struct Container { #[usage(long, default = "500ms")] pub max_delay: moq_tokio::cli::Duration, - /// Accepted only to refuse a nonzero value with a pointer to `ts`, the one format - /// that can mark where a returned broadcast restarts. + /// The released spelling of `export ts --follow`, accepted only to refuse a nonzero value: + /// only TS can mark where a returned broadcast restarts. #[usage(long, default = "0s", hide = true)] pub linger: moq_tokio::cli::Duration, @@ -1190,7 +1192,8 @@ pub struct Buffered { /// drops none as late, but still skips a stalled group once a newer one is ready. #[usage(long, default = "500ms")] pub delay: moq_tokio::cli::Duration, - /// Accepted only to refuse a nonzero value: only TS can mark a broadcast restart. + /// The released spelling of `export ts --follow`, accepted only to refuse a nonzero value: + /// only TS can mark a broadcast restart. #[usage(long, default = "0s", hide = true)] pub linger: moq_tokio::cli::Duration, /// An older spelling of the staleness setting. @@ -1238,12 +1241,12 @@ pub struct Transport { #[usage(long, default = "1s")] pub delay: moq_tokio::cli::Duration, - /// How long to wait for the same publisher instance to come back once it ends (e.g. `10s`), - /// then carry on with the same stream. A replacement exits 1 unless `--stitch` is passed. - /// An export that fails while the broadcast is still up exits without waiting. - /// Also bounds a stitched replacement through its first catalog; zero requires it already available. + /// How long to wait for the same publisher instance, or with `--stitch` a replacement, to + /// come back once the broadcast ends (e.g. `10s`), then carry on with the same stream. + /// Zero follows only an instance already serving. A replacement exits 1 unless `--stitch` is + /// passed. A failure other than the publisher leaving exits at once. #[usage(long, default = "0s")] - pub linger: moq_tokio::cli::Duration, + pub follow: moq_tokio::cli::Duration, /// Follow another publisher instance that replaces the broadcast, as a full program switch: /// a new PMT from its catalog, flagged as a break on every PID. @@ -1259,9 +1262,13 @@ pub struct Transport { /// Write the multiplex an `import ts --passthrough` published back out byte-identical, /// paced on its own PCR, `--delay` behind it. Packets that arrive later than that are /// dropped, leaving a continuity error. - #[usage(long, conflicts("--linger", "--stitch", "--mux-rate"))] + #[usage(long, conflicts("--follow", "--stitch", "--mux-rate"))] pub passthrough: bool, + /// The released spelling of [`Self::follow`]. + #[usage(long = "linger", hide = true)] + linger: Option, + /// The released spelling of [`Self::delay`]. #[usage(long = "max-age", hide = true)] max_age: Option, @@ -1281,6 +1288,9 @@ impl Transport { if self.latency_max.is_some() { found.changed("--latency-max", None, "--delay", BEHIND); } + if self.linger.is_some() { + found.flag("--linger", None, "--follow"); + } found } } @@ -1318,20 +1328,25 @@ mod tests { let _ = Stage::to_kdl(); } - /// Only TS can mark where a returned broadcast restarts, so only `export ts` may linger. + /// Only TS can mark where a returned broadcast restarts, so only `export ts` follows one. + /// The released `--linger` points there, as a rename on `ts` and a refusal elsewhere. #[test] - fn linger_is_ts_only() { - let parse = |format: &str, linger: &str| { - Invocation::try_parse_from(["moq", "--connect", "http://relay", "export", format, "--linger", linger]) - .unwrap() + fn follow_is_ts_only() { + let parse = |format: &str, flag: &str, value: &str| { + Invocation::try_parse_from(["moq", "--connect", "http://relay", "export", format, flag, value]) }; - assert!(parse("ts", "10s").validate().is_ok()); - for format in ["fmp4", "h264", "h265"] { - let err = parse(format, "10s").validate().unwrap_err().to_string(); - assert!(err.contains("--linger"), "{format}: {err}"); + assert!(parse("ts", "--follow", "10s").unwrap().validate().is_ok()); + let Err(err) = parse("ts", "--linger", "10s") else { + panic!("the released spelling must not start a run"); + }; + assert!(err.to_string().contains("--linger -> --follow"), "{err}"); + for format in ["fmp4", "mkv", "flv", "h264", "h265"] { + assert!(parse(format, "--follow", "10s").is_err(), "{format} takes no --follow"); + let err = parse(format, "--linger", "10s").unwrap().validate().unwrap_err().to_string(); + assert!(err.contains("export ts --follow"), "{format}: {err}"); assert!( - parse(format, "0s").validate().is_ok(), - "{format}: no linger is always fine" + parse(format, "--linger", "0s").unwrap().validate().is_ok(), + "{format}: a zero --linger is always fine" ); } } @@ -1536,7 +1551,7 @@ mod tests { let stdout = export.sink.stdout().unwrap(); assert!(stdout.passthrough); assert_eq!(stdout.max_delay, Duration::from_secs(2)); - for flags in [&["--stitch"][..], &["--linger", "10s"], &["--mux-rate", "20000000"]] { + for flags in [&["--stitch"][..], &["--follow", "10s"], &["--mux-rate", "20000000"]] { assert!(parse(flags).is_err(), "{flags:?} must not start a run"); } } @@ -1777,16 +1792,16 @@ mod tests { ); } - /// `export srt` follows its broadcast with the same `--linger` and `--stitch` as `export ts`. + /// `export srt` follows its broadcast with the same `--follow` and `--stitch` as `export ts`. #[test] - fn export_srt_takes_linger_and_stitch() { + fn export_srt_takes_follow_and_stitch() { let cli = Invocation::try_parse_from([ "moq", "export", "srt", "--listen", "[::]:9000", - "--linger", + "--follow", "10s", "--stitch", ]) @@ -1797,7 +1812,7 @@ mod tests { let ExportSink::Srt(args) = &export.sink else { panic!("an export srt stage"); }; - assert_eq!(Duration::from(args.linger), Duration::from_secs(10)); + assert_eq!(Duration::from(args.follow), Duration::from_secs(10)); assert!(args.stitch); assert!( Invocation::try_parse_from(["moq", "import", "srt", "--listen", "[::]:9000", "--stitch"]).is_err(), diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 46763680dbf..7d08d39217c 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -821,7 +821,7 @@ fn spawn_export( let args = SubscribeArgs { format: stdout.format, max_delay: stdout.max_delay, - linger: stdout.linger, + follow: stdout.follow, stitch: stdout.stitch, fragment_duration: stdout.fragment_duration, mux_rate: stdout.mux_rate, diff --git a/rs/moq-cli/src/srt.rs b/rs/moq-cli/src/srt.rs index caa20cef428..a08f65e35d4 100644 --- a/rs/moq-cli/src/srt.rs +++ b/rs/moq-cli/src/srt.rs @@ -53,12 +53,12 @@ pub struct ExportArgs { #[usage(flatten)] pub endpoint: Args, - /// How long to wait for the same publisher instance to come back once it ends (e.g. `10s`), - /// then carry on with the same stream. A replacement ends the stream unless `--stitch` - /// is passed. - /// Also bounds a stitched replacement through its first catalog; zero requires it already available. + /// How long to wait for the same publisher instance, or with `--stitch` a replacement, to + /// come back once the broadcast ends (e.g. `10s`), then carry on with the same stream. + /// Zero follows only an instance already serving. A replacement ends the stream unless + /// `--stitch` is passed. #[usage(long, default = "0s")] - pub linger: moq_tokio::cli::Duration, + pub follow: moq_tokio::cli::Duration, /// Follow another publisher instance that replaces the broadcast, as a full program switch /// on the same SRT connection. @@ -146,7 +146,7 @@ pub async fn listen_export( args: ExportArgs, ) -> anyhow::Result<()> { let mut server = Server::bind(addr, Duration::from(args.endpoint.latency)).await?; - let (follow, stitch) = (args.linger.into(), args.stitch); + let (follow, stitch) = (args.follow.into(), args.stitch); tracing::info!(%addr, %name, "SRT listening (export)"); notify_ready(); @@ -208,7 +208,7 @@ pub async fn connect_export( let client = moq_srt::Client::new(addr, resource) .with_max_delay(args.endpoint.latency.into()) - .with_follow(args.linger.into()) + .with_follow(args.follow.into()) .with_stitch(args.stitch); Ok(client.publish(&origin, &name).await?) } diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index ee20cc95a8c..ffa3f2f7e24 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -181,8 +181,9 @@ pub struct SubscribeArgs { /// this far behind the source (`--delay`). pub max_delay: Duration, - /// How long to wait for the broadcast to come back after it ends (TS only). - pub linger: Duration, + /// How long to wait for the broadcast, or with `stitch` a replacement, to come back after + /// it ends (TS only). + pub follow: Duration, /// Follow another publisher instance replacing the broadcast, as a program switch (TS only). pub stitch: bool, @@ -421,7 +422,7 @@ impl Subscribe { // written as it comes. The path's announcements carry it across the broadcast's // returns, as they drive a player. let mut ts = moq_mux::container::ts::Follower::new(ts)? - .with_linger(self.args.linger) + .with_follow(self.args.follow) .with_stitch(self.args.stitch); // Reports a track that stops reaching the output while the rest keeps flowing, // the way `publish` reports one that stops arriving. diff --git a/rs/moq-cli/tests/integration/export.rs b/rs/moq-cli/tests/integration/export.rs index 48406d71a19..f546e418e1c 100644 --- a/rs/moq-cli/tests/integration/export.rs +++ b/rs/moq-cli/tests/integration/export.rs @@ -1,9 +1,9 @@ -//! `moq export ts --linger` and `--stitch` over a real relay: the export rides out the same +//! `moq export ts --follow` and `--stitch` over a real relay: the export rides out the same //! publisher instance leaving and coming back, switches to a replacement only when asked, and //! exits with the verdict of the broadcast's last end. The other formats fail once the route //! they read dies. //! -//! Each test spawns `moq` processes that feed clips and wait out lingers in real time, tens of +//! Each test spawns `moq` processes that feed clips and wait out follows in real time, tens of //! seconds between them, so they are ignored here and run nightly as `just rs cli-export`. #![cfg(unix)] @@ -123,9 +123,9 @@ fn moq(relay: &Relay, args: &[&str]) -> tokio::process::Command { command } -/// Start `moq export ts --linger `, collecting its output and diagnostics. -fn export(relay: &Relay, linger: &str) -> Export { - export_with(relay, &[], &["--linger", linger]) +/// Start `moq export ts --follow `, collecting its output and diagnostics. +fn export(relay: &Relay, follow: &str) -> Export { + export_with(relay, &[], &["--follow", follow]) } /// A container both `import` and `export` speak, with a clip of it. @@ -220,7 +220,7 @@ fn collect( received.lock().unwrap().push((Instant::now(), output.len())); } Some(fence) = requested.recv() => { - // The follower entered linger only after each old stdout write and flush. + // The follower started following only after each old stdout write and flush. // This task owns the reader, so no read can race the queued-byte count. let mut pending: libc::c_int = 0; // SAFETY: the pipe is live and pending points to a writable c_int. @@ -313,14 +313,14 @@ fn adaptation_flags(packet: &[u8]) -> Option { #[tokio::test] #[ignore = "real-time process test: just rs cli-export"] -async fn a_clean_finish_exits_zero_once_the_linger_expires() { +async fn a_clean_finish_exits_zero_once_the_follow_expires() { let relay = relay().await; let mut export = export(&relay, "1s"); let (mut publisher, feeding) = import(&relay); let stdin = feeding.await.unwrap(); export.output_past(0).await; - // The linger can't start before stdin closes, but it can before the publisher exits. + // The follow can't start before stdin closes, but it can before the publisher exits. let closed = Instant::now(); drop(stdin); assert!(wait(&mut publisher).await.success()); @@ -332,13 +332,13 @@ async fn a_clean_finish_exits_zero_once_the_linger_expires() { ); assert!( closed.elapsed() >= Duration::from_millis(900), - "the export waited out its linger" + "the export waited out its follow" ); } #[tokio::test] #[ignore = "real-time process test: just rs cli-export"] -async fn a_drop_exits_one_once_the_linger_expires() { +async fn a_drop_exits_one_once_the_follow_expires() { let relay = relay().await; let mut export = export(&relay, "1s"); @@ -358,7 +358,7 @@ async fn a_drop_exits_one_once_the_linger_expires() { /// An export that fails on its own, with the broadcast still up, has no return to wait for. #[tokio::test] #[ignore = "real-time process test: just rs cli-export"] -async fn an_export_failure_with_the_broadcast_up_exits_without_lingering() { +async fn an_export_failure_with_the_broadcast_up_exits_without_following() { let relay = relay().await; let export = export(&relay, "20s"); @@ -416,8 +416,8 @@ async fn interrupted(export: &mut Export, publisher: (Child, JoinHandle::new(), "every PID flags the break"); } -/// One publisher reconnects within the linger, continuing its capture under the same epoch +/// One publisher reconnects within the follow, continuing its capture under the same epoch /// and program before delivering a clean final finish. #[tokio::test] #[ignore = "real-time process test: just rs cli-export"] -async fn the_same_instance_within_the_linger_continues() { +async fn the_same_instance_within_the_follow_continues() { let relay = relay().await; let epoch = moq_net::Epoch::mint(); let lite07 = ["--connect-version", LITE_07]; @@ -529,7 +529,7 @@ async fn the_same_instance_within_the_linger_continues() { .with_reconnect(false); let url: url::Url = relay.url.parse().unwrap(); let connection = client.connect(url.clone()).established().await.unwrap(); - let mut export = export_with(&relay, &lite07, &["--linger", "4s"]); + let mut export = export_with(&relay, &lite07, &["--follow", "4s"]); let output = export.output.clone(); // Continue one capture across a transport loss. Replaying its beginning in a fresh @@ -694,7 +694,7 @@ async fn a_standby_before_its_init_segment_leaves_the_fmp4_exports_running_on_li async fn an_old_publisher_that_stays_up_keeps_the_export() { let relay = relay().await; let lite07 = ["--connect-version", LITE_07]; - let mut export = export_with(&relay, &lite07, &["--linger", "10s"]); + let mut export = export_with(&relay, &lite07, &["--follow", "10s"]); let output = export.output.clone(); let (mut old, feeding) = import_with(&relay, &lite07); @@ -926,7 +926,7 @@ async fn redundant_failover(bufferbloat: bool) { export_with( &subscribers, &["--connect-version", LITE_07], - &["--stitch", "--linger", "10s", "--delay", "3s"], + &["--stitch", "--follow", "10s", "--delay", "3s"], ) }) .collect(); diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 6d7e9cb81da..e716503e2fc 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -9084,11 +9084,15 @@ fn finish((broadcast, mut catalog): (moq_net::broadcast::Producer, crate::catalo } /// A failure whose cause is not the publisher leaving is the export's own: it fails at once, -/// however long the follow. moq-transport's PUBLISH_DONE has no code for a lost route, so a -/// relay's copy reads there as an unrecognized one. +/// however long the follow. #[tokio::test(start_paused = true)] async fn a_follower_fails_at_once_on_its_own_failure() { - for cause in [moq_net::Error::Cancel, moq_net::Error::Remote(0)] { + let causes = [ + moq_net::Error::Cancel, + moq_net::Error::Stream(moq_net::StreamError::Internal), + moq_net::Error::MalformedTrack, + ]; + for cause in causes { let origin = crate::source::produce_origin(); let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&moq_net::Epoch::mint())); let track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); @@ -9113,13 +9117,15 @@ async fn a_follower_fails_at_once_on_its_own_failure() { /// A track that lost its route, or whose session ended, says the publisher left even with the /// broadcast still announced: the follower waits out the follow for a return, then fails with -/// that cause. +/// that cause. moq-transport's PUBLISH_DONE has no code for a lost route, so a relay's copy +/// reads there as an unregistered one. #[tokio::test(start_paused = true)] async fn a_follower_waits_out_the_follow_on_a_lost_route() { let causes = [ moq_net::Error::Unroutable, moq_net::Error::Stream(moq_net::StreamError::Unroutable), moq_net::Error::Session(moq_net::SessionError::Cancel), + moq_net::Error::Remote(0), ]; for cause in causes { let origin = crate::source::produce_origin(); diff --git a/rs/moq-mux/src/container/ts/follower.rs b/rs/moq-mux/src/container/ts/follower.rs index cb90bf456c3..03af4174e48 100644 --- a/rs/moq-mux/src/container/ts/follower.rs +++ b/rs/moq-mux/src/container/ts/follower.rs @@ -73,8 +73,9 @@ enum State { /// /// A track failure is followed only when its cause says the publisher left: the route serving /// it was lost (`Unroutable`, which a moq-lite relay resets a copy with once its upstream -/// session ends), or the session carrying it ended. Any other failure is the export's own and -/// fails here at once. +/// session ends), the session carrying it ended, or a moq-transport peer ended it with a code +/// it does not register, since PUBLISH_DONE has none for a lost route. Any other failure is +/// the export's own and fails here at once. /// /// An instance that comes back is awaited until its catalog serves a first snapshot, with no /// deadline but its own end. A return that refuses its catalog waits for the next change, and @@ -419,7 +420,7 @@ impl Follower { } /// Whether a failure says the publisher left rather than the export failing: the route serving -/// the broadcast was lost, or the session carrying it ended. +/// the broadcast was lost, the session carrying it ended, or a moq-transport peer named no cause. fn left(err: &crate::Error) -> bool { let (crate::Error::Moq(err) | crate::Error::Hang(hang::Error::Moq(err))) = err else { return false; @@ -432,5 +433,6 @@ fn left(err: &crate::Error) -> bool { | Error::Session(_) | Error::Transport(_) | Error::IdleTimeout + | Error::Remote(_) ) } diff --git a/rs/moq-srt/src/dial.rs b/rs/moq-srt/src/dial.rs index f0fc8e0008a..cc84e22a2c5 100644 --- a/rs/moq-srt/src/dial.rs +++ b/rs/moq-srt/src/dial.rs @@ -134,10 +134,9 @@ impl Client { self } - /// Wait up to `follow` for the same publisher instance to come back once the broadcast - /// [`publish`](Self::publish) reads ends, carrying on with the same stream. Zero (the - /// default) ends the push at the broadcast's end. - /// A stitched replacement must also supply its first catalog within this bound. + /// Wait up to `follow` for the same publisher instance, or when stitching a replacement, to + /// come back once the broadcast [`publish`](Self::publish) reads ends, carrying on with the + /// same stream. Zero (the default) follows only an instance already serving. pub fn with_follow(mut self, follow: Duration) -> Self { self.follow = follow; self diff --git a/rs/moq-srt/src/server.rs b/rs/moq-srt/src/server.rs index af2c04e9570..8f04a759846 100644 --- a/rs/moq-srt/src/server.rs +++ b/rs/moq-srt/src/server.rs @@ -471,10 +471,9 @@ impl Subscribe { self.0.peer } - /// Wait up to `follow` for the same publisher instance to come back once the broadcast - /// ends, carrying on with the same stream. Zero (the default) closes the SRT stream at - /// the broadcast's end. - /// A stitched replacement must also supply its first catalog within this bound. + /// Wait up to `follow` for the same publisher instance, or when stitching a replacement, to + /// come back once the broadcast ends, carrying on with the same stream. Zero (the default) + /// follows only an instance already serving, else closes the SRT stream. pub fn with_follow(mut self, follow: Duration) -> Self { self.0.follow = follow; self diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index 360dc535d93..23fde702674 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -178,7 +178,7 @@ const MIN_DELAY: Duration = Duration::from_secs(1); pub(crate) struct Options { /// The muxer's delay, at least [`MIN_DELAY`] ([`Subscriber::new`]). pub latency: Duration, - /// How long a return or stitched replacement has to supply its first catalog. + /// How long to wait for the broadcast to come back once it ends. pub follow: Duration, /// Follow another instance replacing the broadcast, as a full program switch. pub stitch: bool, @@ -224,7 +224,7 @@ impl Subscriber { let catalog = source.catalog::(moq_mux::catalog::CatalogFormat::Hang).await?; let export = ts::Export::new(source, catalog).with_delay(options.latency.max(MIN_DELAY)); let follower = ts::Follower::new(export)? - .with_linger(options.follow) + .with_follow(options.follow) .with_stitch(options.stitch); Ok(Some(Self { follower })) } From ace514fdff2ff7ad37752e183c73deb77a6ce778 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 11:32:34 -0700 Subject: [PATCH 04/10] style(mux): format the TS follower Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/src/args.rs | 6 +++++- rs/moq-mux/src/container/ts/export_test.rs | 10 ++++++++-- rs/moq-mux/src/container/ts/follower.rs | 11 +++++++++-- 3 files changed, 22 insertions(+), 5 deletions(-) diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index 415f4a7f515..58914cb3c20 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -1342,7 +1342,11 @@ mod tests { assert!(err.to_string().contains("--linger -> --follow"), "{err}"); for format in ["fmp4", "mkv", "flv", "h264", "h265"] { assert!(parse(format, "--follow", "10s").is_err(), "{format} takes no --follow"); - let err = parse(format, "--linger", "10s").unwrap().validate().unwrap_err().to_string(); + let err = parse(format, "--linger", "10s") + .unwrap() + .validate() + .unwrap_err() + .to_string(); assert!(err.contains("export ts --follow"), "{format}: {err}"); assert!( parse(format, "--linger", "0s").unwrap().validate().is_ok(), diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index e716503e2fc..3aa7a80a152 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -9779,7 +9779,10 @@ async fn a_midstream_stitch_awaits_the_replacement_catalog() { .is_err(), "nothing to mux, follow={follow:?}" ); - assert!(follower.export().cataloged(), "the late snapshot switches, follow={follow:?}"); + assert!( + follower.export().cataloged(), + "the late snapshot switches, follow={follow:?}" + ); assert_eq!(follower.export().instance(), Some(&epoch)); } } @@ -9863,7 +9866,10 @@ async fn a_nested_stitch_keeps_awaiting_the_pending_replacement() { .is_err(), "nothing to mux" ); - assert!(follower.export().cataloged(), "the second replacement served its catalog"); + assert!( + follower.export().cataloged(), + "the second replacement served its catalog" + ); assert_eq!(follower.export().instance(), Some(&epoch)); } diff --git a/rs/moq-mux/src/container/ts/follower.rs b/rs/moq-mux/src/container/ts/follower.rs index 03af4174e48..f38b1d829fa 100644 --- a/rs/moq-mux/src/container/ts/follower.rs +++ b/rs/moq-mux/src/container/ts/follower.rs @@ -292,7 +292,10 @@ impl Follower { return Ok(false); } }; - if !switching.as_ref().is_some_and(|switching| switching.target.is_clone(target)) { + if !switching + .as_ref() + .is_some_and(|switching| switching.target.is_clone(target)) + { tracing::info!(path = %self.path, "broadcast replaced, switching the program to the new instance"); *switching = Some(self.subscribe(target.clone())); } @@ -343,6 +346,8 @@ impl Follower { // An epochless route has no instance to match, so only the broadcast // being exported is its own. None if broadcast.is_clone(&self.export.source().broadcast()) => Serving::Ours, + // Even the same broadcast handed out again is a return: the origin hands + // one out only for an announcement that brings it. Some(epoch) if broadcast.info().epoch.as_ref() == Some(epoch) => match self.priming { true => Serving::Ours, false => Serving::Return(broadcast), @@ -376,7 +381,9 @@ impl Follower { match missing { Missing::Ended { end, deadline } => State::Settling { end, deadline }, Missing::Switch => State::Settling { - end: Err(anyhow::anyhow!("replacement broadcast `{}` ended before serving its catalog", self.path).into()), + end: Err( + anyhow::anyhow!("replacement broadcast `{}` ended before serving its catalog", self.path).into(), + ), deadline: moq_clock::now() + self.follow, }, } From 57acf5d936935a912678b05c9e887b9a7e13fe2c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 12:01:47 -0700 Subject: [PATCH 05/10] docs(cli): a return lost to a dead session is no return The follower records only a refusal that is not the publisher leaving, so a return routed through a dead session ends the export as the broadcast did, not with that session's error. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index f1b65d57ac2..961ce45d632 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -455,9 +455,10 @@ continuity error, and the rest carry on. It takes neither `--follow`, A stdout export ends with the broadcast. `export ts --follow 10s` waits that long for the same publisher instance to come back, and carries on with the same -stream. A return that cannot be read yet, such as one still routed through a -dead session, keeps it waiting for the next announcement; if the follow runs -out first, it exits 1 with that error. A replacement (a restarted publisher, under a new +stream. A return that refuses its catalog keeps it waiting for the next +announcement; if the follow runs out first, it exits 1 with that refusal. One +that goes before serving, such as one still routed through a dead session, is +no return. A replacement (a restarted publisher, under a new [epoch](/concept/moq-lite#publisher-epochs)) exits 1 unless `--stitch` follows it as a full program switch: a new PMT from its catalog, with every PID flagging the break. A stitch waits for the replacement's first catalog snapshot until it From 56c5312c9f68ee8143e0f61a94b39c22626ffb87 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 12:49:17 -0700 Subject: [PATCH 06/10] fix(cli): detach from MoQ before an interrupted stage drops On a shutdown signal the stages dropped before the session closed, so their open tracks went out as dropped (Internal) or not found, which a TS follower takes for the publisher failing. Closing the session first ends them as a lost route, which a relay reports as Unroutable and a follower follows. Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/src/main.rs | 28 ++++++++++++++++------------ 1 file changed, 16 insertions(+), 12 deletions(-) diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 7d08d39217c..d89074c0e3d 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -560,6 +560,7 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res // anything bound a port or dialed out. let mut client = None; let mut connection = None; + let local = tokio::task::LocalSet::new(); let result = async { let Attached { bandwidth, @@ -614,17 +615,20 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res } if locals.is_empty() { - drive(tasks, signals, shutdown).await + drive(&mut tasks, signals, shutdown).await } else { - let local = tokio::task::LocalSet::new(); supervise(&local, locals.into_iter().map(Publish::run), &mut tasks); - local.run_until(drive(tasks, signals, shutdown)).await + local.run_until(drive(&mut tasks, signals, shutdown)).await } } .await; - // Even on a setup error. + // Even on a setup error. Detach before the stages drop: an interrupted stage would end + // its open tracks unfinished, which a reader takes for the publisher failing. Once the + // session is closed, they end as the route lost, which a reader follows. detach(connection, client, &cluster).await; + tasks.shutdown().await; + drop(local); result } @@ -925,12 +929,14 @@ async fn run_stdout( } /// Run until an endpoint ends, keeping all tasks alive while a signalled listener drains. +/// +/// The tasks are left for the caller to shut down once the MoQ side is detached. async fn drive( - mut tasks: JoinSet>, + tasks: &mut JoinSet>, mut signals: Signals, shutdown: Option, ) -> anyhow::Result<()> { - let result = loop { + loop { tokio::select! { biased; () = signals.recv() => { @@ -947,9 +953,7 @@ async fn drive( None => break Ok(()), } } - }; - tasks.shutdown().await; - result + } } /// The listener / HTTP-serving endpoints bridge one named broadcast, so an @@ -1028,7 +1032,7 @@ mod tests { trigger, observer: observer.clone(), }; - let driving = tokio::spawn(drive(tasks, signals, Some(shutdown))); + let driving = tokio::spawn(async move { drive(&mut tasks, signals, Some(shutdown)).await }); observer.started().await; assert!(!driving.is_finished(), "drain ended under an established session"); drop(session); @@ -1284,7 +1288,7 @@ mod tests { supervise(&local, pipelines, &mut tasks); let err = local - .run_until(drive(tasks, Signals::new().unwrap(), None)) + .run_until(drive(&mut tasks, Signals::new().unwrap(), None)) .await .unwrap_err(); assert!(err.to_string().contains("pipeline panicked"), "{err}"); @@ -1301,7 +1305,7 @@ mod tests { supervise(&local, pipelines, &mut tasks); local - .run_until(drive(tasks, Signals::new().unwrap(), None)) + .run_until(drive(&mut tasks, Signals::new().unwrap(), None)) .await .unwrap(); } From 400efdc44092295932827df65e534f87bd83b6b7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 12:51:31 -0700 Subject: [PATCH 07/10] fix(rtmp): end the catalog when a publish hangs up or fails A hangup finished the media but dropped the catalog unfinished, and a failure aborted the media but dropped the catalog, so readers saw Dropped. The catalog now finishes with the media on a hangup and aborts with the same cause on a failure. Co-Authored-By: Claude Opus 5.5 --- rs/moq-rtmp/src/dial.rs | 4 +++- rs/moq-rtmp/src/server.rs | 32 +++++++++++++++++++++++++++++++- 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/rs/moq-rtmp/src/dial.rs b/rs/moq-rtmp/src/dial.rs index 026665798a7..3e730585893 100644 --- a/rs/moq-rtmp/src/dial.rs +++ b/rs/moq-rtmp/src/dial.rs @@ -603,9 +603,10 @@ impl Publisher { Ok(moq_mux::Linger::new([self.broadcast.clone()], self, linger)) } - /// Flush any buffered media, then end the broadcast so the origin unannounces it at once. + /// Flush any buffered media and finish the catalog, then end the broadcast so the origin unannounces it at once. fn close(mut self) -> anyhow::Result<()> { self.importer.finish()?; + self.catalog.finish()?; self.broadcast.close(); Ok(()) } @@ -615,6 +616,7 @@ impl Publisher { /// /// Consumes the publisher and closes the broadcast. fn abort(self, err: moq_net::Error) { + self.catalog.abort(err.clone()); self.importer.abort(err); self.broadcast.close(); } diff --git a/rs/moq-rtmp/src/server.rs b/rs/moq-rtmp/src/server.rs index 4cb03e20d82..2f43f1209df 100644 --- a/rs/moq-rtmp/src/server.rs +++ b/rs/moq-rtmp/src/server.rs @@ -1536,10 +1536,11 @@ impl Publisher { Ok(moq_mux::Linger::new([self.broadcast.clone()], self, linger)) } - /// Flush any buffered media, close out the broadcast's open groups, and end + /// Flush any buffered media, finish the catalog, close out the broadcast's open groups, and end /// the broadcast so the origin unannounces it immediately. fn close(mut self) -> anyhow::Result<()> { self.importer.finish()?; + self.catalog.finish()?; self.broadcast.close(); Ok(()) } @@ -1550,6 +1551,7 @@ impl Publisher { /// /// Consumes the publisher and closes the broadcast. fn abort(self, err: moq_net::Error) { + self.catalog.abort(err.clone()); self.importer.abort(err); self.broadcast.close(); } @@ -1767,6 +1769,34 @@ mod tests { } } + /// A hangup ends the catalog too, so a reader sees the publisher go rather than a + /// producer dropped unfinished. + #[tokio::test(start_paused = true)] + async fn a_hangup_ends_the_catalog() { + let origin = moq_tokio::origin::spawn(); + let consumer = origin.consume(); + let (mut client, server) = tokio::io::duplex(64 * 1024); + let accept = tokio::spawn(accept_publish(server, origin.clone(), Duration::ZERO)); + let catalog = async { + consumer.routed("live/cam0").await.unwrap(); + let broadcast = consumer.request_broadcast("live/cam0", None).await.unwrap(); + let track = broadcast.track(hang::Catalog::DEFAULT_NAME).unwrap(); + track.subscribe(None).await.unwrap() + }; + // Holding the stream past the publish keeps the hangup until the catalog is subscribed. + let (mut catalog, ()) = tokio::join!(catalog, publish_one_keyframe(&mut client, Leave::Hangup)); + drop(client); + accept.await.unwrap(); + + let end = loop { + match catalog.recv_group().await { + Ok(Some(_)) => continue, + end => break end.map(|_| ()), + } + }; + assert!(end.is_ok(), "the catalog ends cleanly: {end:?}"); + } + /// A late subscriber still reads the finished media during the linger, and an encoder /// reconnecting under the same key takes the path without ending it. #[tokio::test(start_paused = true)] From 9c7f6c70485703dafcc3ee4029614694c32b0e61 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 12:54:03 -0700 Subject: [PATCH 08/10] fix(rtc): an ingest that ends without a DELETE aborts its tracks A peer disconnect or an offer that never connected left the sink's tracks to drop unfinished, and every abort skipped the catalog, so readers saw Dropped. Any end but the DELETE now aborts the media and the catalog with the session's cause. Co-Authored-By: Claude Opus 5.5 --- rs/moq-rtc/src/ingest.rs | 1 + rs/moq-rtc/src/server/whip.rs | 28 ++++++++++++++++++++++++++++ rs/moq-rtc/src/session.rs | 12 ++++++------ 3 files changed, 35 insertions(+), 6 deletions(-) diff --git a/rs/moq-rtc/src/ingest.rs b/rs/moq-rtc/src/ingest.rs index 86d1b8805aa..859f3703e31 100644 --- a/rs/moq-rtc/src/ingest.rs +++ b/rs/moq-rtc/src/ingest.rs @@ -69,6 +69,7 @@ impl session::MediaSink for IngestSink { } fn abort(&mut self, err: moq_net::Error) { + self.catalog.clone().abort(err.clone()); self.bridges.abort(err); } diff --git a/rs/moq-rtc/src/server/whip.rs b/rs/moq-rtc/src/server/whip.rs index 134c886dbfa..9e8d6ad78e1 100644 --- a/rs/moq-rtc/src/server/whip.rs +++ b/rs/moq-rtc/src/server/whip.rs @@ -214,6 +214,34 @@ mod tests { offer.to_sdp_string() } + /// A session that ends without a DELETE, here one that never connects, ends its tracks + /// with the transport cause, so a reader sees the publisher go rather than a producer + /// dropped unfinished. + #[tokio::test(start_paused = true)] + async fn a_session_ending_without_a_delete_ends_its_tracks() { + let server = Server::new(crate::server::Config::default()); + let origin = moq_tokio::origin::spawn(); + let consumer = origin.consume(); + let session = accept(&server, &origin, "live/cam0", &offer()).await.unwrap(); + let broadcast = consumer.request_broadcast("live/cam0", None).await.unwrap(); + let mut catalog = broadcast + .track(hang::Catalog::DEFAULT_NAME) + .unwrap() + .subscribe(None) + .await + .unwrap(); + + let ended = session.run().await; + assert!(matches!(ended, Err(crate::Error::IceTimeout)), "{ended:?}"); + let end = loop { + match catalog.recv_group().await { + Ok(Some(_)) => continue, + end => break end.map(|_| ()), + } + }; + assert!(matches!(end, Err(moq_net::Error::Transport(_))), "{end:?}"); + } + /// An encoder reconnecting to the same broadcast while its stale session is still /// open restarts the broadcast at once: each offer is its own epoch, so a fresh /// request reaches the reconnect, while the stale viewer stays on the old one until diff --git a/rs/moq-rtc/src/session.rs b/rs/moq-rtc/src/session.rs index 49324226fe5..e43c6a7d574 100644 --- a/rs/moq-rtc/src/session.rs +++ b/rs/moq-rtc/src/session.rs @@ -45,8 +45,9 @@ pub trait MediaSink: Send { /// loop has already converted the timestamp to microseconds. fn on_frame(&mut self, mid: str0m::media::Mid, frame: codec::Frame) -> Result<()>; - /// Called once when the session ends with a genuine failure, so the sink can - /// abort its tracks with the real cause instead of a bare `Error::Dropped`. + /// Called once when the session ends other than by the peer's clean end, so the sink can + /// abort its tracks, the catalog included, with the real cause instead of a bare + /// `Error::Dropped`. fn abort(&mut self, err: moq_net::Error); /// Called once when the peer ends the session cleanly after its final media: finish @@ -148,11 +149,10 @@ impl Session { pub async fn run(&mut self) -> Result<()> { let result = self.run_loop().await; - // A genuine failure (not a normal peer disconnect or an unconnected offer) - // propagates to subscribers as the real cause; the normal ends just let the - // sink's tracks close as Dropped. + // Any end but the peer's DELETE, a disconnect or an offer that never connected + // included, ends the sink's tracks with its cause, so subscribers see the publisher + // go rather than its tracks dropped unfinished. if let Err(err) = &result - && !matches!(err, Error::SessionClosed | Error::IceTimeout) && let MediaRole::Ingest(sink) = &mut self.role { sink.abort(moq_net::Error::Transport(err.to_string())); From 93615ae498d093735b7300bb6568922a6e470ca9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 12:57:57 -0700 Subject: [PATCH 09/10] test(cli): pin that an interrupted stage ends its tracks with the session Extract the leave-then-stop teardown into a helper and test it over an in-memory session: the old order gives the reader Stream(Internal). Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/src/main.rs | 94 +++++++++++++++++++++++++++++++++++++++--- 1 file changed, 89 insertions(+), 5 deletions(-) diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index d89074c0e3d..acc79b4c981 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -623,13 +623,24 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res } .await; - // Even on a setup error. Detach before the stages drop: an interrupted stage would end - // its open tracks unfinished, which a reader takes for the publisher failing. Once the - // session is closed, they end as the route lost, which a reader follows. - detach(connection, client, &cluster).await; + // Even on a setup error. + stop(detach(connection, client, &cluster), &mut tasks, local).await; + result +} + +/// Leave the MoQ network, then stop the stages. +/// +/// In this order an interrupted stage's open tracks end with the session that served them, +/// which a relay reports as a lost route and a reader follows. Stopping the stages first would +/// end them unfinished, which a reader takes for the publisher failing. +async fn stop( + leave: impl std::future::Future, + tasks: &mut JoinSet>, + local: tokio::task::LocalSet, +) { + leave.await; tasks.shutdown().await; drop(local); - result } /// Close everything [`spawn_moq`] attached. The process exits next, so the relay @@ -1310,6 +1321,79 @@ mod tests { .unwrap(); } + /// An interrupted stage's open track reaches a remote reader as its session closing, which + /// a TS follower follows, not as a track dropped unfinished (`Stream(Internal)`). + #[tokio::test(start_paused = true)] + async fn an_interrupted_stage_ends_its_tracks_as_the_session_closing() { + let mut config = moq_net::transport::memory::Config::default(); + config.protocol = Some("moq-lite-06".into()); + let (client, server) = moq_net::transport::memory::pair(config); + let relay = moq_tokio::origin::spawn(); + let subscriber = relay.clone(); + let accepting = tokio::spawn(async move { + let (session, driver) = moq_net::Server::new() + .with_subscriber(subscriber) + .accept(moq_net::time::Instant::now(), server) + .await + .unwrap(); + tokio::spawn(moq_net::time::run(driver)); + session + }); + let publisher = moq_tokio::origin::spawn(); + let (session, driver) = moq_net::Client::new() + .with_publisher(publisher.consume()) + .connect(moq_net::time::Instant::now(), client) + .await + .unwrap(); + tokio::spawn(moq_net::time::run(driver)); + let _relayed = accepting.await.unwrap(); + + let broadcast = publisher.create_broadcast("demo").unwrap(); + broadcast.announce(Default::default()).unwrap(); + let info = moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI); + let track = broadcast.create_track("live", info).unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(moq_net::Timestamp::ZERO, b"live".as_ref()).unwrap(); + group.finish().unwrap(); + + let reader = relay.consume(); + reader.routed("demo").await.unwrap(); + let mut live = reader + .request_broadcast("demo", None) + .await + .unwrap() + .track("live") + .unwrap() + .subscribe(None) + .await + .unwrap(); + assert!(live.recv_group().await.unwrap().is_some()); + + let local = tokio::task::LocalSet::new(); + let mut tasks = JoinSet::new(); + let stage: Pipeline = Box::pin(async move { + let _open = (broadcast, track); + std::future::pending().await + }); + supervise(&local, [stage], &mut tasks); + stop( + async move { + let _ = session.close().await; + }, + &mut tasks, + local, + ) + .await; + + let end = loop { + match live.recv_group().await { + Ok(Some(_)) => continue, + end => break end.map(|_| ()), + } + }; + assert!(matches!(end, Err(moq_net::Error::Session(_))), "{end:?}"); + } + /// A stream bind failure fails startup before the caller can report READY, and /// `--connect` dials only once the listener is up, so it leaves no live dial to /// drop without its close. From 43a56e6be6f1f582d47f1971ee0b2afb442ab09d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 24 Oct 2026 13:08:32 -0700 Subject: [PATCH 10/10] test(cli): an interrupted publisher's export waits out its follow The drop test only checked the exit code, so an export failing at once on the drop still passed. It now checks the follow ran from the interrupt, which fails with the old teardown order. The MIGRATE entry named export srt --linger, which no release had. Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/MIGRATE.md | 2 +- rs/moq-cli/tests/integration/export.rs | 6 ++++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/rs/moq-cli/MIGRATE.md b/rs/moq-cli/MIGRATE.md index 62517583371..84620196579 100644 --- a/rs/moq-cli/MIGRATE.md +++ b/rs/moq-cli/MIGRATE.md @@ -81,7 +81,7 @@ Rename `--max-age` to `--delay`. ### `moq export ts --linger` is `--follow`, and follows only the same publisher instance -Rename `export ts --linger` and `export srt --linger` to `--follow`; the old spelling stops the run naming it. +Rename `export ts --linger` to `--follow`; the old spelling stops the run naming it. `export srt` takes the same `--follow`. A restarted publisher exits 1 unless `--stitch` follows it as a program switch, and a stitched replacement's first catalog is awaited until it arrives or the replacement goes, so `--follow 0s` no longer refuses a cold one. A failure other than the publisher leaving, such as a codec TS cannot carry, exits at once rather than after a 1 s grace. diff --git a/rs/moq-cli/tests/integration/export.rs b/rs/moq-cli/tests/integration/export.rs index f546e418e1c..52a9113f21d 100644 --- a/rs/moq-cli/tests/integration/export.rs +++ b/rs/moq-cli/tests/integration/export.rs @@ -345,6 +345,8 @@ async fn a_drop_exits_one_once_the_follow_expires() { let (publisher, feeding) = import(&relay); let stdin = feeding.await.unwrap(); export.output_past(0).await; + // The follow can't start before the interrupt. + let interrupted = Instant::now(); interrupt(publisher, stdin).await; let (status, errors) = export.wait().await; @@ -353,6 +355,10 @@ async fn a_drop_exits_one_once_the_follow_expires() { Some(1), "a drop exits 1, got {status}; exporter stderr: {errors}" ); + assert!( + interrupted.elapsed() >= Duration::from_millis(900), + "the export waited out its follow rather than failing on the drop; exporter stderr: {errors}" + ); } /// An export that fails on its own, with the broadcast still up, has no return to wait for.