diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 584d9719d1..25a953b54c 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -309,6 +309,8 @@ See [Authentication](/bin/relay/auth). fetchable, which the [HLS gateway](/bin/hls) depends on. `export --max-delay` (default 500 ms) is how far a stalled group may fall behind the live edge before *this* consumer skips it. Raising the first never delays playback. +At stdin EOF, `import` keeps the finished broadcast up for its max-age, so a +subscriber still catching up can drain it; Ctrl-C ends that early. `export ts` takes `--delay` (default 500 ms) instead, like an SRT receiver's latency: each frame is muxed that long after its decode time, all tracks in diff --git a/quest/m1/README.md b/quest/m1/README.md index 17b74156dc..0e65570470 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -192,6 +192,7 @@ session outcomes, and embedder refusals inside session outcomes. - [moq fetch closes cleanly](/quest/m1/moq-fetch-close.md) - `moq fetch` closes its session before exiting instead of leaving the relay to time it out - [Terminal dial errors](/quest/m1/moq-tokio-terminal-errors.md) - moq-tokio's reconnect loop stops on errors that can never succeed - [Plan: HLS export failure signal](/quest/m1/hls-failure-signal.md) - decide what players see when the exporter's timeline fails, live or history, so a truncated recording never looks complete +- [Publisher linger](/quest/m1/publisher-linger.md) - every finishing publisher keeps its broadcast up for the media max-age so subscribers can drain it ## Related diff --git a/quest/m1/publisher-linger.md b/quest/m1/publisher-linger.md new file mode 100644 index 0000000000..f605d206f0 --- /dev/null +++ b/quest/m1/publisher-linger.md @@ -0,0 +1,31 @@ +# [M] A finishing publisher lingers for max-age + +## Goal + +Every publisher that ends a broadcast cleanly keeps it up for the media +max-age afterwards, so a subscriber still catching up (one that read the +final catalog late, or is draining a backlog) can request and drain the +finished tracks. Cached content stays servable until it is stale. + +## Plan + +Decided (2026-10-11): a retention rule, not a race-hiding sleep. The bound is +the tracks' max-age, never a new constant, and an interrupt (SIGINT/SIGTERM, +or the library equivalent) ends it early. + +`moq import` from stdin already does this (#5287). Extend the rule +to the other finishing publishers: the CLI's gateway imports (HLS, RTMP, SRT, +RTC), `moq-gst` sinks, the FFI and its bindings, and `@moq/publish`. + +Decide where the rule lives. Recommended: the library, so a finished +`broadcast::Producer` (or the session close that follows it) stays servable +for its tracks' max-age without each publisher reimplementing it. Then +`moq import` drops its own linger. Report the public API shape before +landing it. + +Test with a mocked clock: a subscriber that requests a finished track within +max-age gets its media, and the publisher closes once max-age passes. + +## Related + +- [One max_age meaning](/quest/m1/cache-max-age.md) - what max-age means for a superseded group diff --git a/quest/m1/ts-live-clock.md b/quest/m1/ts-live-clock.md index 61015b7884..e4cc47aef5 100644 --- a/quest/m1/ts-live-clock.md +++ b/quest/m1/ts-live-clock.md @@ -55,5 +55,5 @@ public API and wire changes are not expected. - [Passing full CI](https://github.com/moq-dev/moq/actions/runs/38178905198) - same-head contrast, including real live timing - [Same-instance resume](/quest/m1/ts-same-instance-resume.md) - independent full-check failure - [Transport upgrade](/quest/m1/transport-upgrade/README.md) - existing session handover and epoch contract -- [Mux-rate hold](/quest/m1/tstd/mux-rate-hold.md) - #5287 owns initial rate publication and its finite-source behavior +- [Mux-rate hold #5287](https://github.com/moq-dev/moq/pull/5287) - initial rate publication, and a finished import lingering for its max-age - [TS AC-3 scheduling](/quest/m1/ts-atsc-ac3.md) - #5306 owns decoder deadlines and unknown-rate startup overflow diff --git a/quest/m1/tstd/README.md b/quest/m1/tstd/README.md index 16ddc27252..aeeaf2754e 100644 --- a/quest/m1/tstd/README.md +++ b/quest/m1/tstd/README.md @@ -36,7 +36,6 @@ nightly. ## Required - [Send-ahead within the delay](/quest/m1/tstd/send-ahead.md) - total lag is `--delay`, send-ahead included, with a 1 s default -- [Mux-rate hold](/quest/m1/tstd/mux-rate-hold.md) - import publishes its catalog once the mux rate is measured, so export is constant-rate from the start - [PSI runs](/quest/m1/tstd/psi-runs.md) - the PAT and PMT are deduplicated and paced so the system buffers never overflow ## Related diff --git a/quest/m1/tstd/mux-rate-hold.md b/quest/m1/tstd/mux-rate-hold.md deleted file mode 100644 index 62fd283763..0000000000 --- a/quest/m1/tstd/mux-rate-hold.md +++ /dev/null @@ -1,35 +0,0 @@ -# [S] TS import publishes its catalog once the mux rate is measured - -## Goal - -A TS→MoQ→TS export is constant-rate from its first packet. `moq import ts` -holds the catalog until its mux-rate meter settles, so `mpegts.muxRate` is -there from the start rather than appearing about 2 s in. A VBR source that -never settles is published without it once the window closes. - -## Plan - -Decided (2026-10-01). Found in the fixed-delay release -(#4645): the importer's meter (`rs/moq-mux/src/container/ts/mux_rate.rs`) -publishes `mpegts.muxRate` only after a 2 s window agrees. So the export's -first ~2 s go out unpadded, and the harness skips them (`pcr-timing.py` -grades from the first null packet). - -- Hold the whole catalog until the meter settles, or until one `WINDOW` of - PCR time has passed with no rate published, whichever is first: a - one-time ~2 s start-up cost at ingest. The meter itself keeps rolling and - never times out, so the hold needs that explicit bound. Media written in - the meantime is still published. -- A source whose rate never settles (VBR) publishes after that window - without `muxRate`, and the export stays unpadded, as today. -- End of input releases a held catalog without `muxRate` before - `catalog.finish` closes it: a TS shorter than one window never completes - a measurement, and finishing would otherwise close the catalog track with - the reservation unpublished, leaving media with no catalog. -- Remove the harness's "grade from the first null packet" skip, and the - export's grace for a rate that arrives mid-stream, if nothing else needs - them. -- Test: an imported CBR source's first catalog carries `muxRate`, a VBR - source's catalog arrives after the window without it, and a CBR import - shorter than 2 s still publishes its catalog. A TS→TS export passes - `pcr-schedule` from its first PCR. diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index 25ae9ccfbd..8a67d6f6ab 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -709,7 +709,8 @@ pub struct Import { /// wait for a late group). The default suits a segmented egress (HLS/DASH), which may only /// advertise segments that are still fetchable; lower it when nothing reads history and the /// memory matters. Media tracks only -- the catalog and timeline are read at the live edge, - /// which is retained unconditionally. + /// which is retained unconditionally. A stdin import also stays up this long after EOF, so + /// a subscriber still catching up can drain the finished broadcast. #[usage(long)] pub max_age: Option, diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 6cc7dab1ab..8cc75f112a 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -662,6 +662,8 @@ fn spawn_import( let broadcast = origin.create_broadcast(&name).context("failed to create broadcast")?; Publish::new(broadcast, &format, config)? }; + // Media stays servable at EOF until it goes stale, the same max-age relays retain it for. + let publish = publish.linger(max_age.unwrap_or(hang::container::MAX_AGE)); publish.announce(epoch)?; local = Some(publish); } else { diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index 6dc2b6b759..c27ccff3a4 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -290,6 +290,8 @@ pub struct Publish { // tracks and importers don't. `None` for `import ts --program all`, whose // programs each keep and announce their own. broadcast: Option, + // How long a finished stdin import stays up; see [`Self::linger`]. + linger: std::time::Duration, } impl Publish { @@ -314,6 +316,7 @@ impl Publish { catalog: PublishCatalog::Ts(catalog), }, broadcast: Some(broadcast), + linger: Default::default(), }); } @@ -354,6 +357,7 @@ impl Publish { catalog: PublishCatalog::Media(catalog), }, broadcast: Some(broadcast), + linger: Default::default(), }) } @@ -369,6 +373,7 @@ impl Publish { catalog: PublishCatalog::TsPrograms, }, broadcast: None, + linger: Default::default(), } } @@ -407,9 +412,18 @@ impl Publish { audio, }, broadcast: Some(broadcast), + linger: Default::default(), }) } + /// Keep a finished stdin import up for `linger` after EOF, so a subscriber that learns the + /// final catalog late can still request and drain its tracks. Pass the media max-age: the + /// content is servable until it goes stale. SIGINT or SIGTERM ends it early. + pub fn linger(mut self, linger: std::time::Duration) -> Self { + self.linger = linger; + self + } + /// Advertise the broadcast's path under `epoch`, now that the catalog tracks are in place. pub fn announce(&self, epoch: moq_net::Epoch) -> anyhow::Result<()> { let Some(broadcast) = &self.broadcast else { @@ -423,7 +437,7 @@ impl Publish { /// Drive the source until stdin EOF (or the capture devices stop). pub async fn run(self) -> anyhow::Result<()> { match self.source { - Source::Stream { decoder, catalog } => decode(decoder, catalog, tokio::io::stdin()).await, + Source::Stream { decoder, catalog } => decode(decoder, catalog, tokio::io::stdin(), self.linger).await, #[cfg(feature = "capture")] Source::Capture { broadcast, @@ -481,10 +495,12 @@ impl Publish { /// At EOF the media tracks finish, then the catalog does, while it still lists them: the /// renditions retire from the catalog as the decoder drops, which a subscriber would read /// as removed tracks, and a catalog dropped unfinished reads as a publisher that vanished. +/// The finished tracks then stay servable for `linger`. async fn decode( mut decoder: PublishDecoder, mut catalog: PublishCatalog, mut input: impl tokio::io::AsyncRead + Unpin, + linger: std::time::Duration, ) -> anyhow::Result<()> { // An empty BytesMut grows to only 64 bytes and clearing it keeps that capacity. // Stdin dispatches every read to a blocking thread, so those tiny reads can @@ -529,7 +545,18 @@ async fn decode( log.finish(&stats); } match outcome { - Ok(()) => catalog.finish(), + Ok(()) => { + catalog.finish()?; + // The decoder and catalog own the tracks, so holding them keeps the finished + // broadcast servable: a subscriber that saw the final catalog only at EOF still has + // to request its media. + tracing::info!( + ?linger, + "input ended; serving the finished broadcast until its media is stale" + ); + tokio::time::sleep(linger).await; + Ok(()) + } Err(err) => { decoder.abort(moq_net::Error::Transport(err.to_string())); Err(err) @@ -1190,7 +1217,7 @@ pub(crate) mod tests { remaining: BBB, reads: 0, }; - decode(decoder, catalog, &mut input).await.unwrap(); + decode(decoder, catalog, &mut input, Duration::ZERO).await.unwrap(); assert_eq!( input.reads, BBB.len().div_ceil(188) + 1, @@ -1223,7 +1250,7 @@ pub(crate) mod tests { panic!("expected a stream source"); }; let start = std::time::Instant::now(); - decode(decoder, catalog, input).await.unwrap(); + decode(decoder, catalog, input, Duration::ZERO).await.unwrap(); let elapsed = start.elapsed(); let bytes = INPUT.len() * copies; println!( @@ -1246,7 +1273,7 @@ pub(crate) mod tests { let Source::Stream { decoder, catalog } = publish.source else { panic!("expected a stream source"); }; - decode(decoder, catalog, BBB).await.unwrap(); + decode(decoder, catalog, BBB, Duration::ZERO).await.unwrap(); drop(publish.broadcast); let mut last = None; @@ -1264,6 +1291,51 @@ pub(crate) mod tests { assert_eq!(last.audio.renditions.len(), 1, "the audio rendition is still listed"); } + /// After EOF the finished tracks stay servable for the linger, so a subscriber that reads + /// the final catalog only then still gets the media, and the import returns once it ends. + #[tokio::test(start_paused = true)] + async fn a_finished_import_serves_its_tracks_for_the_linger() { + const LINGER: Duration = Duration::from_secs(30); + let broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let publish = Publish::new(broadcast, &PublishFormat::Ts { program: None }, Default::default()).unwrap(); + let mut catalogs = hang::catalog::Catalog::<()>::subscribe(&consumer).await.unwrap(); + + #[allow(irrefutable_let_patterns)] + let Source::Stream { decoder, catalog } = publish.source else { + panic!("expected a stream source"); + }; + let started = tokio::time::Instant::now(); + let mut import = std::pin::pin!(decode(decoder, catalog, BBB, LINGER)); + let name = tokio::select! { + res = &mut import => panic!("the import returned before its linger: {res:?}"), + name = async { + loop { + let catalog = catalogs.next().await.unwrap().expect("a catalog before the finish"); + if let Some(name) = catalog.video.renditions.keys().next() { + break name.clone(); + } + } + } => name, + }; + + // Subscribe only now, as a subscriber that learned the catalog at EOF does. + let track = consumer.track(&name).unwrap().subscribe(None); + let frame = tokio::select! { + res = &mut import => panic!("the import returned before its linger: {res:?}"), + frame = async { + let mut track = track.await.unwrap(); + let mut group = track.recv_group().await.unwrap().expect("a video group"); + group.read_frame().await.unwrap() + } => frame, + }; + assert!(frame.is_some(), "the finished track still serves its media"); + + import.await.unwrap(); + assert!(started.elapsed() >= LINGER, "returned after {:?}", started.elapsed()); + drop(publish.broadcast); + } + /// A PAT listing two programs, then one MP2 PES of each: program 1 on PID `0x61` at 1 s /// with fill bytes `0xAA`/`0xBB`, program 2 on PID `0x71` an hour later with `0xCC`/`0xDD`. pub(crate) fn two_programs() -> Vec { diff --git a/rs/moq-cli/tests/export.rs b/rs/moq-cli/tests/export.rs index 3108c9a1ad..2fb0fd58c6 100644 --- a/rs/moq-cli/tests/export.rs +++ b/rs/moq-cli/tests/export.rs @@ -111,7 +111,8 @@ fn import(relay: &Relay) -> (Child, JoinHandle) { /// [`import`] with `global` flags, such as the session version or the epoch. fn import_with(relay: &Relay, global: &[&str]) -> (Child, JoinHandle) { let mut child = moq(relay, global) - .args(["import", "ts"]) + // A finished import lingers for its max-age; keep that short so the publisher exits promptly. + .args(["import", "--max-age", "1s", "ts"]) .stdin(Stdio::piped()) .spawn() .expect("spawn import"); @@ -218,7 +219,7 @@ async fn an_export_failure_with_the_broadcast_up_exits_without_lingering() { let (mut export, _output) = export(&relay, "20s"); // TS cannot carry AV1, so the export fails on the catalog while the publisher stays up. - let mut publisher = moq(&relay, &["import", "fmp4"]) + let mut publisher = moq(&relay, &["import", "--max-age", "1s", "fmp4"]) .stdin(Stdio::piped()) .spawn() .expect("spawn import"); diff --git a/rs/moq-cli/tests/import.rs b/rs/moq-cli/tests/import.rs index 69bd6c3cbe..dc7552a1c5 100644 --- a/rs/moq-cli/tests/import.rs +++ b/rs/moq-cli/tests/import.rs @@ -8,7 +8,10 @@ use tokio::io::AsyncWriteExt; use tokio::process::Child; const TIMEOUT: Duration = Duration::from_secs(10); -const BBB: &[u8] = include_bytes!("../../moq-mux/src/container/ts/test_data/bbb.ts"); +/// How long a finished import stays up: long enough for a subscriber to request its media +/// after reading the final catalog, short enough that the process exits within `TIMEOUT`. +const MAX_AGE: &str = "5s"; +const BBB: &[u8] = include_bytes!("../../moq-mux/src/container/ts/test_data/scte35/bbb5s.ts"); /// Await `step` while `moq` still holds stdin open, failing as soon as the process exits. /// @@ -22,14 +25,8 @@ async fn running(child: &mut Child, what: &str, step: impl Future } } -#[tokio::test] -async fn import_delivers_the_catalog_finish_at_eof() { - let _ = moq_tokio::crypto::install_default(); - let fixture = moq_relay::test_relay().await.expect("test relay"); - let ready = fixture.relay.ready(); - tokio::spawn(fixture.relay.run()); - ready.wait().await.expect("relay ready"); - +/// Start `moq import ts`, publishing `demo` from stdin to the relay at `url`. +fn import(url: &str, fingerprint: &str) -> Child { let mut command = tokio::process::Command::new(env!("CARGO_BIN_EXE_moq")); // A `MOQ_*` variable in the developer's shell must not reconfigure the child. for (name, _) in std::env::vars_os() { @@ -37,22 +34,36 @@ async fn import_delivers_the_catalog_finish_at_eof() { command.env_remove(name); } } - let mut child = command + command .args([ "--connect", - fixture.url.as_str(), + url, "--connect-tls-fingerprint", - &fixture.fingerprint, + fingerprint, "--broadcast", "demo", "import", + "--max-age", + MAX_AGE, "ts", ]) .stdin(Stdio::piped()) .kill_on_drop(true) .spawn() - .expect("spawn moq"); + .expect("spawn moq") +} + +#[tokio::test] +async fn import_delivers_the_catalog_finish_at_eof() { + let _ = moq_tokio::crypto::install_default(); + let fixture = moq_relay::test_relay().await.expect("test relay"); + let ready = fixture.relay.ready(); + tokio::spawn(fixture.relay.run()); + ready.wait().await.expect("relay ready"); + + let mut child = import(fixture.url.as_str(), &fixture.fingerprint); let mut stdin = child.stdin.take().expect("stdin"); + // A live TS catalog waits for its initial two-second PCR rate measurement. stdin.write_all(BBB).await.expect("write stdin"); let origin = moq_tokio::origin::spawn(); @@ -110,3 +121,90 @@ async fn import_delivers_the_catalog_finish_at_eof() { } } } + +/// About one second of CBR HEVC and Opus: shorter than the two-second mux-rate window, so +/// the catalog is only released at EOF. +const SHORT: &[u8] = include_bytes!("../../moq-mux/src/container/ts/test_data/bbb_cbr.ts") + .split_at(256 * 188) + .0; + +/// A subscriber already waiting when a short import ends still receives its media: the +/// publisher keeps the finished tracks servable for their max-age after EOF. +#[tokio::test] +async fn a_short_import_drains_its_media_to_a_waiting_subscriber() { + let _ = moq_tokio::crypto::install_default(); + let fixture = moq_relay::test_relay().await.expect("test relay"); + let ready = fixture.relay.ready(); + tokio::spawn(fixture.relay.run()); + ready.wait().await.expect("relay ready"); + + let mut child = import(fixture.url.as_str(), &fixture.fingerprint); + let mut stdin = child.stdin.take().expect("stdin"); + + let origin = moq_tokio::origin::spawn(); + let consumer = origin.consume(); + let mut announced = consumer.announced(); + let mut config = moq_tokio::connect::Config::default(); + config.tls.insecure = Some(true); + config.bind = Some("127.0.0.1:0".parse().unwrap()); + let _connection = config + .init(Default::default()) + .expect("client") + .with_subscriber(origin) + .with_reconnect(false) + .connect(fixture.url.clone()) + .established() + .await + .expect("subscriber connects"); + + running(&mut child, "the announce", announced.next()) + .await + .expect("origin closed"); + let broadcast = running(&mut child, "the broadcast", consumer.request_broadcast("demo", None)) + .await + .expect("announced broadcast resolves"); + let mut catalogs = running( + &mut child, + "the catalog subscription", + hang::catalog::Catalog::<()>::subscribe(&broadcast), + ) + .await + .expect("subscribe to the catalog"); + + stdin.write_all(SHORT).await.expect("write stdin"); + drop(stdin); + + let catalog = tokio::time::timeout(TIMEOUT, catalogs.next()) + .await + .expect("no catalog") + .expect("catalog read") + .expect("the import released its catalog"); + let name = catalog + .video + .renditions + .keys() + .next() + .expect("a video rendition") + .clone(); + let mut track = tokio::time::timeout(TIMEOUT, broadcast.track(&name).expect("track").subscribe(None)) + .await + .expect("the video subscription timed out") + .expect("subscribe to the video"); + let mut frames = 0; + while let Some(mut group) = tokio::time::timeout(TIMEOUT, track.recv_group()) + .await + .expect("the video never ended") + .expect("video group") + { + while group.read_frame().await.expect("video frame").is_some() { + frames += 1; + } + } + assert!(frames > 0, "the subscriber received no video"); + + let status = tokio::time::timeout(TIMEOUT, child.wait()) + .await + .expect("moq never exited") + .expect("wait for moq"); + assert!(status.success(), "moq exited with {status}"); +} diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 71d7d8acb7..4edb6a118c 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -49,6 +49,12 @@ use moq_net::Timestamp; /// fails is dropped and counted in [`stats::Snapshot::crc_error`]; the last good table stays /// in force. /// +/// With the `mpegts` catalog extension, the initial catalog waits for a two-second PCR +/// measurement window. This hold applies to every subscriber of the broadcast. +/// A variable-rate window publishes without a rate, as does four seconds of media without a +/// complete PCR window; EOF releases a shorter input too. +/// Imports without that extension do not wait for a rate measurement. +/// /// The selected container applies only to decoded media renditions. Verbatim tracks in the /// `mpegts` catalog section continue to use the legacy Hang container. pub struct Import { @@ -61,6 +67,14 @@ pub struct Import { /// final. A one-shot muxer (fMP4, TS re-export) sees the complete track list in the first /// snapshot rather than a half-converged one. See [`Reserved`](crate::catalog::Reserved). initial_reservation: Option>, + /// Hold the whole catalog until one PCR measurement window completes, or enough media time + /// passes without one, independently of decoder readiness. + mux_rate_reservation: Option>, + /// Media time seen while that hold is active, summed across time-base breaks so a source + /// that keeps resetting its clock still reaches the bound. + held: std::time::Duration, + /// The newest 90 kHz PTS of the current time base counted into `held`. + held_edge: Option, /// The program's timestamp base: the first PES with a PTS anchors it, and every PTS shifts by /// its offset onto the catalog clock. @@ -157,12 +171,16 @@ impl Import { let mut snapshot = catalog.snapshot(); let supports_mpegts = snapshot.ext.mpegts_mut().is_some(); let si = super::si::Capture::new(broadcast.clone(), catalog.clone()); + let mux_rate_reservation = supports_mpegts.then(|| reserved.clone()); Self { broadcast, catalog, container, timebase: reserved.timebase(), initial_reservation: Some(reserved), + mux_rate_reservation, + held: std::time::Duration::ZERO, + held_edge: None, pat: PatReader::default(), pmt_sections: HashMap::new(), crc_errors: BTreeMap::new(), @@ -285,8 +303,17 @@ impl Import { if let Some(now) = self.liveness.now() { self.health.tick(now); } - if self.supports_mpegts && self.mux_rate.pcr(pcr) { - self.record_mux_rate()?; + if self.supports_mpegts { + let ready = self.mux_rate.ready(); + if self.mux_rate.pcr(pcr) { + self.record_mux_rate()?; + } + if self.mux_rate.ready() { + self.mux_rate_reservation = None; + } else if ready { + // A backwards clock voided the measurement. + self.rearm_mux_rate(); + } } } } @@ -403,9 +430,7 @@ impl Import { self.health.pcr_pid(pmt.pcr_pid.map(|pid| pid.as_u16())); // A new clock: the intervals straddling the switch measure nothing. self.liveness.discontinuity(); - if self.mux_rate.discontinuity() { - self.record_mux_rate()?; - } + self.reset_mux_rate()?; } // SCTE-35 is announced by a program-level registration descriptor with @@ -688,6 +713,9 @@ impl Import { // rather than its flush keeps the section clock below on the same offset as the media. let offset = match pes.pts { Some(pts) => { + if self.mux_rate_reservation.is_some() { + self.hold_media(pts); + } let pts = Timestamp::from_scale(pts, 90_000)?; let offset = self.timebase.anchor(pts)?; if self.start_pts.is_none() { @@ -848,6 +876,8 @@ impl Import { /// PCR/PTS rollover sets nothing and is unwrapped rather than reset; neither is a program /// break and neither reaches here. fn timebase_break(&mut self) -> anyhow::Result<()> { + // First, so the salvage below cannot publish the first catalog with the old clock's rate. + self.rearm_mux_rate(); // The bytes still accumulating belong to the old clock, and their continuation is // stamped on the new one, so cut every PES here rather than splicing across. This // runs before the marker so a salvaged tail lands in the closing group, which the @@ -865,6 +895,7 @@ impl Import { self.media_unwrap.discontinuity(); self.last_pts = None; self.start_pts = None; + self.held_edge = None; self.published = false; self.liveness.discontinuity(); if self.mux_rate.discontinuity() { @@ -874,6 +905,43 @@ impl Import { Ok(()) } + /// The program clock changed, so the measured rate no longer holds. + fn reset_mux_rate(&mut self) -> anyhow::Result<()> { + self.rearm_mux_rate(); + if self.mux_rate.discontinuity() { + self.record_mux_rate()?; + } + Ok(()) + } + + /// A clock change voids the startup measurement, so until the first catalog publishes + /// (while the initial reservation still holds it), hold it again for a window on the new + /// clock. + fn rearm_mux_rate(&mut self) { + if self.supports_mpegts && self.mux_rate_reservation.is_none() { + self.mux_rate_reservation = self.initial_reservation.clone(); + } + } + + /// Count `pts` toward the startup hold. A source without a usable PCR never completes a + /// measurement window, so enough media time releases its catalog without a rate. + fn hold_media(&mut self, pts: u64) { + const WRAP: u64 = 1 << 33; + if let Some(edge) = self.held_edge { + // Modulo the 33-bit wrap, so the hold counts across a rollover. A step past half the + // range is a frame behind the edge (reordered, or another stream), not a jump forward. + let step = pts.wrapping_sub(edge) % WRAP; + if step >= WRAP / 2 { + return; + } + self.held += std::time::Duration::from_nanos(step * 1_000_000_000 / 90_000); + } + self.held_edge = Some(pts); + if self.held >= super::mux_rate::UNMEASURED { + self.mux_rate_reservation = None; + } + } + /// Copy the measured multiplex rate into the `mpegts` section, publishing the /// catalog. Called only when the measurement changed, so a steady source never /// republishes. @@ -993,7 +1061,8 @@ impl Import { for pid in pids { self.flush(pid)?; } - // No frame follows to anchor the clock, so publish the declared track set now. + // EOF releases an unmeasured short source as well as its decoder reservations. + self.mux_rate_reservation = None; self.initial_reservation = None; self.release_sections(self.timebase.offset().unwrap_or_default())?; for (pid, stream) in &mut self.streams { @@ -5947,10 +6016,9 @@ pub(super) mod test { assert_eq!(&frame.payload[..], &payload[..], "verbatim PES payload round-trips"); } - /// A program of only verbatim PIDs reserves no rendition, so the held reservation alone keeps - /// its PMT from publishing the catalog before the first PES anchors the clock. + /// A verbatim-only program retains the PES clock while waiting for rate measurement or EOF. #[tokio::test] - async fn verbatim_only_program_publishes_at_the_first_pes() { + async fn verbatim_only_program_retains_its_pes_clock_until_catalog_release() { use crate::catalog::hang::Catalog; use crate::container::ts::catalog::Ext; @@ -5978,12 +6046,11 @@ pub(super) mod test { .unwrap(); let anchored = catalog.snapshot().clock.expect("a clock"); assert_ne!(anchored, provisional, "the first PES anchors the clock"); + assert!(clocks.drain().is_empty(), "rate measurement still holds the catalog"); + import.finish().unwrap(); let published = clocks.drain(); - assert!(!published.is_empty(), "the first PES publishes the catalog"); + assert!(!published.is_empty(), "EOF releases a program without PCR samples"); assert!(published.iter().all(|clock| *clock == Some(anchored)), "{published:?}"); - - import.finish().unwrap(); - assert!(clocks.drain().iter().all(|clock| *clock == Some(anchored))); } /// A TS packet on `pid` carrying only an adaptation field (no payload) that sets @@ -7397,6 +7464,216 @@ pub(super) mod test { p } + #[test] + fn first_ts_catalog_waits_for_the_mux_rate_window() { + use crate::catalog::hang::Catalog; + use crate::container::ts::catalog::Ext; + + for cbr in [true, false] { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(Catalog::::default()), + ) + .unwrap(); + let mut updates = catalog.consume().unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + import + .decode(&synth_pmt(&[(StreamType::Mpeg2Video, PCR_PID)], false)) + .unwrap(); + import.decode(&pes_packet(PCR_PID, 90_000)).unwrap(); + let waiter = kio::Waiter::noop(); + assert!( + updates.poll_next(&waiter).is_pending(), + "catalog preceded rate measurement" + ); + + for i in 0..=50 { + let mut bytes = pcr_packet(PCR_PID, i * 27_000_000 / 25); + let stuffing = if cbr || i < 25 { 99 } else { 19 }; + for _ in 0..stuffing { + bytes.extend_from_slice(&super::super::export::NULL_PACKET); + } + import.decode(&bytes).unwrap(); + if i < 50 { + assert!( + updates.poll_next(&waiter).is_pending(), + "catalog before a full PCR window" + ); + } + } + let std::task::Poll::Ready(Ok(Some(first))) = updates.poll_next(&waiter) else { + panic!("catalog did not release at the end of the first window"); + }; + assert_eq!(first.ext.mpegts.mux_rate, cbr.then_some(3_760_000)); + } + } + + /// A live source with no PCR never completes a window, so media time releases its catalog. + #[test] + fn a_ts_without_pcr_releases_its_catalog_after_four_seconds_of_media() { + use crate::catalog::hang::Catalog; + use crate::container::ts::catalog::Ext; + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(Catalog::::default()), + ) + .unwrap(); + let mut updates = catalog.consume().unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + import + .decode(&synth_pmt(&[(StreamType::Mpeg2Video, PCR_PID)], false)) + .unwrap(); + let waiter = kio::Waiter::noop(); + for frame in 0..100 { + import.decode(&pes_packet(PCR_PID, 90_000 + frame * 3_600)).unwrap(); + assert!(updates.poll_next(&waiter).is_pending(), "catalog after frame {frame}"); + } + import.decode(&pes_packet(PCR_PID, 90_000 + 100 * 3_600)).unwrap(); + let std::task::Poll::Ready(Ok(Some(first))) = updates.poll_next(&waiter) else { + panic!("four seconds of media did not release the catalog"); + }; + assert_eq!(first.ext.mpegts.mux_rate, None); + } + + /// A source whose clock keeps resetting before a window completes still releases its + /// catalog once enough media time has passed in total. + #[test] + fn a_ts_resetting_its_clock_releases_its_catalog_after_four_seconds_of_media() { + use crate::catalog::hang::Catalog; + use crate::container::ts::catalog::Ext; + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(Catalog::::default()), + ) + .unwrap(); + let mut updates = catalog.consume().unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + import + .decode(&synth_pmt(&[(StreamType::Mpeg2Video, PCR_PID)], false)) + .unwrap(); + let waiter = kio::Waiter::noop(); + // Each epoch carries 0.96 s of media, then a flagged PCR restarts the clock. + for epoch in 0..6 { + for frame in 0..25 { + import.decode(&pes_packet(PCR_PID, 90_000 + frame * 3_600)).unwrap(); + } + import.decode(&flag_discontinuity(pcr_packet(PCR_PID, 0))).unwrap(); + let released = updates.poll_next(&waiter).is_ready(); + assert_eq!(released, epoch >= 4, "released after epoch {epoch}"); + if released { + return; + } + } + unreachable!(); + } + + /// A clock break after a measured window but before the first catalog publishes voids the + /// measurement, so the catalog waits for a window on the new clock and carries its rate. + #[test] + fn a_break_before_the_first_catalog_waits_for_a_new_window() { + use crate::catalog::hang::Catalog; + use crate::container::ts::catalog::Ext; + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(Catalog::::default()), + ) + .unwrap(); + let mut updates = catalog.consume().unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + import + .decode(&synth_pmt(&[(StreamType::Mpeg2Video, PCR_PID)], false)) + .unwrap(); + let window = |import: &mut super::Import, stuffing: usize| { + for i in 0..=50 { + let mut bytes = pcr_packet(PCR_PID, i * 27_000_000 / 25); + for _ in 0..stuffing { + bytes.extend_from_slice(&super::super::export::NULL_PACKET); + } + import.decode(&bytes).unwrap(); + } + }; + let waiter = kio::Waiter::noop(); + + // A full window before any PES, then a break, then the first PES. + window(&mut import, 99); + import.decode(&flag_discontinuity(pcr_packet(PCR_PID, 0))).unwrap(); + import.decode(&pes_packet(PCR_PID, 90_000)).unwrap(); + assert!( + updates.poll_next(&waiter).is_pending(), + "the catalog published on a voided measurement" + ); + + window(&mut import, 49); + let std::task::Poll::Ready(Ok(Some(first))) = updates.poll_next(&waiter) else { + panic!("the new clock's window did not release the catalog"); + }; + assert_eq!(first.ext.mpegts.mux_rate, Some(1_880_000)); + } + + /// The hold counts media time across a 33-bit PTS rollover. + #[test] + fn a_ts_without_pcr_releases_its_catalog_across_a_pts_rollover() { + use crate::catalog::hang::Catalog; + use crate::container::ts::catalog::Ext; + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(Catalog::::default()), + ) + .unwrap(); + let mut updates = catalog.consume().unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + import + .decode(&synth_pmt(&[(StreamType::Mpeg2Video, PCR_PID)], false)) + .unwrap(); + let waiter = kio::Waiter::noop(); + // Starts two seconds before the rollover. + let start = (1 << 33) - 180_000; + for frame in 0..100 { + import + .decode(&pes_packet(PCR_PID, (start + frame * 3_600) % (1 << 33))) + .unwrap(); + assert!(updates.poll_next(&waiter).is_pending(), "catalog after frame {frame}"); + } + import + .decode(&pes_packet(PCR_PID, (start + 100 * 3_600) % (1 << 33))) + .unwrap(); + assert!( + updates.poll_next(&waiter).is_ready(), + "four seconds across the rollover did not release the catalog" + ); + } + + #[test] + fn finishing_a_short_ts_releases_its_catalog_without_a_rate() { + use crate::catalog::hang::Catalog; + use crate::container::ts::catalog::Ext; + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(Catalog::::default()), + ) + .unwrap(); + let mut updates = catalog.consume().unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + import + .decode(&synth_pmt(&[(StreamType::Mpeg2Video, PCR_PID)], false)) + .unwrap(); + import.decode(&pes_packet(PCR_PID, 90_000)).unwrap(); + import.decode(&pcr_packet(PCR_PID, 0)).unwrap(); + let waiter = kio::Waiter::noop(); + assert!(updates.poll_next(&waiter).is_pending()); + import.finish().unwrap(); + let std::task::Poll::Ready(Ok(Some(first))) = updates.poll_next(&waiter) else { + panic!("short recording finished without a catalog"); + }; + assert_eq!(first.ext.mpegts.mux_rate, None); + } + /// The catalog follows the clock: a stable multiplex rate is recorded, a source that /// stops holding it clears the record, a fresh stable window records it again, and a /// declared time-base break clears it at once. diff --git a/rs/moq-mux/src/container/ts/mux_rate.rs b/rs/moq-mux/src/container/ts/mux_rate.rs index ec4298b0ff..84d3740145 100644 --- a/rs/moq-mux/src/container/ts/mux_rate.rs +++ b/rs/moq-mux/src/container/ts/mux_rate.rs @@ -16,6 +16,9 @@ const PACKET_BITS: u64 = 188 * 8; pub(super) const PCR_WRAP: u64 = (1 << 33) * 300; /// How much PCR time one window spans before it is judged. const WINDOW: u64 = 2 * PCR_HZ; +/// Media time after which a source that has not completed a window publishes without a rate: +/// twice the window, so a usable PCR clock always completes one first. +pub(super) const UNMEASURED: std::time::Duration = std::time::Duration::from_secs(2 * WINDOW / PCR_HZ); /// Intervals are pooled into samples this long before they are compared. One /// packet of rounding, or a clock a hardware muxer stamps at half-millisecond /// granularity, is a few percent of one 25 ms interval but well inside the @@ -73,6 +76,8 @@ pub(super) struct Meter { window: Vec, /// The rate recorded in the catalog, if any. published: Option, + /// Whether a window has completed since the last discontinuity, even for VBR. + ready: bool, } impl Meter { @@ -81,6 +86,11 @@ impl Meter { self.published } + /// Whether a PCR measurement window has completed on the current clock. + pub fn ready(&self) -> bool { + self.ready + } + /// One transport packet went by. pub fn packet(&mut self) { self.packets += 1; @@ -101,7 +111,7 @@ impl Meter { return self.discontinuity(); } self.sample.add(Interval { packets, ticks }); - if self.sample.ticks < SAMPLE { + if self.sample.ticks < SAMPLE && self.sample.ticks + self.window.iter().map(|s| s.ticks).sum::() < WINDOW { return false; } self.window.push(std::mem::take(&mut self.sample)); @@ -115,6 +125,7 @@ impl Meter { /// itself changed. The window is worthless and so is a rate measured before it. /// Returns whether [`published`](Self::published) changed. pub fn discontinuity(&mut self) -> bool { + self.ready = false; self.packets = 0; self.last = None; self.sample = Interval::default(); @@ -124,6 +135,7 @@ impl Meter { /// Judge the full window and discard it. fn judge(&mut self) -> bool { + self.ready = true; let window = std::mem::take(&mut self.window); let mut whole = Interval::default(); window.iter().for_each(|s| whole.add(*s)); diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index ed69110e45..4017b10bd3 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -426,7 +426,7 @@ mod tests { } /// A two-program multiplex, each program carrying one private PES stream (carried - /// verbatim) and its first PES, which anchors the clock and so publishes the catalog: + /// verbatim) and its first PES plus a complete PCR measurement window: /// program 1's on PID 0x101, program 2's on 0x201. fn two_programs() -> Bytes { let mut ts = psi_packet(0x0000, &pat(&[(1, 0x0100), (2, 0x0200)])); @@ -434,6 +434,23 @@ mod tests { ts.extend_from_slice(&psi_packet(0x0200, &pmt(2, 0x06, 0x0201))); ts.extend_from_slice(&pes_packet(0x0101, 90_000)); ts.extend_from_slice(&pes_packet(0x0201, 90_000)); + // Complete the startup rate window before expecting a live catalog. + for index in 0..=8u64 { + let base = index * 22_500; + for pid in [0x0101u16, 0x0201] { + let mut packet = vec![0x47, (pid >> 8) as u8, pid as u8, 0x20, 183, 0x10]; + packet.extend_from_slice(&[ + (base >> 25) as u8, + (base >> 17) as u8, + (base >> 9) as u8, + (base >> 1) as u8, + ((base & 1) as u8) << 7 | 0x7e, + 0, + ]); + packet.resize(188, 0xff); + ts.extend_from_slice(&packet); + } + } ts.into() } diff --git a/test/interop/interop.sh b/test/interop/interop.sh index 7e6db609e8..43de7f0b77 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -642,8 +642,9 @@ run_publisher() { local lang="$1" broadcast="$2" url="$3" case "$lang" in rust) + # A finished import stays up for its max-age; `stop_publisher` allows 10 s to close. ffmpeg_h264 | RUST_LOG="$CLI_LOG" "$MOQ" --connect "$url" --connect-version "$AUTH_VERSION" \ - --broadcast "$broadcast" import avc3 + --broadcast "$broadcast" import --max-age 1s avc3 ;; python) ffmpeg_h264 | "$PY" "$CLIENTS/python/interop.py" \ diff --git a/test/justfile b/test/justfile index bd71008b6a..83a9fbb8c7 100644 --- a/test/justfile +++ b/test/justfile @@ -127,6 +127,7 @@ drill-sensitivity *args: # Check the subscriber's MPEG-TS output for IRD compliance. [positional-arguments] ts *args: + python3 ./ts/pcr-timing.test.py ./ts/run.sh "$@" # Round-trip the EIT fixtures: the sparse schedule must survive and diff --git a/test/ts/README.md b/test/ts/README.md index 47e495ebe6..4648cde97e 100644 --- a/test/ts/README.md +++ b/test/ts/README.md @@ -47,8 +47,8 @@ The default arm runs `pcr-timing.py` over its capture too, after `compliance.py` for the PCR checks `compliance.py` leaves to it: the value interval (hard), and whether the bytes between consecutive PCRs are the ones the mux rate implies ([`pcr-schedule`](#byte-schedule), which gates whenever the rate is known, as the -generated clip's is). With the rate known it also runs TSDuck's `pcrverify` past -the first three seconds, failing on any PCR more than 500 ns off its byte position +generated clip's is). With the rate known it also runs TSDuck's `pcrverify` from +the first PCR, failing on any PCR more than 500 ns off its byte position at the rate: the `PCR_accuracy_error` an IRD or a TR 101 290 probe reports, which `pcr-schedule`'s packet of slack lets through. @@ -350,16 +350,13 @@ correct 2 Mb/s stream on a 25 ms grid reads −0.74 % and +2.27 % on alternate intervals, which a percentage alone would fail. Give the rate whenever it is known. The estimate is total bytes over total time, -so a transient biases every interval by the same amount: over a 20 s live window, -the exporter's unpadded first half-second pulled it ~3 % low and read every padded -interval after it as off schedule. `run.sh` passes the generated clip's rate -(`--bitrate`) with `--schedule-pct-min 99`, and lets the grader estimate, and only -report, for `--source`. - -With the rate given, the intervals before the first null packet are not graded. The -exporter pads only once its catalog records the rate, which import measures over the -source's first two seconds, so a subscriber that starts with the publisher begins -unpadded; the report counts those intervals as `unpadded_lead`. +so a transient biases every interval by the same amount. `run.sh` passes the generated +clip's rate (`--bitrate`) with `--schedule-pct-min 99`, and lets the grader estimate, +and only report, for `--source`. + +Every PCR interval is graded. Import holds the catalog until its initial two-second +PCR window has measured a stable rate or established that the input is variable-rate. +A constant-rate exporter therefore has the rate before it sends its first PCR. At the default 10 Mb/s the generated clip compresses to almost nothing, so padding dominates and no keyframe outgrows its 31 kB slot. At 2 Mb/s its 13-19 kB keyframes diff --git a/test/ts/pcr-timing.py b/test/ts/pcr-timing.py index 1d55a517a6..6d0491b57d 100755 --- a/test/ts/pcr-timing.py +++ b/test/ts/pcr-timing.py @@ -76,7 +76,6 @@ class Scan: def __init__(self): self.pcr = [] # (packet_index, ticks, arrival_or_None, pid) self.packets = 0 - self.first_null = None # the first null packet: a padded stream's schedule starts there self.bad_sync = 0 self.transport_error = 0 self.cc_errors = [] @@ -100,8 +99,6 @@ def feed(self, p, arrival=None): if p[1] & 0x80: self.transport_error += 1 pid = ((p[1] & 0x1F) << 8) | p[2] - if pid == 0x1FFF and self.first_null is None: - self.first_null = index # ISO 13818-1 2.4.3.3 allows the counter to jump when the adaptation field sets # discontinuity_indicator, and allows one packet to be sent twice, repeating its @@ -538,11 +535,6 @@ def check_schedule(scan, args): as a start that is not yet padded, biases every interval's error by the same amount; pass the rate whenever it is known. - With --mux-rate, the intervals before the first null packet are not graded: an - exporter pads only once its catalog records the rate, which import measures over the - source's first two seconds, so a subscriber that starts with the publisher begins - unpadded. The report counts them. - Report-only unless --schedule-pct-min is given: a VBR stream has no schedule to keep, and `export ts` is VBR unless a mux rate is declared, so only the caller knows whether this property was promised. @@ -552,15 +544,11 @@ def check_schedule(scan, args): required = args.schedule_pct_min if hard else 99.0 graded = [] skipped = 0 - unpadded = 0 for kb, (a, b) in enumerate(zip(scan.pcr, scan.pcr[1:]), start=1): seconds = (b[1] - a[1]) / TICKS_PER_MS / 1000.0 # An interval spanning a signalled new time base measures nothing, as in the value # check. A non-positive one is a duplicate packet (legal) or a backwards clock (the # value check's defect to report); neither has a rate to compare against. - if args.mux_rate and scan.first_null is not None and b[0] <= scan.first_null: - unpadded += 1 - continue if kb in scan.new_base or seconds <= 0: skipped += 1 continue @@ -587,7 +575,6 @@ def check_schedule(scan, args): "other_pcr_pids": scan.other_pcr_pids, "count": len(graded), "skipped": skipped, - "unpadded_lead": unpadded, "rate_bps": round(rate), "rate_source": "declared" if args.mux_rate else "estimated", "aggregate_bps": round(aggregate), diff --git a/test/ts/pcr-timing.test.py b/test/ts/pcr-timing.test.py new file mode 100644 index 0000000000..9ca97ea937 --- /dev/null +++ b/test/ts/pcr-timing.test.py @@ -0,0 +1,24 @@ +import importlib.util +import pathlib +import types +import unittest + +spec = importlib.util.spec_from_file_location("pcr_timing", pathlib.Path(__file__).with_name("pcr-timing.py")) +timing = importlib.util.module_from_spec(spec) +spec.loader.exec_module(timing) + + +class Schedule(unittest.TestCase): + def test_unpadded_start_is_graded(self): + scan = types.SimpleNamespace( + pcr=[(index, i * 1_080_000) for i, index in enumerate([0, 10, 20, 120, 220, 320])], + first_null=20, new_base=set(), pcr_pid=256, other_pcr_pids=set(), + ) + args = types.SimpleNamespace(mux_rate=3_760_000, schedule_pct_min=99, schedule_tolerance_pct=1) + result = timing.check_schedule(scan, args) + self.assertFalse(result[2], result) + self.assertEqual(result[4]["count"], 5) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/ts/run.sh b/test/ts/run.sh index 530449b453..e9d5036099 100755 --- a/test/ts/run.sh +++ b/test/ts/run.sh @@ -358,11 +358,9 @@ harness_endpoint relay "$URL" # SCHEDULE carries the rate pcr-timing.py's schedule check grades against, and gates # on it. The generated clip is muxed at $BITRATE, which is the rate the catalog records -# and the exporter pads to, on a constant-rate schedule from the first null packet on: -# every interval is the bytes the rate implies, to one packet. Left to estimate, the -# grader divides total bytes by the PCR span, and the unpadded start drags that off the -# true rate. A --source capture's rate is not known here, so for that the grader -# estimates it, says so, and only reports. +# and the exporter pads to, on a constant-rate schedule from the first PCR: +# every interval is the bytes the rate implies, to one packet. A --source capture's +# rate is not known here, so for that the grader estimates it, says so, and only reports. SCHEDULE=() [[ -z "$SOURCE" ]] && SCHEDULE=(--mux-rate "$BITRATE" --schedule-pct-min 99) @@ -421,13 +419,14 @@ sleep 1 # at 5). It is the publisher's granularity, not the exporter's, so the harness has # to be well inside it or it grades tsp. # Bounded like the subscriber: if the relay stalls after readiness, `moq import -# ts` could otherwise block `wait "$PUB_PID"` forever. tsp/moq stderr lands in -# pub.log, which the empty-capture handler below dumps on failure. +# ts` could otherwise block `wait "$PUB_PID"` forever. The bound adds the 30 s +# default max-age a finished import stays up for after EOF. tsp/moq stderr lands +# in pub.log, which the empty-capture handler below dumps on failure. echo "### publishing PCR-paced TS -> $BROADCAST" # shellcheck disable=SC2329 # invoked indirectly via 'harness_spawn' publish() { # shellcheck disable=SC2016 # $1..$4 are the child bash -c positionals, not ours. - timeout -k 3 $((DURATION + 20)) bash -c ' + timeout -k 3 $((DURATION + 20 + 30)) bash -c ' tsp -I file "$1" -P regulate --pcr-synchronous --wait-min 5 | "$2" --connect "$3" --broadcast "$4" import ts ' _ "$SRC_TS" "$MOQ" "$URL" "$BROADCAST" @@ -589,13 +588,11 @@ fi # pcr-schedule allows each PCR a packet of slack, but an IRD or a TR 101 290 probe holds # every PCR to within 500 ns of its byte position at the mux rate (PCR_accuracy_error). -# TSDuck's pcrverify measures exactly that. The first seconds are skipped: the export -# starts unpadded until the importer has measured the source's rate and recorded it. +# TSDuck's pcrverify measures exactly that, including the first PCR interval. if [[ ${#SCHEDULE[@]} -gt 0 ]]; then echo echo "### PCR accuracy (tsp -P pcrverify, +/-500 ns at ${BITRATE} b/s)" - SKIP=$((3 * BITRATE / 8 / 188)) - tsp -I file "$SUB_TS" -P skip "$SKIP" -P pcrverify --bitrate "$BITRATE" --absolute --jitter-max 13 \ + tsp -I file "$SUB_TS" -P pcrverify --bitrate "$BITRATE" --absolute --jitter-max 13 \ -O drop >"$HARNESS_RUN/pcrverify.log" 2>&1 || true VERDICT=$(grep -E "PCR OK, " "$HARNESS_RUN/pcrverify.log" | tail -1) echo " ${VERDICT:-no verdict}"