diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 846119999f..b3e0302a7b 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -169,6 +169,10 @@ jobs: if: ${{ !cancelled() }} run: nix develop --command cargo bench --locked -p moq-mux --bench catalog -- --test + - name: TS damage PID benchmark + if: ${{ !cancelled() }} + run: nix develop --command just rs test -p moq-mux benchmark_damage_pid_fanout --run-ignored only --success-output immediate + - name: RTMP chunk-stream benchmark smoke if: ${{ !cancelled() }} run: nix develop --command just rs bench-rtmp --test diff --git a/quest/m1/README.md b/quest/m1/README.md index 7c1a7ac469..e7fd47ec45 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -169,7 +169,6 @@ bench all require its crate. - [T-STD TS export](/quest/m1/tstd/README.md) - `moq export ts` is a proper remux that passes the T-STD buffer model, starting with a fixed `--delay` - [Multi-packet PMT](/quest/m1/ts-pmt-split.md) - `export ts` splits a PMT longer than one packet instead of exiting before its first packet - [ATSC AC-3](/quest/m1/ts-atsc-ac3.md) - `export ts` carries 48 kHz ATSC AC-3 at every A/52 rate up to 640 kb/s instead of aborting from 384 kb/s up -- [TS damage log](/quest/m1/ts-damage-log.md) - a burst of damaged TS units logs a first warning and a periodic summary per PID, not one warning per unit - [Rust non-continuous signal](/quest/m1/rust-continuous.md) - the Rust container consumer reports a frame after a subscribe or discontinuity as non-continuous, like JS, for the tune-in and warmup trims - [Open-GOP leading pictures](/quest/m1/open-gop-leading-pictures.md) - a viewer joining at a recovery point drops the leading pictures it cannot decode; continuous viewers keep them - [Watch decode errors](/quest/m1/watch-decode-error.md) - a WebCodecs error ends the subscription and the element reports it diff --git a/quest/m1/ts-damage-log.md b/quest/m1/ts-damage-log.md deleted file mode 100644 index 633b99630f..0000000000 --- a/quest/m1/ts-damage-log.md +++ /dev/null @@ -1,18 +0,0 @@ -# [XS] A burst of damaged TS units logs a summary, not a flood - -## Goal - -When MPEG-TS ingest refuses damaged units, it warns on the first damaged unit -of a burst per PID, then logs one summary with the count per interval, instead -of one warning per unit for the whole session. Lands on `main` first, then is -cherry-picked to `release`. - -## Plan - -`Import::damage` (`rs/moq-mux/src/container/ts/import.rs`) already counts per -PID in `damaged`, but warns on every unit. Since #4733 (and #5124 on -`release`) a sender streaming garbage no longer ends the session, so the -warning repeats for its lifetime. Decided 2026-10-10: first warning plus a -periodic summary per PID, no new dependency; test it with mocked time. - -Public API: none. Wire: none. diff --git a/rs/moq-mux/src/container/ts/damage.rs b/rs/moq-mux/src/container/ts/damage.rs new file mode 100644 index 0000000000..30c92db4f8 --- /dev/null +++ b/rs/moq-mux/src/container/ts/damage.rs @@ -0,0 +1,166 @@ +//! Bounded warning bursts for damaged transport units. + +use std::collections::BTreeMap; + +use web_async::time::Instant; + +use super::stats; + +#[derive(Default)] +pub(super) struct Log { + bursts: BTreeMap, + next: Option, +} + +struct Burst { + last: Instant, + pending: u64, +} + +impl Burst { + fn flush(&mut self, pid: u16) { + let count = std::mem::take(&mut self.pending); + if count != 0 { + tracing::warn!(pid, count, "TS damaged-unit summary"); + } + } +} + +impl Log { + pub fn record(&mut self, pid: u16, error: &anyhow::Error) { + let now = Instant::now(); + self.flush(now); + let first = match self.bursts.entry(pid) { + std::collections::btree_map::Entry::Occupied(mut entry) => { + let burst = entry.get_mut(); + // A PID's quiet interval can end before the global summary cadence. + let first = now.duration_since(burst.last) >= stats::Log::INTERVAL; + if first { + burst.flush(pid); + } else { + burst.pending += 1; + } + burst.last = now; + first + } + std::collections::btree_map::Entry::Vacant(entry) => { + entry.insert(Burst { last: now, pending: 0 }); + true + } + }; + if first { + tracing::warn!(pid, count = 1, %error, "dropped a damaged TS unit"); + } + self.next.get_or_insert(now + stats::Log::INTERVAL); + } + + pub fn flush(&mut self, now: Instant) { + if self.next.is_none_or(|next| now < next) { + return; + } + self.bursts.retain(|&pid, burst| { + burst.flush(pid); + // A quiet interval ends the burst, so its next damage warns immediately. + now.duration_since(burst.last) < stats::Log::INTERVAL + }); + self.next = (!self.bursts.is_empty()).then_some(now + stats::Log::INTERVAL); + } +} + +impl Drop for Log { + // Input can end through finish, abort, or an error, so the final burst summarizes on drop. + fn drop(&mut self) { + for (&pid, burst) in &mut self.bursts { + burst.flush(pid); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test(start_paused = true)] + #[ignore = "benchmark; runs in nightly"] + async fn benchmark_damage_pid_fanout() { + let error = anyhow::anyhow!("broken PES"); + eprintln!("pids,damaged_units,ns_per_unit"); + for pids in [1_u16, 64, 8192] { + for units in [1, 16, 128] { + let mut log = Log::default(); + for pid in 0..pids { + log.record(pid, &error); + } + for _ in 0..100 { + log.record(pids - 1, &error); + } + // Media time stays paused, so this measures one touched PID's hot path, + // independently of periodic summaries that visit every active burst. + let start = std::time::Instant::now(); + for _ in 0..10_000 { + for _ in 0..units { + log.record(std::hint::black_box(pids - 1), &error); + } + } + let nanos = start.elapsed().as_nanos() as f64 / (10_000 * units) as f64; + eprintln!("{pids},{units},{nanos:.1}"); + } + } + } + + #[tokio::test(start_paused = true)] + #[tracing_test::traced_test] + async fn a_quiet_pid_starts_a_new_burst_between_summary_ticks() { + let mut log = Log::default(); + let error = anyhow::anyhow!("broken PES"); + log.record(97, &error); + tokio::time::advance(stats::Log::INTERVAL / 2).await; + log.record(97, &error); + tokio::time::advance(stats::Log::INTERVAL / 2).await; + log.flush(Instant::now()); + // This PID is quiet for a full interval before the next global summary tick. + tokio::time::advance(stats::Log::INTERVAL / 2).await; + log.record(97, &error); + logs_assert(|lines| { + let first = lines + .iter() + .filter(|line| line.contains("dropped a damaged TS unit")) + .count(); + if first == 2 { + Ok(()) + } else { + Err(format!("expected an immediate new-burst warning, got {first}")) + } + }); + } + + #[tokio::test(start_paused = true)] + #[tracing_test::traced_test] + async fn a_continuous_burst_logs_one_summary_per_interval() { + let mut log = Log::default(); + let error = anyhow::anyhow!("broken PES"); + log.record(97, &error); + for _ in 0..4 { + tokio::time::advance(stats::Log::INTERVAL / 2).await; + log.record(97, &error); + } + drop(log); + logs_assert(|lines| { + let first = lines + .iter() + .filter(|line| line.contains("dropped a damaged TS unit")) + .count(); + let summaries: Vec<_> = lines + .iter() + .filter(|line| line.contains("TS damaged-unit summary")) + .collect(); + if first == 1 && summaries.len() == 3 && summaries[1].contains("count=2") { + Ok(()) + } else { + Err(format!( + "expected one burst and three summaries, got {first}/{summaries:?}" + )) + } + }); + } +} diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 3dd62ef1e8..98c330d3ba 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -79,6 +79,7 @@ pub struct Import { retired_stats: BTreeMap, /// Damaged packets, PES, or access units refused, cumulative across PID remaps. damaged: BTreeMap, + damage_log: super::damage::Log, /// Access units per elementary stream, timed on the program clock. liveness: Liveness, /// In-progress PES reassembly, keyed by elementary PID. @@ -168,6 +169,7 @@ impl Import { streams: HashMap::new(), retired_stats: BTreeMap::new(), damaged: BTreeMap::new(), + damage_log: Default::default(), liveness: Liveness::default(), pending: HashMap::new(), health: Health::default(), @@ -230,6 +232,7 @@ impl Import { /// now completes. The buffer is fully consumed; a trailing partial packet /// (< 188 bytes) is retained for the next call. pub fn decode(&mut self, data: &[u8]) -> anyhow::Result<()> { + self.damage_log.flush(web_async::time::Instant::now()); self.scratch.extend_from_slice(data); // Route one whole packet at a time, so a PMT is parsed (and any PID it declares @@ -774,7 +777,7 @@ impl Import { None => {} } *self.damaged.entry(pid.as_u16()).or_default() += 1; - tracing::warn!(pid = pid.as_u16(), error = %err, "dropped a damaged TS unit"); + self.damage_log.record(pid.as_u16(), err); Ok(()) } @@ -3334,6 +3337,91 @@ pub(super) mod test { assert_eq!(stats.unconfirmed, 1, "the unvouched publication is counted separately"); } + #[tokio::test(start_paused = true)] + #[tracing_test::traced_test] + async fn damaged_unit_warnings_are_aggregated_per_pid() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + let pid = Pid::new(0x61).unwrap(); + let peer = Pid::new(0x62).unwrap(); + let error = anyhow::anyhow!("broken PES"); + for _ in 0..3 { + import.damage(pid, &error).unwrap(); + } + import.damage(peer, &error).unwrap(); + logs_assert(|lines| { + let first = lines + .iter() + .filter(|line| line.contains("dropped a damaged TS unit")) + .count(); + if first == 2 { + Ok(()) + } else { + Err(format!("expected one warning per PID, got {first}")) + } + }); + + tokio::time::advance(super::stats::Log::INTERVAL).await; + import.decode(&[]).unwrap(); + logs_assert(|lines| { + let summary: Vec<_> = lines + .iter() + .filter(|line| line.contains("TS damaged-unit summary")) + .collect(); + if summary.len() == 1 && summary[0].contains("pid=97 ") && summary[0].contains("count=2") { + Ok(()) + } else { + Err(format!("expected two additional drops on PID 97, got {summary:?}")) + } + }); + import.damage(pid, &error).unwrap(); + import.damage(pid, &error).unwrap(); + import.finish().unwrap(); + assert_eq!(import.stats().streams[&0x61].damaged, 5); + drop(import); + logs_assert(|lines| { + let first = lines + .iter() + .filter(|line| line.contains("dropped a damaged TS unit")) + .count(); + let summary = lines + .iter() + .filter(|line| line.contains("TS damaged-unit summary")) + .count(); + if (first, summary) == (3, 2) { + Ok(()) + } else { + Err(format!("expected new burst and final summary, got {first}/{summary}")) + } + }); + } + + #[tokio::test(start_paused = true)] + #[tracing_test::traced_test] + async fn abort_summarizes_the_final_damage_burst() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + let pid = Pid::new(0x61).unwrap(); + let error = anyhow::anyhow!("broken PES"); + for _ in 0..3 { + import.damage(pid, &error).unwrap(); + } + import.abort(moq_net::Error::Cancel); + logs_assert(|lines| { + let summary: Vec<_> = lines + .iter() + .filter(|line| line.contains("TS damaged-unit summary")) + .collect(); + if summary.len() == 1 && summary[0].contains("count=2") { + Ok(()) + } else { + Err(format!("expected the aborted burst's two drops, got {summary:?}")) + } + }); + } + #[test] fn remapping_a_pid_keeps_its_retired_resync_stats() { let mut broadcast = moq_net::broadcast::Info::new().produce(); diff --git a/rs/moq-mux/src/container/ts/mod.rs b/rs/moq-mux/src/container/ts/mod.rs index 6def6e0bb8..d660ba2766 100644 --- a/rs/moq-mux/src/container/ts/mod.rs +++ b/rs/moq-mux/src/container/ts/mod.rs @@ -23,6 +23,7 @@ //! demultiplexing cannot carry (TS-level scrambling, PSI and SI as authored). mod adts; +mod damage; mod export; mod follower; mod health;