From f745ccf5aaf678c4d89fd9609385520d2e50dbf0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 03:59:43 -0800 Subject: [PATCH 1/5] chore(quest): claim quest/m1/export-broadcast-end.md Quest-Claim: 0653d8bd29baefad From 5a85c31e2659be11a45d2ffa6f19c776afe76ead Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 04:02:04 -0800 Subject: [PATCH 2/5] test(mux): drain finished broadcasts through catalog EOF Finish the timing fixtures catalog and require actual export termination on controlled time. Cover catalog and media finishing in either order and document the catalog lifecycle. Co-Authored-By: GPT-6.1 Sol --- quest/m1/README.md | 1 - quest/m1/export-broadcast-end.md | 30 ------------------------------ 2 files changed, 31 deletions(-) delete mode 100644 quest/m1/export-broadcast-end.md diff --git a/quest/m1/README.md b/quest/m1/README.md index a9ed35f1c1f..5bea6bd063d 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -146,7 +146,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. From c61ae69cb3889011ffb596a196cba6fd241169b5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 04:02:15 -0800 Subject: [PATCH 3/5] test(mux): finish catalogs and assert export termination Co-Authored-By: GPT-6.1 Sol --- doc/bin/cli.md | 7 +- .../src/container/ts/export_timing_test.rs | 70 ++++++++++++++----- 2 files changed, 58 insertions(+), 19 deletions(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index e98f611a13f..70dbd12d068 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -521,9 +521,10 @@ 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 +A stdout export ends when its catalog ends and its media has drained. Finishing +only the media tracks keeps it waiting for catalog updates. 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 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..7f41078e925 100644 --- a/rs/moq-mux/src/container/ts/export_timing_test.rs +++ b/rs/moq-mux/src/container/ts/export_timing_test.rs @@ -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. @@ -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,45 @@ 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); + if catalog_first { + live.catalog.finish().unwrap(); + assert!(!leg.poll(), "an ended catalog must not discard live media"); + } + live.finish_media(); + if !catalog_first { + loop { + assert!(!leg.poll(), "finished media must not end an open catalog"); + 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(), + 2 * GOP as usize + ); + 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(), + 2 * GOP as usize + ); + } +} + /// A PES start as a receiver sees it. #[derive(Debug, Clone, Copy)] struct Unit { From cfdde97d53237a06722039c441f8029386efda09 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 04:11:27 -0800 Subject: [PATCH 4/5] test(mux): exercise media after catalog EOF Drain closed groups while their tracks stay open and require every live poll to stay open. Put media-only lifecycle detail in the exporter reference. Co-Authored-By: GPT-6.1 Sol --- doc/bin/cli.md | 8 +-- rs/moq-mux/src/container/ts/export.rs | 3 +- .../src/container/ts/export_timing_test.rs | 49 ++++++++++++------- 3 files changed, 37 insertions(+), 23 deletions(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 70dbd12d068..ec04fcd4e47 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -521,10 +521,10 @@ 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 when its catalog ends and its media has drained. Finishing -only the media tracks keeps it waiting for catalog updates. 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 +A stdout export ends when its catalog ends and its media has drained. 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 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 7f41078e925..ba3bd4be1c7 100644 --- a/rs/moq-mux/src/container/ts/export_timing_test.rs +++ b/rs/moq-mux/src/container/ts/export_timing_test.rs @@ -376,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; @@ -432,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"); } } @@ -473,23 +473,36 @@ async fn end_waits_for_catalog_and_media() { 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); - if catalog_first { + let expected = if catalog_first { live.catalog.finish().unwrap(); - assert!(!leg.poll(), "an ended catalog must not discard live media"); - } - live.finish_media(); - if !catalog_first { - loop { - assert!(!leg.poll(), "finished media must not end an open catalog"); - 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(), - 2 * GOP as usize + // 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 + ); + 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), @@ -500,7 +513,7 @@ async fn end_waits_for_catalog_and_media() { assert_eq!(leg.export.dropped(), 0); assert_eq!( leg.units().iter().filter(|unit| unit.pid == VIDEO_PID).count(), - 2 * GOP as usize + expected ); } } From b6a677c8be3cd0ecf5301351fb53856653bf6e63 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 09:20:23 -0800 Subject: [PATCH 5/5] test(mux): check exported audio and document the passthrough end Scope the CLI end condition to the muxed exports: `export ts --passthrough` ends with its track while the catalog stays open (release.rs). The lifecycle test now also requires every AAC frame out before and only before EOF. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 15 +++--- .../src/container/ts/export_timing_test.rs | 48 ++++++++++++++++++- 2 files changed, 56 insertions(+), 7 deletions(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index ec04fcd4e47..314cf0f80ab 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -521,12 +521,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 when its catalog ends and its media has drained. 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/rs/moq-mux/src/container/ts/export_timing_test.rs b/rs/moq-mux/src/container/ts/export_timing_test.rs index ba3bd4be1c7..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(); @@ -499,6 +499,8 @@ async fn end_waits_for_catalog_and_media() { 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 { @@ -515,6 +517,11 @@ async fn end_waits_for_catalog_and_media() { 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" + ); } } @@ -1233,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