feat(moq-mux,moq-srt): PTS-exposing TS Export API + PCR-paced SRT egress (#1845)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: kixelated

rs/moq-cli/src/subscribe.rs

@@ -326,8 +326,8 @@ impl Subscribe {
 			.await?
 			.with_latency(self.args.max_latency);
 
-		while let Some(chunk) = ts.next().await? {
-			stdout.write_all(&chunk).await?;
+		while let Some(frame) = ts.next().await? {
+			stdout.write_all(&frame.payload).await?;
 			stdout.flush().await?;
 		}
 

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

@@ -1,9 +1,10 @@
 //! MPEG-TS muxer.
 //!
-//! [`Export`] subscribes to a MoQ broadcast and produces a single MPEG-TS byte
-//! stream: PAT/PMT program tables followed by one PES packet per media frame,
-//! packetized into 188-byte TS packets. Video is carried as Annex-B, audio as
-//! ADTS AAC.
+//! [`Export`] subscribes to a MoQ broadcast and produces MPEG-TS, yielding one
+//! [`Frame`] per media frame: PAT/PMT program tables followed by one PES packet,
+//! packetized into 188-byte TS packets. Each frame keeps its media timestamp so
+//! the caller can pace delivery on the media clock. Video is carried as Annex-B,
+//! audio as ADTS AAC.
 //!
 //! Video flows through [`ExportSource`], which normalizes every H.264/H.265
 //! source to length-prefixed NALU plus a resolved avcC/hvcC (parsing in-band
@@ -47,9 +48,11 @@ const PSI_INTERVAL: Duration = Duration::from_millis(500);
 
 /// Subscribe to a broadcast and produce an MPEG-TS byte stream.
 ///
-/// Use [`next`](Self::next) to pull byte chunks: the first chunk is PAT+PMT, then
-/// each subsequent chunk is the TS packets for one media frame (preceded by a
-/// fresh PAT+PMT at video keyframes). Returns `None` when the broadcast ends.
+/// Use [`next`](Self::next) to pull one [`Frame`] per media frame: its `payload`
+/// is the TS packets, stamped with the source `timestamp` and `keyframe` flag.
+/// The leading PAT/PMT rides on the first frame (so it inherits a real
+/// timestamp), and is re-emitted at video keyframes and periodically for
+/// mid-stream tune-in. Returns `None` when the broadcast ends.
 pub struct Export<E: scte35::Catalog = ()> {
 	broadcast: moq_net::BroadcastConsumer,
 	catalog: Option<crate::catalog::Consumer<E>>,
@@ -159,12 +162,19 @@ impl<E: scte35::Catalog> Export<E> {
 		self
 	}
 
-	/// Get the next byte chunk.
-	pub async fn next(&mut self) -> anyhow::Result<Option<Bytes>> {
+	/// Get the next muxed frame.
+	///
+	/// Each [`Frame`] carries the TS packets for one media frame in `payload`,
+	/// stamped with that frame's media `timestamp` and `keyframe` flag so a
+	/// transport can pace delivery on the media clock. The leading PAT/PMT rides
+	/// on the first frame (inheriting its timestamp), and is re-emitted at video
+	/// keyframes and periodically for mid-stream tune-in. Returns `None` when the
+	/// broadcast ends. `duration` is always `None`: the muxer has no use for it.
+	pub async fn next(&mut self) -> anyhow::Result<Option<Frame>> {
 		kio::wait(|waiter| self.poll_next(waiter)).await
 	}
 
-	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<anyhow::Result<Option<Bytes>>> {
+	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<anyhow::Result<Option<Frame>>> {
 		// 1. Drain catalog updates, discovering the track layout.
 		while let Some(catalog) = self.catalog.as_mut() {
 			match catalog.poll_next(waiter)? {
@@ -207,8 +217,10 @@ impl<E: scte35::Catalog> Export<E> {
 			}
 		}
 
-		// 3. Emit the program tables once the layout is resolved and every
-		// track's codec config is ready.
+		// 3. Build the program tables once the layout is resolved and every
+		// track's codec config is ready. The tables aren't emitted here: PSI has
+		// no media time of its own, so `write_frame` prepends them to the first
+		// frame instead, letting the leading PAT/PMT inherit a real timestamp.
 		if self.psi.is_none() {
 			if self.tracks.is_empty() {
 				// No tracks yet. If the catalog is also done, the broadcast is empty.
@@ -226,15 +238,14 @@ impl<E: scte35::Catalog> Export<E> {
 				return Poll::Pending;
 			}
 			self.build_psi()?;
-			let header = self.write_psi()?;
-			return Poll::Ready(Ok(Some(header)));
 		}
 
-		// 4. Emit the smallest-timestamp pending frame as a PES packet.
+		// 4. Emit the smallest-timestamp pending frame as a PES packet (the first
+		// one carries the buffered PAT/PMT).
 		if let Some(name) = self.pick_next_track() {
 			let frame = self.tracks.get_mut(&name).unwrap().pending.take().unwrap();
-			let chunk = self.write_frame(&name, frame)?;
-			return Poll::Ready(Ok(Some(chunk)));
+			let out = self.write_frame(&name, frame)?;
+			return Poll::Ready(Ok(Some(out)));
 		}
 
 		// 5. End of stream once every track has drained and the catalog is closed.
@@ -444,18 +455,7 @@ impl<E: scte35::Catalog> Export<E> {
 		Ok(())
 	}
 
-	/// Serialize a fresh PAT + PMT into a chunk.
-	fn write_psi(&mut self) -> anyhow::Result<Bytes> {
-		let psi = self.psi.as_ref().context("PSI not built")?;
-		let pat = TsPayload::Pat(psi.pat.clone());
-		let pmt = TsPayload::Pmt(psi.pmt.clone());
-
-		let mut out = Vec::with_capacity(2 * TsPacket::SIZE);
-		self.write_packet(&mut out, Pid::PAT, None, pat)?;
-		self.write_packet(&mut out, PMT_PID, None, pmt)?;
-		Ok(Bytes::from(out))
-	}
-
+	/// Name of the track whose pending frame has the smallest timestamp.
 	fn pick_next_track(&self) -> Option<String> {
 		self.tracks
 			.iter()
@@ -464,14 +464,18 @@ impl<E: scte35::Catalog> Export<E> {
 			.map(|(n, _)| n)
 	}
 
-	/// Packetize one media frame into a chunk, re-emitting PAT/PMT before video
-	/// keyframes (and periodically) so receivers can tune in mid-stream.
-	fn write_frame(&mut self, name: &str, frame: Frame) -> anyhow::Result<Bytes> {
+	/// Packetize one media frame into an output [`Frame`], re-emitting PAT/PMT
+	/// before video keyframes (and periodically) so receivers can tune in
+	/// mid-stream. The returned frame keeps the source `timestamp` and `keyframe`
+	/// flag so the caller can pace it.
+	fn write_frame(&mut self, name: &str, frame: Frame) -> anyhow::Result<Frame> {
 		let track = self.tracks.get(name).context("missing track")?;
 		let pid = track.pid;
 		let kind = track.kind.clone();
 		let is_pcr = self.psi.as_ref().is_some_and(|p| p.pcr_pid == pid);
 		let is_video = matches!(kind, Kind::Video(_));
+		let timestamp = frame.timestamp;
+		let keyframe = frame.keyframe;
 
 		// Build the elementary-stream payload for this frame. Video needs the
 		// resolved avcC/hvcC to rewrite length-prefixed NALs as Annex-B. SCTE-35
@@ -525,7 +529,12 @@ impl<E: scte35::Catalog> Export<E> {
 				self.write_pes(&mut out, &unit, &es_payload)?;
 			}
 		}
-		Ok(Bytes::from(out))
+		Ok(Frame {
+			timestamp,
+			duration: None,
+			payload: Bytes::from(out),
+			keyframe,
+		})
 	}
 
 	/// Packetize a PES payload into 188-byte TS packets.

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

@@ -66,10 +66,10 @@ async fn drain_with<E: scte35::Catalog>(mut exporter: Export<E>) -> BytesMut {
 	let mut out = BytesMut::new();
 	// `while let Ok` stops on the first timeout (`Pending`: no more output).
 	while let Ok(res) = tokio::time::timeout(std::time::Duration::from_secs(1), exporter.next()).await {
-		let Some(chunk) = res.expect("exporter error") else {
+		let Some(frame) = res.expect("exporter error") else {
 			break;
 		};
-		out.extend_from_slice(&chunk);
+		out.extend_from_slice(&frame.payload);
 	}
 	out
 }

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

@@ -212,7 +212,7 @@ async fn import_export_import_roundtrip() {
 	let mut out = BytesMut::new();
 	while let Ok(res) = tokio::time::timeout(std::time::Duration::from_secs(1), exporter.next()).await {
 		match res.expect("exporter error") {
-			Some(chunk) => out.extend_from_slice(&chunk),
+			Some(frame) => out.extend_from_slice(&frame.payload),
 			None => break,
 		}
 	}

rs/moq-srt/src/listen.rs

@@ -238,18 +238,36 @@ async fn serve_request(origin: OriginConsumer, path: &str, mut socket: srt_tokio
 	};
 
 	// MPEG-TS is a continuous byte stream, so we coalesce the muxer's per-frame
-	// chunks and slice them on a fixed boundary rather than preserving them.
+	// output and slice it on a fixed boundary rather than preserving frames.
+	//
+	// Pace on the media clock: stamp each SRT payload with the media time of the
+	// frame it carries, anchored to when the first frame went out. SRT's TSBPD
+	// reconstructs that inter-frame spacing at the receiver, so a tune-in keyframe
+	// burst is released at the media rate instead of all at once. (The Instant
+	// passed to `send` is the packet's origin time feeding TSBPD, not a "send now"
+	// instruction; `Instant::now()` would collapse the spacing into a burst.)
+	//
+	// We pace on the frame's presentation timestamp (the only clock the muxer
+	// exposes) while frames transmit in decode order, so a B-frame stream's
+	// per-GOP reorder leaves `send_at` slightly non-monotonic. That's harmless:
+	// `saturating_sub` keeps it >= the anchor, and the receiver reorders from the
+	// PTS/DTS carried inside the TS payload regardless.
+	let anchor = Instant::now();
+	let mut base = None;
+	let mut send_at = anchor;
 	let mut buffer = bytes::BytesMut::new();
-	while let Some(chunk) = subscriber.next().await? {
-		buffer.extend_from_slice(&chunk);
+	while let Some(frame) = subscriber.next().await? {
+		let origin = *base.get_or_insert(frame.timestamp);
+		send_at = anchor + Duration::from(frame.timestamp).saturating_sub(Duration::from(origin));
+
+		buffer.extend_from_slice(&frame.payload);
 		while buffer.len() >= SRT_PAYLOAD {
-			let payload = buffer.split_to(SRT_PAYLOAD).freeze();
-			socket.send((Instant::now(), payload)).await?;
+			socket.send((send_at, buffer.split_to(SRT_PAYLOAD).freeze())).await?;
 		}
 	}
 
 	if !buffer.is_empty() {
-		socket.send((Instant::now(), buffer.freeze())).await?;
+		socket.send((send_at, buffer.freeze())).await?;
 	}
 	socket.close().await?;
 

rs/moq-srt/src/ts.rs

@@ -8,7 +8,7 @@
 //! it back to MPEG-TS for an SRT caller (VLC, ffmpeg) to play.
 
 use bytes::Bytes;
-use moq_mux::container::ts;
+use moq_mux::container::{Frame, ts};
 use moq_net::{BroadcastInfo, OriginConsumer, OriginProducer, OriginPublish};
 
 use crate::Result;
@@ -62,7 +62,8 @@ impl Publisher {
 ///
 /// The mirror of [`Publisher`]: where that demuxes SRT-carried TS into the
 /// origin, this consumes a broadcast from the origin and re-muxes it to TS so an
-/// SRT caller can play it. Pull byte chunks with [`next`](Self::next).
+/// SRT caller can play it. Pull frames with [`next`](Self::next); each carries
+/// the TS bytes plus the media timestamp used to pace delivery.
 pub struct Subscriber {
 	export: ts::Export,
 }
@@ -82,8 +83,9 @@ impl Subscriber {
 		Ok(Some(Self { export }))
 	}
 
-	/// Pull the next MPEG-TS byte chunk, or `None` once the broadcast ends.
-	pub async fn next(&mut self) -> Result<Option<Bytes>> {
+	/// Pull the next muxed frame (TS bytes + media timestamp), or `None` once the
+	/// broadcast ends.
+	pub async fn next(&mut self) -> Result<Option<Frame>> {
 		Ok(self.export.next().await?)
 	}
 }