From 5d7a0b900e7ade4fb16962b4486aaea2c18851c8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 13:37:43 -0700 Subject: [PATCH 1/2] claim quest/m1/hls-first-catalog Co-authored-by: Grok 4.7 From d7f292101fac9a27eb160d8ea5d8c9a4eee017bf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 13:51:23 -0700 Subject: [PATCH 2/2] fix(hls): publish every rendition in the first catalog An HLS import reserved the catalog only when each rendition's init segment loaded, and that reservation ended before the next rendition existed, so the first snapshot listed only the first rendition. Hold a catalog reservation for the whole pass, beside the timeline reservation, and drop it when the pass ends. Co-authored-by: Grok 4.7 --- quest/m1/README.md | 1 - quest/m1/hls-first-catalog.md | 43 ---------- rs/moq-hls/src/import.rs | 157 +++++++++++++++++++++++++++++++++- 3 files changed, 154 insertions(+), 47 deletions(-) delete mode 100644 quest/m1/hls-first-catalog.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 7702a9f298..905a60451c 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -28,7 +28,6 @@ blocks. The quests that gated m0 lines moved under them. ## Required -- [HLS first catalog](/quest/m1/hls-first-catalog.md) - moq-hls import's first catalog lists every rendition from the master playlist, not only the first - [Typed timedness](/quest/m1/typed-timedness.md) - a track is all timed or all untimed, `Info.timescale` is optional, and `Timed` loses its clock parameter, in Rust, JS, and the bindings - [Untimed model](/quest/m1/untimed-model.md) - an untimed moq-net track reaches every subscriber untimed; no receiver fills in arrival time - [JS untimed model](/quest/m1/js-untimed-model.md) - the same in @moq/net diff --git a/quest/m1/hls-first-catalog.md b/quest/m1/hls-first-catalog.md deleted file mode 100644 index 4d7d3b5584..0000000000 --- a/quest/m1/hls-first-catalog.md +++ /dev/null @@ -1,43 +0,0 @@ -# [XS] moq-hls import publishes every rendition in its first catalog - -## Goal - -moq-hls import holds its first catalog until every rendition selected from -the master playlist that loads its init segment in the first pass has -reserved its tracks, so a consumer never sees a first catalog with only the -first rendition. - -Today each rendition's importer, and its `catalog.reserve()`, is created -lazily in `TrackState::ensure_map` (`rs/moq-hls/src/import.rs`) while `step` -ingests renditions one at a time. The first rendition's reservation can -release before the others exist, so the catalog publishes with one rendition -and grows on later updates. Reported by Dryvnt in -[#4824](https://github.com/moq-dev/moq/pull/4824). - -## Plan - -Decided (2026-10-05, maintainer): m1, [XS]. Each importer now releases its -reservation at its first frame, which still lands before the next rendition -is reserved. - -Guidance: - -- A catalog `Reserved` held across at least the first pass is the likely - shape: taken after `ensure_tracks` and before the first `ingest`. -- Don't let a hold outlive its pass. A rendition with no segments yet never - reaches `ensure_map`, and one whose init fetch fails under `OnError::Warn` - never reserves. Neither may withhold the catalog for the whole import, and a - live `Reserved` withholds it (see [Shared import - clock](/quest/m1/shared-clock.md), which rejected sharing offsets through - one for that reason). -- Regression test (fails on `main`): a master playlist with two or more - renditions, for example a video variant and a separate audio rendition - from `file://` fixtures as the existing import tests use. The catalog - consumer's first snapshot lists every rendition. - -Public API: none. Wire: none. A multi-rendition HLS import's first catalog -lists every rendition instead of only the first. - -## Related - -- [Shared import clock](/quest/m1/shared-clock.md) - moq-hls renditions share one `catalog::Input`, whose `reserve()` this hold may use diff --git a/rs/moq-hls/src/import.rs b/rs/moq-hls/src/import.rs index c8ce1a29e7..4d63a50f10 100644 --- a/rs/moq-hls/src/import.rs +++ b/rs/moq-hls/src/import.rs @@ -316,9 +316,9 @@ impl Sink { /// Mint an fMP4 importer that publishes only the roles in `select`. /// fn importer(&self, select: &select::Broadcast) -> Fmp4 { - // `reserve()` (not `clone()`) so the catalog isn't published until every - // importer's tracks resolve, keeping one-shot muxers from seeing a partial - // catalog (moq-mux reservation gate). + // `reserve()` (not `clone()`) so this init's tracks publish together. + // The pass holds a separate reservation across every rendition; this one + // only covers the tracks this init segment declares. Fmp4::new(self.broadcast.clone(), self.catalog.reserve()).with_select(select.clone()) } } @@ -722,6 +722,15 @@ impl Import { async fn step(&mut self, on_error: OnError) -> Result { self.ensure_tracks().await?; + // Hold the catalog across the pass. A rendition reserves only once its init + // segment loads, and that reservation ends at its first frame, before the next + // rendition is ingested. Without this, the first rendition publishes alone. + // The hold ends with the pass: a rendition that has no segments yet, or whose + // init fetch fails, never reserves, and must not withhold the catalog for the + // rest of the import. Every pass holds, not only the first, because a quiet + // playlist publishes nothing and the first catalog may be a later pass. + let _catalog = self.sink.catalog.reserve(); + let mut wrote_segments = 0; let mut target_duration = None; let mut failed = None; @@ -1013,9 +1022,12 @@ fn moq_sequence(discontinuity_sequence: u64, media_sequence: u64) -> Result mod tests { use super::*; use std::path::{Path, PathBuf}; + use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; + use std::task::Poll; use tokio::io::AsyncWriteExt as _; use tokio::net::TcpListener; + use tokio::sync::Notify; static COUNTER: AtomicUsize = AtomicUsize::new(0); @@ -1554,6 +1566,145 @@ mod tests { assert!(hls.audio.is_none()); } + /// The catalog consumer jumps to the newest snapshot, so this looks while the second + /// rendition's playlist is still withheld: on a pass that publishes early, that read is + /// already the first catalog, and it must list every rendition. + #[tokio::test] + async fn first_catalog_lists_every_rendition() { + let (init, fragments) = fmp4_parts(2); + let mut media = init.clone(); + media.extend_from_slice(&fragments[0]); + media.extend_from_slice(&fragments[1]); + let playlist = format!( + "#EXTM3U\n#EXT-X-VERSION:7\n#EXT-X-TARGETDURATION:2\n#EXT-X-MAP:URI=\"media.mp4\",BYTERANGE=\"{}@0\"\n#EXTINF:1,\n#EXT-X-BYTERANGE:{}@{}\nmedia.mp4\n#EXTINF:1,\n#EXT-X-BYTERANGE:{}\nmedia.mp4\n", + init.len(), + fragments[0].len(), + init.len(), + fragments[1].len() + ); + let master = "\ +#EXTM3U +#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID=\"aud\",NAME=\"en\",DEFAULT=YES,URI=\"audio.m3u8\" +#EXT-X-STREAM-INF:BANDWIDTH=1000000,CODECS=\"avc1.4d401f,mp4a.40.2\",AUDIO=\"aud\" +video.m3u8 +"; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let arrived = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let server = serve_gated( + listener, + vec![ + ("/master.m3u8".to_string(), master.as_bytes().to_vec()), + ("/video.m3u8".to_string(), playlist.as_bytes().to_vec()), + ("/audio.m3u8".to_string(), playlist.into_bytes()), + ("/media.mp4".to_string(), media), + ], + "/audio.m3u8".to_string(), + arrived.clone(), + release.clone(), + ); + + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = CatalogProducer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); + let mut consumer = catalog.consume().unwrap(); + let cfg = Config::new(Url::parse(&format!("http://{address}/master.m3u8")).unwrap()); + let mut import = Import::new(broadcast, catalog, cfg).unwrap(); + + // Keep the import alive across the read: dropping it tears the catalog track down, + // and the consumer would see that instead of the snapshot. + let importing = tokio::spawn(async move { + let result = import.init().await; + (result, import) + }); + + // Video has finished ingesting: its playlist, init, and segments all precede this fetch. + arrived.notified().await; + let published_early = match consumer.poll_next(&kio::Waiter::noop()) { + Poll::Pending => false, + Poll::Ready(Ok(Some(snapshot))) => { + assert!( + !snapshot.video.renditions.is_empty() && !snapshot.audio.renditions.is_empty(), + "first catalog listed video {} and audio {}", + snapshot.video.renditions.len(), + snapshot.audio.renditions.len() + ); + true + } + other => panic!("catalog consumer failed before the second rendition: {other:?}"), + }; + + release.notify_one(); + let (result, _import) = importing.await.unwrap(); + result.unwrap(); + + if !published_early { + let snapshot = match consumer.poll_next(&kio::Waiter::noop()) { + Poll::Ready(Ok(Some(snapshot))) => snapshot, + other => panic!("expected the first catalog, got {other:?}"), + }; + assert!(!snapshot.video.renditions.is_empty(), "video rendition missing"); + assert!(!snapshot.audio.renditions.is_empty(), "audio rendition missing"); + } + server.abort(); + } + + /// Answer `routes` over HTTP/1.1. `gate` is held until `release`, which is how the test + /// observes the catalog after one rendition and before the next. + fn serve_gated( + listener: TcpListener, + routes: Vec<(String, Vec)>, + gate: String, + arrived: Arc, + release: Arc, + ) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + loop { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + loop { + let mut chunk = [0; 1024]; + let read = stream.read(&mut chunk).await.unwrap(); + if read == 0 { + break; + } + request.extend_from_slice(&chunk[..read]); + if request.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + if request.is_empty() { + continue; + } + + let path = request_path(&request).unwrap_or(""); + if path == gate { + arrived.notify_one(); + release.notified().await; + } + let body = routes + .iter() + .find(|(name, _)| name == path) + .map(|(_, body)| body.as_slice()) + .unwrap_or_else(|| panic!("unexpected request for {path}: {}", String::from_utf8_lossy(&request))); + let mut response = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ) + .into_bytes(); + response.extend_from_slice(body); + stream.write_all(&response).await.unwrap(); + } + }) + } + + fn request_path(request: &[u8]) -> Option<&str> { + let request = std::str::from_utf8(request).ok()?; + let target = request.lines().next()?.split_whitespace().nth(1)?; + Some(target.split('?').next().unwrap_or(target)) + } + #[test] fn moq_sequence_orders_discontinuities_after_media_sequence() { // A new epoch outranks every segment of the previous one, even a higher media seq.