diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 4985bbabbba..e2fc6db891d 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -35,7 +35,6 @@ Public API: none. Wire: none. ## Required -- [TS decoded content](/quest/m1/test-flakes-2/ts-video-content.md) - CI's TS compliance decoded-content checks fail only on real loss, and the exporter names every frame it drops - [Interop close code](/quest/m1/test-flakes-2/interop-close-code.md) - the browser reliably sees `unauthorized` for a refused session - [Auth refresh mocks time](/quest/m1/test-flakes-2/auth-refresh-mock-time.md) - moq-relay's auth refresh and fixed-lease tests measure a grant's life on one mocked clock - [uring teardown close](/quest/m1/test-flakes-2/uring-teardown-close.md) - moq-uring's published-close teardown test holds under load, fixed at its cause diff --git a/quest/m1/test-flakes-2/ts-video-content.md b/quest/m1/test-flakes-2/ts-video-content.md deleted file mode 100644 index d4d10232bc8..00000000000 --- a/quest/m1/test-flakes-2/ts-video-content.md +++ /dev/null @@ -1,153 +0,0 @@ -# [L] TS compliance fails only on real loss - -## Goal - -CI's `TS compliance` step (`just test ts`, `.github/workflows/ts.yml`) -gives the same verdict on every run of the same code. Its decoded-content -checks (`test/ts/decode.py`) fail only when the exporter really loses or -corrupts content, never because of how the host scheduled the run. Each -frame the exporter drops is counted and logged, so a failure says what was -dropped and why. The checks stay as strict: a captured suffix that loses or -duplicates a picture or an audio frame still fails. - -## Plan - -### Evidence - -CI: seen 2026-10-13 on #5471, #5481 and two unrelated branches within the -same hour, each passing on rerun. Then, from `gh run list --workflow ts.yml` -on 2026-10-14: since #5328 merged (2026-10-13 23:49 UTC), 4 of 41 completed -PR runs failed `video-content` ("decoded pictures are not an intact suffix of -the source"): #5446 at f6629aac8 -([38506130438](https://github.com/moq-dev/moq/actions/runs/38506130438)), -#5148 at 5adfd84a6 -([38502761750](https://github.com/moq-dev/moq/actions/runs/38502761750)), -`quest/m1/cli-duration-defaults` at e7e3eb1c4 -([38502673714](https://github.com/moq-dev/moq/actions/runs/38502673714)), and -#5481's branch `quest/m1/catalog-wall-clock-precision` at 312ac6260 -([38503267980](https://github.com/moq-dev/moq/actions/runs/38503267980)). -The first three do not touch TS; #5481 changed the shared clock this -pipeline reads, so keep it out of any argument that narrows the cause. -Only the default `just test ts` round failed; every later round passed in the -same runs. Main has no run yet: the workflow runs on matching PRs and a -nightly that had not fired. Failures on branches without #5471 or #5481 -point to an older cause, but check a failing capture against those diffs -before ruling them out. - -Every failing capture was 24531556 bytes against 23281544 in passing runs. -The difference, 1250012 bytes, is 1 s at the padded 10 Mb/s rate, so the -failing export started one 25-frame GOP earlier (group 1, not group 2). The -`--hrd` and `--headroom` rounds show the same one-GOP swing between runs but -do not run `decode.py`. Each failing run uploaded a `ts-failure` artifact -(`decode/*/frames.sha256`, `sub.log`), kept 7 days. - -Local, on 2026-10-14, `test/ts/run.sh` on a host at load ~28 on 24 cores -failed 3 of 4 runs, in two other ways: - -- `video-content`, once. The subscriber's `https://` race was won by - WebSocket, then upgraded to WebTransport; the old session "did not drain in - time" (GOAWAY timeout), and the exporter logged "current group evicted; - skipping to next buffered group" (`rs/moq-mux/src/container/consumer.rs`) - and one "frame missed its deadline". About 30 pictures across groups 10 and - 11 were missing, yet FFmpeg decoded the capture without an error. -- `audio-timing` ("capture has a gap or overlap in decoded audio"), twice, - over WebTransport with no upgrade and intact video. Two or three AAC frames - (43 to 64 ms) vanished mid-stream with no warning anywhere, and the - exporter's release clock reported `dropped=0`. - -Then [#5565](https://github.com/moq-dev/moq/issues/5565), on 2026-10-15: -4 of 51 `ts.yml` runs between 2026-10-14 19:57Z and 2026-10-15 03:55Z failed -`video-content` the same way, two on branches that change nothing in moq-net, -moq-relay or moq-mux's consumer: 38634848576 (`fix/tokio-windows-refused-dial`), -38657903232 (#5553), 38662560784 (`quest/m1/papercuts-rs`) and 38677047938 -(#5548's branch). Their artifacts expire about 2026-10-22. Each `sub.log` has -`current group evicted; skipping to next buffered group track="0.avc3" -group=1 error=Hang(Moq(Stream(Old)))`, 17 to 21 ms after a video subscribe -at about 1.98 s, and the capture keeps pictures 25 to about 33, then 50 -onward. #5548 moved the zero-budget conviction from the successor's first -frame to its arrival, essentially the same instant, so it should not change -the rate. Not reproduced locally: 0 of 15 on main (3abf0161c, macOS arm64). - -So in each case the checks saw real loss. The flake is what makes the -exporter lose content, and the loss goes unreported. - -2026-10-15: the same `video-content` failure blocked unrelated #5566 twice in -one run ([38725177110](https://github.com/moq-dev/moq/actions/runs/38725177110), -both attempts), and hit #5553 and #5148 again, so the rate is above the 4 in -41 estimate. In each of #5553's and #5566's `sub.log`, about 20 ms after the -video subscribe: `current group evicted ... group=1 -error=Hang(Moq(Stream(Old)))`, and `relay.log` and `pub.log` show -`max_delay: 0ns` on `0.avc3`. The publisher serves group 1, group 2's first -frame lands about 17 ms later, `poll_drifted` judges group 1 zero behind the -edge, and it resets group 1's stream `Old` before the raised budget reaches -it through the relay. That confirms hypothesis 1's first candidate (analysis -on #5565). Capture size does not separate the cases at a 1 s `--delay`: a -passing local run captured 25156468 bytes, like a failing CI run, so use the -log signature. - -### Hypotheses - -1. Startup truncation (CI). The importer holds the catalog for its 2 s PCR - measurement window (`rs/moq-mux/src/container/ts/mux_rate.rs`), so the - exporter subscribes around group 2's IDR, and whether group 1 is still - the newest is a wall-clock race. A group-1 join can then keep only a - prefix of group 1 before skipping to group 2. Two candidates: - - The live subscription starts with `max_delay = 0` and only relaxes to - the budget after the first frame (`rs/moq-mux/src/container/source.rs`), - so moq-net may expire group 1 partway through. - - The jitter buffer's acquisition judges staleness per frame against its - own read time (`judge`/`queue` in `rs/moq-mux/src/jitter.rs`). The held - IDR can pass while the frames read after the tables were built are - stale, and the track then waits for the next keyframe. Those drops are - `Wait::Acquired`, which is not counted in `dropped` or logged. -2. Transport upgrade (local mode 1). The WebSocket-to-QUIC upgrade's handover - evicts the current group. [#5484](https://github.com/moq-dev/moq/pull/5484) - (open) removes the upgrade. Check whether a GOAWAY drain still loses the - current group's tail, since drains remain. -3. Silent audio loss under load (local mode 2). Each AAC frame is its own - group (`rs/moq-mux/src/container/ts/import.rs`), so a group expired by the - latency budget, or skipped by the consumer, drops one frame with no count. - -Ruled out from reading the code: picture duplication, timestamp rebasing, and -pre-IDR content in group 0 (the importer opens a group only at a keyframe). - -### Steps - -- Reproduce first. Download a `ts-failure` artifact from one of the CI runs - above before it expires (about 2026-10-21) and map the captured pictures - onto the source to confirm hypothesis 1. Locally, loop `test/ts/run.sh` - beside a CPU hog, as above. -- Fail loud: every frame the exporter drops (acquisition, late, or a skipped - or expired group) is counted and logged, and the harness prints that count - beside a failing check. -- Fix each confirmed cause at its source. Recommended for hypothesis 1: an - export starts at a whole group, so acquisition admits or drops a group - entirely rather than emitting a prefix. Keep the zero-budget live-edge - pick (`ExportSource::live`): starting with the full budget would let a - join reach back to an older retained group, adding startup latency and - replay, a user-visible change. Make the group it picks atomic instead. - Each fix lands with a regression that fails without it, on a paused Tokio - clock or the controlled clock, never wall time. The jitter and export - tests in `rs/moq-mux` already run on mocked time. -- Keep the harness on its real-time path; this quest fixes the exporter, not - the grader. If #5484 has landed, rerun on top of it before chasing - hypothesis 2. Never loosen the comparison, widen a timeout or `--delay`, or - add a retry. -- Open question for the implementer to put to the maintainer, once the causes - are known: whether a frame dropped because the host starved the exporter - past its latency budget fails `decode.py`, or fails a separate named check - fed by the exporter's drop count. Recommended: a separate check, so - `video-content` and `audio-content` mean corruption only. - -Public API: none expected; the drop count may surface in the exporter's -summary. Wire: none. - -## Closes - -- [#5565](https://github.com/moq-dev/moq/issues/5565) - the TS compliance video-content flake - -## Related - -- [#5484](https://github.com/moq-dev/moq/pull/5484) - removes the WebSocket-to-QUIC upgrade behind local mode 1 -- [Loss and recovery tests](/quest/m1/ts-reliability/recovery.md) - deliberate MoQ content loss, which this flake hits by accident -- [Controlled time](/quest/m1/time/README.md) - the clock the regressions should run on diff --git a/quest/m1/ts-reliability/README.md b/quest/m1/ts-reliability/README.md index 622110bdd09..eaa018bcd33 100644 --- a/quest/m1/ts-reliability/README.md +++ b/quest/m1/ts-reliability/README.md @@ -102,6 +102,8 @@ remain with their current quests. - [Whole-PES import](/quest/m1/ts-reliability/pes-import.md) - private PES travels whole at its source placement, a missing PTS stays missing, and the `pes` framing is deleted - [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 diff --git a/quest/m1/ts-reliability/audio-lead.md b/quest/m1/ts-reliability/audio-lead.md new file mode 100644 index 00000000000..f7befebe023 --- /dev/null +++ b/quest/m1/ts-reliability/audio-lead.md @@ -0,0 +1,27 @@ +# [S] A trimmed video start keeps audio behind the keyframe + +## Goal + +`moq export ts` begins on a video keyframe even when the video's start group +is dropped at acquisition and its next keyframe has not arrived yet. No audio +or data frame goes out ahead of the first video frame. + +## Plan + +Found reading [#5713](https://github.com/moq-dev/moq/pull/5713), not +reproduced. Until video goes out, `Export::leads_video` +(`rs/moq-mux/src/container/ts/export.rs`) drops non-video frames that present +before the earliest queued video frame. With no video frame queued it falls +back to `video_start`. When the acquisition drops the video's start group as +stale, the video queue is empty until the next keyframe arrives. Audio after +`video_start` then goes out first, by up to the remaining GOP less the delay. +The upstream-cut path that #5565 hit is not affected, because the next +group's keyframe arrives with the gap. + +Recommended fix, settled in #5713: while no video frame has gone out and none +is queued, drop non-video frames as trimmed. Land it only with a regression +that fails without it. Three paused-clock attempts on the `Interleave` rig did +not drop the start group, because a frame's `arrived` is when its source was +last found empty. A fixture likely needs the tables to wait on something +other than the video (a late catalog entry), so the held keyframe is read +well before the rest of its group. diff --git a/quest/m1/ts-reliability/data-break.md b/quest/m1/ts-reliability/data-break.md new file mode 100644 index 00000000000..7ab77386be8 --- /dev/null +++ b/quest/m1/ts-reliability/data-break.md @@ -0,0 +1,29 @@ +# [M] A data track's declared break is not a loss + +## Goal + +A consumer tells a data track's declared discontinuity from media lost +upstream. `moq export ts` stops counting a declared break on a verbatim PES +track (DVB AC-3, teletext, SCTE-35, ID3) in `stats::Export::gaps`, and stops +warning "media skipped upstream" for it. + +## Plan + +Found in [#5713](https://github.com/moq-dev/moq/pull/5713). Audio and video +mark a break with one empty frame, which the consumer reads as `Read::End`. +An empty payload is data on a data track, so `Producer::discontinuity` +(`rs/moq-mux/src/container/producer.rs`) skips a sequence with no object +instead. The consumer reads that skip as `Read::Gap`, the same as a lost +group, so the export's gap count and warning treat it as loss. #5713 +documents the overlap. Its start trim already follows only gaps that cut a +group short, and the break comes after its group ended, so nothing is trimmed. + +Settle the carriage first: the marker must be visible on the wire, and an +empty group (zero objects) already means nothing. Any wire or format change +updates `drafts/` and `js/hang` in the same PR. Extend +`export_drops_an_ac3_frame_cut_by_a_marker` (`rs/moq-mux/src/container/ts/export_test.rs`) +to assert `gaps` stays 0 across the break. + +## Related + +- [Loss and recovery tests](/quest/m1/ts-reliability/recovery.md) - the loss reporting this keeps honest diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 9f9579d76f1..4d0fcd780ee 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -399,11 +399,13 @@ impl Subscribe { // the way `publish` reports one that stops arriving. let mut log = moq_mux::container::ts::stats::Log::default(); let mut sampled = tokio::time::Instant::now(); - // The jitter buffer's late drops and out-of-tolerance count, as last logged. - let mut reported = (0, 0); + // The export's drop, gap and out-of-tolerance counts, as last logged. + let mut reported = (0, 0, 0, 0); let release = |stats: &moq_mux::container::ts::stats::Export| { tracing::info!( dropped = stats.dropped, + trimmed = stats.trimmed, + gaps = stats.gaps, drift_ppm = ?stats.drift, out_of_tolerance = stats.out_of_tolerance, "TS export release clock" @@ -421,8 +423,9 @@ impl Subscribe { if sampled.elapsed() >= moq_mux::container::ts::stats::Log::INTERVAL { sampled = tokio::time::Instant::now(); let stats = ts.export().stats(); - if (stats.dropped, stats.out_of_tolerance) != reported { - reported = (stats.dropped, stats.out_of_tolerance); + let counts = (stats.dropped, stats.trimmed, stats.gaps, stats.out_of_tolerance); + if counts != reported { + reported = counts; release(&stats); } log.sample(stats.into()); diff --git a/rs/moq-mux/benches/jitter.rs b/rs/moq-mux/benches/jitter.rs index ad5ba32f003..95faafd1545 100644 --- a/rs/moq-mux/benches/jitter.rs +++ b/rs/moq-mux/benches/jitter.rs @@ -27,6 +27,7 @@ async fn steer(tracks: u16, floors: u64) -> u64 { decode: Timestamp::from_millis(second * 1_000).unwrap(), restart: 0, skip: 0, + cut: 0, sync: true, leading: false, item: (), diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index 243e5020b47..ed459e587eb 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -182,6 +182,11 @@ impl Consumer { self.group } + /// The track's name. + pub(crate) fn name(&self) -> &str { + self.track.name() + } + /// Read the next frame from the track. /// /// This method handles timestamp decoding, group ordering, and age management diff --git a/rs/moq-mux/src/container/flv/export.rs b/rs/moq-mux/src/container/flv/export.rs index 3ffefcd3b11..a1a7b6fcfc1 100644 --- a/rs/moq-mux/src/container/flv/export.rs +++ b/rs/moq-mux/src/container/flv/export.rs @@ -362,6 +362,7 @@ impl Export { decode: frame.timestamp, restart: track.source.markers(), skip: track.source.skips(), + cut: track.source.cuts(), sync: kind == 1 || frame.keyframe, leading: false, item: frame, @@ -380,6 +381,7 @@ impl Export { read, restart, skip, + cut, sync, leading, item, @@ -393,6 +395,7 @@ impl Export { decode: moq_net::Timestamp::from_millis(u64::from(timestamp))?, restart, skip, + cut, sync, leading, item: (item, timestamp), diff --git a/rs/moq-mux/src/container/mkv/export.rs b/rs/moq-mux/src/container/mkv/export.rs index 963b294613b..30b4aec5405 100644 --- a/rs/moq-mux/src/container/mkv/export.rs +++ b/rs/moq-mux/src/container/mkv/export.rs @@ -381,6 +381,7 @@ impl Export { decode: frame.timestamp, 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/source.rs b/rs/moq-mux/src/container/source.rs index dccea4b3bfc..09c4a701c2c 100644 --- a/rs/moq-mux/src/container/source.rs +++ b/rs/moq-mux/src/container/source.rs @@ -91,6 +91,9 @@ pub(crate) struct ExportSource { absorbed_keyframe: bool, skips: u64, markers: u64, + /// Whether a frame of the group being read went out and the group has not ended since. + open: bool, + cuts: u64, // A stopped encoder may still emit final packets in this group on the old timeline. ends: Option<(u64, u64)>, } @@ -176,6 +179,8 @@ impl ExportSource { absorbed_keyframe: false, skips: 0, markers: 0, + open: false, + cuts: 0, ends: None, }; source.resolve_video_dimensions(&[])?; @@ -211,6 +216,8 @@ impl ExportSource { absorbed_keyframe: false, skips: 0, markers: 0, + open: false, + cuts: 0, ends: None, })) } @@ -239,6 +246,8 @@ impl ExportSource { absorbed_keyframe: false, skips: 0, markers: 0, + open: false, + cuts: 0, ends: None, }) } @@ -300,6 +309,12 @@ impl ExportSource { self.markers } + /// Gaps that lost the rest of a group this source had begun delivering, rather than whole + /// groups after one that ended. + pub fn cuts(&self) -> u64 { + self.cuts + } + /// True if the codec config is resolved (either present in the catalog, /// no transform attached, or the transform has built its record). pub fn header_ready(&self) -> bool { @@ -397,11 +412,21 @@ impl ExportSource { if self.ends.is_some_and(|(end_group, _)| end_group != group) { self.finish_ends(); } + self.open = true; frame } - Some(Item::Read(Read::Gap { .. })) => { + Some(Item::Read(Read::Gap { timestamp, duration })) => { + // Say what the export never receives and where. A data track declares a + // break by skipping a sequence, which reads the same though nothing is lost. + tracing::warn!( + track = consumer.name(), + at = ?std::time::Duration::from(timestamp), + skipped = ?std::time::Duration::from(duration), + "media skipped upstream, or a data track's declared break; resuming at the next group" + ); self.finish_ends(); self.skips += 1; + self.cuts += u64::from(std::mem::take(&mut self.open)); return Poll::Ready(Ok(Some(Event::GroupEnd))); } Some(Item::Read(Read::End { group, end })) => { @@ -414,6 +439,7 @@ impl ExportSource { Some(Item::FrameEnd(end)) => return Poll::Ready(Ok(Some(Event::FrameEnd { end, opens: false }))), Some(Item::GroupEnd) => { self.finish_ends(); + self.open = false; return Poll::Ready(Ok(Some(Event::GroupEnd))); } None => { @@ -677,6 +703,41 @@ mod tests { (track, export) } + /// A gap that loses the rest of a group already begun is a cut; whole groups missing after + /// one that ended are not. + #[tokio::test(start_paused = true)] + async fn only_a_gap_inside_a_group_cuts_it() { + use crate::container::Container as _; + use crate::container::test_util::raw_frame; + + let (track, mut export) = audio_export().await; + let wire = HangContainer::Legacy(crate::container::Kind::Audio); + let group = |sequence: u64, timestamp: u64| { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + wire.write(&mut group, &[raw_frame(timestamp, b"audio", true)]).unwrap(); + group + }; + // Group 1 is missing whole; group 3 is cut short after its first frame. + group(0, 0).finish().unwrap(); + group(2, 40_000).finish().unwrap(); + let cut = group(3, 60_000); + group(4, 80_000).finish().unwrap(); + track.finish().unwrap(); + + let waiter = kio::Waiter::noop(); + let read = |export: &mut ExportSource| { + let Poll::Ready(Ok(Some(frame))) = export.poll_read(&waiter) else { + panic!("frame ready") + }; + (frame.timestamp.as_micros(), export.skips(), export.cuts()) + }; + assert_eq!(read(&mut export), (0, 0, 0)); + assert_eq!(read(&mut export), (40_000, 1, 0), "group 1 is missing whole"); + assert_eq!(read(&mut export), (60_000, 1, 0)); + cut.abort(moq_net::Error::Old).unwrap(); + assert_eq!(read(&mut export), (80_000, 2, 1), "group 3 was cut short"); + } + #[tokio::test(start_paused = true)] async fn a_gap_then_end_reaches_the_exporter_in_order() { use crate::container::Container as _; diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index afa1d7a998e..4124b20a6a9 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -187,6 +187,13 @@ where /// up before it ever configures video. `None` until the tables are built, and for /// programs with no video track (nothing to align to). video_start: Option, + /// Whether a video frame has gone out since [`Self::video_start`] was set. Until one has, + /// 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 ([`jitter::Buffer`]). + video_out: bool, + /// Frames dropped before the jitter buffer: ahead of the first video keyframe, or of their + /// codec config. + trimmed: u64, /// Each elementary stream's access units and silence on the PCR returned ([`Self::stats`]). liveness: super::import::Liveness, } @@ -208,6 +215,8 @@ 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`]). + cut: u64, /// The earliest the frame could have arrived: when its source was last found empty. arrived: web_async::time::Instant, /// When it was read: the latest it could have arrived. @@ -243,9 +252,9 @@ struct Track { empty: web_async::time::Instant, /// The marker count the decode clock runs under. restart: u64, - /// The marker and skip counts of the sources the track read before this one, so the + /// The marker, skip and cut counts of the sources the track read before this one, so the /// counts keep rising across a resubscription. - base: (u64, u64), + base: (u64, u64, u64), finished: bool, pid: u16, kind: Kind, @@ -256,8 +265,12 @@ struct Track { clock: DecodeClock<(Pending, Option)>, /// Loss counter at the last read, excluding declared ends. recovery_gap: Option, - /// Leading pictures below the recovery keyframe have unavailable references. - leading_floor: Option, + /// Leading pictures below the recovery keyframe have unavailable references, and whether + /// the track queued frames ahead of that keyframe, so dropping them leaves a hole. + leading_floor: Option<(Timestamp, bool)>, + /// The source's cut count at the last frame queued. A cut since clears what a track that + /// has not started holds. + held: Option, /// The PTS of the last keyframe queued, which leading pictures present before. keyframe: Option, /// What the catalog and the SPS declare about the video's reordering. @@ -278,6 +291,8 @@ struct Input<'a> { waiting_for_header: bool, video_start: Option, jitter: &'a mut jitter::Buffer, + /// Counts the frames dropped ahead of the tune-in point or their codec config. + trimmed: &'a mut u64, } impl Track { @@ -288,7 +303,7 @@ impl Track { pending: None, empty: web_async::time::Instant::now(), restart: 0, - base: (0, 0), + base: (0, 0, 0), finished: false, pid: 0, kind, @@ -296,6 +311,7 @@ impl Track { clock: DecodeClock::default(), recovery_gap: None, leading_floor: None, + held: None, keyframe: None, timing: Timing::default(), buffer: None, @@ -327,11 +343,16 @@ impl Track { let gap = pending.skip.saturating_sub(pending.restart); let recover = self.recovery_gap.replace(gap).is_none_or(|previous| previous != gap); if pending.frame.keyframe { - self.leading_floor = recover.then_some(pending.frame.timestamp); + let ahead = self.held == Some(pending.cut); + self.leading_floor = recover.then_some((pending.frame.timestamp, ahead)); } - if self.leading_floor.is_some_and(|floor| pending.frame.timestamp < floor) { + if let Some((floor, ahead)) = self.leading_floor + && pending.frame.timestamp < floor + { + jitter.lose(&self.pid, ahead); return Ok(()); } + self.held = Some(pending.cut); if pending.frame.keyframe { self.timing.describe(stream_type, self.source.description(), name); } @@ -353,12 +374,14 @@ impl Track { match self.source.poll_read(waiter)? { Poll::Ready(Some(frame)) => { if input.waiting_for_header && !self.source.header_ready() { + *input.trimmed += 1; continue; } let pending = Pending { frame, restart: self.base.0 + self.source.markers(), skip: self.base.1 + self.source.skips(), + cut: self.base.2 + self.source.cuts(), arrived: self.empty, read: web_async::time::Instant::now(), }; @@ -367,6 +390,7 @@ impl Track { && !is_video && pending.restart == self.restart && pending.frame.timestamp < start { + *input.trimmed += 1; continue; } if input.waiting_for_header { @@ -406,6 +430,7 @@ impl Track { frame, restart, skip, + cut, arrived, read, } = pending; @@ -414,6 +439,7 @@ impl Track { frame, restart, skip, + cut, arrived, read, }; @@ -431,6 +457,7 @@ impl Track { frame, restart, skip, + cut, arrived, read, }; @@ -608,6 +635,7 @@ impl Track { frame, restart, skip, + cut, arrived, read, } = pending; @@ -634,6 +662,7 @@ impl Track { decode: key, restart, skip, + cut, sync: frame.keyframe || !video, leading, item: Queued { @@ -646,7 +675,12 @@ impl Track { }, }; if jitter.push(self.pid, arrival)? == Push::Late { - tracing::warn!(track = %name, dropped = jitter.dropped(), "frame missed its deadline; dropped"); + tracing::warn!( + track = %name, + dropped = jitter.dropped(), + trimmed = jitter.trimmed(), + "frame missed its deadline; dropped" + ); } Ok(()) } @@ -1070,6 +1104,8 @@ where queue: VecDeque::new(), liveness: Default::default(), video_start: None, + video_out: false, + trimmed: 0, mux_rate: None, mux_rate_override: None, } @@ -1224,6 +1260,7 @@ where && track.pending.as_ref().is_some_and(|p| p.frame.timestamp < start) { track.pending = None; + self.trimmed += 1; } } } @@ -1276,22 +1313,26 @@ where ready.generation >= self.generation, "the jitter buffer released an older generation after a newer one" ); - let name = self + let (name, is_video) = self .tracks .iter() .find(|(_, t)| t.pid == ready.track) - .map(|(name, _)| name.clone()) + .map(|(name, t)| (name.clone(), matches!(t.kind, Kind::Video(_)))) .context("frame for an unknown PID")?; - self.last_timestamp = Some(ready.item.frame.timestamp); - let decode = ready.item.decode.as_nanos(); - self.lay_before(ready.item.key.as_nanos())?; - self.mux(&name, ready.item)?; - if self.delay.is_zero() { - // Without a clock to lay slots on, a frame settles every slot before its own: a - // later frame decodes no earlier, so it can be due no earlier than this, less the - // longest drain of any track. - let drain = self.max_drain(); - self.lay(Some(schedule::slot(decode.saturating_sub(drain.as_nanos()))))?; + if self.leads_video(is_video, ready.item.frame.timestamp) { + self.jitter.discard(&ready); + } else { + self.last_timestamp = Some(ready.item.frame.timestamp); + let decode = ready.item.decode.as_nanos(); + self.lay_before(ready.item.key.as_nanos())?; + self.mux(&name, ready.item)?; + if self.delay.is_zero() { + // Without a clock to lay slots on, a frame settles every slot before its own: a + // later frame decodes no earlier, so it can be due no earlier than this, less the + // longest drain of any track. + let drain = self.max_drain(); + self.lay(Some(schedule::slot(decode.saturating_sub(drain.as_nanos()))))?; + } } if self.queue.is_empty() { // Releasing a due frame opens room in its source. Read it before parking: @@ -1305,6 +1346,7 @@ where waiting_for_header: false, video_start: self.video_start, jitter: &mut self.jitter, + trimmed: &mut self.trimmed, }, )?; if !finished && track.finished { @@ -1419,6 +1461,7 @@ where waiting_for_header: self.psi.is_none(), video_start: self.video_start, jitter: &mut self.jitter, + trimmed: &mut self.trimmed, }; for (name, track) in &mut self.tracks { track.fill(name, waiter, &mut input)?; @@ -1665,8 +1708,10 @@ where continue; }; self.stale.remove(name); + // The new subscription counts as cutting whatever group the old one was reading. track.base.0 += track.source.markers(); track.base.1 += track.source.skips() + 1; + track.base.2 += track.source.cuts() + 1; track.source = source; track.finished = false; track.empty = web_async::time::Instant::now(); @@ -1690,6 +1735,15 @@ where pub fn stats(&self) -> super::stats::Export { let mut stats = super::stats::Export { dropped: self.jitter.dropped(), + trimmed: self.jitter.trimmed() + self.trimmed, + gaps: self + .tracks + .values() + .map(|track| { + let skips = track.base.1 + track.source.skips(); + skips - (track.base.0 + track.source.markers()) + }) + .sum(), drift: self.jitter.drift().map(|drift| drift * 1e6), out_of_tolerance: self.jitter.out_of_tolerance(), ..Default::default() @@ -1825,6 +1879,36 @@ where Ok(()) } + /// 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. + fn leads_video(&mut self, is_video: bool, timestamp: Timestamp) -> bool { + let Some(start) = self.video_start else { + return false; + }; + if is_video { + if !self.video_out { + self.video_out = true; + self.video_start = Some(start.max(timestamp)); + } + return false; + } + if self.video_out { + return timestamp < start; + } + let next = self + .tracks + .values() + .filter(|track| matches!(track.kind, Kind::Video(_))) + .filter_map(|track| self.jitter.peek(&track.pid)) + .map(|queued| queued.frame.timestamp) + .min(); + timestamp < next.map_or(start, |next| next.max(start)) + } + /// Discard what has not gone out and restart the program clock. Every rendition /// joins the new generation: no track is fenced across a declared marker, since a /// latency skip on one track says nothing about another's timeline. The continuity @@ -3415,6 +3499,7 @@ mod tests { frame, restart: 0, skip: 0, + cut: 0, arrived: now, read: now, }; diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 1ac00ebca6b..e9816b2895e 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -1022,8 +1022,17 @@ async fn export_trims_leading_pictures_after_each_delivery_break() { group.finish().unwrap(); } track.finish().unwrap(); - let ts = drain(consumer).await; - let (video, _) = collect_pes_pts(&ts); + let mut export = open_export(crate::source::announced(&consumer).await) + .await + .unwrap() + .with_delay(RECORDING_MAX_AGE) + .with_replay(); + let bytes: Vec = drain_frames(&mut export) + .await + .iter() + .flat_map(|frame| frame.payload.iter().copied()) + .collect(); + let (video, _) = collect_pes_pts(&bytes); let expected: Vec<_> = [100, 110, 200, 180, 190, 210, 400, 410] .into_iter() .map(|ms| ms * 90) @@ -1032,6 +1041,12 @@ async fn export_trims_leading_pictures_after_each_delivery_break() { video, expected, "continuous leading pictures survive; only join and later-hole orphans are trimmed" ); + let stats = export.stats(); + assert_eq!( + (stats.dropped, stats.trimmed), + (2, 2), + "the join's orphans precede the output; the hole's are lost in it" + ); } /// A frame dropped for missing its deadline takes the leading pictures of the keyframe the @@ -1591,7 +1606,9 @@ async fn export_skips_an_unconfirmed_ac3_sync_word_at_a_join() { } /// A declared marker breaks the timeline even where the timestamps run on, so a frame left -/// unfinished before it is dropped rather than completed with the first bytes after it. +/// unfinished before it is dropped rather than completed with the first bytes after it. A +/// data track declares the break by skipping a sequence after its group ended, so nothing it +/// held before the break is trimmed. #[tokio::test(start_paused = true)] async fn export_drops_an_ac3_frame_cut_by_a_marker() { let stream: Vec = (0..4).flat_map(ac3_frame).collect(); @@ -5599,6 +5616,158 @@ 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, as a zero-budget live-edge subscribe +/// is cut when the next group lands before the raised budget reaches it (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. +#[tokio::test(start_paused = true)] +async fn a_start_group_cut_short_starts_at_the_next() { + use crate::container::Container as _; + + let delay = Duration::from_secs(1); + 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 five pictures before the next group lands and the publisher + // resets it. + let mut cut = track.create_group(moq_net::group::Info { sequence: 0 }).unwrap(); + for tick in 0..5 { + 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(); + for tick in 5..GOP { + rig.at(tick * VIDEO_US).await; + out.extend(poll_frames(&mut export)); + rig.audio_until(tick * VIDEO_US, &mut export, &mut out); + } + for tick in GOP + 1..4 * GOP { + rig.at(tick * VIDEO_US).await; + 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(); + out.extend(poll_frames(&mut export)); + rig.audio_until(tick * VIDEO_US, &mut export, &mut out); + } + group.finish().unwrap(); + rig.finish(); + out.extend(drain_frames(&mut export).await); + + let ts: Vec = 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, + (GOP..4 * GOP).map(pts).collect::>(), + "the export starts at the next group, whole" + ); + 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 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); +} + +/// 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 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.at(tick * VIDEO_US).await; + out.extend(poll_frames(&mut export)); + if tick < 5 { + rig.audio_until(tick * VIDEO_US, &mut export, &mut out); + } + } + for tick in 10..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(); + } + out.extend(poll_frames(&mut export)); + rig.audio_until(tick * VIDEO_US, &mut export, &mut out); + } + group.finish().unwrap(); + rig.finish(); + out.extend(drain_frames(&mut export).await); + + let ts: Vec = 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 export starts at the next group"); + assert!( + audio.iter().all(|&p| p >= pts(GOP)), + "no audio leads the keyframe: {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 diff --git a/rs/moq-mux/src/container/ts/release.rs b/rs/moq-mux/src/container/ts/release.rs index b1c2c65295e..0e446003094 100644 --- a/rs/moq-mux/src/container/ts/release.rs +++ b/rs/moq-mux/src/container/ts/release.rs @@ -233,6 +233,7 @@ impl Release { decode: frame.timestamp, 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/container/ts/stats.rs b/rs/moq-mux/src/container/ts/stats.rs index b2fdb5d5d4f..b8b61981b76 100644 --- a/rs/moq-mux/src/container/ts/stats.rs +++ b/rs/moq-mux/src/container/ts/stats.rs @@ -174,9 +174,21 @@ impl Class { pub struct Export { /// One row per elementary stream PID the exporter writes, ordered by PID. pub streams: BTreeMap, - /// Frames dropped for arriving after their deadline, and the frames a video track then - /// dropped waiting for its next keyframe or leading the keyframe it resumed at. + /// Frames dropped after their stream's first went out: for arriving after their deadline, + /// for falling due before a new timeline's clock was acquired, and the frames a video track + /// then dropped waiting for its next keyframe or leading the keyframe it resumed at, late + /// or after a hole upstream once the output held frames ahead of the hole. pub dropped: u64, + /// Frames dropped before their stream's first went out, so the output starts on a whole + /// group led by the video keyframe: what a joiner is handed older than the delay, a start + /// group cut short upstream, the first keyframe's leading pictures, and what precedes the + /// first keyframe or the codec config. + /// Nothing is missing from the output for these. + pub trimmed: u64, + /// Times a source skipped media it will never deliver: a group lost or expired upstream, + /// or a resubscription. A data track's declared break is a skipped sequence, so it counts + /// here too, though nothing of it is lost. Each is logged with the span it skipped. + pub gaps: u64, /// How fast the source's clock runs against ours, in parts per million (positive runs /// fast), as last measured. `None` until there is a measurement. pub drift: Option, diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index 107d5bf5160..79369f3fae3 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -24,6 +24,9 @@ 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 @@ -39,6 +42,8 @@ pub(crate) struct Ready { /// frame of one generation goes out before any frame of the next. pub generation: u64, pub item: T, + /// Whether it is the first frame of its track to go out. + pub first: bool, } /// What [`Buffer::push`] did with a frame. @@ -68,7 +73,15 @@ pub(crate) enum Push { /// ahead and audio just in time, and a clock anchored on the video would make every audio /// frame late. A joiner is handed the group it joined at once, already partly old, and /// 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, without counting as late. +/// 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. Whole groups missing after one +/// that ended take nothing: a joiner whose zero-budget live-edge pick has the +/// publisher cut its group short starts at the next group rather than on the cut-off start. +/// What goes before a track's first frame out is [`Buffer::trimmed`], not lost; what drops +/// after it is [`Buffer::dropped`]. /// /// A skipped group leaves the timeline where it was, so it keeps the clock: frames it made /// late are dropped like any other. A source that restarts its timeline (a declared @@ -146,6 +159,7 @@ pub(crate) struct Buffer { leads: BTreeMap>, timer: Option>>, dropped: u64, + trimmed: u64, /// Steps at which the source's clock ran further off ours than the clock may follow. out_of_tolerance: u64, } @@ -301,24 +315,19 @@ struct Floor { struct Track { /// Frames in arrival order, each with its generation, decode time and deadline. queue: VecDeque<(u64, Timestamp, Instant, T)>, - /// The source's restart and skip counters at the last frame. + /// The source's restart, skip and cut 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, and why. - waiting: Option, - /// Why the track resumed at its last sync frame, whose leading frames are dropped too. - resumed: Option, -} - -/// Why a track waits for a sync frame. -#[derive(Clone, Copy, PartialEq, Eq)] -enum Wait { - /// A frame missed its deadline: what it takes with it counts as dropped. - Late, - /// The acquisition dropped what was due before it ended, which is not a loss. - Acquired, + /// Frames are dropped until the next sync frame. + waiting: bool, + /// Whether the track resumed at its last sync frame after waiting, whose leading frames are + /// dropped too. + resumed: bool, + /// Whether a frame of it has gone out. Until one has, a loss takes everything it holds. + started: bool, } impl Buffer { @@ -337,6 +346,7 @@ impl Buffer { leads: BTreeMap::new(), timer: None, dropped: 0, + trimmed: 0, out_of_tolerance: 0, } } @@ -526,30 +536,42 @@ impl Buffer { } /// 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. fn queue(&mut self, key: K, arrival: Arrival, generation: u64, deadline: Option, stale: bool) -> Push { let track = self.tracks.entry(key).or_insert_with(|| Track { queue: VecDeque::new(), restart: arrival.restart, skip: arrival.skip, + cut: arrival.cut, generation, - waiting: None, - resumed: None, + waiting: false, + resumed: false, + started: false, }); + // The rest of the group the track holds will never arrive. + if arrival.cut != track.cut && !track.started { + self.trimmed += track.queue.len() as u64; + track.queue.clear(); + } track.restart = arrival.restart; track.skip = arrival.skip; + track.cut = arrival.cut; track.generation = generation; let push = match deadline { _ if stale => { - track.waiting = Some(Wait::Acquired); + track.waiting = true; Push::Waiting } - _ if track.waiting.is_some() && !arrival.sync => Push::Waiting, - _ if track.resumed.is_some() && arrival.leading => Push::Waiting, + _ if track.waiting && !arrival.sync => Push::Waiting, + _ if track.resumed && arrival.leading => Push::Waiting, Some(deadline) if arrival.arrived <= deadline || self.delay.is_zero() => { if arrival.sync { track.resumed = track.waiting; } - track.waiting = None; + track.waiting = false; track .queue .push_back((generation, arrival.decode, deadline, arrival.item)); @@ -558,12 +580,19 @@ impl Buffer { } // Past the deadline, or so far off the clock that no instant holds it. _ => { - track.waiting = Some(Wait::Late); + track.waiting = true; Push::Late } }; - if push != Push::Queued && track.waiting.or(track.resumed) == Some(Wait::Late) { - self.dropped += 1; + match push { + Push::Queued => {} + _ if track.started => self.dropped += 1, + // A leading frame of the sync frame the track starts at, which stays queued. + _ if !track.waiting => self.trimmed += 1, + _ => { + self.trimmed += 1 + track.queue.len() as u64; + track.queue.clear(); + } } push } @@ -813,7 +842,9 @@ impl Buffer { if let Some((deadline, key)) = &front && (self.delay.is_zero() || now >= *deadline) { - let (generation, _, _, item) = self.tracks.get_mut(key).and_then(|t| t.queue.pop_front()).unwrap(); + let track = self.tracks.get_mut(key).expect("the front's track"); + let (generation, _, _, item) = track.queue.pop_front().expect("the front frame"); + let first = !std::mem::replace(&mut track.started, true); self.released = true; // A previous-generation frame queued now would go out after this one. if self.clock.is_some_and(|clock| clock.generation == generation) { @@ -823,6 +854,7 @@ impl Buffer { track: key.clone(), generation, item, + first, }); } let Some(wake) = front.map(|(deadline, _)| deadline).into_iter().chain(acquired).min() else { @@ -841,6 +873,34 @@ impl Buffer { } } + /// 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) { + if !ready.first { + self.dropped += 1; + return; + } + self.trimmed += 1; + if let Some(track) = self.tracks.get_mut(&ready.track) { + track.started = false; + } + } + + /// Count a frame of `key` the caller dropped before queuing it: a loss once the track + /// started or when the caller queued frames `ahead` of it, else trimmed. + pub fn lose(&mut self, key: &K, ahead: bool) { + match ahead || self.tracks.get(key).is_some_and(|track| track.started) { + true => self.dropped += 1, + false => self.trimmed += 1, + } + } + + /// The frame `key` releases next, if it has one queued. + pub fn peek(&self, key: &K) -> Option<&T> { + let (_, _, _, item) = self.tracks.get(key)?.queue.front()?; + Some(item) + } + /// When the next queued frame is due, or the acquisition under way ends. #[cfg(test)] pub fn next_deadline(&self) -> Option { @@ -872,12 +932,19 @@ impl Buffer { self.acquire.is_none() && self.tracks.values().all(|track| track.queue.is_empty()) } - /// How many frames were dropped, late or lost with one: waiting for a sync frame after it, - /// or leading the sync frame the track resumed at. + /// How many frames were dropped after their track's first went out: late, due before a new + /// generation's acquisition ended, or lost with one of those, waiting for a sync frame after + /// it or leading the sync frame the track resumed at. pub fn dropped(&self) -> u64 { 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. + pub fn trimmed(&self) -> u64 { + self.trimmed + } + /// The source's clock rate off ours, as last measured, as a fraction: positive runs fast. pub fn drift(&self) -> Option { self.steer.source @@ -953,6 +1020,7 @@ mod tests { decode: ms(decode), restart: 0, skip: 0, + cut: 0, sync: true, leading: false, item, @@ -1053,6 +1121,8 @@ mod tests { let mut buffer = Buffer::new(DELAY).replay(); buffer.push(1, arrival(start, 0, "v0")).unwrap(); buffer.push(2, arrival(start, 0, "a0")).unwrap(); + tokio::time::advance(DELAY).await; + assert_eq!(due(&mut buffer), ["v0", "a0"]); let late = start + Duration::from_millis(200); let video = |decode, sync, item| Arrival { sync, @@ -1062,14 +1132,15 @@ mod tests { assert_eq!(buffer.push(1, video(200, false, "v200")).unwrap(), Push::Waiting); assert_eq!(buffer.push(1, video(240, true, "v240")).unwrap(), Push::Queued); assert_eq!(buffer.push(2, arrival(late, 180, "a180")).unwrap(), Push::Queued); - assert_eq!(buffer.dropped(), 2); + assert_eq!((buffer.dropped(), buffer.trimmed()), (2, 0)); tokio::time::advance(Duration::from_secs(1)).await; - assert_eq!(due(&mut buffer), ["v0", "a0", "a180", "v240"]); + assert_eq!(due(&mut buffer), ["a180", "v240"]); } /// The sync frame a track resumes at after a drop loses its leading frames too, since /// they reference the dropped ones. The next sync frame's leading frames are kept. + /// The track has started, so both are losses ([`Buffer::dropped`]). #[tokio::test(start_paused = true)] async fn a_resumed_sync_frame_drops_its_leading_frames() { let start = Instant::now(); @@ -1080,6 +1151,8 @@ mod tests { ..arrival(arrived, decode, item) }; buffer.push(1, video(start, 0, true, false, "v0")).unwrap(); + tokio::time::advance(DELAY).await; + assert_eq!(due(&mut buffer), ["v0"]); let late = start + Duration::from_millis(200); assert_eq!( buffer.push(1, video(late, 40, false, false, "v40")).unwrap(), @@ -1108,7 +1181,7 @@ mod tests { assert_eq!(buffer.dropped(), 2, "the leading frame was lost with the late one"); tokio::time::advance(Duration::from_secs(1)).await; - assert_eq!(due(&mut buffer), ["v0", "v200", "v280", "v320", "v360"]); + assert_eq!(due(&mut buffer), ["v200", "v280", "v320", "v360"]); } /// A restart before anything went out keeps the clock, so the tracks stay on one. @@ -1431,6 +1504,7 @@ mod tests { decode: Timestamp::from_micros(decode.as_micros() as u64).unwrap(), restart: 0, skip: 0, + cut: 0, sync: true, leading: false, item: "v", @@ -1606,6 +1680,7 @@ mod tests { decode: Timestamp::from_micros(decode.as_micros() as u64).unwrap(), restart: 0, skip: 0, + cut: 0, sync: true, leading: false, item: "v", @@ -1829,6 +1904,61 @@ mod tests { tokio::time::advance(DELAY).await; assert_eq!(due(&mut buffer), ["v240"]); assert_eq!(buffer.dropped(), 0, "trimming the join is not a loss"); + 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. + #[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)); + } + + /// The keyframe a joiner holds while the export starts is read early and passes, while the + /// rest of its group, read later, is older than the delay. The keyframe goes with them, + /// rather than out alone ahead of a hole. + #[tokio::test(start_paused = true)] + async fn a_join_drops_its_group_whole() { + let start = Instant::now(); + let mut buffer = Buffer::new(DELAY); + let video = |arrived, decode, sync, item| Arrival { + sync, + ..arrival(arrived, decode, item) + }; + buffer.push(1, video(start, 0, true, "v0")).unwrap(); + tokio::time::advance(Duration::from_millis(250)).await; + let later = start + Duration::from_millis(250); + for (decode, sync, item) in [(40, false, "v40"), (80, false, "v80"), (400, true, "v400")] { + buffer.push(1, video(later, decode, sync, item)).unwrap(); + } + tokio::time::advance(Duration::from_secs(1)).await; + assert_eq!(due(&mut buffer), ["v400"]); + assert_eq!((buffer.dropped(), buffer.trimmed()), (0, 3)); } /// A source that sends one track well ahead of its decode time and another just in time diff --git a/test/ts/README.md b/test/ts/README.md index 1047fb2644c..ebfa7ec4c2d 100644 --- a/test/ts/README.md +++ b/test/ts/README.md @@ -620,6 +620,13 @@ 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 +exporter reported: the `TS export release clock` counts (`dropped` late, +`trimmed` before the output began, `gaps` skipped upstream) and each loss it +logged. + `just test ts-controls` exercises unchanged input, remuxing with new timestamps and PES sizes, timestamp quantization, and a cold join. Negative controls require the intended failure for missing audio, changed pictures, changed audio, a lost diff --git a/test/ts/run.sh b/test/ts/run.sh index 807fb3c1285..754cd61541a 100755 --- a/test/ts/run.sh +++ b/test/ts/run.sh @@ -631,7 +631,15 @@ echo "### captured $(wc -c <"$SUB_TS" | tr -d ' ') bytes -> analyzing" # pair. Other arms retain their specific structural/model oracles until their # acquisition and decoder-priming contracts are covered by fixtures. if [[ -z "$SOURCE$OPEN_GOP$HRD$HEADROOM$LEAK$FULL" ]]; then - python3 "$DIR/decode.py" "$SRC_TS" "$SUB_TS" --output "$HARNESS_RUN/decode" + if ! python3 "$DIR/decode.py" "$SRC_TS" "$SUB_TS" --output "$HARNESS_RUN/decode"; then + # The exporter counts what it dropped (late), trimmed (before its output began), + # and lost upstream (gaps), and logs each loss, so the failure says which it was. + echo >&2 + echo "error: the decoded capture differs from the source; the exporter reported:" >&2 + grep -E 'TS export release clock|media skipped upstream|missed its deadline|group evicted' \ + "$HARNESS_RUN/sub.log" | sed 's/^/ sub: /' >&2 || true + exit 1 + fi else echo "decoded-content comparison: not covered for this profile" fi