fix(moq-mux): a failed stitch returns to a live export; a return during the grace is followed

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)]

rs/moq-mux/src/container/ts/follower.rs

@@ -52,8 +52,10 @@ enum State<E: catalog::Catalog> {
 	/// Resolving the instance serving the path now, to carry the export on into it.
 	Following {
 		resolving: Resolving<E>,
-		/// 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<E: catalog::Catalog + 'static> Follower<E> {
 						self.state = State::Following {
 							resolving: self.resolve(),
 							settling,
+							live: true,
 						};
 						continue;
 					}
@@ -215,7 +218,8 @@ impl<E: catalog::Catalog + 'static> Follower<E> {
 					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<E: catalog::Catalog + 'static> Follower<E> {
 						self.state = State::Following {
 							resolving: self.resolve(),
 							settling: Some((end, deadline)),
+							live: false,
 						};
 						continue;
 					}
@@ -253,6 +258,7 @@ impl<E: catalog::Catalog + 'static> Follower<E> {
 				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<E: catalog::Catalog + 'static> Follower<E> {
 								}
 								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<E: catalog::Catalog + 'static> Follower<E> {
 						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;
 				}
 			}