fix(ts): hold audio behind a stale start group's next keyframe

quest/m1/ts-reliability/README.md

@@ -106,7 +106,6 @@ remain with their current quests.
 - [Section placement](/quest/m1/ts-reliability/section-placement.md) - sections share the PCR placement clock instead of the video PTS
 - [Loss and recovery tests](/quest/m1/ts-reliability/recovery.md) - actual MoQ omissions exercise reacquisition and unaffected tracks
 - [Data-track breaks](/quest/m1/ts-reliability/data-break.md) - a data track's declared break reads as a break, not as media lost upstream
-- [Audio behind the keyframe](/quest/m1/ts-reliability/audio-lead.md) - an export whose video start group is dropped still begins on a keyframe
 - [T-STD TS export](/quest/m1/tstd/README.md) - the semantic mux schedules the content it admits within its supported buffer models
 
 ## Related

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

@@ -1881,10 +1881,11 @@ where
 
 	/// Whether a frame released at `timestamp` presents ahead of the video keyframe the stream
 	/// begins on ([`Self::video_start`]), and so is dropped. A video track whose start group
-	/// was cut before it went out starts at its next keyframe instead ([`jitter::Buffer`]), so
-	/// until video goes out the start is the earliest frame a video track holds. Video goes out
-	/// further ahead of its decode time than the other tracks, so what they release before it
-	/// presents ahead of it too.
+	/// was cut or went stale before it went out starts at its next keyframe instead
+	/// ([`jitter::Buffer`]), so until video goes out the start is the earliest frame a video
+	/// track holds, and with none held, everything leads the keyframe yet to arrive. Video goes
+	/// out further ahead of its decode time than the other tracks, so what they release before
+	/// it presents ahead of it too.
 	fn leads_video(&mut self, is_video: bool, timestamp: Timestamp) -> bool {
 		let Some(start) = self.video_start else {
 			return false;
@@ -1906,7 +1907,7 @@ where
 			.filter_map(|track| self.jitter.peek(&track.pid))
 			.map(|queued| queued.frame.timestamp)
 			.min();
-		timestamp < next.map_or(start, |next| next.max(start))
+		next.is_none_or(|next| timestamp < next.max(start))
 	}
 
 	/// Discard what has not gone out and restart the program clock. Every rendition

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

@@ -5257,6 +5257,8 @@ struct Interleave {
 	source: crate::Source,
 	/// The next audio frame to write.
 	audio_index: u64,
+	/// Audio frames per group.
+	audio_group: u64,
 	/// The paused clock's instant at media time zero, set by the first [`Self::at`].
 	start: Option<tokio::time::Instant>,
 }
@@ -5313,6 +5315,7 @@ impl Interleave {
 			video,
 			audio,
 			audio_index: 0,
+			audio_group: AUDIO_GROUP,
 			start: None,
 		}
 	}
