feat(moq-hls)!: expose a credential-aware serve surface for embedders

rs/moq-hls/src/export/master.rs

@@ -69,7 +69,7 @@ fn group_audio(audio: &[AudioVariant]) -> Vec<AudioGroup<'_>> {
 		.collect()
 }
 
-fn render_video(out: &mut String, variant: &VideoVariant, audio: Option<&AudioGroup<'_>>) {
+fn render_video(out: &mut String, variant: &VideoVariant, audio: Option<&AudioGroup<'_>>, suffix: &str) {
 	let bandwidth = variant
 		.bandwidth
 		.saturating_add(audio.map_or(0, |group| group.bandwidth));
@@ -86,11 +86,17 @@ fn render_video(out: &mut String, variant: &VideoVariant, audio: Option<&AudioGr
 		let _ = write!(line, ",AUDIO=\"{}\"", group.id);
 	}
 	let _ = writeln!(out, "{line}");
-	let _ = writeln!(out, "{}/{}/media.m3u8", Kind::Video.as_str(), variant.name);
+	let _ = writeln!(out, "{}/{}/media.m3u8{suffix}", Kind::Video.as_str(), variant.name);
 }
 
 /// Render the multivariant playlist. The first rendition in each audio codec group is default.
-pub fn render_master(video: &[VideoVariant], audio: &[AudioVariant]) -> String {
+///
+/// `query` is an optional query string (without the leading `?`, e.g. `jwt=<token>`)
+/// appended to every child media-playlist URL, so a credential the master was fetched
+/// with propagates to the rendition playlists a stock player loads next.
+pub fn render_master(video: &[VideoVariant], audio: &[AudioVariant], query: Option<&str>) -> String {
+	let suffix = query.map(|q| format!("?{q}")).unwrap_or_default();
+
 	let mut out = String::new();
 	let _ = writeln!(out, "#EXTM3U");
 	let _ = writeln!(out, "#EXT-X-VERSION:{VERSION}");
@@ -101,7 +107,7 @@ pub fn render_master(video: &[VideoVariant], audio: &[AudioVariant]) -> String {
 			let default = if index == 0 { "YES" } else { "NO" };
 			let _ = writeln!(
 				out,
-				"#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID=\"{}\",NAME=\"{}\",DEFAULT={default},AUTOSELECT=YES,URI=\"{}/{}/media.m3u8\"",
+				"#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID=\"{}\",NAME=\"{}\",DEFAULT={default},AUTOSELECT=YES,URI=\"{}/{}/media.m3u8{suffix}\"",
 				group.id,
 				variant.name,
 				Kind::Audio.as_str(),
@@ -112,10 +118,10 @@ pub fn render_master(video: &[VideoVariant], audio: &[AudioVariant]) -> String {
 
 	for variant in video {
 		if audio_groups.is_empty() {
-			render_video(&mut out, variant, None);
+			render_video(&mut out, variant, None, &suffix);
 		} else {
 			for group in &audio_groups {
-				render_video(&mut out, variant, Some(group));
+				render_video(&mut out, variant, Some(group), &suffix);
 			}
 		}
 	}
@@ -128,7 +134,7 @@ pub fn render_master(video: &[VideoVariant], audio: &[AudioVariant]) -> String {
 				"#EXT-X-STREAM-INF:BANDWIDTH={},CODECS=\"{}\"",
 				variant.bandwidth, variant.codec
 			);
-			let _ = writeln!(out, "{}/{}/media.m3u8", Kind::Audio.as_str(), variant.name);
+			let _ = writeln!(out, "{}/{}/media.m3u8{suffix}", Kind::Audio.as_str(), variant.name);
 		}
 	}
 
@@ -154,7 +160,7 @@ mod tests {
 			codec: "mp4a.40.2".into(),
 		}];
 
-		let out = render_master(&video, &audio);
+		let out = render_master(&video, &audio, None);
 		assert!(out.starts_with("#EXTM3U\n#EXT-X-VERSION:9\n"));
 		assert!(out.contains(
 			"#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID=\"aud\",NAME=\"audio\",DEFAULT=YES,AUTOSELECT=YES,URI=\"audio/audio/media.m3u8\"\n"
@@ -163,6 +169,11 @@ mod tests {
 			"#EXT-X-STREAM-INF:BANDWIDTH=2628000,RESOLUTION=1280x720,CODECS=\"avc1.42c01f,mp4a.40.2\",AUDIO=\"aud\"\n"
 		));
 		assert!(out.contains("\nvideo/video/media.m3u8\n"));
+
+		// A credential rides every child media-playlist URL, audio and video alike.
+		let signed = render_master(&video, &audio, Some("jwt=abc.def"));
+		assert!(signed.contains("URI=\"audio/audio/media.m3u8?jwt=abc.def\"\n"));
+		assert!(signed.contains("\nvideo/video/media.m3u8?jwt=abc.def\n"));
 	}
 
 	#[test]
@@ -192,7 +203,7 @@ mod tests {
 			},
 		];
 
-		let out = render_master(&video, &audio);
+		let out = render_master(&video, &audio, None);
 		assert!(out.contains("GROUP-ID=\"aud-0\",NAME=\"aac-low\",DEFAULT=YES"));
 		assert!(out.contains("GROUP-ID=\"aud-0\",NAME=\"aac-high\",DEFAULT=NO"));
 		assert!(out.contains("GROUP-ID=\"aud-1\",NAME=\"opus\",DEFAULT=YES"));
@@ -208,7 +219,7 @@ mod tests {
 			bandwidth: 128_000,
 			codec: "opus".into(),
 		}];
-		let out = render_master(&[], &audio);
+		let out = render_master(&[], &audio, None);
 		assert!(out.contains("#EXT-X-STREAM-INF:BANDWIDTH=128000,CODECS=\"opus\"\n"));
 		assert!(out.contains("\naudio/audio/media.m3u8\n"));
 	}

