feat(moq-mux): record the TS mux rate and pad export to it

doc/bin/cli.md

@@ -64,6 +64,14 @@ discarding the old mux buffer. The first new clock packet signals the break and
 stdout pacing re-anchors. Every rendition joins the new program generation;
 no track is fenced across the marker.
 
+A constant-rate MPEG-TS source records its multiplex rate in the catalog
+(`mpegts.muxRate`, measured off the PCR clock, null stuffing included), and
+`export ts` pads its output with null packets back to that rate so an IRD or
+groomer receives a constant-rate stream. `--mux-rate 5000000` pads to an explicit
+rate instead, including for a broadcast that recorded none. Media is never delayed
+or dropped to fit: a source that sustains more than the rate overruns it, and a
+VBR source records nothing, so export without either stays unpadded.
+
 fMP4 export writes one fragment per publisher group on each track. Audio follows
 the publisher's cuts; video normally follows GOPs. Closing a group flushes it
 even when the live publisher pauses. `--fragment-duration 2s` caps

drafts/draft-lcurley-moq-mpegts.md

@@ -110,6 +110,7 @@ type Mpegts = {
   "programDescriptors": Descriptor[] | undefined,
   "program": Program | undefined,
   "si": Map<PidString, Si> | undefined,
+  "muxRate": number | undefined,
 }
 ~~~
 
@@ -181,6 +182,13 @@ The key is `table_id` rather than PID because that is the granularity the requir
 
 PAT and PMT are never carried here: they are rebuilt from `program`, `programDescriptors`, and the per-track entries.
 
+## muxRate {#field-mux-rate}
+The rate the source's PCR clock paced the whole multiplex at, in bits per second: every PID, the PSI, and the null packets, measured as the packets between two PCRs over the time they span.
+It is not a sum of the elementary streams, and a track's own bitrate keeps its codec meaning.
+
+A publisher MUST include `muxRate` only while the source holds a constant rate, and MUST omit it for a variable-rate or unpaced source; a value that has become invalid is removed rather than left stale.
+A consumer rebuilding a transport stream SHOULD pad its output with null packets to `muxRate` ({{rebuild}}).
+
 ### SI Track {#si-track}
 Each group is a complete picture of the entry's current sections: one frame per sub-table, each frame that sub-table's sections concatenated verbatim in `section_number` order.
 Sections are self-delimiting through `section_length`, so a frame needs no further framing.
@@ -260,6 +268,7 @@ A consumer rebuilding a transport stream:
 - MUST re-emit each track's `descriptors` as its ES-level descriptors, and `programDescriptors` as the PMT's `program_info`.
 - MUST re-emit each `si` entry's sections byte-for-byte on that entry's PID, reading them from its track ({{si-track}}), at least as often as its `interval` when declared.
 - MUST repacketize each verbatim track per its `framing` and `streamType`, using `streamId` when recorded.
+- SHOULD pad the output with null packets to `muxRate` when present, so the rebuilt stream is constant-rate again; a source that exceeds the rate is passed through rather than delayed or dropped.
 
 With no `program` the consumer synthesizes an identity, and SHOULD then omit any carried `si`, which describes a program that no longer exists.
 
@@ -334,6 +343,7 @@ A broadcast demultiplexed from a DVB transport stream: video and audio described
 - Initial version.
 - The `Si` type is keyed by `table_id` only; the PID lives on the enclosing `si` map.
 - A consumer refuses a catalog with an unrecognized `framing` or an invalid `si` map key.
+- Added `muxRate`, the source's constant multiplex rate.
 
 
 # Acknowledgments

quest/next/README.md

