From 5d62f3b8174032f259e0f4656a077d350d3676a4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 13:57:52 -0700 Subject: [PATCH 01/10] chore(quest): claim mux-rate catalog hold Co-Authored-By: GPT-6 From e7a4666e1023ef124bab27d7905e2162cfef50ce Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 15:05:25 -0700 Subject: [PATCH 02/10] fix(ts): publish the catalog after initial mux rate measurement Co-Authored-By: GPT-6 # Conflicts: # quest/m1/tstd/README.md --- quest/m1/tstd/README.md | 1 - quest/m1/tstd/mux-rate-hold.md | 35 --------- rs/moq-cli/tests/import.rs | 3 +- rs/moq-mux/src/container/ts/import.rs | 99 ++++++++++++++++++++++--- rs/moq-mux/src/container/ts/mux_rate.rs | 10 ++- rs/moq-srt/src/ts.rs | 19 ++++- test/justfile | 1 + test/ts/README.md | 17 ++--- test/ts/pcr-timing.py | 13 ---- test/ts/pcr-timing.test.py | 24 ++++++ test/ts/run.sh | 14 ++-- 11 files changed, 155 insertions(+), 81 deletions(-) delete mode 100644 quest/m1/tstd/mux-rate-hold.md create mode 100644 test/ts/pcr-timing.test.py diff --git a/quest/m1/tstd/README.md b/quest/m1/tstd/README.md index a748afd30c..839ea07bb3 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 ## 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/tests/import.rs b/rs/moq-cli/tests/import.rs index 69bd6c3cbe..b89cc84e70 100644 --- a/rs/moq-cli/tests/import.rs +++ b/rs/moq-cli/tests/import.rs @@ -8,7 +8,7 @@ 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"); +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. /// @@ -53,6 +53,7 @@ async fn import_delivers_the_catalog_finish_at_eof() { .spawn() .expect("spawn moq"); 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(); diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 56b0e02534..bf986f829b 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -61,6 +61,8 @@ 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 TS rate metadata until one PCR measurement window, independently of decoder readiness. + mux_rate_reservation: 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. @@ -156,12 +158,14 @@ 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, pat: PatReader::default(), pmt_sections: HashMap::new(), crc_errors: BTreeMap::new(), @@ -282,8 +286,13 @@ 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 { + if self.mux_rate.pcr(pcr) { + self.record_mux_rate()?; + } + if self.mux_rate.ready() { + self.mux_rate_reservation = None; + } } } } @@ -990,7 +999,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 { @@ -5854,10 +5864,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; @@ -5885,12 +5894,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 @@ -7250,6 +7258,77 @@ 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)); + } + } + + #[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..89020f1005 100644 --- a/rs/moq-mux/src/container/ts/mux_rate.rs +++ b/rs/moq-mux/src/container/ts/mux_rate.rs @@ -73,6 +73,8 @@ pub(super) struct Meter { window: Vec, /// The rate recorded in the catalog, if any. published: Option, + /// Whether the startup measurement window has completed, even for VBR. + ready: bool, } impl Meter { @@ -81,6 +83,11 @@ impl Meter { self.published } + /// Whether the first PCR measurement window has completed. + pub fn ready(&self) -> bool { + self.ready + } + /// One transport packet went by. pub fn packet(&mut self) { self.packets += 1; @@ -101,7 +108,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)); @@ -124,6 +131,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 b3fdf755bb..e388c124a8 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -425,7 +425,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)])); @@ -433,6 +433,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/justfile b/test/justfile index 30114d289d..31d4e8e983 100644 --- a/test/justfile +++ b/test/justfile @@ -118,6 +118,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..a2dab141b0 100644 --- a/test/ts/README.md +++ b/test/ts/README.md @@ -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..e1d8fd230d 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) @@ -589,13 +587,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}" From ba495659c9dc016ce0c1dd0983c6ac094746cc85 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 17:04:05 -0700 Subject: [PATCH 03/10] docs(ts): retain the unresolved startup hold lifecycle Co-Authored-By: GPT-6 Astra --- quest/m1/tstd/README.md | 1 + quest/m1/tstd/mux-rate-hold.md | 57 +++++++++++++++++++++++++++ rs/moq-mux/src/container/ts/import.rs | 7 +++- 3 files changed, 64 insertions(+), 1 deletion(-) create mode 100644 quest/m1/tstd/mux-rate-hold.md diff --git a/quest/m1/tstd/README.md b/quest/m1/tstd/README.md index cd19295ceb..d88fd14ee0 100644 --- a/quest/m1/tstd/README.md +++ b/quest/m1/tstd/README.md @@ -35,6 +35,7 @@ nightly. ## Required +- [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 - [Send-ahead within the delay](/quest/m1/tstd/send-ahead.md) - total lag is `--delay`, send-ahead included, with a 1 s default - [PSI runs](/quest/m1/tstd/psi-runs.md) - aligned keyframes never lay identical PAT/PMT runs back to back, so the system buffer holds diff --git a/quest/m1/tstd/mux-rate-hold.md b/quest/m1/tstd/mux-rate-hold.md new file mode 100644 index 0000000000..cce823d22a --- /dev/null +++ b/quest/m1/tstd/mux-rate-hold.md @@ -0,0 +1,57 @@ +# [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. + +### Progress + +The draft implementation holds the initial catalog until a complete PCR window +or EOF and grades PCR timing from the first interval. The hold applies to every +subscriber of a broadcast whose producer uses the `mpegts` catalog extension. + +A real-relay one-second CBR import exposes an unresolved EOF boundary: the held +catalog arrives as the publisher closes, and a subscriber captures no media. +The same source without the hold captures 0.38 s of its 0.98 s reference. Session +close drains subscriptions already serving the peer; it cannot cover media +subscriptions initiated after that last catalog reaches the subscriber. The +short-input unit test proves catalog publication, not end-to-end delivery. + +Keep this quest open until short-input delivery has a defined lifecycle and a +real-relay regression. Recommended: explicitly preserve a finite broadcast +until the intended consumer has discovered and drained its tracks. The +alternative is to refuse short unmeasured live imports explicitly; silently +exiting successfully with no delivered media is not a supported outcome. Do +not add a sleep or extend a shutdown timeout to hide the subscription race. +A source that never supplies a usable PCR window also needs an explicit +refusal policy rather than an indefinitely unpublished live catalog. diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index bf986f829b..90bd609063 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -49,6 +49,11 @@ 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; 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,7 +66,7 @@ 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 TS rate metadata until one PCR measurement window, independently of decoder readiness. + /// Hold the whole catalog until one PCR measurement window, independently of decoder readiness. mux_rate_reservation: Option>, /// The program's timestamp base: the first PES with a PTS anchors it, and every PTS shifts by From bbaa91b5d436832b379bf7545e3ede31b29a3b18 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 16:57:36 -0700 Subject: [PATCH 04/10] fix(ts): release a PCR-less catalog after four seconds of media A source with no usable PCR never completes a measurement window, so the startup hold now ends after twice the window of media time and publishes without a rate instead of withholding the catalog until EOF. Add the real-relay reproduction of the short-input drain (ignored until the broadcast drain lifecycle lands) and record the decision and the missing demand signal in the quest. Co-Authored-By: Claude Opus 5.5 --- quest/m1/tstd/mux-rate-hold.md | 40 ++++---- rs/moq-cli/tests/import.rs | 117 +++++++++++++++++++++--- rs/moq-mux/src/container/ts/import.rs | 45 ++++++++- rs/moq-mux/src/container/ts/mux_rate.rs | 3 + 4 files changed, 170 insertions(+), 35 deletions(-) diff --git a/quest/m1/tstd/mux-rate-hold.md b/quest/m1/tstd/mux-rate-hold.md index cce823d22a..045cc4a958 100644 --- a/quest/m1/tstd/mux-rate-hold.md +++ b/quest/m1/tstd/mux-rate-hold.md @@ -36,22 +36,24 @@ grades from the first null packet). ### Progress -The draft implementation holds the initial catalog until a complete PCR window -or EOF and grades PCR timing from the first interval. The hold applies to every -subscriber of a broadcast whose producer uses the `mpegts` catalog extension. - -A real-relay one-second CBR import exposes an unresolved EOF boundary: the held -catalog arrives as the publisher closes, and a subscriber captures no media. -The same source without the hold captures 0.38 s of its 0.98 s reference. Session -close drains subscriptions already serving the peer; it cannot cover media -subscriptions initiated after that last catalog reaches the subscriber. The -short-input unit test proves catalog publication, not end-to-end delivery. - -Keep this quest open until short-input delivery has a defined lifecycle and a -real-relay regression. Recommended: explicitly preserve a finite broadcast -until the intended consumer has discovered and drained its tracks. The -alternative is to refuse short unmeasured live imports explicitly; silently -exiting successfully with no delivered media is not a supported outcome. Do -not add a sleep or extend a shutdown timeout to hide the subscription race. -A source that never supplies a usable PCR window also needs an explicit -refusal policy rather than an indefinitely unpublished live catalog. +The draft holds the initial catalog until a complete PCR window, four seconds +of media without one (a source with no usable PCR), or EOF, and grades PCR +timing from the first interval. The hold applies to every subscriber of a +broadcast whose producer uses the `mpegts` catalog extension. + +Decided (2026-10-11): drain to consumers. A short input's media must reach a +subscriber already waiting on the broadcast, and an import never reports +success while delivering zero media. A source that never completes a PCR +window publishes unpadded rather than nothing. + +Open: the drain has no signal to wait on. A one-second import releases its +catalog only at EOF, and the subscriber's media SUBSCRIBE reaches the +publisher after it has left (`subscribe to the video: Dropped`). The ignored +`a_short_import_drains_its_media_to_a_waiting_subscriber` in +`rs/moq-cli/tests/import.rs` reproduces it; un-ignore it once the drain lands. +Demand does not span the catalog-to-media handoff: the catalog subscription +ends at its FIN, a relay releases its upstream at that FIN, and the media +request is a separate stream, so the publisher sees no demand in between. +Holding the catalog unfinished until media demand appears deadlocks a +catalog-only subscriber, which `import_delivers_the_catalog_finish_at_eof` +relies on. Do not add a sleep or extend a shutdown timeout to hide the race. diff --git a/rs/moq-cli/tests/import.rs b/rs/moq-cli/tests/import.rs index b89cc84e70..e70eceb02b 100644 --- a/rs/moq-cli/tests/import.rs +++ b/rs/moq-cli/tests/import.rs @@ -22,14 +22,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,12 +31,12 @@ 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", @@ -51,7 +45,18 @@ async fn import_delivers_the_catalog_finish_at_eof() { .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"); @@ -111,3 +116,91 @@ 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 available until that subscriber drains them. +#[tokio::test] +#[ignore = "fails: the publisher leaves before the media SUBSCRIBE the final catalog prompts arrives; see quest/m1/tstd/mux-rate-hold.md"] +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 aa39ea8c50..5ef3fcd23e 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -51,7 +51,8 @@ use moq_net::Timestamp; /// /// 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; EOF releases a shorter input too. +/// 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 @@ -66,7 +67,8 @@ 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, independently of decoder readiness. + /// 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>, /// The program's timestamp base: the first PES with a PTS anchors it, and every PTS shifts by @@ -704,8 +706,15 @@ impl Import { Some(pts) => { let pts = Timestamp::from_scale(pts, 90_000)?; let offset = self.timebase.anchor(pts)?; - if self.start_pts.is_none() { - self.start_pts = Some(offset.apply(pts)?); + let pts = offset.apply(pts)?; + let start = *self.start_pts.get_or_insert(pts); + // A source without a usable PCR never completes a measurement window, so media + // time releases its catalog without a rate. + if pts + .checked_sub(start) + .is_ok_and(|elapsed| std::time::Duration::from(elapsed) >= super::mux_rate::UNMEASURED) + { + self.mux_rate_reservation = None; } offset } @@ -7455,6 +7464,34 @@ pub(super) mod test { } } + /// 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); + } + #[test] fn finishing_a_short_ts_releases_its_catalog_without_a_rate() { use crate::catalog::hang::Catalog; diff --git a/rs/moq-mux/src/container/ts/mux_rate.rs b/rs/moq-mux/src/container/ts/mux_rate.rs index 89020f1005..f397df41f1 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 From d393ee39ae0c5ae28c00fb69ff26cd0432b58292 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 18:51:59 -0700 Subject: [PATCH 05/10] fix(cli): keep a finished import up for its max-age A short TS import releases its catalog only at EOF, so a waiting subscriber's media requests arrived after the publisher had left and it received nothing. A stdin import now keeps the finished broadcast up for the media max-age after EOF, so cached content stays servable until it is stale; SIGINT or SIGTERM ends that early. Un-ignore the real-relay regression, add a paused-clock unit test, give scripted callers that wait on the import a short max-age or budget, and complete the mux-rate hold quest. The general rule for other publishers and the library is quest/m1/publisher-linger.md. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 2 + quest/m1/README.md | 1 + quest/m1/publisher-linger.md | 31 +++++++++++++ quest/m1/tstd/README.md | 1 - quest/m1/tstd/mux-rate-hold.md | 59 ------------------------ rs/moq-cli/src/args.rs | 3 +- rs/moq-cli/src/main.rs | 2 + rs/moq-cli/src/publish.rs | 82 +++++++++++++++++++++++++++++++--- rs/moq-cli/tests/export.rs | 5 ++- rs/moq-cli/tests/import.rs | 8 +++- test/interop/interop.sh | 3 +- test/ts/run.sh | 7 +-- 12 files changed, 130 insertions(+), 74 deletions(-) create mode 100644 quest/m1/publisher-linger.md delete mode 100644 quest/m1/tstd/mux-rate-hold.md diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 45d0353554..1471298848 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -306,6 +306,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 004d82441b..e7c8d6744a 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -196,6 +196,7 @@ session outcomes, and embedder refusals inside session outcomes. - [moqsrc reconnect](/quest/m1/moqsrc-reconnect.md) - `moqsrc` redials after losing its relay and resumes on the same pads - [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 +- [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/tstd/README.md b/quest/m1/tstd/README.md index 9db322f018..aeeaf2754e 100644 --- a/quest/m1/tstd/README.md +++ b/quest/m1/tstd/README.md @@ -35,7 +35,6 @@ nightly. ## Required -- [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 - [Send-ahead within the delay](/quest/m1/tstd/send-ahead.md) - total lag is `--delay`, send-ahead included, with a 1 s default - [PSI runs](/quest/m1/tstd/psi-runs.md) - the PAT and PMT are deduplicated and paced so the system buffers never overflow diff --git a/quest/m1/tstd/mux-rate-hold.md b/quest/m1/tstd/mux-rate-hold.md deleted file mode 100644 index 045cc4a958..0000000000 --- a/quest/m1/tstd/mux-rate-hold.md +++ /dev/null @@ -1,59 +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. - -### Progress - -The draft holds the initial catalog until a complete PCR window, four seconds -of media without one (a source with no usable PCR), or EOF, and grades PCR -timing from the first interval. The hold applies to every subscriber of a -broadcast whose producer uses the `mpegts` catalog extension. - -Decided (2026-10-11): drain to consumers. A short input's media must reach a -subscriber already waiting on the broadcast, and an import never reports -success while delivering zero media. A source that never completes a PCR -window publishes unpadded rather than nothing. - -Open: the drain has no signal to wait on. A one-second import releases its -catalog only at EOF, and the subscriber's media SUBSCRIBE reaches the -publisher after it has left (`subscribe to the video: Dropped`). The ignored -`a_short_import_drains_its_media_to_a_waiting_subscriber` in -`rs/moq-cli/tests/import.rs` reproduces it; un-ignore it once the drain lands. -Demand does not span the catalog-to-media handoff: the catalog subscription -ends at its FIN, a relay releases its upstream at that FIN, and the media -request is a separate stream, so the publisher sees no demand in between. -Holding the catalog unfinished until media demand appears deadlocks a -catalog-only subscriber, which `import_delivers_the_catalog_finish_at_eof` -relies on. Do not add a sleep or extend a shutdown timeout to hide the race. 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 2c3c9042d7..567fbd160a 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 e70eceb02b..dc7552a1c5 100644 --- a/rs/moq-cli/tests/import.rs +++ b/rs/moq-cli/tests/import.rs @@ -8,6 +8,9 @@ use tokio::io::AsyncWriteExt; use tokio::process::Child; const TIMEOUT: Duration = Duration::from_secs(10); +/// 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. @@ -40,6 +43,8 @@ fn import(url: &str, fingerprint: &str) -> Child { "--broadcast", "demo", "import", + "--max-age", + MAX_AGE, "ts", ]) .stdin(Stdio::piped()) @@ -124,9 +129,8 @@ const SHORT: &[u8] = include_bytes!("../../moq-mux/src/container/ts/test_data/bb .0; /// A subscriber already waiting when a short import ends still receives its media: the -/// publisher keeps the finished tracks available until that subscriber drains them. +/// publisher keeps the finished tracks servable for their max-age after EOF. #[tokio::test] -#[ignore = "fails: the publisher leaves before the media SUBSCRIBE the final catalog prompts arrives; see quest/m1/tstd/mux-rate-hold.md"] 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"); 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/ts/run.sh b/test/ts/run.sh index e1d8fd230d..e9d5036099 100755 --- a/test/ts/run.sh +++ b/test/ts/run.sh @@ -419,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" From abad01577b4a4baf07e0ee63876cb2227acb9d20 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 19:27:18 -0700 Subject: [PATCH 06/10] quest: point the live-clock quest at #5287 instead of the finished hold quest Co-Authored-By: Claude Opus 5.5 --- quest/m1/ts-live-clock.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 From 871db8588228a00a999c675a20e6cb585743d100 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 19:49:48 -0700 Subject: [PATCH 07/10] fix(ts): sum the startup hold's media time across time-base breaks A source that resets its clock more often than the measurement window reset the fallback's start too, so it never released its catalog. Count media time per time base and sum it, so four seconds in total releases the catalog without a rate. Also drop the README's stale three-second pcrverify skip. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 70 ++++++++++++++++++++++++--- test/ts/README.md | 4 +- 2 files changed, 64 insertions(+), 10 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 5ef3fcd23e..00686535bf 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -70,6 +70,11 @@ pub struct Import { /// 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 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. @@ -174,6 +179,8 @@ impl Import { 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(), @@ -707,14 +714,11 @@ impl Import { let pts = Timestamp::from_scale(pts, 90_000)?; let offset = self.timebase.anchor(pts)?; let pts = offset.apply(pts)?; - let start = *self.start_pts.get_or_insert(pts); - // A source without a usable PCR never completes a measurement window, so media - // time releases its catalog without a rate. - if pts - .checked_sub(start) - .is_ok_and(|elapsed| std::time::Duration::from(elapsed) >= super::mux_rate::UNMEASURED) - { - self.mux_rate_reservation = None; + if self.start_pts.is_none() { + self.start_pts = Some(pts); + } + if self.mux_rate_reservation.is_some() { + self.hold_media(pts); } offset } @@ -888,6 +892,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() { @@ -897,6 +902,22 @@ impl Import { Ok(()) } + /// 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: Timestamp) { + // Reordered frames and other streams land behind the edge and add nothing. + if let Some(edge) = self.held_edge { + let Ok(step) = pts.checked_sub(edge) else { + return; + }; + self.held += std::time::Duration::from(step); + } + 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. @@ -7492,6 +7513,39 @@ pub(super) mod test { 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!(); + } + #[test] fn finishing_a_short_ts_releases_its_catalog_without_a_rate() { use crate::catalog::hang::Catalog; diff --git a/test/ts/README.md b/test/ts/README.md index a2dab141b0..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. From d2f38b29b4f2647c555b01449b70c4d7f17840e6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 20:07:40 -0700 Subject: [PATCH 08/10] fix(ts): count the startup hold across a PTS rollover The hold's media time compared catalog-clock timestamps, which are not unwrapped, so a source without PCR that rolled its 33-bit PTS during the hold never released its catalog. Count on the raw 90 kHz clock modulo the wrap instead. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 60 +++++++++++++++++++++------ 1 file changed, 48 insertions(+), 12 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 00686535bf..6cf37a4b9e 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -73,8 +73,8 @@ pub struct Import { /// 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 PTS of the current time base counted into `held`. - held_edge: Option, + /// 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. @@ -711,14 +711,13 @@ 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)?; - let pts = offset.apply(pts)?; if self.start_pts.is_none() { - self.start_pts = Some(pts); - } - if self.mux_rate_reservation.is_some() { - self.hold_media(pts); + self.start_pts = Some(offset.apply(pts)?); } offset } @@ -904,13 +903,16 @@ impl Import { /// 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: Timestamp) { - // Reordered frames and other streams land behind the edge and add nothing. + fn hold_media(&mut self, pts: u64) { + const WRAP: u64 = 1 << 33; if let Some(edge) = self.held_edge { - let Ok(step) = pts.checked_sub(edge) else { + // 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(step); + } + 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 { @@ -7546,6 +7548,40 @@ pub(super) mod test { unreachable!(); } + /// 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; From a5c083fb192a91b9b78f9fb0359ea6edfbd32ec5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 20:22:49 -0700 Subject: [PATCH 09/10] fix(ts): re-arm the startup hold on a clock break before the first catalog A PCR window could complete before the first PES and release the hold; a break before that PES then voided the rate while the catalog published without it. The meter's readiness now resets on a discontinuity, and the hold is restored while the first catalog is still unpublished. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 57 +++++++++++++++++++++++++ rs/moq-mux/src/container/ts/mux_rate.rs | 5 ++- 2 files changed, 60 insertions(+), 2 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 6cf37a4b9e..9ad781f7db 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -304,11 +304,15 @@ impl Import { self.health.tick(now); } 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(); } } } @@ -897,10 +901,19 @@ impl Import { if self.mux_rate.discontinuity() { self.record_mux_rate()?; } + self.rearm_mux_rate(); tracing::debug!("MPEG-TS system time-base discontinuity"); Ok(()) } + /// A break 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) { @@ -7548,6 +7561,50 @@ pub(super) mod test { 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() { diff --git a/rs/moq-mux/src/container/ts/mux_rate.rs b/rs/moq-mux/src/container/ts/mux_rate.rs index f397df41f1..84d3740145 100644 --- a/rs/moq-mux/src/container/ts/mux_rate.rs +++ b/rs/moq-mux/src/container/ts/mux_rate.rs @@ -76,7 +76,7 @@ pub(super) struct Meter { window: Vec, /// The rate recorded in the catalog, if any. published: Option, - /// Whether the startup measurement window has completed, even for VBR. + /// Whether a window has completed since the last discontinuity, even for VBR. ready: bool, } @@ -86,7 +86,7 @@ impl Meter { self.published } - /// Whether the first PCR measurement window has completed. + /// Whether a PCR measurement window has completed on the current clock. pub fn ready(&self) -> bool { self.ready } @@ -125,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(); From 929cb4351f1d6067d1d376ea37cd7135badb4688 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 20:39:30 -0700 Subject: [PATCH 10/10] fix(ts): re-arm the startup hold on a PCR PID switch and before salvage A PMT that moves the PCR PID resets the meter too, and a time-base break salvages pending PES before resetting it, which could publish the first catalog with the old clock's rate. Both now restore the hold first. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/import.rs | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 9ad781f7db..4edb6a118c 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -430,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 @@ -878,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 @@ -901,13 +901,22 @@ impl Import { if self.mux_rate.discontinuity() { self.record_mux_rate()?; } - self.rearm_mux_rate(); tracing::debug!("MPEG-TS system time-base discontinuity"); Ok(()) } - /// A break 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. + /// 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();