fix(ts): restore the scheduled TS workflow

.github/workflows/ts.yml

@@ -78,7 +78,7 @@ jobs:
         if: ${{ !cancelled() }}
         run: nix develop --command just test ts --mpeg2
         env:
-          MOQ_TEST_RUNS: ${{ runner.temp }}/moq-test-runs
+          HARNESS_RUNS: ${{ runner.temp }}/moq-test-runs
         shell: bash -leo pipefail {0}
 
       # A KLV stream joins the program mid-stream and leaves it: the export announces each

doc/concept/stats.md

@@ -101,7 +101,7 @@ source's. `export ts --passthrough` reports it without `mpegts`.
 | Field | Meaning |
 | --- | --- |
 | `dropped` | Frames lost from the output after their stream began: missed their release deadline, or a video track dropped them waiting for its next keyframe. |
-| `trimmed` | Frames dropped before their stream's first went out, so the output starts on a whole group led by a keyframe. Nothing is missing for these. |
+| `trimmed` | Frames dropped before their stream's first went out, so the output starts on a keyframe. Nothing is missing for these. |
 | `gaps` | Times the source skipped media it will never deliver: a group lost or expired upstream, or a resubscription. A declared break is not one. |
 | `margin` | Gauge, milliseconds: the least time any frame arrived before its deadline over the last second. Negative when one missed. |
 | `offset` | Gauge, milliseconds: how far the output clock sits from the source's plus the delay. Positive holds frames longer than the delay. The clock pulls it shut over ten minutes. |

quest/m1/ts-reliability/README.md

