fix(mux): a TS export drains a left publisher's buffered tail before it ends

doc/bin/cli.md

@@ -483,12 +483,13 @@ 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 --follow 10s` waits that
-long for the same publisher instance to come back, and carries on with the same
-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
+A stdout export ends with the broadcast. A publisher that leaves still gets
+what the export buffered of it out on schedule first. `export ts --follow 10s`
+then waits that long for the same publisher instance to come back, and carries
+on with the same 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

quest/m0/nightly/README.md

@@ -35,7 +35,6 @@ Public API: none. Wire: none.
 - [The software renderer tests both capability paths](/quest/m0/nightly/render-capabilities.md) - the software-renderer lane verifies output selection with and without DMA-BUF import support without requiring hardware the runner lacks
 - [The browser publisher completes its fairness sweep](/quest/m0/nightly/publisher-fairness.md) - the publisher fairness benchmark completes its backlog on the hosted ARM runner, yielding browser tasks while serving every group
 - [The browser backend baseline completes on IETF22](/quest/m0/nightly/browser-baseline.md) - the WASM and UniFFI backend baseline completes every supported transport case without an internal error
-- [A stitched restart signals every TS discontinuity](/quest/m0/nightly/stitch-discontinuity.md) - a restarted publisher exported with stitching signals the program break on every affected PID
 - [Resolve the paired catalog fanout regressions](/quest/m0/nightly/paired-shared.md) - the reported catalog fanout regressions are attributed and genuine extra costs are fixed with paired measurements
 - [Resolve the paired relay-session regressions](/quest/m0/nightly/paired-relay.md) - the reported relay-session regressions are attributed and genuine extra costs are fixed with paired measurements
 - [Probing beside Cubic](/quest/m0/nightly/competing-cubic.md) - the `competing_cubic` probe test stays within its deadline budget

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

@@ -106,7 +106,8 @@ fn sanitize_mux_rate(rate: u64) -> Option<u64> {
 /// is re-emitted at video keyframes and periodically for mid-stream tune-in. SI and
 /// whole PES or sections wait until a PCR interval has followed it, since a reader
 /// places them on the program clock and the PMT is what names it.
-/// Returns `None` when the broadcast ends.
+/// Returns `None` when the broadcast ends. A publisher that leaves instead fails the
+/// export with its cause once what was already buffered has gone out on schedule.
 ///
 /// A rendition the catalog moves to another track (a new `track` or `broadcast`) goes on out
 /// of its PID from the new track's first sync frame past what was read, when the PMT can
@@ -133,6 +134,8 @@ where
 {
 	id: u64,
 	retired: bool,
+	/// Why the publisher left, reported once the output it buffered has gone out.
+	left: Option<crate::Error>,
 	expected: Vec<TrackKey>,
 	expected_changed: bool,
 	source: crate::Source,
@@ -1316,6 +1319,7 @@ where
 		Self {
 			id: 0,
 			retired: false,
+			left: None,
 			expected: Vec::new(),
 			expected_changed: false,
 			source,
@@ -1467,11 +1471,7 @@ where
 			self.stop_old(waiter)?;
 		}
 		if let Err(err) = self.prepare(waiter) {
-			if self.pending.is_none() || !super::follower::left(&err) {
-				return Poll::Ready(Err(err));
-			}
-			self.follow_until = Some(moq_clock::now());
-			self.stop_old(waiter)?;
+			self.leave(err)?;
 		}
 		if self.pending.is_some() {
 			loop {
@@ -1643,7 +1643,7 @@ where
 				// a zero-delay slot may need another frame to prove the slot is complete.
 				let track = self.program.tracks.get_mut(&name).expect("the released frame's track");
 				let finished = track.finished;
-				track.fill(
+				let filled = track.fill(
 					&name,
 					waiter,
 					&mut Input {
@@ -1652,8 +1652,12 @@ where
 						jitter: &mut self.jitter,
 						trimmed: &mut self.program.trimmed,
 					},
-				)?;
-				if !finished && track.finished {
+				);
+				let ended = !finished && track.finished;
+				if let Err(err) = filled {
+					self.leave(err)?;
+				}
+				if ended || self.program.retired {
 					self.program
 						.schedule
 						.set_ended(self.program.tracks.values().all(|track| track.finished));
@@ -1726,7 +1730,7 @@ where
 
 		// End of stream once the catalog is closed too: nothing more can appear.
 		if self.pending.is_none() && self.program.catalog_ended && (drained || self.program.tracks.is_empty()) {
-			return Poll::Ready(Ok(None));
+			return Poll::Ready(self.program.left.take().map_or(Ok(None), Err));
 		}
 
 		Poll::Pending
@@ -1921,6 +1925,26 @@ where
 		Ok(true)
 	}
 
+	/// The program playing failed with `err`. Only a publisher that left is survivable: its
+	/// input stops at once, as a follow that runs out stops it, but what it already buffered
+	/// goes out on schedule. Without a replacement, the export then reports `err`, so a
+	/// [`Follower`](super::Follower) waits for a return only after the old output has ended,
+	/// rather than leaving it buffered to go out late, behind the gap, once something returns.
+	///
+	/// The program keeps its catalog, whose format the follower subscribes a return with.
+	fn leave(&mut self, err: crate::Error) -> crate::Result<()> {
+		if !super::follower::left(&err) {
+			return Err(err);
+		}
+		tracing::debug!(target: "moq_mux::container::ts::handoff", phase = "left", now = ?moq_clock::now(), program = self.program.id, committed = self.jitter.is_committed(), %err);
+		if self.pending.is_none() {
+			self.program.left = Some(err);
+		}
+		self.follow_timer = None;
+		self.follow_until = None;
+		self.retire()
+	}
+
 	fn stop_old(&mut self, waiter: &kio::Waiter) -> crate::Result<()> {
 		if self.program.retired {
 			return Ok(());
@@ -1940,10 +1964,16 @@ where
 			return Ok(());
 		}
 		tracing::debug!(target: "moq_mux::container::ts::handoff", phase = "follow-expired", now = ?moq_clock::now(), program = self.program.id, committed = self.jitter.is_committed());
-		self.program.retired = true;
 		self.follow_timer = None;
 		self.follow_until = None;
 		self.program.catalog = None;
+		self.retire()
+	}
+
+	/// Stop the program's input. What it already buffered still goes out, unless a
+	/// committed replacement's switch drops it.
+	fn retire(&mut self) -> crate::Result<()> {
+		self.program.retired = true;
 		self.program.catalog_ended = true;
 		for (name, track) in &mut self.program.tracks {
 			if !track.finished {
@@ -2628,6 +2658,7 @@ where
 		self.program.catalog_ended = false;
 		self.program.cataloged = false;
 		self.program.retired = false;
+		self.program.left = None;
 		self.program.si_flushed = false;
 		for si in self.program.si.values_mut() {
 			si.track.clear();

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

@@ -11602,6 +11602,74 @@ async fn an_expired_follow_stops_healthy_input_but_drains_its_buffered_tail() {
 	assert!(gap > delay, "no input can bridge B's post-deadline wait: {gap:?}");
 }
 
+/// A publisher that leaves still gets what the export buffered of it out on schedule, and only
+/// then does the export report it gone. Reported at once, the tail sat unpolled while a
+/// follower waited, and went out late, behind the gap, ahead of whatever came back.
+#[tokio::test(start_paused = true)]
+async fn a_publisher_that_leaves_drains_its_buffered_tail_before_the_end() {
+	let origin = crate::source::produce_origin();
+	let (old, mut catalog) = publish_live::<()>(&origin, Default::default());
+	let audio = aac_rendition(&mut old.broadcast().clone(), &mut catalog, "a.aac");
+	let mut publisher = Some((catalog, audio));
+	let delay = Duration::from_millis(500);
+	let left = Duration::from_millis(2000);
+	let step = Duration::from_millis(20);
+	let mut export = open_export(crate::source::resolved(&origin, "live").await)
+		.await
+		.unwrap()
+		.with_delay(delay);
+	let start = tokio::time::Instant::now();
+	let mut frames = Vec::new();
+	let end = loop {
+		let elapsed = start.elapsed();
+		if elapsed < left {
+			write_aac(&mut publisher.as_mut().unwrap().1, elapsed.as_millis() as u64);
+		} else if let Some((catalog, audio)) = publisher.take() {
+			audio.abort(moq_net::Error::Unroutable);
+			catalog.abort(moq_net::Error::Unroutable);
+		}
+		tokio::task::yield_now().await;
+		let end = loop {
+			match export.poll_next(&kio::Waiter::noop()) {
+				std::task::Poll::Ready(Ok(Some(frame))) => frames.push((start.elapsed(), frame)),
+				std::task::Poll::Ready(end) => break Some(end),
+				std::task::Poll::Pending => break None,
+			}
+		};
+		if let Some(end) = end {
+			break end;
+		}
+		assert!(
+			elapsed < left + 4 * delay,
+			"the export never reported the publisher gone"
+		);
+		tokio::time::advance(step).await;
+	};
+	let ended = start.elapsed();
+	assert!(
+		matches!(&end, Err(err) if super::follower::left(err)),
+		"the end reports the publisher leaving: {end:?}"
+	);
+	let slices: Vec<Frame> = frames.iter().map(|(_, frame)| frame.clone()).collect();
+	// When each audio PTS went out, after the publisher left.
+	let tail: Vec<(Duration, u64)> = frames
+		.iter()
+		.zip(audio_pts_by_slice(&slices))
+		.filter(|((at, _), _)| *at >= left)
+		.flat_map(|((at, _), pts)| pts.into_iter().map(move |pts| (*at, pts)))
+		.collect();
+	let (last, pts) = *tail.last().expect("a tail went out after the publisher left");
+	assert!(
+		pts >= (60_000 + left.as_millis() as u64 - 100) * 90,
+		"the tail buffered when the publisher left goes out: {tail:?}"
+	);
+	assert!(
+		last >= left + delay - PCR_INTERVAL - step,
+		"the tail goes out on schedule, not in a burst: {tail:?}"
+	);
+	assert!(ended >= last, "the end follows the tail");
+}
+
 /// A replacement that leaves while it acquires is no switch: the healthy old program plays on.
 #[tokio::test(start_paused = true)]
 async fn a_replacement_leaving_before_it_commits_keeps_the_old_program() {