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.