From 0da3a7736e80a7f15ebffb893343278b73d3868a Mon Sep 17 00:00:00 2001 From: t0ms Date: Mon, 12 Oct 2026 14:10:03 +0100 Subject: [PATCH 1/3] feat(mux): export ts --passthrough writes the m2ts track back on a fixed delay `moq export ts --passthrough --delay ` reads the track a broadcast's `m2ts` catalog section names and writes its packets back out unchanged, from the first group it releases. Each object (the packets from one PCR to the next) goes out `--delay` after its PCR time on the TS export's jitter buffer, so the output keeps the source's own PCR pacing and two exporters emit the same bytes. An object past its deadline is dropped and counted, leaving a continuity error; a flagged forward jump opens a new clock instead of being waited out. Co-authored-by: Claude Opus 5.5 --- doc/bin/cli.md | 10 +- rs/moq-cli/src/args.rs | 31 ++ rs/moq-cli/src/main.rs | 1 + rs/moq-cli/src/subscribe.rs | 40 +++ rs/moq-mux/src/container/ts/mod.rs | 7 +- rs/moq-mux/src/container/ts/release.rs | 189 +++++++++++ rs/moq-mux/src/container/ts/release_test.rs | 339 ++++++++++++++++++++ 7 files changed, 615 insertions(+), 2 deletions(-) create mode 100644 rs/moq-mux/src/container/ts/release.rs create mode 100644 rs/moq-mux/src/container/ts/release_test.rs diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 584d9719d1..b438035cdc 100644 --- a/doc/bin/cli.md +++ b/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 @@ -317,6 +317,14 @@ 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`. + 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 diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index 25ae9ccfbd..60db34927e 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -973,6 +973,7 @@ impl ExportSink { stitch: false, fragment_duration: None, mux_rate: None, + passthrough: false, }; Some(match self { Self::Fmp4(args) => Stdout { @@ -990,6 +991,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), @@ -1009,6 +1011,8 @@ pub struct Stdout { pub stitch: bool, pub fragment_duration: Option, pub mux_rate: Option, + /// Write the `m2ts` passthrough track back out (`ts` only). + pub passthrough: bool, } /// Options shared by the stdout container sinks other than `ts`. @@ -1073,6 +1077,12 @@ pub struct Transport { #[usage(long)] pub mux_rate: Option, + /// 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, @@ -1181,6 +1191,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 { let argv = ["moq", "--connect", "http://relay", "--broadcast", "room", "export"]; Invocation::try_parse_from(argv.iter().chain(flags).copied()) diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 6cc7dab1ab..3b82f49504 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -745,6 +745,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, }; diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 042d1742b5..238dcd2d7a 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -194,6 +194,9 @@ pub struct SubscribeArgs { /// overriding the catalog's recorded multiplex rate. pub mux_rate: Option, + /// 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, @@ -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(); diff --git a/rs/moq-mux/src/container/ts/mod.rs b/rs/moq-mux/src/container/ts/mod.rs index d660ba2766..b342a3f513 100644 --- a/rs/moq-mux/src/container/ts/mod.rs +++ b/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; diff --git a/rs/moq-mux/src/container/ts/release.rs b/rs/moq-mux/src/container/ts/release.rs new file mode 100644 index 0000000000..d71c5adfe3 --- /dev/null +++ b/rs/moq-mux/src/container/ts/release.rs @@ -0,0 +1,189 @@ +//! 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>, + /// The `m2ts` track, once a catalog names it. + track: Option<(String, ExportSource)>, + /// When the track was last found empty: an object read since arrived no earlier. + empty: Instant, + 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. + pub async fn new(source: crate::Source, catalog_format: CatalogFormat) -> crate::Result { + let catalog = source.catalog(catalog_format).await?; + let mut release = Self { + source, + catalog: Some(catalog), + track: None, + empty: Instant::now(), + 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 { + 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> { + kio::wait(|waiter| self.poll_next(waiter)).await + } + + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>> { + 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 => { + self.empty = Instant::now(); + break; + } + }; + let arrival = Arrival { + arrived: self.empty, + read: Instant::now(), + 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(()) + } +} diff --git a/rs/moq-mux/src/container/ts/release_test.rs b/rs/moq-mux/src/container/ts/release_test.rs new file mode 100644 index 0000000000..28f8bb0778 --- /dev/null +++ b/rs/moq-mux/src/container/ts/release_test.rs @@ -0,0 +1,339 @@ +//! Tests for [`Release`](super::Release). +//! +//! Each imports a hardware encoder's capture with [`Passthrough`] and releases the track it +//! publishes, either replayed whole or republished object by object on a paused clock. + +use std::time::Duration; + +use bytes::Bytes; +use tokio::time::Instant; + +use super::{Passthrough, Release}; +use crate::catalog::CatalogFormat; +use crate::container::Frame; + +const KYRION: &[u8] = include_bytes!("test_data/kyrion_mpeg2av_ac3.ts"); +const KYRION_H264: &[u8] = include_bytes!("test_data/scte35/kyrion_dirtystart.ts"); + +/// Longer than either capture, so a replay reads every group. +const RECORDING: Duration = Duration::from_secs(30); +/// How long a drain waits for the next object before taking the release to have stalled. +const STALL: Duration = RECORDING.saturating_mul(2); + +/// One object as published. +#[derive(Clone, Debug, PartialEq)] +struct Object { + group: u64, + micros: u64, + payload: Bytes, +} + +/// A capture imported whole, with what keeps its broadcast up. +struct Imported { + _passthrough: Passthrough, + _catalog: crate::catalog::Producer, + _broadcast: moq_net::broadcast::Producer, + consumer: moq_net::broadcast::Consumer, + objects: Vec, +} + +async fn import(data: &[u8]) -> Imported { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let consumer = broadcast.consume(); + let mut passthrough = Passthrough::new(broadcast.clone(), catalog.reserve()).unwrap(); + let name = passthrough.track().to_string(); + passthrough.decode(data).unwrap(); + passthrough.finish().unwrap(); + + let mut track = consumer + .track(&name) + .unwrap() + .subscribe(moq_net::track::Subscription::default().with_max_delay(RECORDING)) + .await + .unwrap(); + let mut objects = Vec::new(); + while let Some(mut group) = tokio::time::timeout(Duration::from_secs(1), track.recv_group()) + .await + .unwrap() + .unwrap() + { + while let Some(frame) = group.read_frame().await.unwrap() { + let frame = hang::container::Frame::decode(frame.payload).unwrap(); + objects.push(Object { + group: group.sequence, + micros: frame.timestamp.as_micros() as u64, + payload: frame.payload, + }); + } + } + Imported { + _passthrough: passthrough, + _catalog: catalog, + _broadcast: broadcast, + consumer, + objects, + } +} + +/// Republish `objects` on a new broadcast's `m2ts` track, in the groups they were published in, +/// each once `at` says it is due after the start and not before the one ahead of it, as a relay +/// delivering them would. +fn republish(objects: Vec, at: impl Fn(usize, &Object) -> Duration + Send + 'static) -> Republished { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let track = broadcast + .unique_track(".m2ts", catalog.track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let name = track.name().to_string(); + catalog + .mutate(|catalog| catalog.m2ts = Some(hang::catalog::M2ts::new(name))) + .unwrap(); + let consumer = broadcast.consume(); + let start = Instant::now(); + let task = tokio::spawn(async move { + let _keep = (&broadcast, &catalog); + let mut group: Option<(u64, moq_net::group::Producer)> = None; + let mut due = Duration::ZERO; + for (i, object) in objects.into_iter().enumerate() { + due = due.max(at(i, &object)); + tokio::time::sleep_until(start + due).await; + if group.as_ref().map(|(sequence, _)| *sequence) != Some(object.group) { + if let Some((_, group)) = group.take() { + group.finish().unwrap(); + } + let info = moq_net::group::Info { sequence: object.group }; + group = Some((object.group, track.create_group(info).unwrap())); + } + let frame = hang::container::Frame { + timestamp: moq_net::Timestamp::from_micros(object.micros).unwrap(), + payload: object.payload, + }; + frame.write_to(&mut group.as_mut().unwrap().1).unwrap(); + } + if let Some((_, group)) = group { + group.finish().unwrap(); + } + track.finish().unwrap(); + std::future::pending::<()>().await; + }); + Republished { consumer, task } +} + +struct Republished { + consumer: moq_net::broadcast::Consumer, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for Republished { + fn drop(&mut self) { + self.task.abort(); + } +} + +/// Every object `release` writes, and when, until it ends or stalls. +async fn drain(mut release: Release) -> (Vec<(Instant, Frame)>, Release) { + let mut released = Vec::new(); + while let Ok(next) = tokio::time::timeout(STALL, release.next()).await { + let Some(frame) = next.expect("release error") else { + break; + }; + released.push((Instant::now(), frame)); + } + (released, release) +} + +fn bytes<'a>(payloads: impl IntoIterator) -> Vec { + payloads + .into_iter() + .flat_map(|payload| payload.iter().copied()) + .collect() +} + +/// How much later than the first object each one went out, less how much later its PCR time is. +fn lag(released: &[(Instant, Frame)]) -> Vec { + let (start, first) = &released[0]; + released + .iter() + .map(|(at, frame)| { + let out = at.duration_since(*start).as_micros() as i128; + out - (frame.timestamp.as_micros() as i128 - first.timestamp.as_micros() as i128) + }) + .collect() +} + +#[tokio::test(start_paused = true)] +async fn a_capture_is_released_byte_identical_on_its_own_pcr() { + for data in [KYRION, KYRION_H264] { + let imported = import(data).await; + let source = crate::source::announced(&imported.consumer).await; + let release = Release::new(source, CatalogFormat::Hang) + .await + .unwrap() + .with_delay(RECORDING) + .with_replay(); + let (released, release) = drain(release).await; + + let out = bytes(released.iter().map(|(_, frame)| &frame.payload)); + assert!( + out == bytes(imported.objects.iter().map(|object| &object.payload)), + "every published object goes out, in order" + ); + assert!(data.ends_with(&out), "the output is the input from the first group on"); + assert_eq!(release.dropped(), 0); + for (i, lag) in lag(&released).into_iter().enumerate() { + assert!(lag.abs() < 1_000, "object {i} goes out {lag} us off its PCR time"); + } + } +} + +#[tokio::test(start_paused = true)] +async fn a_flagged_jump_crosses_without_holding_the_output() { + // The capture twice, the second copy's clock twenty seconds ahead, its first PCR flagged + // as the discontinuity. + const STEP: u64 = 20 * 90_000; + let mut second = KYRION.to_vec(); + let mut pcrs = second + .as_chunks_mut::<188>() + .0 + .iter_mut() + .filter(|pkt| pkt[3] & 0x20 != 0 && pkt[4] > 0 && pkt[5] & 0x10 != 0) + .peekable(); + pcrs.peek_mut().expect("a PCR")[5] |= 0x80; + for pkt in pcrs { + let base = pkt[6..11].iter().fold(0u64, |base, byte| base << 8 | u64::from(*byte)) >> 7; + let base = ((base + STEP) & ((1 << 33) - 1)) << 7 | u64::from(pkt[10] & 0x7f); + pkt[6..11].copy_from_slice(&base.to_be_bytes()[3..]); + } + let input = [KYRION, &second].concat(); + let objects = import(&input).await.objects; + assert!( + objects.windows(2).any(|pair| pair[1].group > pair[0].group + 1), + "the break skips a sequence" + ); + + // The bytes keep coming at the source's pace across the jump: only the PCR leaps. + let mut arrivals = vec![Duration::from_millis(40)]; + for pair in objects.windows(2) { + let step = Duration::from_micros(pair[1].micros - pair[0].micros).min(Duration::from_millis(100)); + arrivals.push(*arrivals.last().unwrap() + step); + } + let republished = republish(objects.clone(), move |i, _| arrivals[i]); + let source = crate::source::announced(&republished.consumer).await; + let release = Release::new(source, CatalogFormat::Hang).await.unwrap(); + let delay = Duration::from_millis(500); + let (released, release) = drain(release.with_delay(delay)).await; + + let out = bytes(released.iter().map(|(_, frame)| &frame.payload)); + assert!( + out == bytes(objects.iter().map(|object| &object.payload)), + "both sides of the jump go out, byte for byte" + ); + assert_eq!(release.dropped(), 0); + let (jump, step) = released + .windows(2) + .enumerate() + .map(|(i, pair)| { + ( + i, + pair[1].1.timestamp.as_micros() as i128 - pair[0].1.timestamp.as_micros() as i128, + ) + }) + .max_by_key(|(_, step)| *step) + .unwrap(); + assert!(step > 10_000_000, "the timestamps carry the jump"); + let pause = released[jump + 1].0.duration_since(released[jump].0); + assert!(pause <= delay, "the output paused {pause:?} at the jump"); +} + +#[tokio::test(start_paused = true)] +async fn releases_fed_with_different_skew_write_the_same_bytes() { + let objects = import(KYRION).await.objects; + let first = objects[0].micros; + let pcr = move |object: &Object| Duration::from_micros(object.micros - first); + // Each object is complete once the next one starts, about a PCR interval after its own PCR. + let steady = republish(objects.clone(), move |_, object| { + pcr(object) + Duration::from_millis(40) + }); + let skewed = republish(objects.clone(), move |i, object| { + pcr(object) + Duration::from_millis(40 + 30 * (i as u64 % 3)) + }); + + let release = |consumer| async move { + let source = crate::source::announced(consumer).await; + let release = Release::new(source, CatalogFormat::Hang).await.unwrap(); + drain(release.with_delay(Duration::from_millis(500))).await + }; + let ((steady_out, steady_release), (skewed_out, skewed_release)) = + tokio::join!(release(&steady.consumer), release(&skewed.consumer)); + + assert_eq!(steady_release.dropped(), 0); + assert_eq!(skewed_release.dropped(), 0); + let payloads = |released: &[(Instant, Frame)]| bytes(released.iter().map(|(_, frame)| &frame.payload)); + assert!(payloads(&steady_out) == payloads(&skewed_out), "the same bytes go out"); + assert!( + payloads(&steady_out) == bytes(objects.iter().map(|object| &object.payload)), + "from the first group" + ); + for released in [&steady_out, &skewed_out] { + for (i, lag) in lag(released).into_iter().enumerate() { + assert!(lag.abs() < 1_000, "object {i} goes out {lag} us off its PCR time"); + } + } +} + +#[tokio::test(start_paused = true)] +async fn an_object_past_its_deadline_is_a_gap_in_that_release_only() { + let objects = import(KYRION).await.objects; + let first = objects[0].micros; + let pcr = move |object: &Object| Duration::from_micros(object.micros - first); + let steady = republish(objects.clone(), move |_, object| { + pcr(object) + Duration::from_millis(40) + }); + // The path stalls for 600 ms from 0.5 s in, then delivers what it held at once. + let stalled = republish(objects.clone(), move |_, object| { + let at = pcr(object) + Duration::from_millis(40); + match at >= Duration::from_millis(500) && at < Duration::from_millis(1100) { + true => Duration::from_millis(1100), + false => at, + } + }); + + let release = |consumer| async move { + let source = crate::source::announced(consumer).await; + let release = Release::new(source, CatalogFormat::Hang).await.unwrap(); + drain(release.with_delay(Duration::from_millis(500))).await + }; + let ((steady_out, steady_release), (stalled_out, stalled_release)) = + tokio::join!(release(&steady.consumer), release(&stalled.consumer)); + + assert_eq!(steady_release.dropped(), 0); + let dropped = stalled_release.dropped() as usize; + assert!(dropped > 0, "the stall makes objects late"); + assert_eq!( + stalled_out.len() + dropped, + steady_out.len(), + "every late object is counted" + ); + + // The stalled release writes the steady one's objects less a run of the late ones. + let gap = steady_out + .iter() + .zip(&stalled_out) + .position(|((_, steady), (_, stalled))| steady.payload != stalled.payload) + .expect("a gap"); + let frames = |released: &[(Instant, Frame)]| -> Vec<(u64, Bytes)> { + released + .iter() + .map(|(_, frame)| (frame.timestamp.as_micros() as u64, frame.payload.clone())) + .collect() + }; + let mut expected = frames(&steady_out); + expected.drain(gap..gap + dropped); + assert!(frames(&stalled_out) == expected, "only the late objects are missing"); + + // The rest still go out on time: on the same clock as before the stall. + for (i, lag) in lag(&stalled_out).into_iter().enumerate() { + assert!(lag.abs() < 1_000, "object {i} goes out {lag} us off its PCR time"); + } +} From 6d8dc5a2ab4785e735b856ee090490a2e57f7351 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 12 Oct 2026 18:51:45 -0700 Subject: [PATCH 2/3] fix(ts): time passthrough arrivals by their read, and refuse the MSF catalog A release parked on the track judged an object that landed past its deadline by the last empty poll, so it went out late instead of being dropped and counted. The stall test missed it because tokio::join! polled the stalled release whenever the steady one woke; each release now drains in its own task. The MSF catalog carries no m2ts section, so a release reading it waited forever. Release::new now refuses it. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 3 +- rs/moq-mux/src/container/ts/release.rs | 20 +++++----- rs/moq-mux/src/container/ts/release_test.rs | 42 ++++++++++++++------- 3 files changed, 41 insertions(+), 24 deletions(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index b438035cdc..c869623f64 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -323,7 +323,8 @@ 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`. +`--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 diff --git a/rs/moq-mux/src/container/ts/release.rs b/rs/moq-mux/src/container/ts/release.rs index d71c5adfe3..32415ff44a 100644 --- a/rs/moq-mux/src/container/ts/release.rs +++ b/rs/moq-mux/src/container/ts/release.rs @@ -26,8 +26,6 @@ pub struct Release { catalog: Option>, /// The `m2ts` track, once a catalog names it. track: Option<(String, ExportSource)>, - /// When the track was last found empty: an object read since arrived no earlier. - empty: Instant, finished: bool, /// An object has gone out. released: bool, @@ -38,13 +36,17 @@ pub struct Release { 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 { + 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, - empty: Instant::now(), finished: false, released: false, delay: Duration::ZERO, @@ -123,14 +125,14 @@ impl Release { break; } Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), - Poll::Pending => { - self.empty = Instant::now(); - break; - } + 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: self.empty, - read: Instant::now(), + arrived: read, + read, decode: frame.timestamp, restart: track.markers(), skip: track.skips(), diff --git a/rs/moq-mux/src/container/ts/release_test.rs b/rs/moq-mux/src/container/ts/release_test.rs index 28f8bb0778..ba6073ac4a 100644 --- a/rs/moq-mux/src/container/ts/release_test.rs +++ b/rs/moq-mux/src/container/ts/release_test.rs @@ -143,6 +143,17 @@ async fn drain(mut release: Release) -> (Vec<(Instant, Frame)>, Release) { (released, release) } +/// Drain a release of `republished` with a 500 ms delay, in a task of its own so that no other +/// release waking polls it. +fn spawn_release(republished: &Republished) -> tokio::task::JoinHandle<(Vec<(Instant, Frame)>, Release)> { + let consumer = republished.consumer.clone(); + tokio::spawn(async move { + let source = crate::source::announced(&consumer).await; + let release = Release::new(source, CatalogFormat::Hang).await.unwrap(); + drain(release.with_delay(Duration::from_millis(500))).await + }) +} + fn bytes<'a>(payloads: impl IntoIterator) -> Vec { payloads .into_iter() @@ -259,13 +270,9 @@ async fn releases_fed_with_different_skew_write_the_same_bytes() { pcr(object) + Duration::from_millis(40 + 30 * (i as u64 % 3)) }); - let release = |consumer| async move { - let source = crate::source::announced(consumer).await; - let release = Release::new(source, CatalogFormat::Hang).await.unwrap(); - drain(release.with_delay(Duration::from_millis(500))).await - }; - let ((steady_out, steady_release), (skewed_out, skewed_release)) = - tokio::join!(release(&steady.consumer), release(&skewed.consumer)); + let (steady_task, skewed_task) = (spawn_release(&steady), spawn_release(&skewed)); + let (steady_out, steady_release) = steady_task.await.unwrap(); + let (skewed_out, skewed_release) = skewed_task.await.unwrap(); assert_eq!(steady_release.dropped(), 0); assert_eq!(skewed_release.dropped(), 0); @@ -299,13 +306,9 @@ async fn an_object_past_its_deadline_is_a_gap_in_that_release_only() { } }); - let release = |consumer| async move { - let source = crate::source::announced(consumer).await; - let release = Release::new(source, CatalogFormat::Hang).await.unwrap(); - drain(release.with_delay(Duration::from_millis(500))).await - }; - let ((steady_out, steady_release), (stalled_out, stalled_release)) = - tokio::join!(release(&steady.consumer), release(&stalled.consumer)); + let (steady_task, stalled_task) = (spawn_release(&steady), spawn_release(&stalled)); + let (steady_out, steady_release) = steady_task.await.unwrap(); + let (stalled_out, stalled_release) = stalled_task.await.unwrap(); assert_eq!(steady_release.dropped(), 0); let dropped = stalled_release.dropped() as usize; @@ -337,3 +340,14 @@ async fn an_object_past_its_deadline_is_a_gap_in_that_release_only() { assert!(lag.abs() < 1_000, "object {i} goes out {lag} us off its PCR time"); } } + +/// The MSF catalog never names the `m2ts` track, so a release reading it would wait forever. +#[tokio::test(start_paused = true)] +async fn an_msf_catalog_is_refused() { + let imported = import(KYRION).await; + let source = crate::source::announced(&imported.consumer).await; + let Err(err) = Release::new(source, CatalogFormat::Msf).await else { + panic!("a release reading the MSF catalog must fail"); + }; + assert!(err.to_string().contains("MSF"), "{err}"); +} From 42047fbfbc3f5ac0676e2c607665c32943750499 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 12 Oct 2026 19:13:17 -0700 Subject: [PATCH 3/3] style(ts): format the MSF refusal Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/release.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/rs/moq-mux/src/container/ts/release.rs b/rs/moq-mux/src/container/ts/release.rs index 32415ff44a..4b9146a648 100644 --- a/rs/moq-mux/src/container/ts/release.rs +++ b/rs/moq-mux/src/container/ts/release.rs @@ -40,7 +40,9 @@ impl Release { /// Fails for [`CatalogFormat::Msf`], which has no `m2ts` section. pub async fn new(source: crate::Source, catalog_format: CatalogFormat) -> crate::Result { if catalog_format == CatalogFormat::Msf { - return Err(anyhow::anyhow!("the MSF catalog has no `m2ts` passthrough track; read the hang catalog").into()); + 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 {