diff --git a/.github/workflows/ts.yml b/.github/workflows/ts.yml index 71d06cb737c..5d97604def1 100644 --- a/.github/workflows/ts.yml +++ b/.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 diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 5bc22c8e491..22e4949520f 100644 --- a/doc/concept/stats.md +++ b/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. | diff --git a/quest/m1/ts-reliability/README.md b/quest/m1/ts-reliability/README.md index 04d1b19d923..c71ae1b8823 100644 --- a/quest/m1/ts-reliability/README.md +++ b/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 diff --git a/quest/m1/ts-reliability/validation-green.md b/quest/m1/ts-reliability/validation-green.md deleted file mode 100644 index 96c2b63bf5d..00000000000 --- a/quest/m1/ts-reliability/validation-green.md +++ /dev/null @@ -1,30 +0,0 @@ -# [M] The TS workflow passes on main - -## Goal - -The scheduled TS workflow (`.github/workflows/ts.yml`) passes on `main`, with -no lane loosened. - -## Plan - -Two breaks, seen in run -[40195146593](https://github.com/moq-dev/moq/actions/runs/40195146593) -(`66af72315b`, 2026-10-25). Last green run: 2026-10-20 at `f13283b302`. - -- **New on 2026-10-25:** private PES, broadcast-sized CPB, full leak, full - multiplex, and headroom each lose 18 or 28 video frames at a regular - interval (`FAIL PID 0x100 h264: 475 frames of 40.00 ms, 18 holes, 1440 ms - missing`). Green on 2026-10-24 at `ea09d025a2`. Bisect - `ea09d025a2..66af72315b`; candidates include - [moq-dev/moq#6534](https://github.com/moq-dev/moq/pull/6534) (assemble TS - video across PES boundaries) and - [moq-dev/moq#6656](https://github.com/moq-dev/moq/pull/6656) (export - follows by cause). -- **Since 2026-10-21:** the congested path (`--headroom --rate 7000000`) - captures only the last second of its 120 s run, and the overflowing path - (`--queue 250ms`, CUBIC) starts 49 s late. Bisect - `f13283b302..dbf06bdcb3`. -- Fix each at its cause; the congestion lanes must keep their bufferbloat - shape and their assertions. - -Public API: none expected. Wire: none. diff --git a/rs/moq-mux/benches/jitter.rs b/rs/moq-mux/benches/jitter.rs index 1c083bc5328..d9950971e7e 100644 --- a/rs/moq-mux/benches/jitter.rs +++ b/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: (), diff --git a/rs/moq-mux/src/container/flv/export.rs b/rs/moq-mux/src/container/flv/export.rs index ac707f3eca7..32f75cc2f59 100644 --- a/rs/moq-mux/src/container/flv/export.rs +++ b/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 Export { 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, diff --git a/rs/moq-mux/src/container/mkv/export.rs b/rs/moq-mux/src/container/mkv/export.rs index 74bc7338b03..30eeb79b91b 100644 --- a/rs/moq-mux/src/container/mkv/export.rs +++ b/rs/moq-mux/src/container/mkv/export.rs @@ -536,7 +536,6 @@ impl Export { 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, diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 0a218af1422..86c84cbb12a 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -220,7 +220,7 @@ where video_start: Option, /// 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, /// 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 diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index a01c4d0d558..ffb83342705 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/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,17 +6251,17 @@ 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::>(), - "the export starts at the next group, whole" + (0..5).chain(GOP..4 * GOP).map(pts).collect::>(), + "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 @@ -6268,61 +6269,30 @@ async fn a_start_group_cut_short_starts_at_the_next() { /// 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); diff --git a/rs/moq-mux/src/container/ts/recovery_test.rs b/rs/moq-mux/src/container/ts/recovery_test.rs index 77fc7f0ed2a..9c890698b98 100644 --- a/rs/moq-mux/src/container/ts/recovery_test.rs +++ b/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); } diff --git a/rs/moq-mux/src/container/ts/release.rs b/rs/moq-mux/src/container/ts/release.rs index a2a3a1e7dc9..92d8bd215db 100644 --- a/rs/moq-mux/src/container/ts/release.rs +++ b/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, diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index e3a9847a584..5f368a81882 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -150,9 +150,6 @@ impl Buffer { { self.writing(key).lose(key, ahead); } - pub fn cut(&mut self, key: &K, cut: u64) { - self.writing(key).cut(key, cut); - } pub fn discard(&mut self, ready: &Ready) { if let Some(Handoff::Committed(committed)) = self.handoff.as_deref_mut() && !committed.keys.contains(&ready.track) @@ -381,9 +378,6 @@ pub(crate) struct Arrival { /// How many times the source's playhead had jumped (a skipped group, or a restart) /// when the frame was read. pub skip: u64, - /// How many times the source had lost the rest of a group it had begun delivering when the - /// frame was read: the skips that cut a group short. - pub cut: u64, /// Whether the frame decodes without the track's earlier frames. pub sync: bool, /// Whether it presents before the track's last sync frame, so it also references frames @@ -434,11 +428,12 @@ pub(crate) enum Push { /// anchoring on its first frame would carry that lag for the whole run. What the anchor /// makes due before the acquisition ended is dropped instead. /// -/// A track starts on a whole group. Until a frame of it has gone out, a loss (a frame -/// dropped here, or the rest of a group its source cut short) takes everything the track -/// holds with it, and the track starts at its next sync frame, so a joiner whose start group -/// was cut short starts at the next group rather than on the cut-off start. Whole groups -/// missing after one that ended take nothing. +/// A track starts on a sync frame. Until a frame of it has gone out, a frame dropped here +/// takes everything the track holds with it, and the track starts at its next sync frame. +/// The rest of a group its source cut short takes nothing: what the track holds of it +/// starts on its sync frame and decodes, and the hole after it is shed like any later one. A +/// path narrower than the media cuts every group, so waiting for a whole one could leave +/// the output dark for the whole run. /// What goes before a track's first frame out is [`Buffer::trimmed`], not lost; what drops /// after it is [`Buffer::dropped`]. /// @@ -574,8 +569,6 @@ struct Acquire { tracks: BTreeMap, /// A track crossed a group boundary while these frames were being read. grouped: bool, - /// Each track's cut count reported by [`Buffer::cut`] while these frames were held. - cuts: BTreeMap, } impl Acquire { @@ -701,10 +694,9 @@ struct Floor { struct Track { /// Frames in arrival order, each with its generation, decode time and deadline. queue: VecDeque<(u64, Decode, Instant, T)>, - /// The source's restart, skip and cut counters at the last frame. + /// The source's restart and skip counters at the last frame. restart: u64, skip: u64, - cut: u64, /// The generation of the last frame. generation: u64, /// Frames are dropped until the next sync frame. @@ -716,19 +708,6 @@ struct Track { started: bool, } -impl Track { - /// Take the source's cut count. The rest of a group cut since will never arrive, so until a - /// frame of the track has gone out, what it holds goes too. Returns how many frames that was. - fn cut(&mut self, cut: u64) -> u64 { - let lost = match cut != self.cut && !self.started { - true => std::mem::take(&mut self.queue).len() as u64, - false => 0, - }; - self.cut = cut; - lost - } -} - impl Engine { pub fn new(delay: Duration) -> Self { Self { @@ -869,7 +848,6 @@ impl Engine { frames: Vec::new(), tracks: BTreeMap::new(), grouped: false, - cuts: BTreeMap::new(), }); let mid = acquire.tracks.entry(key.clone()).or_default(); acquire.grouped |= arrival.sync && *mid; @@ -926,9 +904,6 @@ impl Engine { let (deadline, stale) = self.judge(&clock, &frame, acquiring); self.queue(key, frame, clock.generation, deadline, stale); } - for (key, cut) in acquire.cuts { - self.cut(&key, cut); - } } /// A held frame's deadline on `clock`, and whether it was due before it was read, when @@ -943,9 +918,8 @@ impl Engine { /// Queue a frame on `generation` due at `deadline`, unless it is `stale` or cannot make it. /// - /// Until a frame of the track has gone out, a loss (a frame dropped here, or the rest of a - /// group its source cut short) takes everything the track holds with it, so its output starts on a - /// whole group at its next sync frame rather than on a group's cut-off start. + /// Until a frame of the track has gone out, a frame dropped here takes everything the track + /// holds with it, so its output starts at its next sync frame rather than ahead of a hole. fn queue(&mut self, key: K, arrival: Arrival, generation: u64, deadline: Option, stale: bool) -> Push { if let Some(deadline) = deadline.filter(|_| !stale && !self.delay.is_zero()) { self.spare(arrival.read, since(arrival.arrived, deadline) as f64 / 1e9); @@ -954,13 +928,11 @@ impl Engine { queue: VecDeque::new(), restart: arrival.restart, skip: arrival.skip, - cut: arrival.cut, generation, waiting: false, resumed: false, started: false, }); - self.trimmed += track.cut(arrival.cut); track.restart = arrival.restart; track.skip = arrival.skip; track.generation = generation; @@ -1309,22 +1281,6 @@ impl Engine { } } - /// The source lost the rest of the group `key` was in, its `cut`th such loss - /// ([`Arrival::cut`]), known before the track's next frame: what a track that has not - /// started holds goes with it now, rather than once that frame arrives, which may be after - /// the cut-off start was due. - pub fn cut(&mut self, key: &K, cut: u64) { - if let Some(acquire) = &mut self.acquire - && acquire.tracks.contains_key(key) - { - acquire.cuts.insert(key.clone(), cut); - return; - } - if let Some(track) = self.tracks.get_mut(key) { - self.trimmed += track.cut(cut); - } - } - /// Count a frame the caller dropped after [`Self::poll_next`] released it. The first of its /// track is trimmed and leaves the track not started; a later one is a loss. pub fn discard(&mut self, ready: &Ready) { @@ -1430,8 +1386,8 @@ impl Engine { self.dropped } - /// How many frames were dropped before their track's first went out, so it starts on a whole - /// group: what a joiner is handed older than the delay, and a start group a loss cut short. + /// How many frames were dropped before their track's first went out, so it starts on a sync + /// frame: what a joiner is handed older than the delay, and what it held with them. pub fn trimmed(&self) -> u64 { self.trimmed } @@ -1714,7 +1670,6 @@ mod tests { decode: ms(u64::from(key) * 100_000 + (now - start).as_millis() as u64).into(), restart: 0, skip: 0, - cut: 0, sync: true, leading: false, item: now, @@ -1848,7 +1803,6 @@ mod tests { decode: ms(decode).into(), restart: 0, skip: 0, - cut: 0, sync: true, leading: false, item, @@ -2360,40 +2314,6 @@ mod tests { assert_eq!(buffer.dropped(), 0); } - /// A cut the source reports before the track's next frame takes the start group it left - /// short then, whether the clock is still being acquired or the group is queued, though the - /// next group arrives only after that start was due. - #[tokio::test(start_paused = true)] - async fn a_cut_reported_before_the_next_group_drops_the_start_group() { - for replay in [false, true] { - let mut buffer = Buffer::new(DELAY); - if replay { - buffer = buffer.replay(); - } - let video = |decode, sync, cut, item| Arrival { - sync, - skip: cut, - cut, - ..arrival(moq_clock::now(), decode, item) - }; - buffer.push(1, video(0, true, 0, "v0")).unwrap(); - tokio::time::advance(Duration::from_millis(40)).await; - buffer.push(1, video(40, false, 0, "v40")).unwrap(); - tokio::time::advance(Duration::from_millis(20)).await; - buffer.cut(&1, 1); - tokio::time::advance(DELAY).await; - assert_eq!( - due(&mut buffer), - Vec::<&str>::new(), - "replay {replay}: the cut-off start went out" - ); - buffer.push(1, video(160, true, 1, "v160")).unwrap(); - tokio::time::advance(2 * DELAY).await; - assert_eq!(due(&mut buffer), ["v160"], "replay {replay}"); - assert_eq!((buffer.dropped(), buffer.trimmed()), (0, 2), "replay {replay}"); - } - } - /// Every frame of a generation goes out before the next generation's, even one the new /// clock puts earlier. #[tokio::test(start_paused = true)] @@ -2481,7 +2401,6 @@ mod tests { decode: Timestamp::from_micros(decode.as_micros() as u64).unwrap().into(), restart: 0, skip: 0, - cut: 0, sync: true, leading: false, item: "v", @@ -2669,7 +2588,6 @@ mod tests { decode: Timestamp::from_micros(decode.as_micros() as u64).unwrap().into(), restart: 0, skip: 0, - cut: 0, sync: true, leading: false, item: "v", @@ -2899,36 +2817,31 @@ mod tests { assert_eq!(buffer.trimmed(), 6); } - /// A start group its source cuts short before anything of it went out is dropped whole, - /// so the track starts on the next group rather than on the cut-off start. Whole groups - /// missing after one that ended take nothing with them. + /// A start group its source cuts short before anything of it went out still goes out, from + /// its sync frame, ahead of the hole: a path narrower than the media cuts every group, so a + /// track waiting for a whole one might never start. Nothing is trimmed. #[tokio::test(start_paused = true)] - async fn a_start_group_cut_by_a_gap_is_dropped_whole() { - let start = Instant::now(); - let mut buffer = Buffer::new(DELAY).replay(); - let video = |decode, sync, skip, cut, item| Arrival { - sync, - skip, - cut, - ..arrival(start, decode, item) - }; - buffer.push(1, video(0, true, 0, 0, "v0")).unwrap(); - buffer.push(1, video(40, false, 0, 0, "v40")).unwrap(); - // The rest of the group never arrives: the source cuts it short. - assert_eq!(buffer.push(1, video(120, true, 1, 1, "v120")).unwrap(), Push::Queued); - buffer.push(1, video(160, false, 1, 1, "v160")).unwrap(); - // A whole group is missing after that one ended. - assert_eq!(buffer.push(1, video(280, true, 2, 1, "v280")).unwrap(), Push::Queued); - tokio::time::advance(Duration::from_secs(1)).await; - assert_eq!(due(&mut buffer), ["v120", "v160", "v280"]); - assert_eq!((buffer.dropped(), buffer.trimmed()), (0, 2)); - - // Once a frame went out, a cut is a loss in the output, and the rest still goes out. - buffer.push(1, video(320, false, 2, 1, "v320")).unwrap(); - buffer.push(1, video(400, true, 3, 2, "v400")).unwrap(); - tokio::time::advance(Duration::from_secs(1)).await; - assert_eq!(due(&mut buffer), ["v320", "v400"]); - assert_eq!((buffer.dropped(), buffer.trimmed()), (0, 2)); + async fn a_start_group_cut_short_goes_out() { + for replay in [false, true] { + let mut buffer = Buffer::new(DELAY); + if replay { + buffer = buffer.replay(); + } + let video = |decode, sync, skip, item| Arrival { + sync, + skip, + ..arrival(moq_clock::now(), decode, item) + }; + buffer.push(1, video(0, true, 0, "v0")).unwrap(); + tokio::time::advance(Duration::from_millis(40)).await; + buffer.push(1, video(40, false, 0, "v40")).unwrap(); + // The rest of the group never arrives: the source cuts it short. + tokio::time::advance(Duration::from_millis(80)).await; + buffer.push(1, video(120, true, 1, "v120")).unwrap(); + tokio::time::advance(2 * DELAY).await; + assert_eq!(due(&mut buffer), ["v0", "v40", "v120"], "replay {replay}"); + assert_eq!((buffer.dropped(), buffer.trimmed()), (0, 0), "replay {replay}"); + } } /// The keyframe a joiner holds while the export starts is read early and passes, while the diff --git a/test/ts/README.md b/test/ts/README.md index efdb7afaf97..25e393cb5b4 100644 --- a/test/ts/README.md +++ b/test/ts/README.md @@ -21,7 +21,9 @@ software profiles below; it does not certify a hardware IRD or every TS layout. - `overhead.py` counts the packets a capture spends on clocks and audio framing against its source. - `holes.py` finds media missing from a generated clip's capture, from its - timestamps against the source's. + presentation timestamps against the source's. The export derives each group's + decode timestamps from that group alone, so they step at group boundaries with + nothing lost. - `passthrough.py` places every packet of an `export ts --passthrough` capture at its source position and compares the compliance verdicts. @@ -892,9 +894,9 @@ the same access unit: AAC overlap and perceptual-noise state differ between a cold decoder and a decoder that saw earlier frames. No arbitrary warmup samples are discarded. PES boundaries and TS packetization may change. -The suffix is the exporter's to choose, and it starts on a whole group: a join -whose first group is cut short before anything of it went out starts at the next -one. Anything lost after that fails. When it does, the harness prints what the +The suffix is the exporter's to choose, and it starts on a keyframe: a join +whose start group went stale before anything of it went out starts at the next +one. Anything lost after the first frame out fails, a group cut short included. When it does, the harness prints what the exporter reported: the `TS export release clock` counts (`dropped` late, `trimmed` before the output began, `gaps` skipped upstream) and each loss it logged. diff --git a/test/ts/holes.py b/test/ts/holes.py index 1e5dd51a708..4d0484bd9c9 100755 --- a/test/ts/holes.py +++ b/test/ts/holes.py @@ -4,9 +4,12 @@ ./holes.py capture.ts source.ts [--shed] Every audio and video stream of a generated clip has a constant frame duration, so the step -between consecutive decode timestamps is the same throughout. A capture's step past one and a -half of its source's is a hole: media the round-trip lost, whether the path dropped it, the relay evicted -it, or the exporter missed its deadline. A lost prefix or tail leaves no step behind, so each +between consecutive presentation timestamps, in presentation order, is the same throughout. A +capture's step past one and a half of its source's is a hole: media the round-trip lost, whether +the path dropped it, the relay evicted it, or the exporter missed its deadline. Presentation +timestamps, because the export keeps them while it derives each group's decode timestamps from +that group's frames alone: a reordered group's first frames decode just before its earliest +presentation, not a frame apart, which leaves a decode step at every group with nothing lost. A lost prefix or tail leaves no step behind, so each source stream must also be in the capture, on its PID, ending where the source's does: the export keeps the source's timestamps. It may start later, by up to JOIN: the subscriber joins at the relay's latest group once the catalog is out, which skips the first second or two of @@ -43,15 +46,15 @@ JOIN = Fraction(3) -def step(dts: list[int]) -> int: - """A stream's frame step: its most common decode-timestamp step, or 0 with fewer than two frames.""" - steps = [b - a for a, b in zip(dts, dts[1:])] +def step(times: list[int]) -> int: + """A stream's frame step: its most common step between ascending timestamps, or 0 with fewer than two frames.""" + steps = [b - a for a, b in zip(times, times[1:])] return Counter(steps).most_common(1)[0][0] if steps else 0 -def holes(dts: list[int], frame: int) -> list[tuple[int, int]]: - """The (dts, step) of each step past one and a half frames.""" - return [(a, b - a) for a, b in zip(dts, dts[1:]) if 2 * (b - a) > 3 * frame] +def holes(times: list[int], frame: int) -> list[tuple[int, int]]: + """The (timestamp, step) of each step past one and a half frames between ascending timestamps.""" + return [(a, b - a) for a, b in zip(times, times[1:]) if 2 * (b - a) > 3 * frame] @dataclass @@ -158,7 +161,7 @@ def main() -> None: for pid, src in sorted(source.items()): label = f"PID {pid} {src.info['codec_name']}" # The source's step, not the capture's: a capture that kept every other frame has a step of its own. - frame = step(src.dts) + frame = step(sorted(src.pts)) if frame == 0: sys.exit(f"error: {label} has fewer than two frames in {args.source}") if pid not in capture: @@ -187,7 +190,7 @@ def main() -> None: reasons.append(f"starts {float(late):.3f}s after its source, past the {float(JOIN)}s join") summary = f"{graded.missing} frames ({float(graded.missing * frame * base):.1f} s) shed" else: - found = holes(got.dts, frame) + found = holes(sorted(got.pts), frame) reasons = [f"at {float(at * base):.3f}s: {float(gap * base) * 1000:.0f} ms step" for at, gap in found[:5]] reasons += cuts(src.span, got.span, frame * base) missing = sum(gap - frame for _, gap in found) * base diff --git a/test/ts/holes.test.py b/test/ts/holes.test.py index f41307243ad..bdc7a4e794a 100644 --- a/test/ts/holes.test.py +++ b/test/ts/holes.test.py @@ -40,6 +40,42 @@ def test_too_short(self): self.assertEqual(holes.step([0]), 0) +# Two B-pyramid groups of 25 frames in decode order (I P B b b ...), in 40 ms frames. +PYRAMID = [0] + [i + d for i in range(1, 25, 4) for d in (3, 1, 0, 2)] + + +class PerGroupDecodeTimes(unittest.TestCase): + PTS = [(g * 25 + i) * 3600 for g in range(2) for i in PYRAMID] + # The export decodes each group's first two frames a tick apart just below its earliest + # presentation, so its decode timestamps step at every group with nothing lost. + DTS = [g * 25 * 3600 + (i - 2 if i < 2 else (i - 2) * 3600) for g in range(2) for i in range(25)] + + def test_decode_steps_at_groups_are_not_holes(self): + self.assertTrue(holes.holes(self.DTS, 3600)) + self.assertEqual(holes.holes(sorted(self.PTS), holes.step(sorted(self.PTS))), []) + + def test_a_lost_b_picture_is_a_hole(self): + lost = self.PTS[:10] + self.PTS[11:] + self.assertEqual(len(holes.holes(sorted(lost), holes.step(sorted(self.PTS)))), 1) + + def grade(self, kept: list[int]) -> int: + def stream(index): + dts, pts = [self.DTS[i] for i in index], [self.PTS[i] for i in index] + return holes.Stream( + {"codec_name": "h264", "time_base": "1/90000"}, dts, pts, [True] * len(pts), + tuple(Fraction(t, 90_000) for t in (min(pts), max(pts), dts[-1])), + ) + with patch.object(holes, "probe", side_effect=[{"0x101": stream(kept)}, {"0x101": stream(range(50))}]), \ + patch.object(sys, "argv", ["holes.py", "capture.ts", "source.ts"]), \ + contextlib.redirect_stdout(io.StringIO()), self.assertRaises(SystemExit) as exit: + holes.main() + return exit.exception.code + + def test_main_grades_presentation_timestamps(self): + self.assertEqual(self.grade(range(50)), 0) + self.assertEqual(self.grade([i for i in range(50) if i != 10]), 1) + + class Shed(unittest.TestCase): # Four groups of 25 frames, each opening on a sync frame, with each P picture decoded ahead # of the B picture it presents after.