diff --git a/doc/bin/cli.md b/doc/bin/cli.md index af7a2515717..1fc887a5d2b 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -483,12 +483,13 @@ continuity error, and the rest carry on. It takes neither `--follow`, `--stitch` nor `--mux-rate`. Only the hang catalog names the track, so an `.msf` broadcast needs `--catalog-format hang`. -A stdout export ends with the broadcast. `export ts --follow 10s` waits that -long for the same publisher instance to come back, and carries on with the same -stream. A return that refuses its catalog keeps it waiting for the next -announcement; if the follow runs out first, it exits 1 with that refusal. One -that goes before serving, such as one still routed through a dead session, is -no return. A replacement (a restarted publisher, under a new +A stdout export ends with the broadcast. A publisher that leaves still gets +what the export buffered of it out on schedule first. `export ts --follow 10s` +then waits that long for the same publisher instance to come back, and carries +on with the same stream. A return that refuses its catalog keeps it waiting +for the next announcement; if the follow runs out first, it exits 1 with that +refusal. One that goes before serving, such as one still routed through a dead +session, is no return. A replacement (a restarted publisher, under a new [epoch](/concept/moq-lite#publisher-epochs)) exits 1 unless `--stitch` follows it as a full program switch: a new PMT from its catalog, with every PID flagging the break. A stitch waits for the replacement's first catalog snapshot until it diff --git a/quest/m0/nightly/README.md b/quest/m0/nightly/README.md index 5d6a764872d..d4073bc8b77 100644 --- a/quest/m0/nightly/README.md +++ b/quest/m0/nightly/README.md @@ -35,7 +35,6 @@ Public API: none. Wire: none. - [The software renderer tests both capability paths](/quest/m0/nightly/render-capabilities.md) - the software-renderer lane verifies output selection with and without DMA-BUF import support without requiring hardware the runner lacks - [The browser publisher completes its fairness sweep](/quest/m0/nightly/publisher-fairness.md) - the publisher fairness benchmark completes its backlog on the hosted ARM runner, yielding browser tasks while serving every group - [The browser backend baseline completes on IETF22](/quest/m0/nightly/browser-baseline.md) - the WASM and UniFFI backend baseline completes every supported transport case without an internal error -- [A stitched restart signals every TS discontinuity](/quest/m0/nightly/stitch-discontinuity.md) - a restarted publisher exported with stitching signals the program break on every affected PID - [Resolve the paired catalog fanout regressions](/quest/m0/nightly/paired-shared.md) - the reported catalog fanout regressions are attributed and genuine extra costs are fixed with paired measurements - [Resolve the paired relay-session regressions](/quest/m0/nightly/paired-relay.md) - the reported relay-session regressions are attributed and genuine extra costs are fixed with paired measurements - [Probing beside Cubic](/quest/m0/nightly/competing-cubic.md) - the `competing_cubic` probe test stays within its deadline budget diff --git a/quest/m0/nightly/stitch-discontinuity.md b/quest/m0/nightly/stitch-discontinuity.md deleted file mode 100644 index a0ffdd33eb5..00000000000 --- a/quest/m0/nightly/stitch-discontinuity.md +++ /dev/null @@ -1,20 +0,0 @@ -# [M] A stitched restart signals every TS discontinuity - -## Goal - -A restarted publisher exported with stitching signals the program break on every affected PID. - -## Plan - -[The CLI-export job](https://github.com/moq-dev/moq/actions/runs/40698329026/job/122851001566) fails -`export::a_restarted_publisher_is_a_program_switch_with_stitch` at -"every PID flags the break": expected PIDs `[256, 257, 17]`, found none. - -Reproduce on current main and trace the replacement program through the -follower and transport-stream muxer. Fix discontinuity signaling at its source -without weakening the end-to-end assertion. Coordinate with -[moq-dev/moq#7049](https://github.com/moq-dev/moq/pull/7049), which changes -how a stitched switch drops the old buffered tail; it does not claim this -PID-signaling failure. - -Public API: none expected. Wire: none expected. diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 1084f09a5f4..843ca561702 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -106,7 +106,8 @@ fn sanitize_mux_rate(rate: u64) -> Option { /// is re-emitted at video keyframes and periodically for mid-stream tune-in. SI and /// whole PES or sections wait until a PCR interval has followed it, since a reader /// places them on the program clock and the PMT is what names it. -/// Returns `None` when the broadcast ends. +/// Returns `None` when the broadcast ends. A publisher that leaves instead fails the +/// export with its cause once what was already buffered has gone out on schedule. /// /// A rendition the catalog moves to another track (a new `track` or `broadcast`) goes on out /// of its PID from the new track's first sync frame past what was read, when the PMT can @@ -133,6 +134,8 @@ where { id: u64, retired: bool, + /// Why the publisher left, reported once the output it buffered has gone out. + left: Option, expected: Vec, expected_changed: bool, source: crate::Source, @@ -1316,6 +1319,7 @@ where Self { id: 0, retired: false, + left: None, expected: Vec::new(), expected_changed: false, source, @@ -1467,11 +1471,7 @@ where self.stop_old(waiter)?; } if let Err(err) = self.prepare(waiter) { - if self.pending.is_none() || !super::follower::left(&err) { - return Poll::Ready(Err(err)); - } - self.follow_until = Some(moq_clock::now()); - self.stop_old(waiter)?; + self.leave(err)?; } if self.pending.is_some() { loop { @@ -1643,7 +1643,7 @@ where // a zero-delay slot may need another frame to prove the slot is complete. let track = self.program.tracks.get_mut(&name).expect("the released frame's track"); let finished = track.finished; - track.fill( + let filled = track.fill( &name, waiter, &mut Input { @@ -1652,8 +1652,12 @@ where jitter: &mut self.jitter, trimmed: &mut self.program.trimmed, }, - )?; - if !finished && track.finished { + ); + let ended = !finished && track.finished; + if let Err(err) = filled { + self.leave(err)?; + } + if ended || self.program.retired { self.program .schedule .set_ended(self.program.tracks.values().all(|track| track.finished)); @@ -1726,7 +1730,7 @@ where // End of stream once the catalog is closed too: nothing more can appear. if self.pending.is_none() && self.program.catalog_ended && (drained || self.program.tracks.is_empty()) { - return Poll::Ready(Ok(None)); + return Poll::Ready(self.program.left.take().map_or(Ok(None), Err)); } Poll::Pending @@ -1921,6 +1925,26 @@ where Ok(true) } + /// The program playing failed with `err`. Only a publisher that left is survivable: its + /// input stops at once, as a follow that runs out stops it, but what it already buffered + /// goes out on schedule. Without a replacement, the export then reports `err`, so a + /// [`Follower`](super::Follower) waits for a return only after the old output has ended, + /// rather than leaving it buffered to go out late, behind the gap, once something returns. + /// + /// The program keeps its catalog, whose format the follower subscribes a return with. + fn leave(&mut self, err: crate::Error) -> crate::Result<()> { + if !super::follower::left(&err) { + return Err(err); + } + tracing::debug!(target: "moq_mux::container::ts::handoff", phase = "left", now = ?moq_clock::now(), program = self.program.id, committed = self.jitter.is_committed(), %err); + if self.pending.is_none() { + self.program.left = Some(err); + } + self.follow_timer = None; + self.follow_until = None; + self.retire() + } + fn stop_old(&mut self, waiter: &kio::Waiter) -> crate::Result<()> { if self.program.retired { return Ok(()); @@ -1940,10 +1964,16 @@ where return Ok(()); } tracing::debug!(target: "moq_mux::container::ts::handoff", phase = "follow-expired", now = ?moq_clock::now(), program = self.program.id, committed = self.jitter.is_committed()); - self.program.retired = true; self.follow_timer = None; self.follow_until = None; self.program.catalog = None; + self.retire() + } + + /// Stop the program's input. What it already buffered still goes out, unless a + /// committed replacement's switch drops it. + fn retire(&mut self) -> crate::Result<()> { + self.program.retired = true; self.program.catalog_ended = true; for (name, track) in &mut self.program.tracks { if !track.finished { @@ -2628,6 +2658,7 @@ where self.program.catalog_ended = false; self.program.cataloged = false; self.program.retired = false; + self.program.left = None; self.program.si_flushed = false; for si in self.program.si.values_mut() { si.track.clear(); diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 38576a15883..53e91af2c96 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -11602,6 +11602,74 @@ async fn an_expired_follow_stops_healthy_input_but_drains_its_buffered_tail() { assert!(gap > delay, "no input can bridge B's post-deadline wait: {gap:?}"); } +/// A publisher that leaves still gets what the export buffered of it out on schedule, and only +/// then does the export report it gone. Reported at once, the tail sat unpolled while a +/// follower waited, and went out late, behind the gap, ahead of whatever came back. +#[tokio::test(start_paused = true)] +async fn a_publisher_that_leaves_drains_its_buffered_tail_before_the_end() { + let origin = crate::source::produce_origin(); + let (old, mut catalog) = publish_live::<()>(&origin, Default::default()); + let audio = aac_rendition(&mut old.broadcast().clone(), &mut catalog, "a.aac"); + let mut publisher = Some((catalog, audio)); + let delay = Duration::from_millis(500); + let left = Duration::from_millis(2000); + let step = Duration::from_millis(20); + let mut export = open_export(crate::source::resolved(&origin, "live").await) + .await + .unwrap() + .with_delay(delay); + let start = tokio::time::Instant::now(); + let mut frames = Vec::new(); + let end = loop { + let elapsed = start.elapsed(); + if elapsed < left { + write_aac(&mut publisher.as_mut().unwrap().1, elapsed.as_millis() as u64); + } else if let Some((catalog, audio)) = publisher.take() { + audio.abort(moq_net::Error::Unroutable); + catalog.abort(moq_net::Error::Unroutable); + } + tokio::task::yield_now().await; + let end = loop { + match export.poll_next(&kio::Waiter::noop()) { + std::task::Poll::Ready(Ok(Some(frame))) => frames.push((start.elapsed(), frame)), + std::task::Poll::Ready(end) => break Some(end), + std::task::Poll::Pending => break None, + } + }; + if let Some(end) = end { + break end; + } + assert!( + elapsed < left + 4 * delay, + "the export never reported the publisher gone" + ); + tokio::time::advance(step).await; + }; + let ended = start.elapsed(); + assert!( + matches!(&end, Err(err) if super::follower::left(err)), + "the end reports the publisher leaving: {end:?}" + ); + let slices: Vec = frames.iter().map(|(_, frame)| frame.clone()).collect(); + // When each audio PTS went out, after the publisher left. + let tail: Vec<(Duration, u64)> = frames + .iter() + .zip(audio_pts_by_slice(&slices)) + .filter(|((at, _), _)| *at >= left) + .flat_map(|((at, _), pts)| pts.into_iter().map(move |pts| (*at, pts))) + .collect(); + let (last, pts) = *tail.last().expect("a tail went out after the publisher left"); + assert!( + pts >= (60_000 + left.as_millis() as u64 - 100) * 90, + "the tail buffered when the publisher left goes out: {tail:?}" + ); + assert!( + last >= left + delay - PCR_INTERVAL - step, + "the tail goes out on schedule, not in a burst: {tail:?}" + ); + assert!(ended >= last, "the end follows the tail"); +} + /// A replacement that leaves while it acquires is no switch: the healthy old program plays on. #[tokio::test(start_paused = true)] async fn a_replacement_leaving_before_it_commits_keeps_the_old_program() {