feat(gateway)!: align embedding APIs

quest/m1/README.md

@@ -31,7 +31,6 @@ the transport line in m2 assumes a single stack.
 - [Bindings announce match](/quest/m1/api-origin-scopes.md) - every binding takes a pattern scope and reports the announce match with its captures
 - [PathPrefixes](/quest/m1/api-path-prefixes.md) - the unused moq_net::PathPrefixes type is deleted before the release
 - [Rendition ownership](/quest/m1/api-mux-rendition.md) - one handle publishes a media track and reports its estimate, instead of five
-- [Gateway types](/quest/m1/api-gateways.md) - no `anyhow` in a gateway `Error`, `PathOwned` prefixes, `Duration` segments, `moq_rtc::Server::new(config)`, an SRT reject with a reason
 - [Cluster -01](/quest/m1/cluster-01/README.md) - rs/moq-net and js/net speak the revised cluster extension (HOP_ID, REQUEST_UPDATE repricing) and -01 is published
 - [API review gate](/quest/m1/api-review-gate.md) - each `api-*` quest above is landed or deferred by the maintainer before the merge PR opens
 - [Merge dev](/quest/m1/merge-dev.md) - dev lands on main with a closing keyword for every issue it fixed

quest/m1/api-review-gate.md

@@ -17,8 +17,7 @@ file is deleted on completion) or deletes the quest with a note in
 quest is deleted too. No code.
 
 The list: [Announce event](/quest/m1/api-net-announce.md),
-[Rendition ownership](/quest/m1/api-mux-rendition.md),
-[Gateway types](/quest/m1/api-gateways.md).
+[Rendition ownership](/quest/m1/api-mux-rendition.md).
 
 ## Related
 

quest/m2/gateway-embed.md

@@ -33,7 +33,6 @@ Public API: additive. Wire: none.
 
 ## Required
 
-- [Gateway types](/quest/m1/api-gateways.md) - the constructors this extends
 - [Merge dev](/quest/m1/merge-dev.md) - starts on main
 
 ## Related

quest/m2/qos/stats/schema.md

@@ -20,10 +20,10 @@ what a subscriber received and played, per audio and video. The relay is
   `[<tier>/]sessions.json[.z]`, a `BTreeMap<String, Presence>` keyed by auth
   root, read through `Consumer::sessions` whatever `E` is; a client publishes
   it only when it holds sessions worth counting.
-- An exact-path mode. `ProducerConfig` treats its path as a prefix and
+- An exact-path mode. `produce::Config` treats its path as a prefix and
   advertises `<prefix>/node[/<node>]`, so a client asking for
   `room/alice.stats` would publish `room/alice.stats/node`, which no longer
-  ends in `.stats`. `ProducerConfig::at(path)` publishes the broadcast at
+  ends in `.stats`. `produce::Config::at(path)` publishes the broadcast at
   exactly that path with no category segment, refusing a path that does not
   end in `.stats`; the relay keeps the prefix layout. Test the advertised
   path for both modes.

rs/moq-cli/src/hls.rs

@@ -2,10 +2,12 @@
 //! DASH over HTTP from MoQ broadcasts (export), fetching media groups on demand.
 
 use std::net::SocketAddr;
+use std::path::PathBuf;
 
 use anyhow::Context;
 use axum::http::Method;
 use hang::moq_net;
+use url::Url;
 
 use crate::moq::{ImportTarget, notify_ready};
 
@@ -60,6 +62,7 @@ pub async fn import(target: ImportTarget, playlist: String) -> anyhow::Result<()
 		.announce(Default::default())
 		.context("failed to announce broadcast")?;
 
+	let playlist = playlist_url(&playlist)?;
 	let mut importer = moq_hls::import::Import::new(producer, catalog, moq_hls::import::Config::new(playlist))?;
 
 	tracing::info!(%name, "importing HLS");
@@ -69,6 +72,20 @@ pub async fn import(target: ImportTarget, playlist: String) -> anyhow::Result<()
 	Ok(importer.run().await?)
 }
 
+fn playlist_url(playlist: &str) -> anyhow::Result<Url> {
+	if playlist.starts_with("http://") || playlist.starts_with("https://") {
+		return Url::parse(playlist).context("invalid HLS playlist URL");
+	}
+
+	let path = PathBuf::from(playlist);
+	let absolute = if path.is_absolute() {
+		path
+	} else {
+		std::env::current_dir()?.join(path)
+	};
+	Url::from_file_path(&absolute).map_err(|_| anyhow::anyhow!("invalid HLS playlist path: {}", absolute.display()))
+}
+
 /// Serve HLS and DASH over HTTP for the single broadcast `name` (reached at
 /// `/<name>/master.m3u8` and `/<name>/manifest.mpd`); other broadcasts in the
 /// Origin are not served.

rs/moq-cli/src/rtc.rs

