feat(mux): report each TS elementary stream's access units at export

quest/m1/ts-stats-module.md

@@ -31,10 +31,9 @@ Decided while planning the follow-ups of
 - Update moq-cli's `publish.rs`, the only consumer that names these types.
   moq-srt only uses `stats::Log`.
 - Main quests still add fields under the old names:
-  [TS PSI reassembly](/quest/m1/ts-psi-reassembly.md),
-  [TS import health](/quest/m2/ts-import-health.md) and
-  [TS export liveness](/quest/m2/ts-export-liveness.md). The rename carries
-  those fields at merge time. Update the type names in the quests still open
+  [TS PSI reassembly](/quest/m1/ts-psi-reassembly.md) and
+  [TS import health](/quest/m2/ts-import-health.md). `Export::stats` (#4577)
+  also returns the old `ts::Stats`. The rename carries those at merge time. Update the type names in the quests still open
   when this lands, including [media stats schema](/quest/m1/stats/schema.md)
   and [Rust reporters](/quest/m1/stats/rust.md).
 

quest/m2/README.md

@@ -98,7 +98,6 @@ waiting on an upstream release waits in [m4](/quest/m4/README.md).
 - [GOP overhead](/quest/m2/gop-overhead.md) - price the I-frames a short GOP pays for, deciding whether a long GOP plus a keyframe request is worth designing
 - [Per-program SI](/quest/m2/ts-program-si.md) - a selected TS program's broadcast carries only its own service's SDT and EIT
 - [TS import health](/quest/m2/ts-import-health.md) - `moq import ts` counts the TR 101 290 errors of the feed it receives, PCR and PTS graded on its own values
-- [TS export liveness](/quest/m2/ts-export-liveness.md) - `moq export ts` reports each elementary stream's access units and quiet time, catching a per-track stall
 - [TS health stats](/quest/m2/ts-health-stats.md) - the TS counters ride the stats plumbing beside the media counters
 - [Teleoperation](/quest/m2/teleop/README.md) - MoQ carries robot video down and control up on one session as a library capability
 - [Media QA on other engines](/quest/m2/browser-media-qa-engines.md) - the media harness measures a Firefox or WebKit player over the fallback and names what each engine lacks

quest/m2/ts-health-stats.md

@@ -43,4 +43,3 @@ Decided while planning [#1838](https://github.com/moq-dev/moq/issues/1838):
 - [Media stats schema](/quest/m1/stats/schema.md) - the snapshot the counters flatten into
 - [Rust reporters](/quest/m1/stats/rust.md) - `moq import --stats` and the stats interval
 - [TS import health](/quest/m2/ts-import-health.md) - the ingest counters
-- [TS export liveness](/quest/m2/ts-export-liveness.md) - the egress rows

rs/moq-cli/src/subscribe.rs

@@ -322,6 +322,10 @@ impl Subscribe {
 		// hand-rolled poll instead of `ts.next()`.
 		let mut delivery = Delivery::new(self.args.max_age);
 		let linger = self.args.linger;
+		// Reports a track that stops reaching the output while the rest keeps flowing,
+		// 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();
 		loop {
 			let end = loop {
 				let mut waited = false;
@@ -341,6 +345,11 @@ impl Subscribe {
 				};
 				delivery.update(&frame, ts.discontinuity());
 				delivery.deliver(&frame, waited, &mut stdout).await?;
+
+				if sampled.elapsed() >= moq_mux::container::ts::stats::Log::INTERVAL {
+					sampled = tokio::time::Instant::now();
+					log.sample(ts.stats());
+				}
 			};
 
 			// Any end waits out the linger, and on expiry the last one is the result: a

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

@@ -58,7 +58,7 @@ const PSI_INTERVAL: Duration = Duration::from_millis(500);
 /// table asking for faster repetition than this still gets it.
 const SI_REVISION_INTERVAL: Duration = Duration::from_secs(1);
 /// Emit a PCR on every crossing of this media-time grid ([`Export::emit`]).
-/// TR 101 290 flags a gap over 40 ms; broadcast muxes emit every 25-40 ms.
+/// TR 101 290 V1.4.1 flags a gap over 100 ms; broadcast muxes emit every 25-40 ms.
 pub(super) const PCR_INTERVAL: Duration = Duration::from_millis(25);
 /// How many missed PCR slots to backfill at most: one second's worth. Frames
 /// coarser than the grid cross several slots at a time and every one is filled so
@@ -158,8 +158,9 @@ pub struct Export<E: catalog::Catalog = ()> {
 	/// Offsets into [`pending`](Self::pending) where a keyframe's packets begin, so
 	/// the output frame carrying one keeps the flag.
 	keyframes: Vec<usize>,
-	/// Output frames ready to hand out, one per grid slot the last span covered.
-	queue: VecDeque<Frame>,
+	/// Output frames ready to hand out, one per grid slot the last span covered, each with
+	/// what it tells [`Self::stats`] once returned.
+	queue: VecDeque<(Frame, Tally)>,
 	/// Continuity counter of the last packet emitted on the PCR PID. A clock packet
 	/// carries no payload, so it repeats whatever preceded it on the wire rather
 	/// than advancing the counter ([`Export::pcr_at`]).
@@ -197,6 +198,24 @@ pub struct Export<E: catalog::Catalog = ()> {
 	/// 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<Timestamp>,
+	/// Offsets into [`pending`](Self::pending) where each access unit's packets begin, with
+	/// its PID, so the output frame carrying it counts it.
+	units: Vec<(usize, u16)>,
+	/// Clock packets written into the output frame being laid out ([`Self::push`]).
+	tally: Tally,
+	/// Each elementary stream's access units and silence on the PCR returned ([`Self::stats`]).
+	liveness: super::import::Liveness,
+}
+
+/// What a queued output frame tells [`Export::stats`], applied only once the frame is
+/// returned: a rewind drops the queue unwritten.
+#[derive(Default)]
+struct Tally {
+	/// The frame's clock packets in order: the PCR in 27 MHz ticks, and whether it flags a
+	/// new time base.
+	pcrs: Vec<(u64, bool)>,
+	/// The PID of each access unit whose packets begin in the frame, after its clock packets.
+	units: Vec<u16>,
 }
 
 struct Pending {
@@ -408,6 +427,23 @@ enum Kind {
 	},
 }
 
+impl Kind {
+	/// The track suffix [`Import`](super::Import) gives a PID carrying this kind, so a stream's
+	/// row is named alike at both edges.
+	fn suffix(&self) -> &'static str {
+		match self {
+			Kind::Video(StreamType::H265) => ".hev1",
+			Kind::Video(_) => ".avc3",
+			Kind::Aac { .. } => ".aac",
+			Kind::Opus { .. } => ".opus",
+			Kind::Mp2 { .. } => ".mp2",
+			Kind::Ac3 => ".ac3",
+			Kind::Eac3 => ".eac3",
+			Kind::Verbatim { .. } => ".ts",
+		}
+	}
+}
+
 /// The null stuffing owed to the multiplex rate ([`Export::stuff`]).
 #[derive(Default)]
 struct Stuffing {
@@ -698,6 +734,9 @@ impl<E: catalog::Catalog> Export<E> {
 			mux_rate: None,
 			mux_rate_override: None,
 			stuffing: Stuffing::default(),
+			units: Vec::new(),
+			tally: Tally::default(),
+			liveness: Default::default(),
 		})
 	}
 
@@ -819,8 +858,7 @@ impl<E: catalog::Catalog> Export<E> {
 		// own values imply and lets the caller's pacer release each at the instant
 		// it asserts. See [`Self::advance`].
 		loop {
-			if let Some(out) = self.queue.pop_front() {
-				self.emitted_epoch = self.epoch;
+			if let Some(out) = self.pop() {
 				return Poll::Ready(Ok(Some(out)));
 			}
 			let Some(name) = self.pick_next_track(waiter) else {
@@ -877,8 +915,7 @@ impl<E: catalog::Catalog> Export<E> {
 		let drained = !self.tracks.is_empty() && self.tracks.values().all(|t| t.finished);
 		if drained {
 			self.emit(None)?;
-			if let Some(out) = self.queue.pop_front() {
-				self.emitted_epoch = self.epoch;
+			if let Some(out) = self.pop() {
 				return Poll::Ready(Ok(Some(out)));
 			}
 			// SI emission rides media frames, so a snapshot that arrived behind the
@@ -899,6 +936,22 @@ impl<E: catalog::Catalog> Export<E> {
 		Poll::Pending
 	}
 
+	/// Return the next queued frame, committing what it carries to the output boundary.
+	fn pop(&mut self) -> Option<Frame> {
+		let (out, tally) = self.queue.pop_front()?;
+		self.emitted_epoch = self.epoch;
+		for (pcr, discontinuity) in tally.pcrs {
+			if discontinuity {
+				self.liveness.discontinuity();
+			}
+			self.liveness.written_pcr(pcr);
+		}
+		for pid in tally.units {
+			self.liveness.delivered(pid, 1);
+		}
+		Some(out)
+	}
+
 	/// The trailing SI frame for end of stream: every entry's current sections,
 	/// re-emitted once. Duplicates of already-emitted sections are harmless
 	/// repetitions; `None` when there are no tables or no program was ever built.
@@ -1257,6 +1310,30 @@ impl<E: catalog::Catalog> Export<E> {
 		self.emitted_epoch
 	}
 
+	/// Snapshot the access units each elementary stream has written, and how long each has
+	/// been quiet on the PCR the output carries.
+	///
+	/// A track stalled upstream stops advancing its row while the PSI and the other PIDs
+	/// keep flowing, which nothing graded on the output bytes alone can see. Empty until the
+	/// program tables are built. Cheap enough to poll per frame.
+	pub fn stats(&self) -> super::Stats {
+		let mut stats = super::Stats::default();
+		if self.psi.is_none() {
+			return stats;
+		}
+		for track in self.tracks.values() {
+			let (units, quiet) = self.liveness.stream(track.pid);
+			let row = super::StreamStats {
+				track: track.kind.suffix(),
+				units,
+				quiet,
+				..Default::default()
+			};
+			stats.streams.insert(track.pid, row);
+		}
+		stats
+	}
+
 	/// Carry on with the broadcast that replaced the one this export was reading.
 	///
 	/// Call it once [`Source::returned`](crate::Source::returned) has resolved, after
@@ -1301,6 +1378,8 @@ impl<E: catalog::Catalog> Export<E> {
 		}
 		self.pending.clear();
 		self.keyframes.clear();
+		self.units.clear();
+		self.tally = Tally::default();
 		self.queue.clear();
 		self.watermark = None;
 		// The stall carries across with its budget, and so does a held frame's
@@ -1499,6 +1578,9 @@ impl<E: catalog::Catalog> Export<E> {
 			es_info,
 		};
 
+		for track in &tracks {
+			self.liveness.register(track.pid);
+		}
 		self.psi = Some(Psi {
 			pat,
 			pmt,
@@ -1664,6 +1746,8 @@ impl<E: catalog::Catalog> Export<E> {
 			}
 		}
 
+		// A malformed section writes nothing, so the unit counts only if bytes went in.
+		let written = out.len();
 		match es_payload {
 			// Section-framed verbatim (SCTE-35, ...) rides in private sections, not PES;
 			// carry the bytes verbatim.
@@ -1689,6 +1773,9 @@ impl<E: catalog::Catalog> Export<E> {
 				self.write_pes(&mut out, &unit, &es_payload)?;
 			}
 		}
+		if out.len() > written {
+			self.units.push((self.pending.len() + written, pid));
+		}
 		if keyframe {
 			self.keyframes.push(self.pending.len());
 		}
@@ -1752,6 +1839,7 @@ impl<E: catalog::Catalog> Export<E> {
 		self.span_counters = None;
 		let bytes = std::mem::take(&mut self.pending);
 		let keyframes = std::mem::take(&mut self.keyframes);
+		let units = std::mem::take(&mut self.units);
 		let Some(to) = self.low.take() else { return Ok(()) };
 		let packets = bytes.len() / TsPacket::SIZE;
 
@@ -1814,7 +1902,7 @@ impl<E: catalog::Catalog> Export<E> {
 			};
 			self.send(&mut payload, &bytes[cut * TsPacket::SIZE..next * TsPacket::SIZE]);
 			self.stuff(index, &mut payload);
-			self.push(at, payload, &keyframes, cut, next);
+			self.push(at, payload, &keyframes, &units, cut, next);
 			cut = next;
 			at = boundary;
 			let before = counter_before(&bytes, cut * TsPacket::SIZE, pcr_pid, self.pcr_cc);
@@ -1824,7 +1912,7 @@ impl<E: catalog::Catalog> Export<E> {
 		}
 
 		self.send(&mut payload, &bytes[cut * TsPacket::SIZE..]);
-		self.push(at, payload, &keyframes, cut, packets);
+		self.push(at, payload, &keyframes, &units, cut, packets);
 		if let Some(cc) = counter_before(&bytes, bytes.len(), pcr_pid, None) {
 			self.pcr_cc = Some(cc);
 		}
@@ -1891,19 +1979,34 @@ impl<E: catalog::Catalog> Export<E> {
 	}
 
 	/// Queue one output frame, unless it would be empty. `from`..`to` are the packet
-	/// indices it carries, which decide whether a keyframe begins in it.
-	fn push(&mut self, timestamp: Timestamp, payload: Vec<u8>, keyframes: &[usize], from: usize, to: usize) {
+	/// indices it carries, which decide whether a keyframe, or which access units, begin in it.
+	fn push(
+		&mut self,
+		timestamp: Timestamp,
+		payload: Vec<u8>,
+		keyframes: &[usize],
+		units: &[(usize, u16)],
+		from: usize,
+		to: usize,
+	) {
 		if payload.is_empty() {
 			return;
 		}
 		let (from, to) = (from * TsPacket::SIZE, to * TsPacket::SIZE);
 		let keyframe = keyframes.iter().any(|&at| at >= from && at < to);
-		self.queue.push_back(Frame {
+		let mut tally = std::mem::take(&mut self.tally);
+		tally.units = units
+			.iter()
+			.filter(|&&(at, _)| at >= from && at < to)
+			.map(|&(_, pid)| pid)
+			.collect();
+		let frame = Frame {
 			timestamp,
 			duration: None,
 			payload: Bytes::from(payload),
 			keyframe,
-		});
+		};
+		self.queue.push_back((frame, tally));
 	}
 
 	/// The clock packet for grid slot `index`, and record that the slot is served.
@@ -1938,9 +2041,11 @@ impl<E: catalog::Catalog> Export<E> {
 		};
 		self.last_pcr = Some(index);
 		let mut packet = pcr_packet(pcr_pid, ticks, cc)?;
-		if std::mem::take(&mut self.pcr_discontinuity) {
+		let discontinuity = std::mem::take(&mut self.pcr_discontinuity);
+		if discontinuity {
 			packet[5] |= 0x80;
 		}
+		self.tally.pcrs.push(((ticks & TS_TIMESTAMP_MASK) * 300, discontinuity));
 		Ok(packet)
 	}