From 3addc4eb8968666fc65d4200af5e6ae5d1e27545 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 00:01:46 -0700 Subject: [PATCH 1/6] chore(quest): claim quest/m1/ts-reliability/validation-green.md Quest-Claim: 4a780c9ac85907ea From df07ab5634aae7e5606c35a9400bc3d90c21775f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 00:41:25 -0700 Subject: [PATCH 2/6] test(ts): grade holes on presentation timestamps; fix MPEG-2 lane's run dir variable The declared decode clock (#6453) derives each group's DTS from that group's frames alone, so a reordered group's first frames decode a tick apart below its earliest PTS. holes.py read the resulting DTS step at each group as missing media; grade gaps in presentation order instead, as --shed already does. The MPEG-2 lane set MOQ_TEST_RUNS, which binaries refuse since #6633. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ts.yml | 2 +- test/ts/holes.py | 25 ++++++++++++++----------- test/ts/holes.test.py | 19 +++++++++++++++++++ 3 files changed, 34 insertions(+), 12 deletions(-) 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/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..550f8f4a061 100644 --- a/test/ts/holes.test.py +++ b/test/ts/holes.test.py @@ -40,6 +40,25 @@ 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) + + 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. From 61c5b53dc741f4a65cc70b90210c6f75a433b9ae Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 02:22:31 -0700 Subject: [PATCH 3/6] fix(mux): start an export on a cut-short start group's prefix A path narrower than the media cuts every group, so a track waiting for a whole start group could stay dark for the whole run: the congested TS lanes started up to the last second, or one group past their join. What a track holds of a cut group starts on its sync frame and decodes, so it goes out ahead of the hole like any later cut. The cut counters the jitter buffer kept only for that trim go with it. Co-Authored-By: Claude Opus 5.5 --- doc/concept/stats.md | 2 +- rs/moq-mux/benches/jitter.rs | 3 - rs/moq-mux/src/container/flv/export.rs | 3 - rs/moq-mux/src/container/mkv/export.rs | 1 - rs/moq-mux/src/container/ts/export.rs | 14 +- rs/moq-mux/src/container/ts/export_test.rs | 82 ++++------ rs/moq-mux/src/container/ts/recovery_test.rs | 8 +- rs/moq-mux/src/container/ts/release.rs | 1 - rs/moq-mux/src/jitter.rs | 156 +++++-------------- test/ts/README.md | 10 +- 10 files changed, 78 insertions(+), 202 deletions(-) diff --git a/doc/concept/stats.md b/doc/concept/stats.md index a7954b0bfd8..6db4610835e 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -102,7 +102,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/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 7d7360e73d7..a8e4e35d5f7 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -218,7 +218,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 @@ -249,7 +249,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. @@ -512,14 +512,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, @@ -730,9 +729,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 @@ -769,7 +768,6 @@ impl Track { decode: key.into(), restart, skip, - cut, sync, leading, item: Queued { @@ -2598,7 +2596,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 63a4e9c03e6..d730eebab0b 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -5940,11 +5940,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); @@ -6002,17 +6003,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 @@ -6020,61 +6021,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..25ec9cc3c98 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 when the catalog lands"); 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 0cd02b5be23..18be04e9f73 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) @@ -368,9 +365,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 @@ -421,11 +415,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`]. /// @@ -561,8 +556,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 { @@ -688,10 +681,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. @@ -703,19 +695,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 { @@ -856,7 +835,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; @@ -913,9 +891,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 @@ -930,9 +905,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); @@ -941,13 +915,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; @@ -1296,22 +1268,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) { @@ -1417,8 +1373,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 } @@ -1733,7 +1689,6 @@ mod tests { decode: ms(decode).into(), restart: 0, skip: 0, - cut: 0, sync: true, leading: false, item, @@ -2245,40 +2200,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)] @@ -2366,7 +2287,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", @@ -2554,7 +2474,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", @@ -2784,36 +2703,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. From 0f193c45443c7ceb1f1d3a5b9a32f221dbba7634 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 02:45:25 -0700 Subject: [PATCH 4/6] quest: finish the TS workflow repair Co-Authored-By: Claude Opus 5.5 --- quest/m1/ts-reliability/README.md | 1 - quest/m1/ts-reliability/validation-green.md | 30 -------------------- rs/moq-mux/src/container/ts/recovery_test.rs | 2 +- 3 files changed, 1 insertion(+), 32 deletions(-) delete mode 100644 quest/m1/ts-reliability/validation-green.md diff --git a/quest/m1/ts-reliability/README.md b/quest/m1/ts-reliability/README.md index c908d9e6d33..04e56d983dd 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 revised PAT never leaves a broadcast meaning another program -- [TS workflow passes](/quest/m1/ts-reliability/validation-green.md) - the scheduled TS lanes pass on main again, with no lane loosened - [Export SI order](/quest/m1/ts-reliability/export-si-order.md) - a TS export sends SI and media after its PMT and a PCR interval, so a hop keeps first emissions - [A stitched switch returns to --delay](/quest/m1/ts-reliability/stitch-catch-up.md) - `--stitch` drops the old program's tail once the replacement acquires, so latency settles back to `--delay` 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/src/container/ts/recovery_test.rs b/rs/moq-mux/src/container/ts/recovery_test.rs index 25ec9cc3c98..9c890698b98 100644 --- a/rs/moq-mux/src/container/ts/recovery_test.rs +++ b/rs/moq-mux/src/container/ts/recovery_test.rs @@ -1619,7 +1619,7 @@ async fn a_late_catalog_starts_on_the_group_in_flight() { .await; let video = run.ledger(true); let first = video.units.iter().find(|unit| unit.output).unwrap().group; - assert_eq!(first, 4, "the export starts on the group in flight when the catalog lands"); + assert_eq!(first, 4, "the export starts on the group in flight"); assert_eq!(run.stats.release.dropped, 0); } From 5fa32f0662883e8ca498a36d594fbe1ea5be43b1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 08:39:28 -0700 Subject: [PATCH 5/6] test(ts): grade per-group decode steps through holes.main() Co-Authored-By: Claude Opus 5.5 --- test/ts/holes.test.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/test/ts/holes.test.py b/test/ts/holes.test.py index 550f8f4a061..bdc7a4e794a 100644 --- a/test/ts/holes.test.py +++ b/test/ts/holes.test.py @@ -58,6 +58,23 @@ 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 From ffce73adcb6d03b2599cbd81b5f125880583b634 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 13:39:47 -0700 Subject: [PATCH 6/6] fix(moq-mux): drop the removed cut field from a test merged from main Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/jitter.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index 29a8546c47f..5f368a81882 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -1670,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,