@@ -61,8 +61,8 @@ pub async fn listen_import(target: ImportTarget, listen: Listen) -> anyhow::Resu
 	let mut config = server_config(&listen);
 	config.max_age = target.max_age;
 	config.bandwidth = target.bandwidth;
-	let server = moq_rtc::Server::new(config, publisher, target.origin.consume());
-	serve(server.publish_router(), "WHIP", listen).await
+	let server = moq_rtc::Server::new(config);
+	serve(server.publish_router(publisher), "WHIP", listen).await
 }
 
 /// WHEP server: serve WebRTC plays of `name` from the Origin (export).
@@ -73,11 +73,8 @@ pub async fn listen_export(origin: moq_net::origin::Consumer, name: String, list
 	let subscriber = origin
 		.scope("", &scope)
 		.with_context(|| format!("failed to scope origin to broadcast `{name}`"))?;
-	// A WHEP server only reads; it still needs a publisher handle for the shared
-	// glue, so hand it an unused, empty Origin producer.
-	let publisher = moq_tokio::origin::spawn();
-	let server = moq_rtc::Server::new(server_config(&listen), publisher, subscriber);
-	serve(server.subscribe_router(), "WHEP", listen).await
+	let server = moq_rtc::Server::new(server_config(&listen));
+	serve(server.subscribe_router(subscriber), "WHEP", listen).await
 }
 
 /// Restrict a producer to the single broadcast `name` so a WHIP peer can only publish it.

rs/moq-cli/src/srt.rs

@@ -6,7 +6,7 @@ use std::time::Duration;
 
 use anyhow::Context;
 use hang::moq_net;
-use moq_srt::{Request, Server};
+use moq_srt::{Reject, Request, Server};
 use moq_tokio::RedactedUrl;
 use url::Url;
 
@@ -62,7 +62,7 @@ pub async fn listen_import(target: ImportTarget, addr: SocketAddr, latency: Dura
 			}
 			Request::Subscribe(subscribe) => {
 				tokio::spawn(async move {
-					let _ = subscribe.reject().await;
+					let _ = subscribe.reject(Reject::Forbidden).await;
 				});
 			}
 			_ => {}
@@ -96,7 +96,7 @@ pub async fn listen_export(
 			}
 			Request::Publish(publish) => {
 				tokio::spawn(async move {
-					let _ = publish.reject().await;
+					let _ = publish.reject(Reject::Forbidden).await;
 				});
 			}
 			_ => {}
@@ -113,11 +113,11 @@ pub async fn connect_import(target: ImportTarget, url: Url, latency: Duration) -
 	tracing::info!(url = %RedactedUrl::new(&url), %name, "SRT client pulling");
 	notify_ready();
 
-	let mut config = moq_srt::dial::Config::new(addr, resource);
-	config.latency = latency;
-	config.max_age = target.max_age;
-	config.bandwidth = target.bandwidth;
-	Ok(moq_srt::dial::pull(&config, &target.origin, name).await?)
+	let client = moq_srt::Client::new(addr, resource)
+		.with_latency(latency)
+		.with_max_age(target.max_age)
+		.with_bandwidth(target.bandwidth);
+	Ok(client.pull(&target.origin, name).await?)
 }
 
 /// Push a broadcast from the Origin to a remote SRT server (export).
@@ -131,9 +131,8 @@ pub async fn connect_export(
 	tracing::info!(url = %RedactedUrl::new(&url), %name, "SRT client pushing");
 	notify_ready();
 
-	let mut config = moq_srt::dial::Config::new(addr, resource);
-	config.latency = latency;
-	Ok(moq_srt::dial::publish(&config, &origin, &name).await?)
+	let client = moq_srt::Client::new(addr, resource).with_latency(latency);
+	Ok(client.publish(&origin, &name).await?)
 }
 
 /// Parse `srt://host:port?streamid=<resource>` into a resolved address and resource.

rs/moq-hls/Cargo.toml

@@ -17,9 +17,6 @@ default = ["server"]
 server = ["dep:axum"]
 
 [dependencies]
-# Always required (import + export library).
-anyhow = { workspace = true, features = ["backtrace"] }
-
 # Only needed by the HTTP export server (gated by `server`).
 axum = { workspace = true, optional = true }
 bytes = { workspace = true }

rs/moq-hls/src/error.rs