@@ -5351,20 +5354,25 @@ impl Interleave {
 	/// polling `export` after each.
 	fn audio_until(&mut self, until: u64, export: &mut Export<crate::catalog::Consumer>, out: &mut Vec<Frame>) {
 		while self.audio_index * AUDIO_US < until {
-			let index = self.audio_index;
-			self.audio
-				.write(Frame {
-					timestamp: Timestamp::from_micros(index * AUDIO_US).unwrap(),
-					duration: None,
-					payload: Bytes::from_iter((0..180u16).map(|i| (i ^ index as u16) as u8)),
-					keyframe: index.is_multiple_of(AUDIO_GROUP),
-				})
-				.unwrap();
-			self.audio_index += 1;
+			self.audio_frame();
 			out.extend(poll_frames(export));
 		}
 	}
 
+	/// Write the next audio frame.
+	fn audio_frame(&mut self) {
+		let index = self.audio_index;
+		self.audio
+			.write(Frame {
+				timestamp: Timestamp::from_micros(index * AUDIO_US).unwrap(),
+				duration: None,
+				payload: Bytes::from_iter((0..180u16).map(|i| (i ^ index as u16) as u8)),
+				keyframe: index.is_multiple_of(self.audio_group),
+			})
+			.unwrap();
+		self.audio_index += 1;
+	}
+
 	fn finish(&mut self) {
 		self.video.finish().unwrap();
 		self.audio.finish().unwrap();
@@ -5605,7 +5613,9 @@ async fn tune_in_does_not_wait_on_dropped_audio() {
 	let mut export = rig.export(delay).await;
 	let mut out = Vec::new();
 	rig.audio_until(3 * GOP * VIDEO_US, &mut export, &mut out);
-	for tick in GOP..2 * GOP {
+	// Under a delay's worth of the group, read at once: the whole group would leave its
+	// keyframe stale, and the audio would wait on the next.
+	for tick in GOP..GOP + 6 {
 		rig.video(tick);
 	}
 	out.extend(poll_frames(&mut export));
@@ -5768,6 +5778,50 @@ async fn audio_dropped_ahead_of_a_moved_start_is_not_started() {
 	assert!(stats.trimmed > 0);
 }
 
+/// A joiner's start group older than the delay goes stale at acquisition and is dropped
+/// whole, so the video holds nothing until its next keyframe lands. The audio due meanwhile
+/// presents after the dropped keyframe but before the next one, and is dropped rather than
+/// leading it.
+#[tokio::test(start_paused = true)]
+async fn a_stale_start_group_holds_the_audio_behind_the_next_keyframe() {
+	let delay = Duration::from_millis(100);
+	let mut rig = Interleave::new().await;
+	// A group per audio frame, so the joiner's audio starts at the live edge, fresher than the
+	// video's start group.
+	rig.audio_group = 1;
+	// Join 160 ms into the first group: older than the delay, with most of it still to come.
+	let join = 4;
+	for tick in 0..join {
+		rig.at(tick * VIDEO_US).await;
+		rig.video(tick);
+		while rig.audio_index * AUDIO_US < (tick + 1) * VIDEO_US {
+			rig.audio_frame();
+		}
+	}
+	let mut export = open_export(rig.source.clone()).await.unwrap().with_delay(delay);
+	let mut out = Vec::new();
+	for tick in join..3 * GOP {
+		rig.at(tick * VIDEO_US).await;
+		rig.video(tick);
+		out.extend(poll_frames(&mut export));
+		rig.audio_until((tick + 1) * VIDEO_US, &mut export, &mut out);
+	}
+	rig.finish();
+	out.extend(drain_frames(&mut export).await);
+
+	let ts: Vec<u8> = out.iter().flat_map(|f| f.payload.iter().copied()).collect();
+	let (video, audio) = collect_pes_pts(&ts);
+	let pts = |tick: u64| tick * VIDEO_US * 90 / 1_000;
+	assert_eq!(video.first(), Some(&pts(GOP)), "the stale start group is dropped");
+	assert!(
+		audio.iter().all(|&p| p >= pts(GOP)),
+		"no audio leads the keyframe the export starts at: {audio:?}"
+	);
+	let stats = export.stats();
+	assert_eq!(stats.dropped, 0, "nothing was lost from the output");
+	assert!(stats.trimmed > 0);
+}
+
 /// A section lost before the cycle wraps commits an observed subset; the next
 /// cycle must *converge* to the full set rather than flip-flop between subsets.
 /// The repetition fast-path skips sections already active, so without the

rs/moq-srt/src/ts.rs

@@ -658,7 +658,13 @@ mod tests {
 		let expected_cue = cue.payload;
 		assert_eq!(expected_cue[0], 0xFC, "a verbatim splice_info_section (table_id 0xFC)");
 
-		let mut subscriber = Subscriber::new(&origin.consume(), "ingest", Options::default())
+		// The egress joins the clip's last group, read at once and two seconds long. Under the
+		// default delay its keyframe is stale, and the egress waits on the next one.
+		let options = Options {
+			latency: Duration::from_secs(3),
+			..Default::default()
+		};
+		let mut subscriber = Subscriber::new(&origin.consume(), "ingest", options)
 			.await
 			.unwrap()
 			.expect("the ingest broadcast is available for SRT egress");