From 8f906391f6fa29bc7a75902522c235f1492d09a4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 13:55:47 -0700 Subject: [PATCH 1/4] chore(quest): claim TS damage log aggregation Co-Authored-By: GPT-6 From 776de77afc8e138a31e009188478c0ae06f426f6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 14:22:55 -0700 Subject: [PATCH 2/4] fix(mux): aggregate TS damage warnings by PID burst Co-Authored-By: GPT-6 --- quest/m1/README.md | 1 - quest/m1/ts-damage-log.md | 18 ------ rs/moq-mux/src/container/ts/damage.rs | 87 +++++++++++++++++++++++++++ rs/moq-mux/src/container/ts/import.rs | 65 +++++++++++++++++++- rs/moq-mux/src/container/ts/mod.rs | 1 + 5 files changed, 152 insertions(+), 20 deletions(-) delete mode 100644 quest/m1/ts-damage-log.md create mode 100644 rs/moq-mux/src/container/ts/damage.rs diff --git a/quest/m1/README.md b/quest/m1/README.md index 8498025f09..ceee76456b 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -171,7 +171,6 @@ bench all require its crate. - [Encoder colour](/quest/m1/color-model.md) - every moq-video encode path signals the colour its output actually has, or refuses instead of mislabelling - [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 -- [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 - [TS adaptation field length](/quest/m1/ts-adaptation-only.md) - a TS adaptation field with the wrong length for its packet (183 adaptation-only, at most 182 with a payload) is refused as damage - [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 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..583c62c569 --- /dev/null +++ b/rs/moq-mux/src/container/ts/damage.rs @@ -0,0 +1,87 @@ +//! 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 Log { + pub fn record(&mut self, pid: u16, error: &anyhow::Error) { + let now = Instant::now(); + self.flush(now, false); + match self.bursts.entry(pid) { + std::collections::btree_map::Entry::Occupied(mut entry) => { + let burst = entry.get_mut(); + burst.last = now; + burst.pending += 1; + } + std::collections::btree_map::Entry::Vacant(entry) => { + tracing::warn!(pid, count = 1, %error, "dropped a damaged TS unit"); + entry.insert(Burst { last: now, pending: 0 }); + } + } + self.next.get_or_insert(now + stats::Log::INTERVAL); + } + + pub fn flush(&mut self, now: Instant, finish: bool) { + if !finish && self.next.is_none_or(|next| now < next) { + return; + } + self.bursts.retain(|&pid, burst| { + let count = std::mem::take(&mut burst.pending); + if count != 0 { + tracing::warn!(pid, count, "TS damaged-unit summary"); + } + // A quiet interval ends the burst, so its next damage warns immediately. + !finish && now.duration_since(burst.last) < stats::Log::INTERVAL + }); + self.next = (!self.bursts.is_empty()).then_some(now + stats::Log::INTERVAL); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[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); + } + log.flush(Instant::now(), true); + 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 56b0e02534..511f98f396 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(), false); 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(()) } @@ -1001,6 +1004,7 @@ impl Import { section.finish()?; } self.si.finish(self.last_pts.unwrap_or(Timestamp::ZERO))?; + self.damage_log.flush(web_async::time::Instant::now(), true); Ok(()) } @@ -3329,6 +3333,65 @@ 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); + 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}")) + } + }); + } + #[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; From 80848433ea57959fe03875ac09d82e8e84989e72 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 20:22:07 -0700 Subject: [PATCH 3/4] fix(mux): recognize quiet PID bursts between summary ticks Co-Authored-By: GPT-6 Astra --- .github/workflows/nightly.yml | 4 ++ rs/moq-mux/src/container/ts/damage.rs | 84 ++++++++++++++++++++++++--- 2 files changed, 81 insertions(+), 7 deletions(-) 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/rs/moq-mux/src/container/ts/damage.rs b/rs/moq-mux/src/container/ts/damage.rs index 583c62c569..94225db411 100644 --- a/rs/moq-mux/src/container/ts/damage.rs +++ b/rs/moq-mux/src/container/ts/damage.rs @@ -17,20 +17,39 @@ struct Burst { 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, false); - match self.bursts.entry(pid) { + 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; - burst.pending += 1; + first } std::collections::btree_map::Entry::Vacant(entry) => { - tracing::warn!(pid, count = 1, %error, "dropped a damaged TS unit"); 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); } @@ -40,10 +59,7 @@ impl Log { return; } self.bursts.retain(|&pid, burst| { - let count = std::mem::take(&mut burst.pending); - if count != 0 { - tracing::warn!(pid, count, "TS damaged-unit summary"); - } + burst.flush(pid); // A quiet interval ends the burst, so its next damage warns immediately. !finish && now.duration_since(burst.last) < stats::Log::INTERVAL }); @@ -55,6 +71,60 @@ impl Log { 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(), false); + // 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() { From bde36c79b18d010e2cab7d8c8d3825112db58a1e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 09:36:44 -0700 Subject: [PATCH 4/4] fix(mux): summarize the final TS damage burst on every terminal path abort and an early error in finish skipped the final flush, losing the last burst's pending count. Flush on drop instead of in finish. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/damage.rs | 21 +++++++++++++------ rs/moq-mux/src/container/ts/import.rs | 29 +++++++++++++++++++++++++-- 2 files changed, 42 insertions(+), 8 deletions(-) diff --git a/rs/moq-mux/src/container/ts/damage.rs b/rs/moq-mux/src/container/ts/damage.rs index 94225db411..30c92db4f8 100644 --- a/rs/moq-mux/src/container/ts/damage.rs +++ b/rs/moq-mux/src/container/ts/damage.rs @@ -29,7 +29,7 @@ impl Burst { impl Log { pub fn record(&mut self, pid: u16, error: &anyhow::Error) { let now = Instant::now(); - self.flush(now, false); + self.flush(now); let first = match self.bursts.entry(pid) { std::collections::btree_map::Entry::Occupied(mut entry) => { let burst = entry.get_mut(); @@ -54,19 +54,28 @@ impl Log { self.next.get_or_insert(now + stats::Log::INTERVAL); } - pub fn flush(&mut self, now: Instant, finish: bool) { - if !finish && self.next.is_none_or(|next| now < next) { + 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. - !finish && now.duration_since(burst.last) < stats::Log::INTERVAL + 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::*; @@ -108,7 +117,7 @@ mod tests { tokio::time::advance(stats::Log::INTERVAL / 2).await; log.record(97, &error); tokio::time::advance(stats::Log::INTERVAL / 2).await; - log.flush(Instant::now(), false); + 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); @@ -135,7 +144,7 @@ mod tests { tokio::time::advance(stats::Log::INTERVAL / 2).await; log.record(97, &error); } - log.flush(Instant::now(), true); + drop(log); logs_assert(|lines| { let first = lines .iter() diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 9649843d4d..98c330d3ba 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -232,7 +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(), false); + 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 @@ -1004,7 +1004,6 @@ impl Import { section.finish()?; } self.si.finish(self.last_pts.unwrap_or(Timestamp::ZERO))?; - self.damage_log.flush(web_async::time::Instant::now(), true); Ok(()) } @@ -3380,6 +3379,7 @@ pub(super) mod test { 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() @@ -3397,6 +3397,31 @@ pub(super) mod test { }); } + #[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();