fix(hls): publish every rendition in the first catalog

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

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<StepOutcome> {
 		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<u64>
 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<u8>)>,
+		gate: String,
+		arrived: Arc<Notify>,
+		release: Arc<Notify>,
+	) -> 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.