rs/moq-hls/src/export/mod.rs

@@ -118,7 +118,14 @@ impl Broadcaster {
 	}
 
 	/// Render the multivariant (master) playlist from the current renditions.
-	pub fn master_playlist(&self) -> String {
+	///
+	/// `query` is an optional query string (without the leading `?`, e.g. `jwt=<token>`)
+	/// appended to every child media-playlist URL, so a credential the master was fetched
+	/// with propagates to the rendition playlists a stock player loads next. It is an
+	/// argument rather than a [`Config`] field because one broadcaster fans out to viewers
+	/// holding different tokens; a credential in `Config` would embed one viewer's in
+	/// another's playlist.
+	pub fn master_playlist(&self, query: Option<&str>) -> String {
 		let mut video = Vec::new();
 		let mut audio = Vec::new();
 		for rendition in self.renditions.snapshot() {
@@ -137,7 +144,7 @@ impl Broadcaster {
 				}),
 			}
 		}
-		master::render_master(&video, &audio)
+		master::render_master(&video, &audio, query)
 	}
 
 	/// Whether the current catalog contains no servable renditions (serve path).
@@ -254,7 +261,7 @@ mod tests {
 			.expect("rendition discovered from the catalog");
 		let _ = tokio::time::timeout(Duration::from_secs(5), rendition.playable()).await;
 
-		let master = broadcaster.master_playlist();
+		let master = broadcaster.master_playlist(None);
 		assert!(master.contains("video/video0/media.m3u8"), "master lists the rendition");
 
 		let playlist = rendition.playlist();
@@ -265,11 +272,16 @@ mod tests {
 		assert_eq!(playlist.target_duration, 2, "ceil of the longest timeline gap");
 		assert!(!playlist.finished);
 
-		let rendered = render_media(&playlist);
+		let rendered = rendition.media_playlist(None).expect("playable");
 		assert!(rendered.contains("#EXT-X-MAP:URI=\"init.mp4\"\n"));
 		assert!(rendered.contains("seg/0.m4s\n"));
 		assert!(rendered.contains("seg/1.m4s\n"));
 
+		// The same render, but carrying a credential into every child URL.
+		let signed = rendition.media_playlist(Some("jwt=abc.def")).expect("playable");
+		assert!(signed.contains("#EXT-X-MAP:URI=\"init.mp4?jwt=abc.def\"\n"));
+		assert!(signed.contains("seg/0.m4s?jwt=abc.def\n"));
+
 		let init = rendition.init().await.unwrap().expect("init segment");
 		assert_eq!(&init[4..8], b"ftyp");
 

rs/moq-hls/src/export/playlist.rs

@@ -36,13 +36,19 @@ pub(crate) struct Segment {
 }
 
 /// Render a media playlist for one rendition from a [`Snapshot`].
-pub(crate) fn render_media(snapshot: &Snapshot) -> String {
+///
+/// `query` is an optional query string (without the leading `?`, e.g. `jwt=<token>`)
+/// appended to every child URL (the init map and each segment), so a stock player that
+/// does not replay request headers still carries a credential on its follow-up requests.
+pub(crate) fn render_media(snapshot: &Snapshot, query: Option<&str>) -> String {
+	let suffix = query.map(|q| format!("?{q}")).unwrap_or_default();
+
 	let mut out = String::new();
 	let _ = writeln!(out, "#EXTM3U");
 	let _ = writeln!(out, "#EXT-X-VERSION:{VERSION}");
 	let _ = writeln!(out, "#EXT-X-TARGETDURATION:{}", snapshot.target_duration);
 	let _ = writeln!(out, "#EXT-X-MEDIA-SEQUENCE:{}", snapshot.media_sequence);
-	let _ = writeln!(out, "#EXT-X-MAP:URI=\"init.mp4\"");
+	let _ = writeln!(out, "#EXT-X-MAP:URI=\"init.mp4{suffix}\"");
 
 	for (index, segment) in snapshot.segments.iter().enumerate() {
 		if index == 0
@@ -55,7 +61,7 @@ pub(crate) fn render_media(snapshot: &Snapshot) -> String {
 			);
 		}
 		let _ = writeln!(out, "#EXTINF:{:.5},", segment.duration);
-		let _ = writeln!(out, "seg/{}.m4s", segment.group);
+		let _ = writeln!(out, "seg/{}.m4s{suffix}", segment.group);
 	}
 
 	if snapshot.finished {
@@ -90,7 +96,7 @@ mod tests {
 			program_date_time: Some(SystemTime::UNIX_EPOCH + Duration::from_millis(1_751_846_400_123)),
 		};
 
-		let out = render_media(&snapshot);
+		let out = render_media(&snapshot, None);
 		assert!(out.starts_with("#EXTM3U\n#EXT-X-VERSION:6\n"));
 		assert!(out.contains("#EXT-X-TARGETDURATION:2\n"));
 		assert!(out.contains("#EXT-X-MEDIA-SEQUENCE:10\n"));
@@ -99,6 +105,12 @@ mod tests {
 		assert!(out.contains("#EXTINF:2.00000,\nseg/10.m4s\n"));
 		assert!(out.contains("#EXTINF:1.96000,\nseg/11.m4s\n"));
 		assert!(!out.contains("#EXT-X-ENDLIST"));
+
+		// A credential rides every child URL so a header-less player keeps sending it.
+		let signed = render_media(&snapshot, Some("jwt=abc.def"));
+		assert!(signed.contains("#EXT-X-MAP:URI=\"init.mp4?jwt=abc.def\"\n"));
+		assert!(signed.contains("\nseg/10.m4s?jwt=abc.def\n"));
+		assert!(signed.contains("\nseg/11.m4s?jwt=abc.def\n"));
 	}
 
 	#[test]
@@ -114,7 +126,7 @@ mod tests {
 			program_date_time: None,
 		};
 
-		let out = render_media(&snapshot);
+		let out = render_media(&snapshot, None);
 		assert!(out.contains("#EXT-X-ENDLIST\n"));
 		assert!(!out.contains("PROGRAM-DATE-TIME"));
 	}

rs/moq-hls/src/export/rendition.rs

@@ -83,8 +83,7 @@ pub struct Rendition {
 	live: Arc<segments::Producer>,
 	/// The largest `EXT-X-TARGETDURATION` advertised so far, in seconds. Latched monotonically:
 	/// HLS forbids a live playlist's target duration from changing, so it never shrinks even
-	/// after a long segment evicts from the window. Read only by the serve path's `playlist`.
-	#[cfg_attr(not(feature = "server"), allow(dead_code))]
+	/// after a long segment evicts from the window.
 	target_duration: std::sync::atomic::AtomicU64,
 	/// The init segment, built on first request.
 	init: tokio::sync::Mutex<Option<Bytes>>,
@@ -185,15 +184,28 @@ impl Rendition {
 		self.watcher.abort();
 	}
 
-	/// Resolve once the playlist is renderable ([`is_playable`](Self::is_playable)). Bounding
-	/// the wait is the caller's policy (the serve path wraps this in its own timeout).
-	#[cfg_attr(not(feature = "server"), allow(dead_code))]
-	pub(crate) async fn playable(&self) {
+	/// Resolve once the playlist is renderable, i.e. once [`media_playlist`](Self::media_playlist)
+	/// would return `Some`. Bounding the wait is the caller's policy (the serve path wraps this
+	/// in its own timeout).
+	pub async fn playable(&self) {
 		kio::wait(|waiter| self.live.poll_playable(waiter)).await;
 	}
 
+	/// Render this rendition's media playlist from the current timeline window, or `None` when
+	/// there is nothing to serve yet (no complete segment, and the broadcast has not ended) --
+	/// a playlist with no segments confuses players, so a server should treat `None` as "not
+	/// ready" rather than serve it.
+	///
+	/// `query` is an optional query string (without the leading `?`, e.g. `jwt=<token>`)
+	/// appended to every child URL (the init map and each segment), so a stock player that does
+	/// not replay request headers still carries a credential on its follow-up requests. It is an
+	/// argument rather than a [`Config`](super::Config) field because one broadcaster fans out
+	/// to viewers holding different tokens.
+	pub fn media_playlist(&self, query: Option<&str>) -> Option<String> {
+		self.is_playable().then(|| super::render_media(&self.playlist(), query))
+	}
+
 	/// Render the media playlist from the current timeline window.
-	#[cfg_attr(not(feature = "server"), allow(dead_code))]
 	pub(crate) fn playlist(&self) -> Snapshot {
 		let window = self.live.window();
 
@@ -228,7 +240,6 @@ impl Rendition {
 
 	/// Whether the playlist has anything to serve yet (at least one complete segment, or the
 	/// broadcast already ended).
-	#[cfg_attr(not(feature = "server"), allow(dead_code))]
 	pub(crate) fn is_playable(&self) -> bool {
 		self.live.is_playable()
 	}
@@ -301,7 +312,7 @@ impl Rendition {
 	/// Fetches every group the segment covers (audio timelines skip groups, so a segment may span
 	/// several) and encodes them as a single CMAF fragment. `None` when the segment isn't in the
 	/// playlist window or its groups already left the relay cache.
-	pub(crate) async fn segment(&self, sequence: u64) -> Result<Option<Bytes>> {
+	pub async fn segment(&self, sequence: u64) -> Result<Option<Bytes>> {
 		let Some((start, end)) = self.live.segment_groups(sequence) else {
 			return Ok(None);
 		};

rs/moq-hls/src/server/mod.rs

@@ -84,7 +84,7 @@ impl Server {
 	/// Get or create the [`Broadcaster`] for `name`, resolving the broadcast from
 	/// the relay (waiting briefly for its announcement). Returns `None` if the
 	/// broadcast never shows up.
-	pub(crate) async fn broadcaster(&self, name: &str) -> Option<Arc<Broadcaster>> {
+	pub async fn broadcaster(&self, name: &str) -> Option<Arc<Broadcaster>> {
 		{
 			let mut broadcasters = self.inner.broadcasters.lock().unwrap();
 			if let Some(existing) = broadcasters.get(name) {

rs/moq-hls/src/server/routes.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use axum::Router;
-use axum::extract::{Path, State};
+use axum::extract::{Path, RawQuery, State};
 use axum::http::{StatusCode, header};
 use axum::response::{IntoResponse, Response};
 use axum::routing::get;
@@ -29,20 +29,23 @@ pub fn router(server: Server) -> Router {
 		.with_state(server)
 }
 
-async fn master(State(server): State<Server>, Path(broadcast): Path<String>) -> Response {
+async fn master(State(server): State<Server>, Path(broadcast): Path<String>, RawQuery(query): RawQuery) -> Response {
 	let Some(broadcaster) = server.broadcaster(&broadcast).await else {
 		return not_found();
 	};
 	let _ = tokio::time::timeout(READY_TIMEOUT, broadcaster.ready()).await;
 	if broadcaster.is_empty() {
 		return not_found();
 	}
-	m3u8(broadcaster.master_playlist())
+	// Propagate whatever query reached the master (e.g. a credential a wrapping
+	// middleware required) down to the child media-playlist URLs.
+	m3u8(broadcaster.master_playlist(query.as_deref()))
 }
 
 async fn media(
 	State(server): State<Server>,
 	Path((broadcast, kind, rendition)): Path<(String, String, String)>,
+	RawQuery(query): RawQuery,
 ) -> Response {
 	let Some(rendition) = rendition_for(&server, &broadcast, &kind, &rendition).await else {
 		return not_found();
@@ -51,9 +54,6 @@ async fn media(
 	// A playlist with no segments confuses players; give the timeline a moment to index the
 	// first complete segment before answering.
 	let _ = tokio::time::timeout(READY_TIMEOUT, rendition.playable()).await;
-	if !rendition.is_playable() {
-		return not_found();
-	}
 
 	// The playlist references init.mp4 via EXT-X-MAP. Make sure it's actually buildable before
 	// advertising it (an inline-codec init needs a keyframe group fetched first), so a player
@@ -64,7 +64,10 @@ async fn media(
 		Err(err) => return server_error(err),
 	}
 
-	m3u8(crate::export::render_media(&rendition.playlist()))
+	match rendition.media_playlist(query.as_deref()) {
+		Some(playlist) => m3u8(playlist),
+		None => not_found(),
+	}
 }
 
 async fn init(