diff --git a/doc/bin/cli.md b/doc/bin/cli.md index fab9bdecf50..f600f66bd13 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -526,11 +526,15 @@ 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. 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 +A stdout export ends when its catalog ends and its media has drained, so +finishing only the media tracks leaves it waiting for catalog updates. +`export ts --passthrough` instead ends with its track, once the packets it +buffered are out, even while the catalog stays open; only a catalog that ends +without naming the track ends it first. 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 diff --git a/quest/m1/README.md b/quest/m1/README.md index 63ad9130f00..3501f6632e0 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -145,7 +145,6 @@ per-frame CPU growth since #5926. - [moq-c at parity](/quest/m1/c/README.md) - the hand-written `moq-c` ships at the cut with every m1 binding feature moq-ffi gains - [Bindings see a moq-transport run end as an end](/quest/m1/ffi-ietf-run-end.md) - an IETF publisher ending its run reaches moq-ffi consumers as an end, not `Internal` - [Byte-stream renditions form a ladder](/quest/m1/stream-ladder.md) - moq-ffi stream producers link as `lower` and disable unfundable rungs like frame producers -- [moq export ends when the broadcast ends](/quest/m1/export-broadcast-end.md) - the export finishes on its own, proven without a give-up timeout - [A relay answers a SUBSCRIBE on a copy it kept after a fetch](/quest/m1/relay-fetched-copy-subscribe.md) - the downstream SUBSCRIBE_OK waits for the upstream edge instead of hanging - [A job's exact route continues its claim's readers](/quest/m1/announce-takeover.md) - a job's route inherits its claim's identity, replacing the ANNOUNCE_TAKEOVER predecessor design - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges with absolute or relative ends, in either order, and replaces FETCH; merged ranges never starve a subscriber; relays fill misses by range, including over moq-transport diff --git a/quest/m1/export-broadcast-end.md b/quest/m1/export-broadcast-end.md deleted file mode 100644 index 9711cf638d5..00000000000 --- a/quest/m1/export-broadcast-end.md +++ /dev/null @@ -1,30 +0,0 @@ -# [S] moq export ends when the broadcast ends - -## Goal - -`moq export` finishes on its own once the broadcast it reads ends. A test -proves this without a give-up timeout. - -## Plan - -Planned 2026-11-01 (maintainer, follow-up planning). Found while iterating -[moq-dev/moq#6987](https://github.com/moq-dev/moq/pull/6987): - -- In all 20 drains of its export timing tests, the export never reported the - end of the broadcast after the video and audio tracks finished. An older - version of the test helper ended fine, so something on main changed this. -- Main's copy of the helper gives up after advancing the clock 10 ms, which - hides the problem. -- It may only be the export waiting on the catalog track, which those tests - never close. - -First find out which it is. - -- **If it's a real bug,** an export that never ends after a finished - broadcast, fix it. -- **If the export correctly waits for the catalog,** make the tests finish - their catalog. Document in `doc/bin/cli.md` (or wherever `moq export` is - documented) that export ends when the catalog does. - -Either way, the timing tests then read to the export's real end, and the -10 ms give-up is removed. diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 67eb50234d6..430474037ee 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. A publisher that leaves instead fails the +/// Returns `None` once the catalog ends and its media drains. Finishing only the media +/// tracks keeps it waiting for catalog updates. 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 diff --git a/rs/moq-mux/src/container/ts/export_timing_test.rs b/rs/moq-mux/src/container/ts/export_timing_test.rs index 1392ff3828e..22671a3cf81 100644 --- a/rs/moq-mux/src/container/ts/export_timing_test.rs +++ b/rs/moq-mux/src/container/ts/export_timing_test.rs @@ -325,7 +325,7 @@ impl Live { .write(Frame { timestamp: Timestamp::from_micros((index * AUDIO_US).strict_add_signed(self.offset)).unwrap(), duration: None, - payload: Bytes::from_iter((0..180u16).map(|i| (i ^ index as u16) as u8)), + payload: audio_payload(index), keyframe: index.is_multiple_of(AUDIO_GROUP), }) .unwrap(); @@ -347,15 +347,21 @@ impl Live { /// End the broadcast and let every leg render what it holds. fn finish(&mut self, legs: &mut [&mut Leg]) { + self.finish_media(); + self.catalog.finish().unwrap(); + for leg in legs.iter_mut() { + leg.drain(&self.clock); + } + } + + /// End the media tracks while the catalog can still announce more. + fn finish_media(&mut self) { if let Some(mut group) = self.group.take() { group.finish().unwrap(); } self.stalled.clear(); self.video.finish().unwrap(); self.audio.finish().unwrap(); - for leg in legs.iter_mut() { - leg.drain(&self.clock); - } } /// Run the clock to `at`, letting every leg read whenever it would have on its own. @@ -370,7 +376,7 @@ impl Live { let now = kio::clock::now(); for leg in legs.iter_mut() { if leg.wake().is_some_and(|wake| wake <= now) { - leg.poll(); + assert!(!leg.poll(), "export ended while the broadcast is live"); if !leg.every.is_zero() { while leg.next_poll <= now { leg.next_poll += leg.every; @@ -426,7 +432,7 @@ impl Leg { /// nothing woke finds nothing new. fn poll_now(&mut self) { if self.every.is_zero() && self.woken.0.load(Ordering::Relaxed) { - self.poll(); + assert!(!self.poll(), "export ended while the broadcast is live"); } } @@ -443,21 +449,14 @@ impl Leg { false } - /// Read to the end of the broadcast, or until nothing is due and 10 ms pass with no end. - /// - /// Only `clock` fires the export's timers, so the 10 ms is an advance of it: a timeout armed - /// on it would never fire. + /// Read to the end of the broadcast, advancing only to scheduled output. fn drain(&mut self, clock: &kio::clock::Deadlines) { while !self.poll() { - if let Some(deadline) = self.export.next_due() { - clock.advance(deadline.max(clock.now())); - continue; - } - clock.advance(clock.now() + Duration::from_millis(10)); - if self.export.next_due().is_none() { - self.poll(); - break; - } + let deadline = self + .export + .next_due() + .expect("export parked without ending or scheduled output"); + clock.advance(deadline.max(clock.now())); } } @@ -467,6 +466,65 @@ impl Leg { } } +/// Media drains while the catalog stays open; EOF needs both, in either finish order. +#[tokio::test(start_paused = true)] +async fn end_waits_for_catalog_and_media() { + for catalog_first in [false, true] { + let mut live = Live::new(1.0, 0).await; + let mut leg = live.join(Duration::ZERO).await; + live.run(2 * GOP, &mut [&mut leg], |_| false); + let expected = if catalog_first { + live.catalog.finish().unwrap(); + // Catalog EOF fixes the layout, but the named tracks can still publish media. + live.run(3 * GOP, &mut [&mut leg], |_| false); + // Close the groups so their tails flush, leaving both tracks open for more. + live.group.take().unwrap().finish().unwrap(); + live.audio.cut(None).unwrap(); + 3 * GOP as usize + } else { + live.finish_media(); + 2 * GOP as usize + }; + loop { + assert!( + !leg.poll(), + "EOF before both catalog and media finish: catalog_first {catalog_first}" + ); + let Some(deadline) = leg.export.next_due() else { + break; + }; + live.clock.advance(deadline.max(live.clock.now())); + } + assert_eq!( + leg.units().iter().filter(|unit| unit.pid == VIDEO_PID).count(), + expected + ); + let audio: Vec = (0..live.audio_index).map(audio_payload).collect(); + assert_eq!(aac_frames(&leg), audio, "catalog_first {catalog_first}: audio drained"); + if catalog_first { + live.finish_media(); + } else { + assert!(!leg.woken.0.load(Ordering::Relaxed), "the exporter is parked"); + live.catalog.finish().unwrap(); + assert!( + leg.woken.0.load(Ordering::Relaxed), + "catalog EOF must wake the exporter" + ); + } + leg.drain(&live.clock); + assert_eq!(leg.export.dropped(), 0); + assert_eq!( + leg.units().iter().filter(|unit| unit.pid == VIDEO_PID).count(), + expected + ); + assert_eq!( + aac_frames(&leg), + audio, + "catalog_first {catalog_first}: no audio after EOF" + ); + } +} + /// A PES start as a receiver sees it. #[derive(Debug, Clone, Copy)] struct Unit { @@ -1182,6 +1240,45 @@ async fn live_audio_shares_its_pes() { } } +/// The raw AAC of the fixture's audio frame `index`. +fn audio_payload(index: u64) -> Bytes { + Bytes::from_iter((0..180u16).map(|i| (i ^ index as u16) as u8)) +} + +/// Each AAC frame a leg rendered, in output order, with its ADTS header removed. +fn aac_frames(leg: &Leg) -> Vec { + let bytes: Vec = leg + .out + .iter() + .flat_map(|(_, frame)| frame.payload.iter().copied()) + .collect(); + let mut reader = TsPacketReader::new(Cursor::new(bytes)); + let mut pes: Vec> = Vec::new(); + while let Some(packet) = reader.read_ts_packet().unwrap() { + if packet.header.pid.as_u16() != AUDIO_PID { + continue; + } + match packet.payload { + Some(TsPayload::PesStart(start)) => pes.push(start.data.to_vec()), + Some(TsPayload::PesContinuation(rest)) => { + pes.last_mut().expect("a PES start first").extend_from_slice(&rest) + } + _ => {} + } + } + let mut frames = Vec::new(); + for data in &pes { + // One PES carries several frames. + let mut data = &data[..]; + while !data.is_empty() { + let header = super::adts::Header::parse(data).unwrap(); + frames.push(Bytes::copy_from_slice(&data[header.header_len..header.frame_len])); + data = &data[header.frame_len..]; + } + } + frames +} + /// Each video frame's DTS (else PTS) by its PTS, in 90 kHz ticks, as a leg rendered it. fn decode_times(leg: &Leg) -> std::collections::BTreeMap { let bytes: Vec = leg