From ed21aa2805b8707a1b138ad57ab5304c80fc3c5c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 02:15:21 -0700 Subject: [PATCH 1/9] chore(quest): claim quest/m1/dash-init-ready.md Quest-Claim: 0963cf37a45eb43c From ac3115d4c850880e0cc2c9ca9fe353fde1bb0999 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 02:26:46 -0700 Subject: [PATCH 2/9] fix(hls): never hold a DASH manifest on one rendition's init Rendering the manifest used to await every rendition's init serially, disabled ones included, so an inline-codec rendition bootstrapping from a still-open group held up the whole manifest. Rendering now starts each enabled rendition's missing init in the background (one build at a time per rendition) and lists only the renditions whose init exists; a later refresh picks up the rest. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/dash-init-ready.md | 29 ------ rs/moq-hls/src/export/mod.rs | 154 ++++++++++++++++++++++++----- rs/moq-hls/src/export/rendition.rs | 41 ++++++-- rs/moq-hls/src/server/routes.rs | 2 - 5 files changed, 167 insertions(+), 60 deletions(-) delete mode 100644 quest/m1/dash-init-ready.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 9ad0bfc1971..6dd80f08146 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -158,7 +158,6 @@ Catalog HLS field. - [HLS URL redaction](/quest/m1/hls-url-redaction.md) - no moq-hls error prints a signed URL's query or userinfo - [HLS ranged responses](/quest/m1/hls-range-206.md) - a byte-range segment accepts only a matching `206` and never buffers past its length - [HLS redirect base](/quest/m1/hls-redirect-base.md) - relative playlist URIs resolve against a same-origin redirect's final URL -- [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 - [HLS dot-only names](/quest/m1/hls-dot-names.md) - a `.` or `..` rendition is refused instead of advertising a URL outside its directory - [MSF duplicate init](/quest/m1/msf-duplicate-init.md) - an MSF catalog refuses a repeated `initDataList` id instead of resolving it last-wins - [Image hardening](/quest/m1/image-hardening.md) - the relay image runs unprivileged and its builder stage is pinned 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 97be0f11c38..72e49986b66 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 @@ -296,8 +299,9 @@ 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(); let representations = match rendition.representations() { Ok(representations) => representations, // Its own playlist answers the error, while its siblings stay listed. @@ -339,20 +343,6 @@ 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. - #[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"); - } - } - } - /// Resolve once the broadcast has something listable (its first complete segment, or an /// already-ended timeline). Every rendition's window is fed from the same timeline, so the /// first rendition's readiness stands in for the broadcast's. Bounding the wait is the @@ -1405,7 +1395,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(); @@ -1548,8 +1538,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()); @@ -1621,6 +1611,126 @@ 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. + async fn inline_sibling() -> ( + moq_net::origin::Producer, + moq_net::broadcast::Producer, + 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(); + video1.push(&gop(sequence)).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(); + } + + // 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. + #[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; + 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}"); + } + + // 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; + } + // With every rendition disabled no reference watcher is left, so the source ending must // still end every playlist and recording cursor. #[tokio::test(start_paused = true)] diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index e6630cd7ec5..9e2c5cab2d7 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -289,8 +289,9 @@ pub struct Rendition { /// 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<()>, + /// 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>, } impl Rendition { @@ -378,7 +379,7 @@ impl Rendition { watcher: Mutex::new(None), media, run: Mutex::default(), - building: tokio::sync::Mutex::new(()), + building: Arc::new(tokio::sync::Mutex::new(())), }) } @@ -410,7 +411,7 @@ impl Rendition { watcher: Mutex::new(None), media, run: Mutex::default(), - building: tokio::sync::Mutex::new(()), + building: Arc::new(tokio::sync::Mutex::new(())), }) } @@ -989,7 +990,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)) } @@ -1004,6 +1005,29 @@ impl Rendition { 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. + pub(crate) fn start_init(self: &Arc) { + let Ok(binding) = self.authorize() else { + return; + }; + if self.built_init(&self.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 rendition = self.clone(); + tokio::spawn(async move { + let _building = building; + if let Err(err) = rendition.bootstrap(&binding).await { + tracing::warn!(rendition = %rendition.name, %err, "failed to build init segment"); + } + }); + } + async fn load_init(&self) -> Result>> { let binding = self.authorize()?; // A transmuxed segment can cache the init while another bootstrap holds `building`, @@ -1012,6 +1036,11 @@ 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)); @@ -1029,7 +1058,7 @@ impl Rendition { 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 { + 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? diff --git a/rs/moq-hls/src/server/routes.rs b/rs/moq-hls/src/server/routes.rs index 88190a2ec42..8e8009659d9 100644 --- a/rs/moq-hls/src/server/routes.rs +++ b/rs/moq-hls/src/server/routes.rs @@ -91,8 +91,6 @@ async fn manifest(server: &Server, broadcast: &str, query: Option<&str>) -> Resp // A manifest whose timelines are all empty confuses players; give the broadcast a moment // 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; match broadcaster.manifest(query) { Ok(Some(manifest)) => mpd(manifest), Ok(None) => not_found(), From f1c7af679601b15fdfb6344ccebd1cc3a5d21b33 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 02:36:17 -0700 Subject: [PATCH 3/9] fix(hls): bootstrap an inline init from a closed keyframe group The init bootstrap read the newest indexed keyframe group, which can be the one the publisher is still writing, so the build waited out the group. It now tries the newest keyframe group whose span already reached a later group first, then the newest one, and moves on when a candidate is gone from the cache. Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/mod.rs | 99 ++++++++++++++++++++++++------ rs/moq-hls/src/export/rendition.rs | 55 +++++++++-------- rs/moq-hls/src/export/segments.rs | 33 +++++----- rs/moq-hls/src/export/spans.rs | 35 ++++++++--- 4 files changed, 149 insertions(+), 73 deletions(-) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 72e49986b66..44e9cb636c0 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -1619,7 +1619,8 @@ mod tests { } /// `timeline_pair`, with `video1` learning its geometry from media so its init needs a - /// keyframe group, and both timelines indexing four groups. + /// 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::broadcast::Producer, @@ -1638,8 +1639,18 @@ mod tests { 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]) } @@ -1669,8 +1680,31 @@ mod tests { .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. + // 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; @@ -1872,41 +1906,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 mut broadcast = origin.create_broadcast("live").expect("publish allowed"); - let (mut catalog, mut registration, _media) = publish_vp8(&mut broadcast); - 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.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 9e2c5cab2d7..9ea17859c34 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -1049,33 +1049,36 @@ impl Rendition { 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, so bootstrap it from an indexed one: a closed + // group first, since reading an open one waits for the publisher to finish it. + if let Some(bytes) = muxer.init()? { + return Ok(Some(self.cache_init(run.epoch, bytes))); + } + 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); + } + if groups.is_empty() { + return Ok(None); + } + let Some((broadcast, track)) = self.track(binding).await? else { + return Ok(None); }; - Ok(Some(self.cache_init(run.epoch, bytes))) + for sequence in 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 muxer, &mut group, &broadcast).await?.is_err() { + continue; + } + if let Some(bytes) = muxer.init()? { + return Ok(Some(self.cache_init(run.epoch, bytes))); + } + } + Ok(None) } /// Fetch and transmux the segment a [`Resource::Segment`](super::Resource::Segment) URL names. diff --git a/rs/moq-hls/src/export/segments.rs b/rs/moq-hls/src/export/segments.rs index d5a6f05f038..02773b045e5 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 @@ -822,14 +823,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 From dec12ce907de5b056de544133a3eba9a89e88ea0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 02:41:01 -0700 Subject: [PATCH 4/9] fix(hls): wait for one video init before the first DASH manifest A player keeps the adaptation sets of the first manifest it loads, and a fresh edge's first render lacked any inline-codec video because the init build it started could not finish before the render. The manifest route now starts every enabled video rendition's build and waits, within its existing READY_TIMEOUT, until one of them has an init. Any single built video init ends the wait, so a stuck rendition never holds a sibling. The run state moves onto a kio producer so caching the init wakes it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/mod.rs | 20 +++++++++++++++++ rs/moq-hls/src/export/rendition.rs | 35 +++++++++++++++++++++-------- rs/moq-hls/src/export/renditions.rs | 23 +++++++++++++++++++ rs/moq-hls/src/server/routes.rs | 23 +++++++++++++++++++ 4 files changed, 92 insertions(+), 9 deletions(-) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 44e9cb636c0..cd3027ad3ee 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -343,6 +343,23 @@ impl Broadcaster { ))) } + /// Start every enabled video rendition's init build, then resolve once one of them has its + /// init, or none is enabled. 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 video_init(&self) { + for rendition in self.renditions.snapshot() { + if rendition.kind == Kind::Video && rendition.is_enabled() { + rendition.start_init(); + } + } + self.renditions.video_init().await; + } + /// Resolve once the broadcast has something listable (its first complete segment, or an /// already-ended timeline). Every rendition's window is fed from the same timeline, so the /// first rendition's readiness stands in for the broadcast's. Bounding the wait is the @@ -1716,6 +1733,9 @@ mod tests { 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}"); diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index 9ea17859c34..b064d669ff6 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -287,8 +287,8 @@ pub struct Rendition { /// [`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, + /// 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>, @@ -378,7 +378,7 @@ impl Rendition { catalog: source.broadcast(), watcher: Mutex::new(None), media, - run: Mutex::default(), + run: kio::Producer::default(), building: Arc::new(tokio::sync::Mutex::new(())), }) } @@ -410,7 +410,7 @@ impl Rendition { catalog: source.broadcast(), watcher: Mutex::new(None), media, - run: Mutex::default(), + run: kio::Producer::default(), building: Arc::new(tokio::sync::Mutex::new(())), }) } @@ -492,7 +492,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, @@ -504,7 +504,7 @@ impl Rendition { 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, @@ -514,7 +514,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. @@ -811,7 +828,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)); } @@ -998,7 +1015,7 @@ 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)); } diff --git a/rs/moq-hls/src/export/renditions.rs b/rs/moq-hls/src/export/renditions.rs index 96ef3975dc7..cecbc42afeb 100644 --- a/rs/moq-hls/src/export/renditions.rs +++ b/rs/moq-hls/src/export/renditions.rs @@ -511,6 +511,29 @@ impl Producer { .await; } + /// Resolve once an enabled video rendition has its init, or none is enabled. + #[cfg_attr( + not(feature = "server"), + allow(dead_code, reason = "only the `server` feature uses it") + )] + pub(crate) async fn video_init(&self) { + let _ = kio::wait(|waiter| { + self.state.poll_ref(waiter, |current| { + let mut video = current + .renditions + .values() + .filter(|rendition| rendition.kind == Kind::Video && rendition.is_enabled()) + .peekable(); + if video.peek().is_none() || video.any(|rendition| rendition.poll_init(waiter).is_ready()) { + Poll::Ready(()) + } else { + 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/server/routes.rs b/rs/moq-hls/src/server/routes.rs index 8e8009659d9..091a7fee8be 100644 --- a/rs/moq-hls/src/server/routes.rs +++ b/rs/moq-hls/src/server/routes.rs @@ -91,6 +91,10 @@ async fn manifest(server: &Server, broadcast: &str, query: Option<&str>) -> Resp // A manifest whose timelines are all empty confuses players; give the broadcast a moment // 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. + // 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(), @@ -530,6 +534,25 @@ 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 mut 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, 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 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. From 82d92bfcb0ec877407d3f8d844f8a4b6f869ae34 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 09:10:45 -0700 Subject: [PATCH 5/9] fix(hls): tie the background init build to its rendition and wait for a listable video - The build holds only owned media handles and a weak rendition, and is aborted when the rendition drops, so a build parked on an open group no longer keeps a torn-down export (or its timeline watcher) alive. - A manifest stays dynamic while an enabled rendition's init is still building. - The first-manifest wait needs an advertised video with its init, and keeps waiting after the catalog closes. Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/mod.rs | 118 ++++++++++++++++++- rs/moq-hls/src/export/rendition.rs | 171 ++++++++++++++++++---------- rs/moq-hls/src/export/renditions.rs | 22 +++- 3 files changed, 246 insertions(+), 65 deletions(-) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 675846b096e..d6e318c0def 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -290,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. @@ -302,6 +304,7 @@ impl Broadcaster { // 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. @@ -318,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. @@ -1757,6 +1761,116 @@ mod tests { 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); + } + + // 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"); + } + // 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() { diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index b064d669ff6..6d6429be667 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,8 +304,8 @@ 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 @@ -292,6 +315,8 @@ pub struct Rendition { /// 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>>, } impl Rendition { @@ -377,9 +402,10 @@ impl Rendition { spans: Arc::new(Spans::new()), catalog: source.broadcast(), watcher: Mutex::new(None), - media, + media: Arc::new(media), run: kio::Producer::default(), building: Arc::new(tokio::sync::Mutex::new(())), + build: Mutex::new(None), }) } @@ -409,9 +435,10 @@ impl Rendition { spans: Arc::new(Spans::new()), catalog: source.broadcast(), watcher: Mutex::new(None), - media, + media: Arc::new(media), run: kio::Producer::default(), building: Arc::new(tokio::sync::Mutex::new(())), + build: Mutex::new(None), }) } @@ -952,29 +979,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 { @@ -1025,24 +1029,51 @@ impl Rendition { /// 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; }; - if self.built_init(&self.run()).is_some() { + 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 rendition = self.clone(); - tokio::spawn(async move { + 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 = tokio::spawn(async move { let _building = building; - if let Err(err) = rendition.bootstrap(&binding).await { - tracing::warn!(rendition = %rendition.name, %err, "failed to build init segment"); + 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>> { @@ -1063,39 +1094,30 @@ impl Rendition { 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 an indexed one: a closed - // group first, since reading an open one waits for the publisher to finish it. + // `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); } - if groups.is_empty() { - return Ok(None); - } - let Some((broadcast, track)) = self.track(binding).await? else { - return Ok(None); - }; - for sequence in 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 muxer, &mut group, &broadcast).await?.is_err() { - continue; - } - if let Some(bytes) = muxer.init()? { - return Ok(Some(self.cache_init(run.epoch, bytes))); - } - } - Ok(None) + (!groups.is_empty()).then_some(InitRead { muxer, groups }) } /// Fetch and transmux the segment a [`Resource::Segment`](super::Resource::Segment) URL names. @@ -1156,7 +1178,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); }; @@ -1283,11 +1305,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(); + } } } diff --git a/rs/moq-hls/src/export/renditions.rs b/rs/moq-hls/src/export/renditions.rs index cecbc42afeb..f254fa20519 100644 --- a/rs/moq-hls/src/export/renditions.rs +++ b/rs/moq-hls/src/export/renditions.rs @@ -511,27 +511,37 @@ impl Producer { .await; } - /// Resolve once an enabled video rendition has its init, or none is enabled. + /// Resolve once an enabled video rendition has its init and is + /// [advertised](Rendition::is_advertised), so a manifest lists it, or none is enabled. #[cfg_attr( not(feature = "server"), allow(dead_code, reason = "only the `server` feature uses it") )] pub(crate) async fn video_init(&self) { - let _ = kio::wait(|waiter| { - self.state.poll_ref(waiter, |current| { + 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()) .peekable(); - if video.peek().is_none() || video.any(|rendition| rendition.poll_init(waiter).is_ready()) { + if video.peek().is_none() + || video.any(|rendition| { + 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; + .await } /// Enroll a freshly-created rendition in the timeline feed, replaying the recent history From 0c9f7681c06274c5dabd3b71b1877c2a70895570 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 11:21:27 -0700 Subject: [PATCH 6/9] fix(hls): release a stuck init build when a new generation starts A background init build parked on the old run's open group kept holding the build lock after a restart, so the new run could never start its own and stayed out of the manifest. Restarting now aborts it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/mod.rs | 25 +++++++++++++++++++++++++ rs/moq-hls/src/export/rendition.rs | 5 +++++ 2 files changed, 30 insertions(+) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index bc39d10163d..e801dcf1c6d 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -1791,6 +1791,31 @@ mod tests { 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. diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index 6d6429be667..251b866f751 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -528,6 +528,11 @@ 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(); From 0e71d4061b4b8efdd8595be16bd3f8369468a35b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 11:29:16 -0700 Subject: [PATCH 7/9] fix(hls): run the background init build on the rendition's runtime The synchronous manifest render started the init build with tokio::spawn, which panics when an embedder renders from a thread outside the runtime. A rendition now keeps the handle of the runtime it was built on, where its timeline watcher already runs, and spawns the build there. Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/mod.rs | 23 +++++++++++++++++++++++ rs/moq-hls/src/export/rendition.rs | 7 ++++++- 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index e801dcf1c6d..e32d440d640 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -1791,6 +1791,29 @@ mod tests { 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)] diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index 251b866f751..3753b42f92b 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -317,6 +317,9 @@ pub struct Rendition { 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 { @@ -406,6 +409,7 @@ impl Rendition { run: kio::Producer::default(), building: Arc::new(tokio::sync::Mutex::new(())), build: Mutex::new(None), + runtime: tokio::runtime::Handle::current(), }) } @@ -439,6 +443,7 @@ impl Rendition { run: kio::Producer::default(), building: Arc::new(tokio::sync::Mutex::new(())), build: Mutex::new(None), + runtime: tokio::runtime::Handle::current(), }) } @@ -1060,7 +1065,7 @@ impl Rendition { let media = self.media.clone(); let rendition = Arc::downgrade(self); let name = self.name.clone(); - let build = tokio::spawn(async move { + let build = self.runtime.spawn(async move { let _building = building; match read.read(&media, &binding).await { Ok(Some(bytes)) => { From e46e2ec5eba2f07373df869db531dffeec02d27a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 14:44:50 -0700 Subject: [PATCH 8/9] fix(hls): leave refused videos out of the first-video wait A video whose media is denied or whose timeline failed can never be listed, so counting it made every MPD request wait out READY_TIMEOUT when only audio was playable. The wait now skips any rendition whose listings refuse, and wakes when one starts refusing mid-wait. Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/mod.rs | 2 +- rs/moq-hls/src/export/rendition.rs | 47 +++++++++++++++++++++++++ rs/moq-hls/src/export/renditions.rs | 9 +++-- rs/moq-hls/src/export/segments.rs | 15 ++++++++ rs/moq-hls/src/server/routes.rs | 53 +++++++++++++++++++++++++++++ 5 files changed, 123 insertions(+), 3 deletions(-) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 2e9e0ab5393..e78fe439153 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -348,7 +348,7 @@ impl Broadcaster { } /// Start every enabled video rendition's init build, then resolve once one of them has its - /// init, or none is enabled. A DASH player picks its adaptation sets from the first manifest + /// init, or none is left that could be listed. 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( diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index 3753b42f92b..f08de3d2c24 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -750,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. @@ -1657,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 1d936e1a212..34b713537c6 100644 --- a/rs/moq-hls/src/export/renditions.rs +++ b/rs/moq-hls/src/export/renditions.rs @@ -513,7 +513,8 @@ impl Producer { } /// Resolve once an enabled video rendition has its init and is - /// [advertised](Rendition::is_advertised), so a manifest lists it, or none is enabled. + /// [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. #[cfg_attr( not(feature = "server"), allow(dead_code, reason = "only the `server` feature uses it") @@ -524,7 +525,11 @@ impl Producer { let mut video = current .renditions .values() - .filter(|rendition| rendition.kind == Kind::Video && rendition.is_enabled()) + .filter(|rendition| { + rendition.kind == Kind::Video + && rendition.is_enabled() + && rendition.poll_refused(waiter).is_pending() + }) .peekable(); if video.peek().is_none() || video.any(|rendition| { diff --git a/rs/moq-hls/src/export/segments.rs b/rs/moq-hls/src/export/segments.rs index 81616887edb..35037bcf8c7 100644 --- a/rs/moq-hls/src/export/segments.rs +++ b/rs/moq-hls/src/export/segments.rs @@ -480,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. diff --git a/rs/moq-hls/src/server/routes.rs b/rs/moq-hls/src/server/routes.rs index a6946e85ad3..404cb6a5412 100644 --- a/rs/moq-hls/src/server/routes.rs +++ b/rs/moq-hls/src/server/routes.rs @@ -364,6 +364,59 @@ mod tests { 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. From bce6607d33e1c6a8930b8efb06e5fa3add1476ac Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 25 Oct 2026 15:06:55 -0700 Subject: [PATCH 9/9] fix(hls): start video init builds from the first-video wait The wait started each build once, up front, so an inline video whose timeline arrived mid-wait never built its init and the request timed out into an audio-only manifest. The wait now starts builds on every wake. Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/mod.rs | 53 ++++++++++++++++++++++++----- rs/moq-hls/src/export/renditions.rs | 10 ++++-- 2 files changed, 51 insertions(+), 12 deletions(-) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index e78fe439153..40b634063d4 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -347,20 +347,15 @@ impl Broadcaster { ))) } - /// Start every enabled video rendition's init build, then resolve once one of them has its - /// init, or none is left that could be listed. 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. + /// 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 video_init(&self) { - for rendition in self.renditions.snapshot() { - if rendition.kind == Kind::Video && rendition.is_enabled() { - rendition.start_init(); - } - } self.renditions.video_init().await; } @@ -1919,6 +1914,46 @@ mod tests { .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() { diff --git a/rs/moq-hls/src/export/renditions.rs b/rs/moq-hls/src/export/renditions.rs index 34b713537c6..f572a02dc89 100644 --- a/rs/moq-hls/src/export/renditions.rs +++ b/rs/moq-hls/src/export/renditions.rs @@ -512,9 +512,10 @@ impl Producer { .await; } - /// Resolve once an enabled video rendition 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. + /// 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") @@ -533,6 +534,9 @@ impl Producer { .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(())