@@ -99,7 +99,6 @@ catalog linger (#5305).
 ## Required
 
 - [PAT changes](/quest/m1/ts-reliability/pat-lifecycle.md) - a running export follows a moved PMT PID
-- [TS workflow passes](/quest/m1/ts-reliability/validation-green.md) - the scheduled TS lanes pass on main again, with no lane loosened
 - [A demoted subscriber priority reaches the publisher](/quest/m1/ts-reliability/stitch-catch-up.md) - a mid-stream priority demotion reaches the publisher on lite-07 and moq-transport, including an old instance after an epoch switch
 
 ## Related

rs/moq-mux/benches/jitter.rs

@@ -29,7 +29,6 @@ async fn steer(tracks: u16, floors: u64) -> u64 {
 				decode: Timestamp::from_millis(second * 1_000).unwrap().into(),
 				restart: 0,
 				skip: 0,
-				cut: 0,
 				sync: true,
 				leading: false,
 				item: (),
@@ -91,7 +90,6 @@ fn acquisition(c: &mut Criterion) {
 											decode: Timestamp::from_millis(frame * 20).unwrap().into(),
 											restart: 0,
 											skip: 0,
-											cut: 0,
 											sync: frame % 15 == 0,
 											leading: false,
 											item: (),
@@ -149,7 +147,6 @@ fn handoff(c: &mut Criterion) {
 												.into(),
 												restart: 0,
 												skip: 0,
-												cut: 0,
 												sync: frame % 15 == 0,
 												leading: false,
 												item: (),

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

@@ -286,7 +286,6 @@ impl FlvTrack {
 			read,
 			restart,
 			skip,
-			cut,
 			sync,
 			leading,
 			item,
@@ -300,7 +299,6 @@ impl FlvTrack {
 				decode: moq_net::Timestamp::from_millis(u64::from(timestamp))?.into(),
 				restart,
 				skip,
-				cut,
 				sync,
 				leading,
 				item: Tag {
@@ -514,7 +512,6 @@ impl<S: Stream> Export<S> {
 							decode: frame.timestamp.into(),
 							restart: track.source.markers(),
 							skip: track.source.skips(),
-							cut: track.source.cuts(),
 							sync: kind == 1 || frame.keyframe,
 							leading: false,
 							item: frame,

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

@@ -536,7 +536,6 @@ impl<S: Stream> Export<S> {
 						decode: frame.timestamp.into(),
 						restart: track.source.markers(),
 						skip: track.source.skips(),
-						cut: track.source.cuts(),
 						sync: track.kind == TrackKind::Audio || frame.keyframe,
 						leading: false,
 						item: frame,

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

@@ -220,7 +220,7 @@ where
 	video_start: Option<Timestamp>,
 	/// The jitter buffer key of the first video frame out since `video_start` was set.
 	/// Until one has gone, what the other tracks release is dropped: a video track whose
-	/// start group was cut before it went out starts at its next keyframe instead
+	/// start group went stale before it went out starts at its next keyframe instead
 	/// ([`jitter::Buffer`]).
 	video_out: Option<Timestamp>,
 	/// Frames dropped before the jitter buffer: ahead of the first video keyframe, or of their
@@ -251,7 +251,7 @@ struct Pending {
 	restart: u64,
 	/// How many times its playhead had jumped, markers included ([`jitter::Arrival::skip`]).
 	skip: u64,
-	/// How many of those jumps cut a group short ([`jitter::Arrival::cut`]).
+	/// How many of those jumps cut a group short ([`ExportSource::cuts`]).
 	cut: u64,
 	/// When it landed in this node's cache ([`moq_net::frame::Frame::arrived`]), so a stall
 	/// that holds the export back too does not hide that it came late.
@@ -514,14 +514,13 @@ impl Track {
 					}
 					self.queue(name, pending, input.jitter)?;
 				}
-				// A cut takes a start group it left short now, not with the next group, which may
-				// arrive after that start was due, or with the stream's end.
+				// A cut settles what the decode clock holds of the group it left short now, not with
+				// the next group, which may arrive after those frames were due, or with the stream's end.
 				Poll::Ready(Some(Event::Cut { observed })) => {
 					self.latest = self.latest.max(observed);
 					self.settle = input.waiting_for_header;
 					if !input.waiting_for_header {
 						self.release(name, input.jitter, true)?;
-						input.jitter.cut(&self.key, self.base.2 + self.source.cuts());
 					}
 				}
 				Poll::Ready(Some(Event::GroupEnd)) => self.settle = true,
@@ -732,9 +731,9 @@ impl Track {
 			frame,
 			restart,
 			skip,
-			cut,
 			arrived,
 			read,
+			..
 		} = pending;
 		let dts = dts.filter(|&dts| dts != i128::from(to_ticks(frame.timestamp)));
 		let decode = dts
@@ -771,7 +770,6 @@ impl Track {
 			decode: key.into(),
 			restart,
 			skip,
-			cut,
 			sync,
 			leading,
 			item: Queued {
@@ -2602,7 +2600,7 @@ where
 
 	/// Whether a frame of `kind` released now presents ahead of the video keyframe the stream
 	/// begins on (`video_start`), and so is dropped. A video track whose start group
-	/// was cut or went stale before it went out starts at its next keyframe instead
+	/// 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

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

@@ -6188,11 +6188,12 @@ async fn tune_in_does_not_wait_on_dropped_audio() {
 	assert!(!out.is_empty(), "the tune-in waited on audio it had dropped");
 }
 
-/// A join whose first group the publisher cuts short (moq-dev/moq#5565) starts at the next
-/// group: nothing of the cut-off start goes out, the next keyframe leads the audio, and the
-/// rest is intact.
+/// A join whose first group the publisher cuts short (moq-dev/moq#5565) starts on it all the
+/// same: what arrived of it goes out from its keyframe, ahead of the hole, then the rest is
+/// intact. A path narrower than the media cuts every group, so waiting for a whole one might
+/// never start the export.
 #[tokio::test(start_paused = true)]
-async fn a_start_group_cut_short_starts_at_the_next() {
+async fn a_start_group_cut_short_goes_out() {
 	use crate::container::Container as _;
 
 	let delay = Duration::from_secs(1);
@@ -6250,79 +6251,48 @@ async fn a_start_group_cut_short_starts_at_the_next() {
 	let pts = |tick: u64| tick * VIDEO_US * 90 / 1_000;
 	assert_eq!(
 		video,
-		(GOP..4 * GOP).map(pts).collect::<Vec<_>>(),
-		"the export starts at the next group, whole"
+		(0..5).chain(GOP..4 * GOP).map(pts).collect::<Vec<_>>(),
+		"the export starts on the cut-off group, then resumes at the next"
 	);
 	assert!(
-		audio.iter().all(|&p| p >= pts(GOP)),
+		audio.iter().all(|&p| p >= pts(0)),
 		"no audio leads the keyframe the export starts at: {audio:?}"
 	);
 	let stats = export.stats().release;
 	assert_eq!(stats.dropped, 0, "nothing that went out lost a frame");
 	assert_eq!(stats.gaps, 1, "the cut is counted");
-	assert!(stats.trimmed >= 5, "the cut-off start is counted: {}", stats.trimmed);
+	assert_eq!(stats.trimmed, 0, "nothing of the cut-off start is trimmed");
 }
 
 /// Audio released while the video's start moves to its next group is dropped ahead of the
 /// keyframe, and its track has still sent nothing: audio that then misses its deadline is
 /// trimmed with it, not counted as lost from the output.
 #[tokio::test(start_paused = true)]
 async fn audio_dropped_ahead_of_a_moved_start_is_not_started() {
-	use crate::container::Container as _;
-
-	let delay = Duration::from_millis(200);
+	let delay = Duration::from_millis(100);
 	let mut rig = Interleave::new().await;
-	let mut export = rig.export(delay).await;
-	let track = rig.video.track().clone();
-	let wire = HangContainer::Legacy(crate::container::Kind::Video);
-	let picture = |tick: u64| {
-		let keyframe = tick.is_multiple_of(GOP);
-		let slice = vec![if keyframe { 0x65u8 } else { 0x41 }; 400];
-		Frame {
-			timestamp: Timestamp::from_micros(tick * VIDEO_US).unwrap(),
-			duration: None,
-			payload: length_prefixed(&[&slice]),
-			keyframe,
-		}
-	};
-	let mut out = Vec::new();
-
-	// The first group delivers three pictures, then the next group lands, sent ahead, and the
-	// publisher resets the first.
-	let mut cut = track.create_group(moq_net::group::Info { sequence: 0 }).unwrap();
-	for tick in 0..3 {
-		rig.at(tick * VIDEO_US).await;
-		wire.write(&mut cut, &[picture(tick)]).unwrap();
-		out.extend(poll_frames(&mut export));
-		rig.audio_until(tick * VIDEO_US, &mut export, &mut out);
-	}
-	let mut group = track.create_group(moq_net::group::Info { sequence: 1 }).unwrap();
-	wire.write(&mut group, &[picture(GOP)]).unwrap();
-	cut.abort(moq_net::Error::Old).unwrap();
-	// The audio flows long enough for some of it to go out and be dropped ahead of the
-	// keyframe, then stalls past its deadline.
-	for tick in 3..10 {
+	rig.audio_group = 1;
+	// Join 160 ms into the first group, older than the delay, so the start moves to the next.
+	let join = 4;
+	for tick in 0..join {
 		rig.at(tick * VIDEO_US).await;
-		out.extend(poll_frames(&mut export));
-		if tick < 5 {
-			rig.audio_until(tick * VIDEO_US, &mut export, &mut out);
+		rig.video(tick);
+		while rig.audio_index * AUDIO_US < (tick + 1) * VIDEO_US {
+			rig.audio_frame();
 		}
 	}
-	for tick in 10..3 * GOP {
+	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;
-		if tick > GOP {
-			if tick.is_multiple_of(GOP) {
-				group.finish().unwrap();
-				group = track
-					.create_group(moq_net::group::Info { sequence: tick / GOP })
-					.unwrap();
-			}
-			wire.write(&mut group, &[picture(tick)]).unwrap();
-		}
+		rig.video(tick);
 		out.extend(poll_frames(&mut export));
-		rig.audio_until(tick * VIDEO_US, &mut export, &mut out);
+		// The audio flows long enough for some of it to go out and be dropped ahead of the
+		// keyframe, then stalls past its deadline.
+		if !(6..10).contains(&tick) {
+			rig.audio_until((tick + 1) * VIDEO_US, &mut export, &mut out);
+		}
 	}
-	group.finish().unwrap();
 	rig.finish();
 	out.extend(drain_frames(&mut export).await);
 

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

@@ -1603,10 +1603,10 @@ async fn a_join_starts_on_a_whole_group() {
 }
 
 /// The catalog names the tracks only 2.3 s in, while the group in flight is cut short: the export
-/// waits for it, starts on the next whole group with the tables ahead of it, and loses nothing
-/// after.
+/// starts on what arrived of it, from its keyframe with the tables ahead of it, and loses
+/// nothing after the cut.
 #[tokio::test(start_paused = true)]
-async fn a_late_catalog_starts_on_a_whole_group() {
+async fn a_late_catalog_starts_on_the_group_in_flight() {
 	let run = recover(Arm {
 		catalog: Duration::from_millis(2300),
 		..arm(Profile::Closed, |slot| {
@@ -1619,7 +1619,7 @@ async fn a_late_catalog_starts_on_a_whole_group() {
 	.await;
 	let video = run.ledger(true);
 	let first = video.units.iter().find(|unit| unit.output).unwrap().group;
-	assert_eq!(first, 5, "the export starts on the first whole group after the catalog");
+	assert_eq!(first, 4, "the export starts on the group in flight");
 	assert_eq!(run.stats.release.dropped, 0);
 }
 

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

@@ -162,7 +162,6 @@ impl Release {
 					decode: frame.timestamp.into(),
 					restart: track.markers(),
 					skip: track.skips(),
-					cut: track.cuts(),
 					// The output opens on a group, at a random access point; from then on every
 					// object goes out on its own, so one dropped late is a gap of its packets only.
 					sync: frame.keyframe || self.released,