feat(moq-cli): add --pace to emit TS output, following the live edge

doc/bin/cli.md

@@ -187,6 +187,20 @@ ffmpeg -i input.mp4 -c copy -f mpegts - | \
 moq-cli subscribe --url https://relay.example.com --broadcast my-stream --format ts | ffplay -
 ```
 
+By default a retained broadcast is written out as fast as it can be read. Add
+`--pace` to emit the TS stream at its real-time rate instead, like ffmpeg's
+`-re`, which is what a downstream player or re-publish expects:
+
+```bash
+moq-cli subscribe --url https://relay.example.com --broadcast my-stream --format ts --pace \
+    | ffmpeg -i - -c copy -f mpegts udp://239.0.0.1:1234
+```
+
+Pacing follows the live edge: it holds at most `--max-latency` of buffer (default
+500ms) ahead of the playhead, so a retained broadcast streams at its media rate
+while a live source stays near the edge instead of letting the initial burst
+become permanent latency.
+
 TS export carries H.264 / H.265 as Annex-B and AAC as ADTS. Both in-band
 (avc3 / hev1) and out-of-band (avc1 / hvc1, e.g. from an fMP4 import) video
 sources work: the parameter sets are read from the bitstream or the catalog

rs/moq-cli/src/main.rs

@@ -283,6 +283,8 @@ async fn run_subscribe(
 	broadcast: String,
 	args: SubscribeArgs,
 ) -> anyhow::Result<()> {
+	args.validate()?;
+
 	let origin = moq_net::Origin::random().produce();
 	let consumer = origin.consume();
 

rs/moq-cli/src/subscribe.rs

@@ -43,6 +43,15 @@ pub struct SubscribeArgs {
 	#[arg(long, default_value = "500ms", value_parser = humantime::parse_duration)]
 	pub max_latency: Duration,
 
+	/// Pace output at the media's real-time rate, like ffmpeg's `-re`.
+	///
+	/// Without this, a retained broadcast is written out as fast as it can be read.
+	/// With it, frames are emitted on the media clock (anchored to the first frame),
+	/// which is what a downstream player or re-publish expects. Only `--format ts`
+	/// supports pacing today.
+	#[arg(long)]
+	pub pace: bool,
+
 	/// Cap the output fragment duration (e.g. `2s`, `500ms`).
 	///
 	/// By default a fragment covers one GOP (rolled over on video keyframes).
@@ -61,6 +70,15 @@ pub struct SubscribeArgs {
 }
 
 impl SubscribeArgs {
+	/// Reject flag combinations that don't apply to the chosen format, before we
+	/// bother connecting to the relay.
+	pub fn validate(&self) -> anyhow::Result<()> {
+		if self.pace && !matches!(self.format, SubscribeFormat::Ts) {
+			anyhow::bail!("--pace is only supported with --format ts");
+		}
+		Ok(())
+	}
+
 	/// Resolve the catalog format, falling back to detection from the broadcast
 	/// name suffix and then to the default.
 	pub fn catalog_format(&self, broadcast: &str) -> CatalogFormat {
@@ -135,6 +153,7 @@ impl Subscribe {
 
 	async fn run_ts(self) -> anyhow::Result<()> {
 		let mut stdout = tokio::io::stdout();
+		let pace = self.args.pace;
 
 		// TS emits PAT/PMT then a continuous PES stream (re-emitting PAT/PMT at
 		// keyframes for tune-in). Avc3/Hev1 sources pass through as Annex-B; AAC
@@ -145,6 +164,13 @@ impl Subscribe {
 			moq_mux::container::ts::Export::with_ts(self.broadcast, self.catalog)?.with_latency(self.args.max_latency);
 
 		while let Some(frame) = ts.next().await? {
+			// `--pace`: hold each frame until it's due. The muxer paces on its decode clock
+			// and follows the live edge with up to `--max-latency` of buffer, so a retained
+			// broadcast drains in real time while a live source stays near the edge.
+			if pace {
+				tokio::time::sleep_until(frame.pace.into()).await;
+			}
+
 			stdout.write_all(&frame.payload).await?;
 			stdout.flush().await?;
 		}
@@ -170,3 +196,26 @@ impl Subscribe {
 		Ok(())
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	fn args(format: SubscribeFormat, pace: bool) -> SubscribeArgs {
+		SubscribeArgs {
+			format,
+			max_latency: Duration::from_millis(500),
+			pace,
+			fragment_duration: None,
+			catalog: None,
+		}
+	}
+
+	#[test]
+	fn pace_requires_ts_format() {
+		assert!(args(SubscribeFormat::Ts, true).validate().is_ok());
+		assert!(args(SubscribeFormat::Fmp4, true).validate().is_err());
+		// Without --pace, any format is fine.
+		assert!(args(SubscribeFormat::Fmp4, false).validate().is_ok());
+	}
+}

rs/moq-mux/src/container/mod.rs

@@ -16,6 +16,7 @@ use bytes::Bytes;
 
 mod consumer;
 pub(crate) mod jitter;
+mod pace;
 mod producer;
 mod source;
 
@@ -27,6 +28,7 @@ pub mod mkv;
 pub mod ts;
 
 pub use consumer::Consumer;
+pub(crate) use pace::Pacer;
 pub use producer::Producer;
 pub(crate) use source::ExportSource;
 

rs/moq-mux/src/container/pace.rs

@@ -0,0 +1,162 @@
+//! Pace media output on a media clock, following the live edge.
+
+use std::time::{Duration, Instant};
+
+use super::Timestamp;
+
+/// Maps media (decode) timestamps onto the wall clock so a caller can emit frames at
+/// the source's real-time rate while bounding how far behind the live edge it falls.
+///
+/// The first frame anchors the media clock to "now"; every later frame is due at
+/// `anchor + (timestamp - base)`. Sleeping until [`Pacer::due`] before emitting a frame
+/// drains a retained broadcast at its media rate, like ffmpeg's `-re`, instead of as
+/// fast as it can be read.
+///
+/// To keep a bursty or faster-than-real source from accruing unbounded latency, the
+/// timeline holds at most `lead` of buffer ahead of now: when a frame would be due
+/// further out than that (a tune-in burst delivers a whole GOP at once, or the source
+/// drifts ahead of wall-clock), the anchor jumps forward to the live edge so the buffer
+/// never exceeds `lead`. A frame that merely trails the edge (network jitter, a
+/// reordered B-frame) keeps its earlier instant.
+///
+/// `lead` is the target buffer, typically the subscription's max latency. With `lead`
+/// = 0 the timeline never leads now, which is what an SRT egress wants (it stamps each
+/// payload's TSBPD origin time and the receiver owns the jitter buffer); with `lead` >
+/// 0 the caller sleeps to pace output and holds that much buffer itself.
+#[derive(Default)]
+pub(crate) struct Pacer {
+	anchor: Option<Anchor>,
+}
+
+/// The media-clock anchor: `base`'s media time maps to `at` on the wall clock.
+struct Anchor {
+	at: Instant,
+	base: Timestamp,
+}
+
+impl Pacer {
+	/// A pacer that anchors on the first frame it sees.
+	pub(crate) fn new() -> Self {
+		Self::default()
+	}
+
+	/// The wall-clock instant `timestamp` is due, holding at most `lead` of buffer
+	/// ahead of now. Reads the clock itself; see [`Self::at`] for the testable core.
+	pub(crate) fn due(&mut self, timestamp: Timestamp, lead: Duration) -> Instant {
+		self.at(timestamp, lead, Instant::now())
+	}
+
+	/// [`Self::due`] with an explicit `now`, so the mapping is deterministic in tests.
+	fn at(&mut self, timestamp: Timestamp, lead: Duration, now: Instant) -> Instant {
+		let anchor = self.anchor.get_or_insert(Anchor {
+			at: now,
+			base: timestamp,
+		});
+
+		let due = match timestamp.checked_sub(anchor.base) {
+			Ok(ahead) => anchor.at + Duration::from(ahead),
+			Err(_) => anchor
+				.at
+				.checked_sub(Duration::from(anchor.base - timestamp))
+				.unwrap_or(anchor.at),
+		};
+
+		// Never schedule more than `lead` ahead: when the source outruns that, re-anchor
+		// the live edge to `now + lead` so the buffer stays bounded. Re-anchoring only ever
+		// moves forward, so a frame that trails the edge keeps its earlier instant.
+		let cap = now + lead;
+		if due > cap {
+			anchor.at = cap;
+			anchor.base = timestamp;
+			cap
+		} else {
+			due
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	fn ms(m: u64) -> Timestamp {
+		Timestamp::from_micros(m * 1_000).unwrap()
+	}
+
+	#[test]
+	fn paces_on_the_media_clock() {
+		// Within the lead budget, frames pace on the media clock: a frame 40ms later in
+		// media is due 40ms after the anchor, however quickly it was read. This is the
+		// `-re` case (the caller sleeps in lockstep, so the cap never trips).
+		let mut pacer = Pacer::new();
+		let start = Instant::now();
+		let lead = Duration::from_millis(500);
+
+		assert_eq!(
+			pacer.at(ms(1_000), lead, start),
+			start,
+			"the first frame anchors to now"
+		);
+		assert_eq!(
+			pacer.at(ms(1_040), lead, start + Duration::from_millis(1)),
+			start + Duration::from_millis(40),
+			"output is paced on the media clock, not arrival time"
+		);
+	}
+
+	#[test]
+	fn re_anchors_past_the_lead_budget() {
+		// A tune-in burst: a whole GOP is read at once, so the newest frame's media time
+		// runs far past `now`. It re-anchors to `now + lead` rather than scheduling
+		// seconds out, so the buffer never exceeds the target latency.
+		let mut pacer = Pacer::new();
+		let start = Instant::now();
+		let lead = Duration::from_millis(500);
+		pacer.at(ms(1_000), lead, start);
+
+		let now = start + Duration::from_millis(2);
+		let edge = pacer.at(ms(4_132), lead, now);
+		assert_eq!(edge, now + lead, "the live edge is held `lead` ahead of now");
+
+		// A later frame paces off the re-anchored edge. The caller slept until the edge
+		// was due, so `now` has advanced to it; the 40ms-newer frame is due 40ms past it.
+		let now = now + lead;
+		assert_eq!(
+			pacer.at(ms(4_172), lead, now),
+			now + Duration::from_millis(40),
+			"subsequent frames pace off the re-anchored edge"
+		);
+	}
+
+	#[test]
+	fn trailing_frame_keeps_its_instant() {
+		// A reordered B-frame whose timestamp dips below the edge maps into the past, so
+		// the caller's sleep is a no-op and it's emitted immediately. No re-anchor.
+		let mut pacer = Pacer::new();
+		let start = Instant::now();
+		let lead = Duration::from_millis(500);
+		pacer.at(ms(1_000), lead, start);
+
+		assert_eq!(
+			pacer.at(ms(967), lead, start + Duration::from_millis(5)),
+			start - Duration::from_millis(33),
+		);
+	}
+
+	#[test]
+	fn zero_lead_never_leads_now() {
+		// `lead` = 0 is the SRT egress policy: the timeline is capped at now, so a burst
+		// re-anchors to now (the newest frame is the live edge) and nothing is stamped
+		// into the future.
+		let mut pacer = Pacer::new();
+		let start = Instant::now();
+		pacer.at(ms(1_000), Duration::ZERO, start);
+
+		let now = start + Duration::from_millis(2);
+		assert_eq!(
+			pacer.at(ms(4_132), Duration::ZERO, now),
+			now,
+			"the live edge paces to now"
+		);
+	}
+}