diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 198c614493..50c3cf8906 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 @@ -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 diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index 0bcfe61cfa..9afe02c023 100644 --- a/rs/moq-cli/src/args.rs +++ b/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, 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`. @@ -1075,6 +1079,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, @@ -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 { 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 8cc75f112a..63b56f11b8 100644 --- a/rs/moq-cli/src/main.rs +++ b/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, }; 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..4b9146a648 --- /dev/null +++ b/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>, + /// 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 { + 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 { + 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 => 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(()) + } +} 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..ba6073ac4a --- /dev/null +++ b/rs/moq-mux/src/container/ts/release_test.rs @@ -0,0 +1,353 @@ +//! 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) +} + +/// 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() + .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 (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); + 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 (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; + 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"); + } +} + +/// 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}"); +}