diff --git a/quest/m1/README.md b/quest/m1/README.md index fed911f4ff8..8c336b62209 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -142,7 +142,6 @@ Catalog HLS field. - [Generated JS API reference](/quest/m1/js-api-reference.md) - the doc site renders an API reference for every `@moq/*` package from API Extractor's doc model - [Interop coverage](/quest/m1/interop-coverage/README.md) - the riskiest moq-transport features moq-net exposes get cross-implementation tests over drafts 18 and 22 - [AMF0 exact reads](/quest/m1/rtmp-amf0-exact-reads.md) - a truncated RTMP string fails with `UnexpectedEof` instead of parsing NUL padding -- [DASH init readiness](/quest/m1/dash-init-ready.md) - a DASH manifest lists the renditions whose init exists instead of waiting on one that is stuck - [Calibrate the paired nightly benchmarks](/quest/m1/bench-ci.md) - A/A and deliberately slowed Nightly runs set the paired comparison's threshold, so its regression alerts can be trusted - [HLS takeover leak](/quest/m1/hls-takeover-leak.md) - a replaced instance's recording cursor never yields the replacement's media, and ends or fails instead of stalling - [HLS and DASH segment conformance](/quest/m1/hls-conformance.md) - nightly ffprobe checks that advertised timing, sync flags, and bytes match the media served diff --git a/quest/m1/dash-init-ready.md b/quest/m1/dash-init-ready.md deleted file mode 100644 index 36364e4f91f..00000000000 --- a/quest/m1/dash-init-ready.md +++ /dev/null @@ -1,29 +0,0 @@ -# [S] A DASH manifest never waits on one rendition's init - -## Goal - -Rendering a DASH manifest never blocks on a rendition whose init segment is -not known yet. It lists every enabled rendition whose init exists and leaves -out the rest until a later refresh. Today `build_inits` in -`rs/moq-hls/src/export/mod.rs` awaits each rendition's `latest_init` serially, -disabled ones included, after the readiness wait and with no bound of its own, -so an inline-codec rendition bootstrapping from a still-open group holds up -the whole manifest despite healthy alternatives. - -## Plan - -From the 2026-10-24 moq-hls audit. Decided with the maintainer on 2026-10-24: -report current state, rather than build inits concurrently under the existing -readiness wait, which still lets one stuck rendition hold the manifest. - -- Consider enabled renditions only, and gather their inits concurrently - without letting any one of them gate the render. -- A rendition whose init is not built yet is left out of this MPD (the render - already drops a rendition with no init) and appears once its init exists. -- The `/init.*.mp4` route has the same unbounded wait; it answers only once - the init exists. -- Tests with mocked time: one rendition stuck on an open group, the manifest - renders its siblings immediately, and the stuck one appears after its group - closes; a disabled rendition's init is never built. - -Public API: none. Wire: none. diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 9667da1f6ad..40b634063d4 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -15,7 +15,8 @@ //! handoff continues past the last advertised record sequence, starting a DASH period. Every //! other rendition resolves each segment against its own timeline. //! An inline-parameter-set codec with no catalog `description` GETs one keyframe -//! group on the first playlist render to build its init, then caches it. +//! group on the first playlist render (in the background on a manifest render) to build its +//! init, then caches it. //! //! The same machinery serves two kinds of consumer: //! @@ -273,8 +274,10 @@ impl Broadcaster { /// in [`master_playlist`](Self::master_playlist). /// /// A disabled rendition is left out, as in the master playlist. Each representation names - /// its init by a hash of the bytes, so a rendition is left out until its init is known. An inline-parameter-set codec only learns it from media: await - /// each rendition's [`init`](Rendition::init) first. + /// its init by a hash of the bytes, so a rendition is left out until its init is known. An + /// inline-parameter-set codec only learns it from media, so rendering starts building each + /// missing init in the background and a later render lists that rendition; no render waits + /// on one. /// /// Fails with [`TimelineFailed`](crate::Error::TimelineFailed) once the reference timeline /// failed, or [`CatalogFailed`](crate::Error::CatalogFailed) once the catalog did, rather @@ -287,6 +290,8 @@ impl Broadcaster { let mut video = Vec::new(); let mut audio = Vec::new(); let mut availability_start = None; + // An enabled rendition whose init is still building, which a later render lists. + let mut pending = false; for rendition in self.renditions.snapshot() { // Every rendition shares the catalog's one root clock, so any of them can // supply the declared wall anchor. @@ -296,8 +301,10 @@ impl Broadcaster { if !rendition.is_enabled() { continue; } - // Without its init there is nothing to name in `initialization`; the serve path - // builds every init before rendering (see `build_inits`). + // Without its init there is nothing to name in `initialization`, so it is left out + // of this render. + rendition.start_init(); + pending |= rendition.init_pending(); let representations = match rendition.representations() { Ok(representations) => representations, // Its own playlist answers the error, while its siblings stay listed. @@ -314,8 +321,9 @@ impl Broadcaster { if representations().all(|rep| rep.segments.is_empty()) { return Ok(None); } - // Static only once every representation is final, so none stops being refreshed early. - let finished = representations().all(|rep| rep.ended); + // Static only once every representation is final and no init is still building, so + // neither a listed nor a pending rendition stops being refreshed early. + let finished = !pending && representations().all(|rep| rep.ended); if !finished { // A dynamic presentation needs pts 0 anchored to the wall clock. The fallback is // set whenever a record has been pushed, which listing a segment implies. @@ -339,18 +347,16 @@ impl Broadcaster { ))) } - /// Build every rendition's init segment, so the manifest can name each by its hash. A - /// rendition whose init can't be built yet is left out of the render. + /// Build video inits until one enabled video rendition is listable, or none is left that + /// could be. A DASH player picks its adaptation sets from the first manifest it loads, so that + /// one must not lack video only because no init was built yet, while any one built init is + /// enough to render at once. Bounding the wait is the caller's policy. #[cfg_attr( not(feature = "server"), allow(dead_code, reason = "only the `server` feature uses it") )] - pub(crate) async fn build_inits(&self) { - for rendition in self.renditions.snapshot() { - if let Err(err) = rendition.latest_init().await { - tracing::warn!(rendition = %rendition.name, %err, "failed to build init segment"); - } - } + pub(crate) async fn video_init(&self) { + self.renditions.video_init().await; } /// Resolve once the broadcast has something listable (its first complete segment, or an @@ -1410,7 +1416,7 @@ mod tests { assert_eq!(after.segments[1].segment, 2); assert!(after.segments[1].discontinuity); assert!(!after.finished); - broadcaster.build_inits().await; + build_inits(&broadcaster).await; let playlist = sibling.media_playlist(None).unwrap().unwrap(); assert_eq!(playlist.matches("#EXT-X-DISCONTINUITY\n").count(), 1, "{playlist}"); let mpd = broadcaster.manifest(None).unwrap().unwrap(); @@ -1554,8 +1560,8 @@ mod tests { "one numbering on every edge" ); if resumed == 3 { - saw.build_inits().await; - after.build_inits().await; + build_inits(&saw).await; + build_inits(&after).await; let saw_playlist = saw_video1.media_playlist(None).unwrap().unwrap(); assert!(saw_playlist.contains("#EXT-X-MEDIA-SEQUENCE:4\n"), "{saw_playlist}"); assert_eq!(saw_playlist, after_video1.media_playlist(None).unwrap().unwrap()); @@ -1628,6 +1634,361 @@ mod tests { assert!(!master.contains("video/video1/"), "{master}"); } + /// Build every rendition's init, as a playlist or segment request does. + async fn build_inits(broadcaster: &Broadcaster) { + for rendition in broadcaster.renditions.snapshot() { + rendition.latest_init().await.unwrap(); + } + } + + /// `timeline_pair`, with `video1` learning its geometry from media so its init needs a + /// keyframe group, and both timelines indexing four groups. `video1`'s last record is split + /// inside group 3, as for a group still open past the timeline's ceiling. + async fn inline_sibling() -> ( + moq_net::origin::Producer, + moq_net::origin::Publication, + moq_mux::catalog::Producer, + [moq_mux::timeline::Producer; 2], + ) { + let (origin, broadcast, mut catalog) = timeline_pair().await; + catalog + .mutate(|catalog| { + let config = catalog.video.renditions.get_mut("video1").unwrap(); + config.coded_width = None; + config.coded_height = None; + }) + .unwrap(); + let (_, mut video0) = timeline_track(&broadcast, "video0"); + let (_, mut video1) = timeline_track(&broadcast, "video1"); + for sequence in 0..4 { + video0.push(&gop(sequence)).unwrap(); + } + for sequence in 0..3 { + video1.push(&gop(sequence)).unwrap(); + } + let split = hang::timeline::Record::new( + 3, + 6_000, + 2_000, + hang::timeline::Position::group(3), + hang::timeline::Position::new(3, 1), + ); + video1.push(&split).unwrap(); + (origin, broadcast, catalog, [video0, video1]) + } + + /// Render the manifest until it lists `video1`. + async fn listed_in_manifest(broadcaster: &Broadcaster) -> String { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let manifest = broadcaster.manifest(None).unwrap().unwrap(); + if manifest.contains("id=\"video/video1\"") { + return manifest; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("video1 is listed once its init is built") + } + + /// Write one VP8 keyframe at `sequence`'s [`gop`] pts into `group`. + fn write_keyframe(group: &mut moq_net::group::Producer, sequence: u64) { + let frame = vp8_frame(sequence * 2_000_000, true); + hang::container::Frame { + timestamp: frame.timestamp, + payload: frame.payload, + } + .write_to(group) + .unwrap(); + } + + // An inline init bootstraps from the newest closed keyframe group, read from cache, rather + // than waiting on the newest group the publisher is still writing. + #[tokio::test(start_paused = true)] + async fn an_init_bootstraps_from_a_closed_group() { + let (origin, broadcast, _catalog, _timelines) = inline_sibling().await; + let media = broadcast.create_track("video1", timed()).unwrap(); + for sequence in 0..3 { + let mut group = media.create_group(sequence.into()).unwrap(); + write_keyframe(&mut group, sequence); + group.finish().unwrap(); + } + let mut open = media.create_group(3u64.into()).unwrap(); + write_keyframe(&mut open, 3); + + let broadcaster = Broadcaster::new(crate::resolved(&origin, "live").await, Config::default()); + broadcaster.advertised().await; + let tag = segments::tag(&(Kind::Video, "video0".into(), "video0".into())); + listed_under(&broadcaster.rendition(Kind::Video, "video0").unwrap(), &tag, 3).await; + listed_in_manifest(&broadcaster).await; + drop(open); + } + + // A rendition whose init bootstraps from a still-open group never holds up the manifest: its + // sibling renders at once, and it joins once the group closes. Its closed groups already + // left the cache, so the open one is the only candidate left. + #[tokio::test(start_paused = true)] + async fn a_stuck_init_leaves_the_manifest_to_its_siblings() { + let (origin, broadcast, _catalog, _timelines) = inline_sibling().await; + let media = broadcast.create_track("video1", timed()).unwrap(); + let mut open = media.create_group(3u64.into()).unwrap(); + write_keyframe(&mut open, 3); + + let broadcaster = Broadcaster::new(crate::resolved(&origin, "live").await, Config::default()); + broadcaster.advertised().await; + let tag = segments::tag(&(Kind::Video, "video0".into(), "video0".into())); + listed_under(&broadcaster.rendition(Kind::Video, "video0").unwrap(), &tag, 3).await; + tokio::time::timeout(Duration::from_millis(1), broadcaster.video_init()) + .await + .expect("one video init is enough to render"); + let manifest = broadcaster.manifest(None).unwrap().expect("video0 renders"); + assert!(manifest.contains("id=\"video/video0\""), "{manifest}"); + assert!(!manifest.contains("id=\"video/video1\""), "{manifest}"); + + // The background build is parked on the open group, and every render stays current. + tokio::time::sleep(Duration::from_secs(1)).await; + let manifest = broadcaster.manifest(None).unwrap().expect("video0 renders"); + assert!(!manifest.contains("id=\"video/video1\""), "{manifest}"); + + open.finish().unwrap(); + let manifest = listed_in_manifest(&broadcaster).await; + assert!(manifest.contains("id=\"video/video0\""), "{manifest}"); + } + + // A background build parked on an open group holds neither its rendition nor the export's + // timeline watcher, so dropping the broadcaster releases both. + #[tokio::test(start_paused = true)] + async fn a_stuck_init_build_never_outlives_its_rendition() { + let (origin, broadcast, _catalog, _timelines) = inline_sibling().await; + let media = broadcast.create_track("video1", timed()).unwrap(); + let mut open = media.create_group(3u64.into()).unwrap(); + write_keyframe(&mut open, 3); + + let broadcaster = Broadcaster::new(crate::resolved(&origin, "live").await, Config::default()); + broadcaster.advertised().await; + let tag = segments::tag(&(Kind::Video, "video0".into(), "video0".into())); + listed_under(&broadcaster.rendition(Kind::Video, "video0").unwrap(), &tag, 3).await; + let stuck = broadcaster.rendition(Kind::Video, "video1").unwrap(); + stuck.start_init(); + tokio::time::sleep(Duration::from_secs(1)).await; + assert!(stuck.init_pending(), "the build is parked on the open group"); + let weak = Arc::downgrade(&stuck); + drop(stuck); + let watcher = broadcaster.timeline_watcher.clone(); + + drop(broadcaster); + assert!(weak.upgrade().is_none(), "the build kept its rendition alive"); + assert!( + watcher.lock().unwrap().is_none(), + "the build kept the timeline watcher running" + ); + drop(open); + } + + // A synchronous render may run on a thread outside the runtime, and still starts the + // background init build there. + #[tokio::test(start_paused = true)] + async fn a_manifest_renders_off_the_runtime() { + let (origin, broadcast, _catalog, _timelines) = inline_sibling().await; + let media = broadcast.create_track("video1", timed()).unwrap(); + let mut open = media.create_group(3u64.into()).unwrap(); + write_keyframe(&mut open, 3); + + let broadcaster = Broadcaster::new(crate::resolved(&origin, "live").await, Config::default()); + broadcaster.advertised().await; + let tag = segments::tag(&(Kind::Video, "video0".into(), "video0".into())); + listed_under(&broadcaster.rendition(Kind::Video, "video0").unwrap(), &tag, 3).await; + let render = broadcaster.clone(); + std::thread::spawn(move || render.manifest(None).map(|_| ())) + .join() + .expect("the render panicked") + .unwrap(); + let video1 = broadcaster.rendition(Kind::Video, "video1").unwrap(); + assert!(video1.init_pending(), "the render started no init build"); + drop(open); + } + + // A new generation releases a build parked on the old run's open group, so the new run can + // build its own init. + #[tokio::test(start_paused = true)] + async fn a_new_generation_releases_a_stuck_init_build() { + let (origin, broadcast, _catalog, _timelines) = inline_sibling().await; + let media = broadcast.create_track("video1", timed()).unwrap(); + let mut open = media.create_group(3u64.into()).unwrap(); + write_keyframe(&mut open, 3); + + let broadcaster = Broadcaster::new(crate::resolved(&origin, "live").await, Config::default()); + broadcaster.advertised().await; + broadcaster.set_generation(Some("run1")).unwrap(); + let tag = segments::tag(&(Kind::Video, "video0".into(), "video0".into())); + listed_under(&broadcaster.rendition(Kind::Video, "video0").unwrap(), &tag, 3).await; + let stuck = broadcaster.rendition(Kind::Video, "video1").unwrap(); + stuck.start_init(); + tokio::time::sleep(Duration::from_secs(1)).await; + assert!(stuck.init_pending(), "the build is parked on the open group"); + + broadcaster.set_generation(Some("run2")).unwrap(); + tokio::time::sleep(Duration::from_secs(1)).await; + assert!(!stuck.init_pending(), "the old run's build still holds the init lock"); + drop(open); + } + + // An ended broadcast stays dynamic while an enabled rendition's init is still building, so a + // player keeps refreshing until it is listed instead of loading a static presentation + // without it. + #[tokio::test(start_paused = true)] + async fn an_ended_manifest_stays_dynamic_while_an_init_builds() { + let (origin, broadcast, mut catalog, mut timelines) = inline_sibling().await; + let media = broadcast.create_track("video1", timed()).unwrap(); + let mut open = media.create_group(3u64.into()).unwrap(); + write_keyframe(&mut open, 3); + + let broadcaster = Broadcaster::new(crate::resolved(&origin, "live").await, Config::default()); + broadcaster.advertised().await; + let rendition = broadcaster.rendition(Kind::Video, "video0").unwrap(); + let tag = segments::tag(&(Kind::Video, "video0".into(), "video0".into())); + listed_under(&rendition, &tag, 3).await; + for timeline in &mut timelines { + timeline.finish().unwrap(); + } + catalog.finish().unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + while !rendition.snapshot().finished { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the playlist ends with the source"); + + let manifest = broadcaster.manifest(None).unwrap().expect("video0 renders"); + assert!(!manifest.contains("id=\"video/video1\""), "{manifest}"); + assert!(!manifest.contains("type=\"static\""), "{manifest}"); + + open.finish().unwrap(); + let manifest = listed_in_manifest(&broadcaster).await; + assert!(manifest.contains("type=\"static\""), "{manifest}"); + } + + // The first manifest waits for a video it can list: not for an init alone, which a sibling + // with no listed segment has at once, and not only until the catalog closes. + #[tokio::test(start_paused = true)] + async fn the_first_video_wait_needs_a_listable_video() { + let (origin, broadcast, mut catalog) = timeline_pair().await; + catalog + .mutate(|catalog| { + let config = catalog.video.renditions.get_mut("video0").unwrap(); + config.coded_width = None; + config.coded_height = None; + }) + .unwrap(); + // The inline reference indexes media; video1's init builds from the catalog alone, but + // its timeline resolves no segment. + let (_video0_track, mut video0) = timeline_track(&broadcast, "video0"); + let (_video1_track, _video1) = timeline_track(&broadcast, "video1"); + for sequence in 0..3 { + video0.push(&gop(sequence)).unwrap(); + } + let media = broadcast.create_track("video0", timed()).unwrap(); + // Its earlier groups already left the cache, so the open one is the only candidate. + let mut open = media.create_group(2u64.into()).unwrap(); + write_keyframe(&mut open, 2); + + let broadcaster = Broadcaster::new(crate::resolved(&origin, "live").await, Config::default()); + let _ = tokio::time::timeout(Duration::from_secs(5), broadcaster.ready()).await; + let tag = segments::tag(&(Kind::Video, "video0".into(), "video0".into())); + listed_under(&broadcaster.rendition(Kind::Video, "video0").unwrap(), &tag, 1).await; + catalog.finish().unwrap(); + settle().await; + + let mut waiting = std::pin::pin!(broadcaster.video_init()); + assert!( + tokio::time::timeout(Duration::from_secs(1), waiting.as_mut()) + .await + .is_err(), + "resolved without a listable video" + ); + open.finish().unwrap(); + tokio::time::timeout(Duration::from_secs(1), waiting) + .await + .expect("video0 is listable once its init is built"); + } + + // An inline video with nothing indexed when the wait starts has nothing to build from yet. + // The wait starts its build once the timeline arrives, rather than timing out without video. + #[tokio::test(start_paused = true)] + async fn the_first_video_wait_builds_an_init_indexed_while_waiting() { + let (origin, broadcast, mut catalog) = timeline_pair().await; + catalog + .mutate(|catalog| { + let config = catalog.video.renditions.get_mut("video0").unwrap(); + config.coded_width = None; + config.coded_height = None; + }) + .unwrap(); + // video1's init builds from the catalog alone, but its timeline resolves no segment. + let (_video0_track, mut video0) = timeline_track(&broadcast, "video0"); + let (_video1_track, _video1) = timeline_track(&broadcast, "video1"); + let media = broadcast.create_track("video0", timed()).unwrap(); + + let broadcaster = Broadcaster::new(crate::resolved(&origin, "live").await, Config::default()); + while broadcaster.rendition(Kind::Video, "video0").is_none() { + moq_clock::sleep(Duration::from_millis(10)).await; + } + let mut waiting = std::pin::pin!(broadcaster.video_init()); + assert!( + moq_clock::timeout(Duration::from_secs(1), waiting.as_mut()) + .await + .is_err(), + "resolved without a listable video" + ); + + for sequence in 0..3 { + let mut group = media.create_group(sequence.into()).unwrap(); + write_keyframe(&mut group, sequence); + group.finish().unwrap(); + video0.push(&gop(sequence)).unwrap(); + } + moq_clock::timeout(Duration::from_secs(1), waiting) + .await + .expect("the wait builds video0's init once its timeline arrives"); + } + + // The manifest leaves a disabled rendition out, so it never fetches media for its init. + #[tokio::test(start_paused = true)] + async fn a_disabled_rendition_never_builds_its_init() { + let (origin, broadcast, mut catalog, _timelines) = inline_sibling().await; + catalog + .mutate(|catalog| catalog.video.renditions.get_mut("video1").unwrap().enabled = false) + .unwrap(); + let media = broadcast.create_track("video1", timed()).unwrap(); + for sequence in 0..4 { + let mut group = media.create_group(sequence.into()).unwrap(); + write_keyframe(&mut group, sequence); + group.finish().unwrap(); + } + + let broadcaster = Broadcaster::new(crate::resolved(&origin, "live").await, Config::default()); + broadcaster.advertised().await; + let tag = segments::tag(&(Kind::Video, "video0".into(), "video0".into())); + listed_under(&broadcaster.rendition(Kind::Video, "video0").unwrap(), &tag, 3).await; + for _ in 0..3 { + assert!(broadcaster.manifest(None).unwrap().is_some()); + tokio::time::sleep(Duration::from_secs(1)).await; + } + let disabled = broadcaster.rendition(Kind::Video, "video1").unwrap(); + assert!(!disabled.snapshot().segments.is_empty()); + assert!( + disabled.media_playlist(None).unwrap().is_none(), + "no init was built for the disabled rendition" + ); + + catalog + .mutate(|catalog| catalog.video.renditions.get_mut("video1").unwrap().enabled = true) + .unwrap(); + listed_in_manifest(&broadcaster).await; + } + // A client resolves a `.` or `..` URL segment away and the router refuses an empty one, so // renditions with those names are never admitted, while their siblings keep serving. #[tokio::test(start_paused = true)] @@ -1826,41 +2187,66 @@ mod tests { assert_eq!(&init[4..8], b"ftyp"); } - /// An init bootstrap already blocked on the unfinished tail must not hold back a later - /// request once a completed segment has cached the init. + /// An init bootstrap already blocked on an open group must not hold back a later request + /// once a completed segment has cached the init. #[tokio::test(start_paused = true)] async fn a_blocked_init_bootstrap_does_not_hide_a_cached_init() { - let origin = produce_origin(); - let broadcast = origin.create_broadcast("live").expect("publish allowed"); - let (mut catalog, mut registration, _media) = publish_vp8(&mut broadcast.broadcast().clone()); - registration - .set(hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8)) + let (origin, broadcast, mut catalog) = timeline_pair().await; + catalog + .mutate(|catalog| { + let config = catalog.video.renditions.get_mut("video1").unwrap(); + config.coded_width = None; + config.coded_height = None; + }) .unwrap(); - broadcast.announce(moq_net::origin::Route::default()).unwrap(); - settle().await; + let (_video0_track, mut video0) = timeline_track(&broadcast, "video0"); + let (_video1_track, mut video1) = timeline_track(&broadcast, "video1"); + let media = broadcast.create_track("video1", timed()).unwrap(); + // At first only the open group 0 is indexed, split before it finished. + let mut open = media.create_group(0u64.into()).unwrap(); + write_keyframe(&mut open, 0); + video0.push(&gop(0)).unwrap(); + let split = hang::timeline::Record::new( + 0, + 0, + 2_000, + hang::timeline::Position::group(0), + hang::timeline::Position::new(0, 1), + ); + video1.push(&split).unwrap(); + let broadcaster = Broadcaster::new(crate::resolved(&origin, "live").await, Config::default()); broadcaster.ready().await; - let rendition = broadcaster.rendition(Kind::Video, "video0").expect("rendition"); - - catalog.finish().unwrap(); - while !rendition.snapshot().finished { - tokio::task::yield_now().await; - } + tokio::time::sleep(Duration::from_millis(100)).await; + let rendition = broadcaster.rendition(Kind::Video, "video1").expect("rendition"); let mut blocked = std::pin::pin!(rendition.latest_init()); assert!( tokio::time::timeout(Duration::from_millis(10), blocked.as_mut()) .await .is_err(), - "the bootstrap reads the unfinished trailing group" + "the bootstrap reads the open group" ); - let segment = rendition.segments().next().await.unwrap().expect("completed segment"); - assert!(!segment.media.expect("media, not a gap").is_empty()); + // Closed groups follow, and transmuxing one caches the init. + for sequence in 1..3 { + let mut group = media.create_group(sequence.into()).unwrap(); + write_keyframe(&mut group, sequence); + group.finish().unwrap(); + video1.push(&gop(sequence)).unwrap(); + } + for sequence in 1..4 { + video0.push(&gop(sequence)).unwrap(); + } + let tag = segments::tag(&(Kind::Video, "video0".into(), "video0".into())); + listed_under(&broadcaster.rendition(Kind::Video, "video0").unwrap(), &tag, 3).await; + let segment = rendition.numbered(&tag, 1).await.unwrap().expect("completed segment"); + assert!(!segment.is_empty()); tokio::time::timeout(Duration::from_secs(1), rendition.latest_init()) .await .expect("init must not wait behind the blocked bootstrap") .unwrap() .expect("transmuxed segment has an init"); + drop(open); } // A reconfigure that changes the init bytes changes the init URL, and the old URL stops diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index e6630cd7ec5..f08de3d2c24 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -196,6 +196,29 @@ impl Media { } } + /// The media track handle on `binding`, with that bound broadcast so a fetch can tell a + /// gone publisher from a live failure. + /// + /// `binding` is the handle captured with the rows being served. Collecting the answer + /// never looks the path up ad hoc: a same-path republish installs a new broadcast whose + /// group numbering restarts, and those bytes must not be served under rows produced for + /// the publisher it replaced. + async fn track( + &self, + binding: &moq_mux::Binding, + ) -> Result> { + let track = kio::wait(|waiter| self.poll_track(binding, waiter)).await; + // The bind can be refused while this read waits. Preserve that exact instance's + // authorization error; other failed or disconnected references stay missing. + if track.is_none() + && let Poll::Ready(Err(err @ moq_mux::Error::Moq(moq_net::Error::Unauthorized))) = + binding.poll_broadcast(&kio::Waiter::noop()) + { + return Err(err.into()); + } + Ok(track) + } + fn poll_track( &self, binding: &moq_mux::Binding, @@ -281,16 +304,22 @@ pub struct Rendition { /// Feeds [`spans`](Self::spans); aborted when the rendition drops. watcher: Mutex>>, /// The broadcast serving this rendition's media. A sibling is bound when the rendition is - /// created, and ends the rendition when it ends. - media: Media, + /// created, and ends the rendition when it ends. Shared with a background init build. + media: Arc, /// The current publisher run: the embedder's label every segment URL carries (see /// [`Broadcaster::set_generation`](super::Broadcaster::set_generation)) and the init, built /// on first request. A catalog reconfigure rebuilds the whole rendition, while a new /// generation drops the init, since a restarted inline-codec publisher may carry new - /// parameter sets under the same catalog. - run: Mutex, - /// Serializes init builds, so concurrent first requests fetch the keyframe group once. - building: tokio::sync::Mutex<()>, + /// parameter sets under the same catalog. Shared state so a render can wait for the init. + run: kio::Producer, + /// Serializes init builds, so concurrent first requests fetch the keyframe group once. Shared + /// so a background build (see [`start_init`](Self::start_init)) holds it for its whole run. + building: Arc>, + /// The background init build; aborted when the rendition drops. + build: Mutex>>, + /// The runtime the rendition was built on, which runs the background init build even when a + /// synchronous render calls from another thread. + runtime: tokio::runtime::Handle, } impl Rendition { @@ -376,9 +405,11 @@ impl Rendition { spans: Arc::new(Spans::new()), catalog: source.broadcast(), watcher: Mutex::new(None), - media, - run: Mutex::default(), - building: tokio::sync::Mutex::new(()), + media: Arc::new(media), + run: kio::Producer::default(), + building: Arc::new(tokio::sync::Mutex::new(())), + build: Mutex::new(None), + runtime: tokio::runtime::Handle::current(), }) } @@ -408,9 +439,11 @@ impl Rendition { spans: Arc::new(Spans::new()), catalog: source.broadcast(), watcher: Mutex::new(None), - media, - run: Mutex::default(), - building: tokio::sync::Mutex::new(()), + media: Arc::new(media), + run: kio::Producer::default(), + building: Arc::new(tokio::sync::Mutex::new(())), + build: Mutex::new(None), + runtime: tokio::runtime::Handle::current(), }) } @@ -491,7 +524,7 @@ impl Rendition { /// List every segment URL under `generation` from now on, keeping the run's rows and init. pub(crate) fn label(&self, generation: Option>) { - self.run.lock().expect("run lock poisoned").generation = generation; + self.run_mut().generation = generation; } /// Start a new publisher run labeled `generation`, dropping the previous run's rows, records, @@ -500,10 +533,15 @@ impl Rendition { /// The rows go first: renders read the run before the rows, so one that sees the new label /// never lists the old run's segments under it. pub(crate) fn restart(&self, generation: Option>) { + // A build parked on the old run's open group would hold `building`, so the new run could + // never start its own. + if let Some(build) = self.build.lock().expect("build lock poisoned").take() { + build.abort(); + } self.live.clear(); // The old run's own records would bootstrap an inline init from its media. self.spans.clear(); - let mut run = self.run.lock().expect("run lock poisoned"); + let mut run = self.run_mut(); *run = Run { generation, init: None, @@ -513,7 +551,24 @@ impl Rendition { } fn run(&self) -> Run { - self.run.lock().expect("run lock poisoned").clone() + self.run.read().clone() + } + + fn run_mut(&self) -> kio::Mut<'_, Run> { + self.run + .write() + .unwrap_or_else(|_| unreachable!("the run is never closed")) + } + + /// Poll until the current run's init is built. + pub(crate) fn poll_init(&self, waiter: &kio::Waiter) -> Poll<()> { + match self.run.poll_ref(waiter, |run| match run.init { + Some(_) => Poll::Ready(()), + None => Poll::Pending, + }) { + Poll::Ready(_) => Poll::Ready(()), + Poll::Pending => Poll::Pending, + } } /// The generation every segment URL currently carries. @@ -695,6 +750,22 @@ impl Rendition { self.live.failure().is_none() && self.check().is_err() } + /// Poll until listings of this rendition refuse (see [`check`](Self::check)), waking when its + /// media is denied or either timeline fails. + #[cfg_attr( + not(feature = "server"), + allow(dead_code, reason = "only the `server` feature uses it") + )] + pub(crate) fn poll_refused(&self, waiter: &kio::Waiter) -> Poll<()> { + self.poll_media(waiter); + let _ = self.spans.poll_failed(waiter); + let _ = self.live.poll_failed(waiter); + match self.check() { + Err(_) => Poll::Ready(()), + Ok(()) => Poll::Pending, + } + } + fn snapshot_from_window(&self, window: segments::Window, finished: bool, generation: Option>) -> Snapshot { // HLS requires EXT-X-TARGETDURATION to be at least every EXTINF, rounded up to whole // seconds. The catalog duration is nominal, while actual records can be longer. @@ -810,7 +881,7 @@ impl Rendition { /// Record a start listed in window `resets`. Resets only grow, so a delayed latch for an /// older window never replaces a newer one. fn latch(&self, resets: u64) { - let mut run = self.run.lock().expect("run lock poisoned"); + let mut run = self.run_mut(); run.startable = run.startable.max(Some(resets)); } @@ -934,29 +1005,6 @@ impl Rendition { Poll::Ready(track.map(|(_, track)| track.subscribe(subscription).into_inner())) } - /// The media track handle on `binding`, with that bound broadcast so a fetch can tell a - /// gone publisher from a live failure. - /// - /// `binding` is the handle captured with the rows being served. Collecting the answer - /// never looks the path up ad hoc: a same-path republish installs a new broadcast whose - /// group numbering restarts, and those bytes must not be served under rows produced for - /// the publisher it replaced. - async fn track( - &self, - binding: &moq_mux::Binding, - ) -> Result> { - let track = kio::wait(|waiter| self.media.poll_track(binding, waiter)).await; - // The bind can be refused while this read waits. Preserve that exact instance's - // authorization error; other failed or disconnected references stay missing. - if track.is_none() - && let Poll::Ready(Err(err @ moq_mux::Error::Moq(moq_net::Error::Unauthorized))) = - binding.poll_broadcast(&kio::Waiter::noop()) - { - return Err(err.into()); - } - Ok(track) - } - /// The media track's name in its broadcast, which can differ from the rendition's ID. fn track_name(&self) -> &str { match &self.config { @@ -989,7 +1037,7 @@ impl Rendition { if let Some(init) = &run.init { return Some(init.clone()); } - // A muxer error surfaces from `init()`, which the serve path awaits before rendering. + // A muxer error surfaces from `bootstrap`, which builds from media. let bytes = self.muxer().ok()?.init().ok()??; Some(self.cache_init(run.epoch, bytes)) } @@ -997,13 +1045,63 @@ impl Rendition { /// Cache `bytes` as the init of the run numbered `epoch`, unless a restart has since /// replaced that run. fn cache_init(&self, epoch: u64, bytes: Bytes) -> Arc { - let mut run = self.run.lock().expect("run lock poisoned"); + let mut run = self.run_mut(); if run.epoch != epoch { return Arc::new(Init::new(bytes)); } run.init.get_or_insert_with(|| Arc::new(Init::new(bytes))).clone() } + /// Build the init in the background unless it is built or a build is already running, so a + /// later [`representations`](Self::representations) lists this rendition without any render + /// waiting on its media. + /// + /// The build never holds the rendition itself, and is aborted when the rendition drops, so + /// one stuck on an open group cannot keep a torn-down export's state alive. + pub(crate) fn start_init(self: &Arc) { + let Ok(binding) = self.authorize() else { + return; + }; + let run = self.run(); + if self.built_init(&run).is_some() { + return; + } + // Taken before spawning, so renders never queue tasks behind a build stuck on an open group. + let Ok(building) = self.building.clone().try_lock_owned() else { + return; + }; + let read = match self.muxer().map(|muxer| self.init_read(muxer)) { + Ok(Some(read)) => read, + Ok(None) => return, + Err(err) => { + tracing::warn!(rendition = %self.name, %err, "failed to build init segment"); + return; + } + }; + let media = self.media.clone(); + let rendition = Arc::downgrade(self); + let name = self.name.clone(); + let build = self.runtime.spawn(async move { + let _building = building; + match read.read(&media, &binding).await { + Ok(Some(bytes)) => { + if let Some(rendition) = rendition.upgrade() { + rendition.cache_init(run.epoch, bytes); + } + } + Ok(None) => {} + Err(err) => tracing::warn!(rendition = %name, %err, "failed to build init segment"), + } + }); + *self.build.lock().expect("build lock poisoned") = Some(build); + } + + /// Whether the init is missing while a build of it is running, so a later render may list + /// this rendition. + pub(crate) fn init_pending(&self) -> bool { + self.run().init.is_none() && self.building.try_lock().is_err() + } + async fn load_init(&self) -> Result>> { let binding = self.authorize()?; // A transmuxed segment can cache the init while another bootstrap holds `building`, @@ -1012,43 +1110,42 @@ impl Rendition { return Ok(Some(init)); } let _building = self.building.lock().await; + self.bootstrap(&binding).await + } + + /// Build and cache the init, holding `building`. + async fn bootstrap(&self, binding: &moq_mux::Binding) -> Result>> { let run = self.run(); if let Some(init) = run.init { return Ok(Some(init)); } - let mut muxer = self.muxer()?; - // An out-of-band codec builds its init straight from the catalog; an inline one returns - // `None` until a keyframe group is read, so bootstrap it from the newest one indexed. - let bytes = match muxer.init()? { - Some(bytes) => bytes, - None => { - let reference = (self.kind, self.name.clone(), self.track_name().to_string()); - let latest = self.spans.latest_keyframe_group(); - let Some(sequence) = latest.or_else(|| self.live.latest_keyframe_group(&reference)) else { - return Ok(None); - }; - let Some((broadcast, track)) = self.track(&binding).await? else { - return Ok(None); - }; - let Ok(mut group) = fetch(&track, hang::timeline::Position::group(sequence), None, &broadcast).await? - else { - return Ok(None); - }; - // A cache eviction mid-read leaves the init unbuildable for now, not an error. - if read_group(&mut muxer, &mut group, &broadcast).await?.is_err() { - return Ok(None); - } - let Some(bytes) = muxer.init()? else { - return Ok(None); - }; - bytes - } + // `None` until a keyframe group is read. + let muxer = self.muxer()?; + if let Some(bytes) = muxer.init()? { + return Ok(Some(self.cache_init(run.epoch, bytes))); + } + let Some(read) = self.init_read(muxer) else { + return Ok(None); + }; + let Some(bytes) = read.read(&self.media, binding).await? else { + return Ok(None); }; Ok(Some(self.cache_init(run.epoch, bytes))) } + /// The indexed keyframe groups an inline-codec init is read from, or `None` while there are + /// none. + fn init_read(&self, muxer: Muxer) -> Option { + let mut groups = self.spans.keyframe_groups(); + if groups.is_empty() { + let reference = (self.kind, self.name.clone(), self.track_name().to_string()); + groups = self.live.keyframe_groups(&reference); + } + (!groups.is_empty()).then_some(InitRead { muxer, groups }) + } + /// Fetch and transmux the segment a [`Resource::Segment`](super::Resource::Segment) URL names. /// /// Fetches every frame the segment resolves to on this rendition (an audio segment packs many @@ -1107,7 +1204,7 @@ impl Rendition { return Ok(None); } - let Some((broadcast, track)) = self.track(&binding).await? else { + let Some((broadcast, track)) = self.media.track(&binding).await? else { return Ok(None); }; @@ -1234,11 +1331,46 @@ async fn watch_spans( } } +/// An inline-codec init still to be read from media: owned, so a background build waits on it +/// without holding its rendition. +struct InitRead { + muxer: Muxer, + /// Keyframe groups to try in order: a closed group first, since reading an open one waits + /// for the publisher to finish it. + groups: Vec, +} + +impl InitRead { + /// Read the groups until one yields the init. + async fn read(mut self, media: &Media, binding: &moq_mux::Binding) -> Result> { + let Some((broadcast, track)) = media.track(binding).await? else { + return Ok(None); + }; + for sequence in self.groups { + // A group gone from the cache, before or mid-read, leaves the next candidate to try. + let Ok(mut group) = fetch(&track, hang::timeline::Position::group(sequence), None, &broadcast).await? + else { + continue; + }; + if read_group(&mut self.muxer, &mut group, &broadcast).await?.is_err() { + continue; + } + if let Some(bytes) = self.muxer.init()? { + return Ok(Some(bytes)); + } + } + Ok(None) + } +} + impl Drop for Rendition { fn drop(&mut self) { if let Some(watcher) = self.watcher.lock().expect("watcher lock poisoned").take() { watcher.abort(); } + if let Some(build) = self.build.lock().expect("build lock poisoned").take() { + build.abort(); + } } } @@ -1541,4 +1673,35 @@ mod tests { assert!(!is_cache_miss(&moq_net::Error::Dropped)); assert!(!is_cache_miss(&moq_net::Error::Remote(0))); } + + // A wait on a rendition's refusal wakes when either timeline fails mid-wait, so the + // manifest's first-video wait drops a video that can no longer be listed. + #[tokio::test(start_paused = true)] + async fn a_refusal_wait_wakes_on_a_timeline_failure() { + for own in [true, false] { + let origin = moq_tokio::origin::spawn(); + let broadcast = origin.create_broadcast("live").unwrap(); + let source = moq_mux::Source::new(origin.consume(), broadcast.consume()); + let config = VideoConfig::new(hang::catalog::VideoCodec::VP8); + let rendition = + Arc::new(Rendition::video("video".into(), &config, &source, Archive::default(), None).unwrap()); + // On its own task, so only a registered wake polls it again. + let refused = tokio::spawn({ + let rendition = rendition.clone(); + async move { kio::wait(|waiter| rendition.poll_refused(waiter)).await } + }); + moq_clock::sleep(Duration::from_millis(10)).await; + assert!(!refused.is_finished(), "no timeline failed yet"); + let err = crate::Error::Moq(moq_net::Error::Transport("connection lost".into())); + if own { + rendition.spans.fail(err, false); + } else { + rendition.fail(err); + } + moq_clock::timeout(Duration::from_secs(1), refused) + .await + .expect("the failure wakes the wait") + .unwrap(); + } + } } diff --git a/rs/moq-hls/src/export/renditions.rs b/rs/moq-hls/src/export/renditions.rs index c69c417b6ff..f572a02dc89 100644 --- a/rs/moq-hls/src/export/renditions.rs +++ b/rs/moq-hls/src/export/renditions.rs @@ -512,6 +512,48 @@ impl Producer { .await; } + /// Start every enabled video rendition's init build, then resolve once one of them has its + /// init and is [advertised](Rendition::is_advertised), so a manifest lists it, or none is left + /// that listings do not refuse: a denied or failed one can never be listed. A build that has + /// nothing to read from yet starts on the wake that indexes its media. + #[cfg_attr( + not(feature = "server"), + allow(dead_code, reason = "only the `server` feature uses it") + )] + pub(crate) async fn video_init(&self) { + kio::wait(|waiter| { + let poll = self.state.poll_ref(waiter, |current| { + let mut video = current + .renditions + .values() + .filter(|rendition| { + rendition.kind == Kind::Video + && rendition.is_enabled() + && rendition.poll_refused(waiter).is_pending() + }) + .peekable(); + if video.peek().is_none() + || video.any(|rendition| { + // Rerun on every wake: `poll_refused` registered `waiter` on both timelines, + // so the rows a build needs wake this wait to start it. + rendition.start_init(); + rendition.poll_init(waiter).is_ready() && rendition.poll_advertised(waiter).is_ready() + }) { + Poll::Ready(()) + } else { + Poll::Pending + } + }); + match poll { + Poll::Ready(Ok(())) => Poll::Ready(()), + // A closed catalog adds no rendition, but the existing ones' inits keep arriving, + // and the predicate registered `waiter` on each of them. + Poll::Ready(Err(_)) | Poll::Pending => Poll::Pending, + } + }) + .await + } + /// Enroll a freshly-created rendition in the timeline feed, replaying the recent history /// (and the ended/failed/closed markers) so its window matches its siblings', and start following its /// own timeline. diff --git a/rs/moq-hls/src/export/segments.rs b/rs/moq-hls/src/export/segments.rs index 645b8fdc1ac..35037bcf8c7 100644 --- a/rs/moq-hls/src/export/segments.rs +++ b/rs/moq-hls/src/export/segments.rs @@ -434,17 +434,18 @@ impl Producer { .map(|row| row.segment) } - /// The newest group of `reference` a row starts on a keyframe, used to bootstrap an init - /// segment for inline-parameter-set codecs. - pub fn latest_keyframe_group(&self, reference: &ReferenceKey) -> Option { + /// The groups of `reference` a row starts on a keyframe, to bootstrap an inline-parameter-set + /// codec's init from, in the order to try them (see [`super::spans::bootstrap_groups`]). + pub fn keyframe_groups(&self, reference: &ReferenceKey) -> Vec { let state = self.state.read(); - state - .rows - .iter() - .rev() - .filter(|row| *row.reference == *reference) - .find(|row| row.starts_sync()) - .map(|row| row.frames.start.group) + super::spans::bootstrap_groups( + state + .rows + .iter() + .rev() + .filter(|row| *row.reference == *reference && row.starts_sync()) + .map(|row| row.frames.clone()), + ) } /// Whether the playlist has anything to serve yet (at least one segment, or the broadcast @@ -479,6 +480,21 @@ impl Producer { } } + /// Poll until the reference timeline fails. + #[cfg_attr( + not(feature = "server"), + allow(dead_code, reason = "only the `server` feature uses it") + )] + pub fn poll_failed(&self, waiter: &kio::Waiter) -> Poll<()> { + match self.state.poll_ref(waiter, |state| match state.failed { + Some(_) => Poll::Ready(()), + None => Poll::Pending, + }) { + Poll::Ready(Ok(())) => Poll::Ready(()), + _ => Poll::Pending, + } + } + /// Poll `f` over the listed rows, waking on the next window change while it is pending, and /// return the [`resets`](Self::resets) those rows belong to. A closed window stays pending: no /// new row can make `f` ready. @@ -820,14 +836,10 @@ mod tests { ); assert_eq!(live.row(&tag, 7), None, "unknown segments miss"); assert_eq!(live.row("00000000", 0), None, "another reference's numbering misses"); - assert_eq!( - live.latest_keyframe_group(&video), - Some(0), - "row 1 does not start on a keyframe" - ); - assert_eq!( - live.latest_keyframe_group(&(Kind::Audio, "video".to_string(), "video".to_string())), - None + assert_eq!(live.keyframe_groups(&video), [0], "row 1 does not start on a keyframe"); + assert!( + live.keyframe_groups(&(Kind::Audio, "video".to_string(), "video".to_string())) + .is_empty() ); } diff --git a/rs/moq-hls/src/export/spans.rs b/rs/moq-hls/src/export/spans.rs index df50f53a991..77de4f2582d 100644 --- a/rs/moq-hls/src/export/spans.rs +++ b/rs/moq-hls/src/export/spans.rs @@ -184,17 +184,34 @@ impl Spans { } } - /// The newest group known to start with a keyframe, used to bootstrap an init segment for - /// inline-parameter-set codecs. - pub fn latest_keyframe_group(&self) -> Option { + /// The keyframe groups to bootstrap an inline-parameter-set codec's init from, in the order + /// to try them (see [`bootstrap_groups`]). + pub fn keyframe_groups(&self) -> Vec { let state = self.state.read(); - state - .records - .iter() - .rev() - .find(|record| starts_sync(record)) - .map(|record| record.start.group) + bootstrap_groups( + state + .records + .iter() + .rev() + .filter(|record| starts_sync(record)) + .map(|record| record.start..record.end), + ) + } +} + +/// Order keyframe group spans, newest first, for an init bootstrap: the newest group whose span +/// already reached a later group, which is closed and so read from cache at once, then the +/// newest group when that one may still be open. +pub(crate) fn bootstrap_groups(mut spans: impl Iterator>) -> Vec { + let Some(newest) = spans.next() else { + return Vec::new(); + }; + let closed = |span: &Range| span.end.group > span.start.group; + if closed(&newest) { + return vec![newest.start.group]; } + let older = spans.find(closed).map(|span| span.start.group); + older.into_iter().chain([newest.start.group]).collect() } /// Whether `record` starts at a group start that is a sync point, so a player can begin decoding diff --git a/rs/moq-hls/src/server/routes.rs b/rs/moq-hls/src/server/routes.rs index d5109355688..404cb6a5412 100644 --- a/rs/moq-hls/src/server/routes.rs +++ b/rs/moq-hls/src/server/routes.rs @@ -92,7 +92,9 @@ async fn manifest(server: &Server, broadcast: &str, query: Option<&str>) -> Resp // to index its first complete segment before answering. let _ = moq_clock::timeout(READY_TIMEOUT, broadcaster.playable()).await; // Each representation names its init by hash, which an inline codec only learns from media. - broadcaster.build_inits().await; + // A player keeps the adaptation sets of the first manifest it loads, so wait for one video + // init rather than every init: a stuck rendition never holds a sibling that can play. + let _ = moq_clock::timeout(READY_TIMEOUT, broadcaster.video_init()).await; match broadcaster.manifest(query) { Ok(Some(manifest)) => mpd(manifest), Ok(None) => not_found(), @@ -342,6 +344,79 @@ mod tests { wait_listed(&app, "/live/master.m3u8", "video/video0/media.m3u8").await; } + /// A player keeps the adaptation sets of the first manifest it loads, so a fresh edge's first + /// manifest already lists an inline-codec video rendition, whose init is only known from media. + #[tokio::test(start_paused = true)] + async fn the_first_manifest_lists_inline_video() { + let origin = moq_tokio::origin::spawn(); + let broadcast = origin.create_broadcast("live").expect("publish"); + broadcast.announce(run()).expect("announce"); + let inline = hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8); + let (_catalog, _registration, _track, mut media) = + publish_video(&mut broadcast.broadcast().clone(), inline, timed()); + write_three_gops(&mut media); + + let app = Server::new(origin.consume(), crate::export::Config::default()).router(); + let response = oneshot(app, "/live/manifest.mpd").await; + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let body = String::from_utf8(body.to_vec()).unwrap(); + assert!(body.contains("id=\"video/video0\""), "{body}"); + } + + /// A denied video can never be listed, so it never holds the manifest until the first-video + /// wait times out: the playable audio is served at once. + #[tokio::test(start_paused = true)] + async fn a_denied_video_never_holds_the_manifest() { + let origin = moq_tokio::origin::spawn(); + let broadcast = origin.create_broadcast("room/live").unwrap(); + broadcast.announce(run()).unwrap(); + let secret = origin.create_broadcast("room/secret").unwrap(); + secret.announce(Default::default()).unwrap(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast.broadcast().clone(), Default::default()).unwrap(); + let reserved = catalog.reserve(); + let mut config = video_config(); + config.broadcast = Some(moq_net::path::Relative::new("./secret").to_owned()); + let mut video = reserved + .video( + secret.create_track("video0", timed()).unwrap(), + moq_mux::catalog::hang::Container::Legacy(moq_mux::container::Kind::Video), + config, + ) + .unwrap(); + let mut audio = reserved + .audio( + broadcast.create_track("audio0", timed()).unwrap(), + moq_mux::catalog::hang::Container::Legacy(moq_mux::container::Kind::Data), + hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, 2), + ) + .unwrap(); + drop(reserved); + write_three_gops(&mut video); + for micros in [0, 2_000_000, 4_000_000] { + let keyframe = audio.needs_keyframe(); + audio.write(vp8_frame(micros, keyframe)).unwrap(); + audio.cut(None).unwrap(); + } + + let restricted = origin + .consume() + .scope( + "", + &moq_net::Patterns::from("room/live".parse::().unwrap()), + ) + .unwrap(); + let app = Server::new(restricted, Default::default()).router(); + let start = tokio::time::Instant::now(); + let response = oneshot(app, "/room/live/manifest.mpd").await; + assert!(start.elapsed() < READY_TIMEOUT, "waited {:?}", start.elapsed()); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let body = String::from_utf8(body.to_vec()).unwrap(); + assert!(body.contains("id=\"audio/audio0\""), "{body}"); + assert!(!body.contains("id=\"video/video0\""), "{body}"); + } + /// A restarted publisher gets URLs of its own, whose numbering starts over: the master lists /// the replacement, while a player already on the old instance keeps its playlist until that /// instance's broadcast closes.