@@ -143,7 +143,6 @@ Implementation that depends on a dev contract requires
 - [Intra-refresh GOPs](/quest/next/intra-refresh/README.md) - video with periodic intra refresh publishes, imports, and tunes in cleanly with one group per sweep and a catalog `warmup`
 - [Audio warmup](/quest/next/audio-warmup.md) - a viewer joining an Opus rendition mid-stream never hears the unconverged first 80 ms
 - [#3021](/quest/next/3021-moq-gst-anchor-generated-media-timelines-to-wall-clock.md) - GStreamer maps every pad onto one continuous broadcast clock across source restarts
-- [Mux rate](/quest/next/ts-mux-rate.md) - moq ts: the source multiplex rate is recorded at import and export pads to it, so a CBR stream leaves as one
 - [#2779](/quest/next/2779-moq-export-ts-continuity-counters-are-numbered-from.md) - moq export ts: continuity counters are numbered from process state, so two exporters of the same broadcast emit streams that can never be compared
 - [#2829](/quest/next/2829-moq-export-ts-the-audio-video-interleave-is-decided-by.md) - moq export ts: the audio/video interleave is decided by arrival timing, so two exporters of one broadcast render the same media in different orders
 - [#3489](/quest/next/3489-ts-import-stream-liveness.md) - moq import ts: every elementary stream reports its access units and how long it has been quiet

rs/moq-cli/src/args.rs

@@ -792,9 +792,8 @@ impl Export {
 		}
 		match &self.sink {
 			ExportSink::Fmp4(args) | ExportSink::Mkv(args) => found.extend(args.container.deprecated()),
-			ExportSink::Ts(args) | ExportSink::Flv(args) | ExportSink::H264(args) | ExportSink::H265(args) => {
-				found.extend(args.deprecated())
-			}
+			ExportSink::Ts(args) => found.extend(args.container.deprecated()),
+			ExportSink::Flv(args) | ExportSink::H264(args) | ExportSink::H265(args) => found.extend(args.deprecated()),
 			ExportSink::Hls(hls) => found.extend(hls.tls.deprecated()),
 			ExportSink::Rtmp(rtmp) if rtmp.latency_max.is_some() => {
 				found.flag("--latency-max", None, "--max-age");
@@ -814,7 +813,7 @@ pub enum ExportSink {
 	/// Matroska / WebM to stdout.
 	Mkv(Fragmented),
 	/// MPEG-TS to stdout.
-	Ts(Container),
+	Ts(Transport),
 	/// FLV / RTMP container to stdout.
 	Flv(Container),
 	/// H.264 Annex-B elementary stream to stdout.
@@ -832,30 +831,49 @@ pub enum ExportSink {
 }
 
 impl ExportSink {
-	/// The stdout container format plus its latency and fragment cap, when this
-	/// sink writes to stdout (the container formats). The fragment cap is
-	/// fmp4/mkv-only.
-	pub fn stdout(&self) -> Option<(SubscribeFormat, std::time::Duration, Option<Duration>)> {
+	/// Whether this sink writes to stdout (the container formats).
+	pub fn is_stdout(&self) -> bool {
+		self.stdout().is_some()
+	}
+
+	/// The stdout container format and its options, when this sink writes to
+	/// stdout. The fragment cap is fmp4/mkv-only and the mux rate is TS-only.
+	pub fn stdout(&self) -> Option<Stdout> {
+		let container = |format, container: &Container| Stdout {
+			format,
+			max_age: container.max_age.into_std(),
+			fragment_duration: None,
+			mux_rate: None,
+		};
 		Some(match self {
-			Self::Fmp4(args) => (
-				SubscribeFormat::Fmp4,
-				args.container.max_age.into_std(),
-				args.fragment_duration.map(crate::duration::Duration::into_std),
-			),
-			Self::Mkv(args) => (
-				SubscribeFormat::Mkv,
-				args.container.max_age.into_std(),
-				args.fragment_duration.map(crate::duration::Duration::into_std),
-			),
-			Self::Ts(args) => (SubscribeFormat::Ts, args.max_age.into_std(), None),
-			Self::Flv(args) => (SubscribeFormat::Flv, args.max_age.into_std(), None),
-			Self::H264(args) => (SubscribeFormat::H264, args.max_age.into_std(), None),
-			Self::H265(args) => (SubscribeFormat::H265, args.max_age.into_std(), None),
+			Self::Fmp4(args) => Stdout {
+				fragment_duration: args.fragment_duration.map(crate::duration::Duration::into_std),
+				..container(SubscribeFormat::Fmp4, &args.container)
+			},
+			Self::Mkv(args) => Stdout {
+				fragment_duration: args.fragment_duration.map(crate::duration::Duration::into_std),
+				..container(SubscribeFormat::Mkv, &args.container)
+			},
+			Self::Ts(args) => Stdout {
+				mux_rate: args.mux_rate,
+				..container(SubscribeFormat::Ts, &args.container)
+			},
+			Self::Flv(args) => container(SubscribeFormat::Flv, args),
+			Self::H264(args) => container(SubscribeFormat::H264, args),
+			Self::H265(args) => container(SubscribeFormat::H265, args),
 			_ => return None,
 		})
 	}
 }
 
+/// A stdout sink's format and the options that apply to it.
+pub struct Stdout {
+	pub format: SubscribeFormat,
+	pub max_age: Duration,
+	pub fragment_duration: Option<Duration>,
+	pub mux_rate: Option<u64>,
+}
+
 /// Options shared by every stdout container sink.
 #[derive(usage::Args, Clone)]
 #[usage(unknown_flags = "error", args_override_self = false)]
@@ -879,6 +897,20 @@ impl Container {
 	}
 }
 
+/// The MPEG-TS stdout container: [`Container`] plus null padding.
+#[derive(usage::Args, Clone)]
+#[usage(unknown_flags = "error", args_override_self = false)]
+pub struct Transport {
+	#[usage(flatten)]
+	pub container: Container,
+
+	/// Pad the output with null packets to this constant rate, in bits per second.
+	/// Defaults to the multiplex rate the catalog recorded from a constant-rate
+	/// source (`mpegts.muxRate`); without either the output is unpadded.
+	#[usage(long)]
+	pub mux_rate: Option<u64>,
+}
+
 /// The fmp4 / mkv stdout containers: [`Container`] plus a fragment cap.
 #[derive(usage::Args, Clone)]
 #[usage(unknown_flags = "error", args_override_self = false)]

rs/moq-cli/src/main.rs

@@ -499,7 +499,7 @@ async fn run_stages(moq: MoqSide, stages: Vec<Command>, net: Net) -> anyhow::Res
 				}
 			}
 			Command::Export(export) => {
-				if export.sink.stdout().is_some() {
+				if export.sink.is_stdout() {
 					claim("stdout", &mut stdout, &name)?;
 				}
 				spawn_export(&origin, export, name, &mut tasks)?;
@@ -658,11 +658,12 @@ fn spawn_export(
 		reject_listener_cors(&rtc.cors, "export rtc")?;
 	}
 
-	if let Some((format, max_age, fragment_duration)) = export.sink.stdout() {
+	if let Some(stdout) = export.sink.stdout() {
 		let args = SubscribeArgs {
-			format,
-			max_age,
-			fragment_duration,
+			format: stdout.format,
+			max_age: stdout.max_age,
+			fragment_duration: stdout.fragment_duration,
+			mux_rate: stdout.mux_rate,
 			catalog: export.catalog_format,
 			select: export.select,
 		};

rs/moq-cli/src/subscribe.rs

@@ -142,6 +142,10 @@ pub struct SubscribeArgs {
 	/// Cap the output duration: publisher groups by default for fMP4, video GOPs for MKV.
 	pub fragment_duration: Option<Duration>,
 
+	/// Pad MPEG-TS output with null packets to this rate, in bits per second,
+	/// overriding the catalog's recorded multiplex rate.
+	pub mux_rate: Option<u64>,
+
 	/// Catalog format for track discovery (default: detect from the broadcast suffix).
 	pub catalog: Option<CatalogFormatArg>,
 
@@ -299,6 +303,9 @@ impl Subscribe {
 		let mut ts = moq_mux::container::ts::Export::with_ts(self.source, self.catalog)
 			.await?
 			.with_max_age(self.args.max_age);
+		if let Some(mux_rate) = self.args.mux_rate {
+			ts = ts.with_mux_rate(mux_rate);
+		}
 
 		// A TS byte stream carries no per-frame timing, so delivery time is the only
 		// carrier of each frame's spacing: the exporter slices its output on the PCR

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

@@ -7,7 +7,9 @@
 //! subtitles, private data, ...), the program-level PMT descriptors, the program
 //! identity ([`Program`]), and the standalone SI table map ([`SiEntry`]).
 //! Demuxed media tracks keep their codec config in the base `video`/`audio`
-//! sections; only their MPEG-TS identity lands here.
+//! sections; only their MPEG-TS identity lands here. A constant-rate source also
+//! records its multiplex rate ([`Mpegts::mux_rate`]), so export can pad the rebuilt
+//! stream back to it.
 //!
 //! The section is specified by `drafts/draft-lcurley-moq-mpegts.md` and rides the
 //! root of either catalog: the hang track alongside `video`/`audio`, and the MSF
@@ -116,12 +118,24 @@ pub struct Mpegts {
 	#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
 	#[serde_as(as = "BTreeMap<DisplayFromStr, BTreeMap<DisplayFromStr, _>>")]
 	pub si: BTreeMap<u16, BTreeMap<u8, SiEntry>>,
+
+	/// The rate the PCR clock paces the whole multiplex at, in bits per second:
+	/// every PID plus PSI plus null stuffing, measured between PCRs on import. Not a
+	/// sum of elementary streams; each rendition's `bitrate` keeps its codec meaning.
+	/// Present only while the source is constant-rate, so a VBR or file-paced input
+	/// leaves it absent. Export pads its output with null packets to this rate.
+	#[serde(default, skip_serializing_if = "Option::is_none")]
+	pub mux_rate: Option<u64>,
 }
 
 impl Mpegts {
 	/// True when the section carries nothing, so it's omitted from the catalog.
 	pub fn is_empty(&self) -> bool {
-		self.tracks.is_empty() && self.program_descriptors.is_empty() && self.program.is_none() && self.si.is_empty()
+		self.tracks.is_empty()
+			&& self.program_descriptors.is_empty()
+			&& self.program.is_none()
+			&& self.si.is_empty()
+			&& self.mux_rate.is_none()
 	}
 }
 
@@ -324,6 +338,21 @@ mod test {
 		assert_eq!(serde_json::to_string(&ext).unwrap(), "{}");
 	}
 
+	#[test]
+	fn mux_rate_alone_is_not_empty() {
+		// The rate is the only field of a media-only CBR source, so it must keep the
+		// section alive rather than be dropped with it.
+		let mpegts = Mpegts {
+			mux_rate: Some(2_500_000),
+			..Default::default()
+		};
+		assert!(!mpegts.is_empty());
+		let json = serde_json::to_string(&Ext { mpegts }).unwrap();
+		assert_eq!(json, r#"{"mpegts":{"muxRate":2500000}}"#);
+		// An absent rate is omitted, never `null`.
+		assert_eq!(serde_json::to_string(&Ext::default()).unwrap(), "{}");
+	}
+
 	#[test]
 	fn section_roundtrip() {
 		let mut mpegts = Mpegts::default();
@@ -352,10 +381,15 @@ mod test {
 			tag: 0x05,
 			data: Bytes::from_static(b"CUEI"),
 		});
+		mpegts.mux_rate = Some(2_500_000);
 
 		let json = serde_json::to_string(&Ext { mpegts: mpegts.clone() }).unwrap();
 		// Descriptor bytes are base64 ("CUEI" -> "Q1VFSQ==").
 		assert!(json.contains("\"Q1VFSQ==\""), "descriptor data is base64: {json}");
+		assert!(
+			json.contains("\"muxRate\":2500000"),
+			"mux rate is a bare integer: {json}"
+		);
 
 		let parsed: Ext = serde_json::from_str(&json).unwrap();
 		assert_eq!(parsed.mpegts, mpegts, "mpegts section round-trips");