fix(mux): aggregate TS damage warnings per PID burst

.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

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

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<u16, Burst>,
+	next: Option<Instant>,
+}
+
+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:?}"
+				))
+			}
+		});
+	}
+}

rs/moq-mux/src/container/ts/import.rs

@@ -79,6 +79,7 @@ pub struct Import<E: catalog::Catalog = ()> {
 	retired_stats: BTreeMap<u16, stats::Stream>,
 	/// Damaged packets, PES, or access units refused, cumulative across PID remaps.
 	damaged: BTreeMap<u16, u64>,
+	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<E: catalog::Catalog> Import<E> {
 			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<E: catalog::Catalog> Import<E> {
 	/// 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<E: catalog::Catalog> Import<E> {
 			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();

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;