feat(mux): export ts --passthrough writes the m2ts track back on a fixed delay

doc/bin/cli.md

@@ -79,7 +79,7 @@ logs feed health (stalled streams, TR 101 290 errors) without changing what it
 publishes. `--passthrough` publishes the multiplex whole instead, every packet
 in order on one [`m2ts`](/concept/hang#transport-streams) track with every
 program aboard, for what demultiplexing cannot carry, such as a scrambled
-service; `export ts` does not read it. FLV covers H.264 + AAC.
+service; `export ts --passthrough` writes it back out. FLV covers H.264 + AAC.
 
 ## Export
 
@@ -319,6 +319,15 @@ arriving after its deadline is dropped, and video resumes at its next
 keyframe. The output's PCR follows the source's clock within what ISO/IEC
 13818-1 allows. `--delay 0` writes frames in arrival order and drops nothing.
 
+`export ts --passthrough` reads the track an `import ts --passthrough` published
+and writes its packets back out unchanged, from its first group on, each run
+between two PCRs `--delay` after the first one's time on the source clock. The
+output is exactly as conformant as the source, and two exporters emit the same
+bytes. A run that arrives after its deadline is dropped and counted, leaving a
+continuity error, and the rest carry on. It takes neither `--linger`,
+`--stitch` nor `--mux-rate`. Only the hang catalog names the track, so an
+`.msf` broadcast needs `--catalog-format hang`.
+
 A stdout export ends with the broadcast. `export ts --linger 10s` waits that
 long for the same publisher instance to come back, and carries on with the same
 stream. A replacement (a restarted publisher, under a new

rs/moq-cli/src/args.rs

@@ -974,6 +974,7 @@ impl ExportSink {
 			stitch: false,
 			fragment_duration: None,
 			mux_rate: None,
+			passthrough: false,
 		};
 		Some(match self {
 			Self::Fmp4(args) => Stdout {
@@ -991,6 +992,7 @@ impl ExportSink {
 				stitch: args.stitch,
 				fragment_duration: None,
 				mux_rate: args.mux_rate,
+				passthrough: args.passthrough,
 			},
 			Self::Flv(args) => container(SubscribeFormat::Flv, args),
 			Self::H264(args) => container(SubscribeFormat::H264, args),
@@ -1010,6 +1012,8 @@ pub struct Stdout {
 	pub stitch: bool,
 	pub fragment_duration: Option<Duration>,
 	pub mux_rate: Option<u64>,
+	/// Write the `m2ts` passthrough track back out (`ts` only).
+	pub passthrough: bool,
 }
 
 /// Options shared by the stdout container sinks other than `ts`.
@@ -1075,6 +1079,12 @@ pub struct Transport {
 	#[usage(long)]
 	pub mux_rate: Option<u64>,
 
+	/// Write the multiplex an `import ts --passthrough` published back out byte-identical,
+	/// paced on its own PCR, `--delay` behind it. Packets that arrive later than that are
+	/// dropped, leaving a continuity error.
+	#[usage(long, conflicts("--linger", "--stitch", "--mux-rate"))]
+	pub passthrough: bool,
+
 	/// The released spelling of [`Self::delay`].
 	#[usage(long = "max-age", hide = true)]
 	max_age: Option<crate::duration::Duration>,
@@ -1183,6 +1193,27 @@ mod tests {
 		}
 	}
 
+	/// `export ts --passthrough` writes the source's own packets, so it neither switches
+	/// programs, waits for a return, nor pads to a rate.
+	#[test]
+	fn export_ts_passthrough_writes_the_source_alone() {
+		let parse = |args: &[&str]| {
+			let mut argv = vec!["moq", "export", "--broadcast", "b", "ts", "--passthrough"];
+			argv.extend_from_slice(args);
+			Invocation::try_parse_from(argv)
+		};
+		let cli = parse(&["--delay", "2s"]).unwrap();
+		let Command::Export(export) = &cli.stages[0] else {
+			panic!("not an export");
+		};
+		let stdout = export.sink.stdout().unwrap();
+		assert!(stdout.passthrough);
+		assert_eq!(stdout.max_delay, Duration::from_secs(2));
+		for flags in [&["--stitch"][..], &["--linger", "10s"], &["--mux-rate", "20000000"]] {
+			assert!(parse(flags).is_err(), "{flags:?} must not start a run");
+		}
+	}
+
 	fn export(flags: &[&str]) -> Result<Invocation, ParseError> {
 		let argv = ["moq", "--connect", "http://relay", "--broadcast", "room", "export"];
 		Invocation::try_parse_from(argv.iter().chain(flags).copied())

rs/moq-cli/src/main.rs

@@ -747,6 +747,7 @@ fn spawn_export(
 			stitch: stdout.stitch,
 			fragment_duration: stdout.fragment_duration,
 			mux_rate: stdout.mux_rate,
+			passthrough: stdout.passthrough,
 			catalog: export.catalog_format,
 			select: export.select,
 		};

rs/moq-cli/src/subscribe.rs

@@ -194,6 +194,9 @@ pub struct SubscribeArgs {
 	/// overriding the catalog's recorded multiplex rate.
 	pub mux_rate: Option<u64>,
 
+	/// Write the `m2ts` passthrough track back out instead of muxing the media (TS only).
+	pub passthrough: bool,
+
 	/// Catalog format for track discovery (default: detect from the broadcast suffix).
 	pub catalog: Option<CatalogFormatArg>,
 
@@ -369,6 +372,9 @@ impl Subscribe {
 				self.path
 			)
 		})?;
+		if self.args.passthrough {
+			return self.run_ts_passthrough().await;
+		}
 
 		// 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
@@ -432,6 +438,40 @@ impl Subscribe {
 		}
 	}
 
+	/// Write the multiplex the broadcast's `m2ts` track carries, each run of packets as its
+	/// time on the source's PCR comes round.
+	async fn run_ts_passthrough(self) -> anyhow::Result<()> {
+		let mut stdout = tokio::io::stdout();
+		let mut release = moq_mux::container::ts::Release::new(self.source, self.catalog)
+			.await?
+			.with_delay(self.args.max_delay);
+		let log = |release: &moq_mux::container::ts::Release| {
+			tracing::info!(
+				dropped = release.dropped(),
+				drift_ppm = ?release.drift().map(|drift| drift * 1e6),
+				out_of_tolerance = release.out_of_tolerance(),
+				"TS passthrough release clock"
+			);
+		};
+		let mut sampled = tokio::time::Instant::now();
+		let mut reported = (0, 0);
+		while let Some(object) = release.next().await? {
+			stdout.write_all(&object.payload).await?;
+			stdout.flush().await?;
+
+			if sampled.elapsed() >= moq_mux::container::ts::stats::Log::INTERVAL {
+				sampled = tokio::time::Instant::now();
+				let counts = (release.dropped(), release.out_of_tolerance());
+				if counts != reported {
+					reported = counts;
+					log(&release);
+				}
+			}
+		}
+		log(&release);
+		Ok(())
+	}
+
 	async fn run_flv(self) -> anyhow::Result<()> {
 		let mut stdout = tokio::io::stdout();
 

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

@@ -20,7 +20,8 @@
 //!
 //! [`Passthrough`] is the other shape: it publishes the multiplex whole, every packet
 //! verbatim on one track named by the hang catalog's `m2ts` section, for what
-//! demultiplexing cannot carry (TS-level scrambling, PSI and SI as authored).
+//! demultiplexing cannot carry (TS-level scrambling, PSI and SI as authored), and
+//! [`Release`] writes that track back out on a fixed delay.
 
 mod adts;
 mod damage;
@@ -31,6 +32,7 @@ mod import;
 mod mux_rate;
 mod programs;
 mod psi;
+mod release;
 mod schedule;
 mod si;
 
@@ -45,6 +47,7 @@ pub use follower::Follower;
 pub use import::*;
 pub use passthrough::Passthrough;
 pub use programs::Programs;
+pub use release::Release;
 
 pub mod passthrough;
 pub mod stats;
@@ -57,3 +60,5 @@ mod export_timing_test;
 mod import_test;
 #[cfg(test)]
 mod passthrough_test;
+#[cfg(test)]
+mod release_test;

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

@@ -0,0 +1,193 @@
+//! Writes a [`Passthrough`](super::Passthrough) track back out as the multiplex it carries.
+
+use std::task::Poll;
+use std::time::Duration;
+
+use web_async::time::Instant;
+
+use crate::catalog::{CatalogFormat, Stream};
+use crate::container::{ExportSource, Frame};
+use crate::jitter::{self, Arrival, Push};
+
+/// Writes the multiplex a broadcast's [`m2ts`](hang::catalog::M2ts) track carries back out
+/// byte-identical, each object a fixed delay after its PCR time, like an SRT receiver.
+///
+/// The [importer](super::Passthrough) stamps each object, the packets from one PCR to the
+/// next, with the PCR time of its first byte on the catalog clock. The release holds every
+/// object until that time plus the delay on the jitter buffer's clock, which follows the
+/// source's, so the output keeps the source's PCR pacing whatever the arrival jitter, and two
+/// releases fed the same objects write the same bytes. An object that arrives past its
+/// deadline is dropped and counted; the output shows a continuity error there, since nothing
+/// is rewritten, and carries on with the next object. A flagged forward jump, which the
+/// importer marks by skipping a group sequence, opens a new clock once it leaps further than
+/// the delay, so the output does not wait it out; the bytes carry the jump.
+pub struct Release {
+	source: crate::Source,
+	catalog: Option<crate::catalog::Consumer<()>>,
+	/// The `m2ts` track, once a catalog names it.
+	track: Option<(String, ExportSource)>,
+	finished: bool,
+	/// An object has gone out.
+	released: bool,
+	delay: Duration,
+	jitter: jitter::Buffer<(), Frame>,
+	replay: bool,
+}
+
+impl Release {
+	/// Subscribe to the catalog of `source`, waiting for the `m2ts` section that names its track.
+	///
+	/// Fails for [`CatalogFormat::Msf`], which has no `m2ts` section.
+	pub async fn new(source: crate::Source, catalog_format: CatalogFormat) -> crate::Result<Self> {
+		if catalog_format == CatalogFormat::Msf {
+			return Err(
+				anyhow::anyhow!("the MSF catalog has no `m2ts` passthrough track; read the hang catalog").into(),
+			);
+		}
+		let catalog = source.catalog(catalog_format).await?;
+		let mut release = Self {
+			source,
+			catalog: Some(catalog),
+			track: None,
+			finished: false,
+			released: false,
+			delay: Duration::ZERO,
+			jitter: jitter::Buffer::new(Duration::ZERO),
+			replay: false,
+		};
+		release.jitter = release.jitter();
+		Ok(release)
+	}
+
+	/// Write each object this long after its PCR time. An object that arrives later is dropped.
+	/// Defaults to [`Duration::ZERO`], which holds nothing.
+	pub fn with_delay(mut self, delay: Duration) -> Self {
+		self.delay = delay;
+		self.jitter = self.jitter();
+		self
+	}
+
+	/// Read the track from the oldest group the delay still reaches instead of the live edge,
+	/// and anchor the clock on the first object, for a test that writes a broadcast whole first.
+	#[cfg(test)]
+	pub(super) fn with_replay(mut self) -> Self {
+		self.replay = true;
+		self.jitter = self.jitter();
+		self
+	}
+
+	fn jitter(&self) -> jitter::Buffer<(), Frame> {
+		let mut jitter = jitter::Buffer::new(self.delay);
+		#[cfg(test)]
+		if self.replay {
+			jitter = jitter.replay();
+		}
+		jitter.expect([()]);
+		jitter
+	}
+
+	/// How many objects were dropped for missing their deadline.
+	pub fn dropped(&self) -> u64 {
+		self.jitter.dropped()
+	}
+
+	/// The source's clock rate off ours, as last measured, as a fraction: positive runs fast.
+	pub fn drift(&self) -> Option<f64> {
+		self.jitter.drift()
+	}
+
+	/// How many times the source's clock was measured further off ours than the release follows.
+	pub fn out_of_tolerance(&self) -> u64 {
+		self.jitter.out_of_tolerance()
+	}
+
+	/// The next object, at its time: whole TS packets, verbatim, stamped with the PCR time of
+	/// its first byte. Returns `None` once the track, or a catalog that never named one, ends.
+	pub async fn next(&mut self) -> crate::Result<Option<Frame>> {
+		kio::wait(|waiter| self.poll_next(waiter)).await
+	}
+
+	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<crate::Result<Option<Frame>>> {
+		while let Some(catalog) = self.catalog.as_mut() {
+			match catalog.poll_next(waiter)? {
+				Poll::Ready(Some(snapshot)) => self.update_catalog(&snapshot)?,
+				Poll::Ready(None) => self.catalog = None,
+				Poll::Pending => break,
+			}
+		}
+
+		if !self.finished
+			&& let Some((name, track)) = self.track.as_mut()
+		{
+			loop {
+				let frame = match track.poll_read(waiter) {
+					Poll::Ready(Ok(Some(frame))) => frame,
+					Poll::Ready(Ok(None)) => {
+						self.finished = true;
+						break;
+					}
+					Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+					Poll::Pending => break,
+				};
+				// The track wakes the release as each object lands, so the read times its arrival.
+				// The last empty poll would let one landing past its deadline out late instead.
+				let read = Instant::now();
+				let arrival = Arrival {
+					arrived: read,
+					read,
+					decode: frame.timestamp,
+					restart: track.markers(),
+					skip: track.skips(),
+					// The output opens on a group, at a random access point; from then on every
+					// object goes out on its own, so one dropped late is a gap of its packets only.
+					sync: frame.keyframe || self.released,
+					item: frame,
+				};
+				if self.jitter.push((), arrival)? == Push::Late {
+					tracing::warn!(track = %name, dropped = self.jitter.dropped(), "object missed its deadline; dropped");
+				}
+			}
+		}
+
+		if let Poll::Ready(ready) = self.jitter.poll_next(waiter) {
+			self.released = true;
+			return Poll::Ready(Ok(Some(ready.item)));
+		}
+		let ended = match &self.track {
+			Some(_) => self.finished,
+			None => self.catalog.is_none(),
+		};
+		match ended && self.jitter.is_empty() {
+			true => Poll::Ready(Ok(None)),
+			false => Poll::Pending,
+		}
+	}
+
+	/// Subscribe to the track the catalog's `m2ts` section names, the first time it names one.
+	fn update_catalog(&mut self, catalog: &crate::catalog::hang::Catalog<()>) -> crate::Result<()> {
+		let Some(m2ts) = &catalog.m2ts else {
+			if self.track.is_none() && !(catalog.video.renditions.is_empty() && catalog.audio.renditions.is_empty()) {
+				return Err(anyhow::anyhow!(
+					"the broadcast is demultiplexed, with no `m2ts` passthrough track to release"
+				)
+				.into());
+			}
+			return Ok(());
+		};
+		match &self.track {
+			None => {
+				let source = ExportSource::for_stream(&self.source, &m2ts.track, self.delay / 2)?;
+				let source = match self.replay {
+					true => source,
+					false => source.live(),
+				};
+				self.track = Some((m2ts.track.clone(), source));
+			}
+			Some((name, _)) if *name != m2ts.track => {
+				return Err(anyhow::anyhow!("the `m2ts` track moved from `{name}` to `{}`", m2ts.track).into());
+			}
+			Some(_) => {}
+		}
+		Ok(())
+	}
+}