From 01326c106c0606ebd4948b47446214aeab41de32 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 12:33:30 -0700 Subject: [PATCH] fix(mux): a failed stitch returns to a live export; a return during the grace is followed A stitch that fails to resolve while a return still awaits its catalog now goes back to running that export, rather than settling and leaving it unpolled. A failure whose broadcast went and came back before the follower looked clears the grace and follows the return instead of failing as the export's own. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/export_test.rs | 89 ++++++++++++++++++++++ rs/moq-mux/src/container/ts/follower.rs | 22 ++++-- 2 files changed, 105 insertions(+), 6 deletions(-) diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index a9cfc5e2a6..d77ca6e704 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -7679,6 +7679,95 @@ async fn a_follower_stitches_onto_the_next_replacement_when_one_goes_before_it_r assert!(matches!(end, Ok(None)), "{end:?}"); } +/// 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. +#[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 start = tokio::time::Instant::now(); + finish(first); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "lingering" + ); + + // Back, with a catalog yet to deliver its first snapshot. + let second = publish_bare(&origin, Some(&epoch)); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "awaiting the return's catalog" + ); + + // Replaced, but the replacement's catalog request is never answered before it goes. + let third = origin + .publish( + "live", + moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), + ) + .unwrap(); + let unanswered = third.dynamic(); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "resolving the replacement" + ); + drop((third, unanswered)); + + finish(second); + let end = follower.next().await; + assert!(matches!(end, Ok(None)), "{end:?}"); + assert_eq!(start.elapsed(), Duration::from_secs(3) + linger); +} + +/// 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. +#[tokio::test(start_paused = true)] +async fn a_follower_follows_a_return_that_came_back_during_the_grace() { + 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)); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "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)); + let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&epoch)); + let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + assert!( + tokio::time::timeout(Duration::from_secs(3), follower.next()) + .await + .is_err(), + "carried on into the return" + ); + + track.finish().unwrap(); + finish((broadcast, catalog)); + let end = follower.next().await; + assert!(matches!(end, Ok(None)), "{end:?}"); +} + /// A replacement that stays announced but refuses its catalog fails a stitch loudly, rather /// than leaving the export on the instance it replaced. #[tokio::test(start_paused = true)] diff --git a/rs/moq-mux/src/container/ts/follower.rs b/rs/moq-mux/src/container/ts/follower.rs index c13b0a387e..2f28b13116 100644 --- a/rs/moq-mux/src/container/ts/follower.rs +++ b/rs/moq-mux/src/container/ts/follower.rs @@ -52,8 +52,10 @@ enum State { /// Resolving the instance serving the path now, to carry the export on into it. Following { resolving: Resolving, - /// How the export ended and when the linger runs out, or `None` for a stitch mid-stream. + /// How the export ended and when the linger runs out, as in `Running` or `Settling`. settling: Option<(crate::Result<()>, Instant)>, + /// The export is still running, a stitch mid-stream, so it carries on if the switch does not. + live: bool, }, /// Nothing follows. Done, @@ -171,6 +173,7 @@ impl Follower { self.state = State::Following { resolving: self.resolve(), settling, + live: true, }; continue; } @@ -215,7 +218,8 @@ impl Follower { deadline, } => { self.poll_announced(waiter); - if grace.is_some() && self.serving != Serving::Ours { + // 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); } @@ -240,6 +244,7 @@ impl Follower { self.state = State::Following { resolving: self.resolve(), settling: Some((end, deadline)), + live: false, }; continue; } @@ -253,6 +258,7 @@ impl Follower { State::Following { mut resolving, settling, + live, } => { // Announcements wait until the export is on the instance they describe. if let Poll::Ready(resolved) = waiter.poll_future(resolving.as_mut()) { @@ -271,13 +277,13 @@ impl Follower { } tracing::warn!(path = %self.path, %err, "broadcast went before it resolved, still waiting"); self.serving = Serving::Gone; - self.state = match settling { - Some((end, deadline)) => State::Settling { + self.state = match (live, settling) { + (false, Some((end, deadline))) => State::Settling { end, grace: None, deadline, }, - None => State::Running { settling: None }, + (_, settling) => State::Running { settling }, }; continue; } @@ -300,7 +306,11 @@ impl Follower { tracing::info!(path = %self.path, linger = ?self.linger, "broadcast did not return"); return Poll::Ready(end.clone().map(|()| None)); } - self.state = State::Following { resolving, settling }; + self.state = State::Following { + resolving, + settling, + live, + }; return Poll::Pending; } }