@@ -39,14 +39,6 @@ pub enum Error {
 	#[error("mux: {0}")]
 	Mux(#[from] moq_mux::Error),
 
-	/// The playlist argument looked like an HTTP(S) URL but failed to parse.
-	#[error("invalid playlist URL")]
-	InvalidPlaylistUrl,
-
-	/// The playlist argument was a local path that could not be made into a `file://` URL.
-	#[error("invalid file path")]
-	InvalidFilePath,
-
 	/// A `file://` URL could not be turned back into a filesystem path.
 	#[error("invalid file URL")]
 	InvalidFileUrl,
@@ -127,10 +119,6 @@ pub enum Error {
 	/// I/O error while reading a local playlist or segment.
 	#[error("io: {0}")]
 	Io(std::sync::Arc<std::io::Error>),
-
-	/// Catch-all for gateway logic that reports via `anyhow`.
-	#[error("{0}")]
-	Other(std::sync::Arc<anyhow::Error>),
 }
 
 impl Error {
@@ -159,12 +147,6 @@ impl From<std::io::Error> for Error {
 	}
 }
 
-impl From<anyhow::Error> for Error {
-	fn from(err: anyhow::Error) -> Self {
-		Error::Other(std::sync::Arc::new(err))
-	}
-}
-
 /// Convenience alias for results from the HLS gateway.
 pub type Result<T> = std::result::Result<T, Error>;
 

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

@@ -740,7 +740,7 @@ mod tests {
 		let playlist = rendition.playlist();
 		assert_eq!(playlist.segments.len(), 2, "the live-edge group is not listed");
 		assert_eq!(playlist.segments[0].segment, 0);
-		assert_eq!(playlist.segments[0].duration, 2.0);
+		assert_eq!(playlist.segments[0].duration, Duration::from_secs(2));
 		assert_eq!(playlist.segments[1].segment, 1);
 		assert_eq!(
 			playlist.target_duration, 2,
@@ -1107,7 +1107,7 @@ mod tests {
 		let _ = tokio::time::timeout(Duration::from_secs(5), rendition.playable()).await;
 
 		let playlist = rendition.playlist();
-		assert_eq!(playlist.segments[0].duration, 3.0);
+		assert_eq!(playlist.segments[0].duration, Duration::from_secs(3));
 		assert_eq!(
 			playlist.target_duration, 3,
 			"no bound was declared, so the target duration must still cover the 3s segment"
@@ -1179,7 +1179,11 @@ mod tests {
 		let audio_segments: Vec<u64> = audio_playlist.segments.iter().map(|s| s.segment).collect();
 		assert_eq!(video_segments, vec![0, 1]);
 		assert_eq!(audio_segments, vec![0, 1], "audio lists the same aligned segments");
-		assert_eq!(audio_playlist.segments[0].duration, 2.0, "cut at the video boundary");
+		assert_eq!(
+			audio_playlist.segments[0].duration,
+			Duration::from_secs(2),
+			"cut at the video boundary"
+		);
 
 		// The same URI names the same span of content time on either rendition.
 		let rendered = audio_rendition.media_playlist(None).expect("playable");
@@ -1926,7 +1930,7 @@ mod tests {
 			.expect("a segment, not end");
 		assert_eq!(first.segment, 0);
 		assert_eq!(&first.media[4..8], b"moof", "the segment carries its transmuxed media");
-		assert_eq!(first.duration, 2.0);
+		assert_eq!(first.duration, Duration::from_secs(2));
 		assert!(!first.discontinuity, "a clean start is not a discontinuity");
 
 		let second = segments.next().await.unwrap().expect("second segment");

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

@@ -7,7 +7,7 @@
 //! directory.
 
 use std::fmt::Write;
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
 
 /// fMP4 segments via `EXT-X-MAP` require protocol version 6.
 const VERSION: u32 = 6;
@@ -35,8 +35,8 @@ pub(crate) struct Snapshot {
 pub(crate) struct Segment {
 	/// The aligned segment number; the URI is `seg/{segment}.m4s`.
 	pub segment: u64,
-	/// `EXTINF` duration in seconds.
-	pub duration: f64,
+	/// `EXTINF` duration.
+	pub duration: Duration,
 	/// The rendition has no content for this span (`EXT-X-GAP`): the segment keeps its slot in
 	/// the aligned numbering, but a player should not request it.
 	pub gap: bool,
@@ -82,7 +82,7 @@ pub(crate) fn render_media(snapshot: &Snapshot, query: Option<&str>) -> String {
 		if segment.gap {
 			let _ = writeln!(out, "#EXT-X-GAP");
 		}
-		let _ = writeln!(out, "#EXTINF:{:.5},", segment.duration);
+		let _ = writeln!(out, "#EXTINF:{:.5},", segment.duration.as_secs_f64());
 		let _ = writeln!(out, "seg/{}.m4s{suffix}", segment.segment);
 	}
 
@@ -107,13 +107,13 @@ mod tests {
 			segments: vec![
 				Segment {
 					segment: 10,
-					duration: 2.0,
+					duration: Duration::from_secs(2),
 					gap: false,
 					discontinuity: false,
 				},
 				Segment {
 					segment: 11,
-					duration: 1.96,
+					duration: Duration::from_millis(1960),
 					gap: false,
 					discontinuity: false,
 				},
@@ -146,7 +146,7 @@ mod tests {
 			media_sequence: 0,
 			segments: vec![Segment {
 				segment: 0,
-				duration: 4.0,
+				duration: Duration::from_secs(4),
 				gap: false,
 				discontinuity: false,
 			}],