refactor(tokio): make API shapes type-safe

dart/moq_ffi/lib/src/moq.dart

@@ -12108,7 +12108,7 @@ void _checkApiChecksums() {
   if (uniffi_moq_ffi_checksum_method_moqrequest_query() != 23842) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
-  if (uniffi_moq_ffi_checksum_method_moqrequest_reject() != 57471) {
+  if (uniffi_moq_ffi_checksum_method_moqrequest_reject() != 2829) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
   if (uniffi_moq_ffi_checksum_method_moqrequest_set_consume() != 45399) {

go/wrapper/server.go

@@ -76,7 +76,7 @@ func (r *Request) Accept(ctx context.Context) (*Session, error) {
 	return &Session{inner: inner}, nil
 }
 
-// Reject refuses the session with an HTTP status code (default convention: 404).
+// Reject refuses the session with an application error code; 401 and 403 map to unauthorized.
 func (r *Request) Reject(ctx context.Context, code uint16) error {
 	return runErr(ctx, r.inner.Cancel, func(ctx context.Context) error {
 		return r.inner.Reject(ctx, code)

py/moq-rs/README.md

@@ -133,7 +133,7 @@ client = moq.Client(
   - `.url`, `.path`, `.query`, `.transport`. The query-free path is uniform across transports; the root or missing path is `""`. The encoded query may contain credentials.
   - `.set_publish(origin)`, `.set_consume(origin)`. Per-request overrides, captured at `accept()`. Raise if the request is already answered, cancelled, or currently accepting.
   - `await .accept() → Session`. Complete the handshake (hold the result to keep the connection alive).
-  - `await .reject(code)`. Reject with an HTTP status code.
+  - `await .reject(code)`. Reject with an application error code; 401 and 403 map to unauthorized.
   - `.cancel()`. Cancel an in-flight `accept()`/`reject()` call.
 - **`Session`**. An established connection. Holding it keeps the connection alive; it is also an `async with` context manager that shuts down on exit.
   - `await .closed()`. Wait until the session closes.

py/moq-rs/moq/server.py

@@ -20,10 +20,10 @@ class Request:
     """Wraps MoqRequest, an incoming session that can be accepted or rejected.
 
     Use `await request.accept()` to complete the handshake, or
-    `await request.reject(code)` to reject with an HTTP status code.
+    `await request.reject(code)` to reject with an application error code.
 
     Dropping a Request without responding closes the underlying connection
-    silently; call `reject(code)` to send an explicit HTTP status.
+    silently; call `reject(code)` to send an explicit MoQ error.
     """
 
     def __init__(self, inner: MoqRequest) -> None:
@@ -75,7 +75,9 @@ async def accept(self) -> Session:
         return Session(await self._inner.accept())
 
     async def reject(self, code: int) -> None:
-        """Reject the session with the given HTTP status code.
+        """Reject the session with the given application error code.
+
+        Codes 401 and 403 map to the protocol's unauthorized error.
 
         Raises `Error.AlreadyResponded` if `accept()` or `reject()` has already
         been called.

quest/m1/README.md

@@ -32,7 +32,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
 - [Route cost](/quest/m1/api-route-cost.md) - `Route::with_hop` and `Cost: From<(u64, u64)>` go; ffi and libmoq build `Hops` and `Cost::from_warm_cold`
-- [moq-tokio shapes](/quest/m1/api-tokio-shapes.md) - a `Drop` on `Listener`, a worker `Member` that cannot be cross-wired, `std::time::Duration` fields, one construction idiom, no six-argument merge
 - [Catalog types](/quest/m1/api-hang-catalog.md) - `hang::Catalog<E>` is the one section list, `Clock` holds a `Timestamp`, `Timeline` folds into `Archive`
 - [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

quest/m1/api-review-gate.md

@@ -19,7 +19,6 @@ quest is deleted too. No code.
 The list: [Announce event](/quest/m1/api-net-announce.md),
 [Origin scoping](/quest/m1/api-net-origin.md),
 [Route cost](/quest/m1/api-route-cost.md),
-[moq-tokio shapes](/quest/m1/api-tokio-shapes.md),
 [Catalog types](/quest/m1/api-hang-catalog.md),
 [Rendition ownership](/quest/m1/api-mux-rendition.md),
 [Gateway types](/quest/m1/api-gateways.md),

quest/m2/relay-embed.md

@@ -19,8 +19,7 @@ All additive on `moq-relay` and `moq-tokio`, so on main after the merge:
 - `Config::parse_and_merge` and the `settings` registry become public
   (`moq_relay::settings()` plus a `merge_into` that composes an embedder's
   registry), so the edge's `config.rs` merge and its six clone-and-restore
-  fields go. This waits on the `cli::Merge` shape from
-  [moq-tokio shapes](/quest/m1/api-tokio-shapes.md).
+  fields go.
 - `auth::Config::public_grant()` and `is_empty()` are public; the edge
   `mem::take`s the two pattern lists to rebuild the union.
 - `Relay::with_listeners(self, impl IntoIterator<Item = accept::Health>)` so
@@ -53,6 +52,5 @@ Public API: additive. Wire: none.
 
 ## Related
 
-- [moq-tokio shapes](/quest/m1/api-tokio-shapes.md) - the merge shape this reuses
 - [Auth embedder](/quest/m2/auth-embedder.md) - the admission half of the same surface
 - [Server close](/quest/m2/moq-server-close.md) - the listener lifetime that sits beside `with_listeners`

rs/libmoq/src/api.rs

@@ -1101,13 +1101,13 @@ pub extern "C" fn moq_client_defaults() -> moq_client_config {
 		dst.websocket_delay_ms = millis(websocket.delay);
 		dst.has_websocket_delay = true;
 
-		dst.backoff_initial_us = micros(config.connect.backoff.initial());
+		dst.backoff_initial_us = micros(config.connect.backoff.initial);
 		dst.has_backoff_initial = true;
-		dst.backoff_multiplier = config.connect.backoff.multiplier();
+		dst.backoff_multiplier = config.connect.backoff.multiplier;
 		dst.has_backoff_multiplier = true;
-		dst.backoff_max_us = micros(config.connect.backoff.max());
+		dst.backoff_max_us = micros(config.connect.backoff.max);
 		dst.has_backoff_max = true;
-		dst.backoff_timeout_us = micros(config.connect.backoff.timeout());
+		dst.backoff_timeout_us = micros(config.connect.backoff.timeout);
 		dst.has_backoff_timeout = true;
 
 		let quic = config.quic.resolve();

rs/libmoq/src/client.rs

@@ -53,19 +53,19 @@ pub unsafe fn parse_client(config: Option<&moq_client_config>) -> Result<Config,
 		out.connect.bind = Some(addr);
 	}
 	if config.has_connect_timeout {
-		out.connect.timeout = std::time::Duration::from_millis(config.connect_timeout_ms).into();
+		out.connect.timeout = std::time::Duration::from_millis(config.connect_timeout_ms);
 	}
 	if config.has_failover_delay {
-		out.connect.race = std::time::Duration::from_millis(config.failover_delay_ms).into();
+		out.connect.race = std::time::Duration::from_millis(config.failover_delay_ms);
 	}
 	if config.has_resolution_delay {
-		out.connect.resolution_delay = std::time::Duration::from_millis(config.resolution_delay_ms).into();
+		out.connect.resolution_delay = std::time::Duration::from_millis(config.resolution_delay_ms);
 	}
 	if config.has_websocket_enabled {
 		out.connect.websocket.enabled = Some(config.websocket_enabled);
 	}
 	if config.has_websocket_delay {
-		out.connect.websocket.delay = std::time::Duration::from_millis(config.websocket_delay_ms).into();
+		out.connect.websocket.delay = std::time::Duration::from_millis(config.websocket_delay_ms);
 	}
 
 	// TLS. `insecure` needs no flag: false is both "unset" and "verify".
@@ -90,27 +90,27 @@ pub unsafe fn parse_client(config: Option<&moq_client_config>) -> Result<Config,
 
 	// Reconnect backoff
 	if config.has_backoff_initial {
-		out.connect.backoff.initial = std::time::Duration::from_micros(config.backoff_initial_us).into();
+		out.connect.backoff.initial = std::time::Duration::from_micros(config.backoff_initial_us);
 	}
 	if config.has_backoff_multiplier {
 		out.connect.backoff.multiplier = config.backoff_multiplier;
 	}
 	if config.has_backoff_max {
-		out.connect.backoff.max = std::time::Duration::from_micros(config.backoff_max_us).into();
+		out.connect.backoff.max = std::time::Duration::from_micros(config.backoff_max_us);
 	}
 	if config.has_backoff_timeout {
-		out.connect.backoff.timeout = std::time::Duration::from_micros(config.backoff_timeout_us).into();
+		out.connect.backoff.timeout = std::time::Duration::from_micros(config.backoff_timeout_us);
 	}
 
 	// QUIC
 	if config.has_quic_max_streams {
 		out.quic.max_streams = Some(config.quic_max_streams);
 	}
 	if config.has_quic_idle_timeout {
-		out.quic.idle_timeout = std::time::Duration::from_millis(config.quic_idle_timeout_ms).into();
+		out.quic.idle_timeout = std::time::Duration::from_millis(config.quic_idle_timeout_ms);
 	}
 	if config.has_quic_keep_alive {
-		out.quic.keep_alive = std::time::Duration::from_millis(config.quic_keep_alive_ms).into();
+		out.quic.keep_alive = std::time::Duration::from_millis(config.quic_keep_alive_ms);
 	}
 	if config.has_quic_gso {
 		out.quic.gso = Some(config.quic_gso);

rs/libmoq/src/test.rs

@@ -3974,7 +3974,7 @@ fn parsed(config: &moq_client_config) -> crate::client::Config {
 fn a_null_config_dials_with_the_defaults() {
 	let defaults = crate::client::Config::default();
 	let parsed = unsafe { crate::parse_client(None) }.expect("NULL is the defaults");
-	assert_eq!(parsed.connect.backoff.initial(), defaults.connect.backoff.initial());
+	assert_eq!(parsed.connect.backoff.initial, defaults.connect.backoff.initial);
 	assert_eq!(
 		parsed.connect.websocket.resolve().enabled,
 		defaults.connect.websocket.resolve().enabled
@@ -3989,13 +3989,10 @@ fn a_zeroed_config_is_the_defaults() {
 	let defaults = crate::client::Config::default();
 	let parsed = parsed(&client_config());
 
-	assert_eq!(parsed.connect.backoff.initial(), defaults.connect.backoff.initial());
-	assert_eq!(
-		parsed.connect.backoff.multiplier(),
-		defaults.connect.backoff.multiplier()
-	);
-	assert_eq!(parsed.connect.backoff.max(), defaults.connect.backoff.max());
-	assert_eq!(parsed.connect.backoff.timeout(), defaults.connect.backoff.timeout());
+	assert_eq!(parsed.connect.backoff.initial, defaults.connect.backoff.initial);
+	assert_eq!(parsed.connect.backoff.multiplier, defaults.connect.backoff.multiplier);
+	assert_eq!(parsed.connect.backoff.max, defaults.connect.backoff.max);
+	assert_eq!(parsed.connect.backoff.timeout, defaults.connect.backoff.timeout);
 	assert_eq!(
 		parsed.connect.websocket.resolve().enabled,
 		defaults.connect.websocket.resolve().enabled
@@ -4034,16 +4031,16 @@ fn defaults_report_what_a_zeroed_config_dials() {
 	assert!(config.has_backoff_initial);
 	assert_eq!(
 		config.backoff_initial_us,
-		expected.connect.backoff.initial().as_micros() as u64
+		expected.connect.backoff.initial.as_micros() as u64
 	);
 	assert!(config.has_backoff_multiplier);
-	assert_eq!(config.backoff_multiplier, expected.connect.backoff.multiplier());
+	assert_eq!(config.backoff_multiplier, expected.connect.backoff.multiplier);
 	assert!(config.has_backoff_max);
-	assert_eq!(config.backoff_max_us, expected.connect.backoff.max().as_micros() as u64);
+	assert_eq!(config.backoff_max_us, expected.connect.backoff.max.as_micros() as u64);
 	assert!(config.has_backoff_timeout);
 	assert_eq!(
 		config.backoff_timeout_us,
-		expected.connect.backoff.timeout().as_micros() as u64
+		expected.connect.backoff.timeout.as_micros() as u64
 	);
 
 	let websocket = expected.connect.websocket.resolve();
@@ -4071,7 +4068,7 @@ fn defaults_report_what_a_zeroed_config_dials() {
 	// And what it reports must round-trip: dialing with it is dialing with the defaults.
 	let expected = crate::client::Config::default();
 	let reparsed = parsed(&config);
-	assert_eq!(reparsed.connect.backoff.initial(), expected.connect.backoff.initial());
+	assert_eq!(reparsed.connect.backoff.initial, expected.connect.backoff.initial);
 	assert_eq!(
 		reparsed.connect.websocket.resolve().enabled,
 		expected.connect.websocket.resolve().enabled
@@ -4089,12 +4086,12 @@ fn zero_with_a_flag_set_is_a_real_value() {
 	config.has_quic_keep_alive = true;
 
 	let explicit = parsed(&config);
-	assert_eq!(explicit.connect.backoff.timeout(), std::time::Duration::ZERO);
+	assert_eq!(explicit.connect.backoff.timeout, std::time::Duration::ZERO);
 	assert_eq!(explicit.quic.keep_alive, std::time::Duration::ZERO);
 
 	// Without the flags the same zeroes mean nothing at all.
 	let defaults = parsed(&client_config());
-	assert_ne!(defaults.connect.backoff.timeout(), std::time::Duration::ZERO);
+	assert_ne!(defaults.connect.backoff.timeout, std::time::Duration::ZERO);
 	assert_eq!(defaults.quic.keep_alive, std::time::Duration::from_secs(5));
 }
 
@@ -4221,14 +4218,11 @@ fn config_quic_and_backoff_knobs_apply() {
 
 	let parsed = parsed(&config);
 	assert_eq!(
-		parsed.connect.backoff.initial(),
+		parsed.connect.backoff.initial,
 		std::time::Duration::from_micros(500_000)
 	);
-	assert_eq!(parsed.connect.backoff.multiplier(), 3);
-	assert_eq!(
-		parsed.connect.backoff.max(),
-		std::time::Duration::from_micros(10_000_000)
-	);
+	assert_eq!(parsed.connect.backoff.multiplier, 3);
+	assert_eq!(parsed.connect.backoff.max, std::time::Duration::from_micros(10_000_000));
 	assert_eq!(parsed.quic.max_streams, Some(4096));
 	assert_eq!(parsed.quic.idle_timeout, std::time::Duration::from_millis(15_000));
 	assert_eq!(parsed.quic.gso, Some(false));

rs/moq-bench/Cargo.toml

@@ -49,7 +49,6 @@ humantime = { workspace = true }
 moq-net = { workspace = true }
 moq-tokio = { workspace = true, default-features = false, features = ["aws-lc-rs"] }
 rand = { workspace = true }
-rustls = { version = "0.23", features = ["aws-lc-rs"], default-features = false }
 serde = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = ["full"] }

rs/moq-bench/src/config.rs

@@ -29,16 +29,16 @@ pub struct Config {
 
 	/// Spread connection and subscription startup over this duration to avoid a thundering herd.
 	#[usage(long, env = "MOQ_BENCH_STARTUP", default = "10s", setting = "startup")]
-	pub startup: moq_tokio::cli::Duration,
+	pub(crate) startup: crate::duration::Duration,
 
 	/// Stop the benchmark after this duration. Runs until interrupted if unset.
 	#[usage(long, env = "MOQ_BENCH_DURATION", setting = "duration")]
 	#[serde(default, skip_serializing_if = "Option::is_none")]
-	pub duration: Option<moq_tokio::cli::Duration>,
+	pub(crate) duration: Option<crate::duration::Duration>,
 
 	/// How often to log throughput stats.
 	#[usage(long, env = "MOQ_BENCH_REPORT", default = "1s", setting = "report")]
-	pub report: moq_tokio::cli::Duration,
+	pub(crate) report: crate::duration::Duration,
 
 	/// Number of connections (A) to establish. Rolled once for the whole run.
 	#[usage(long, env = "MOQ_BENCH_CONNECTIONS", default = "1", setting = "connections")]
@@ -150,7 +150,7 @@ impl Config {
 	/// Refused in `parse_and_merge`, before anything reads the config: those
 	/// spellings land on hidden fields that nothing honors, so continuing would dial
 	/// with settings the command line never asked for.
-	fn deprecated(&self) -> moq_tokio::Deprecated {
+	fn deprecated(&self) -> moq_tokio::cli::Deprecated {
 		let mut deprecated = self.client.deprecated();
 		deprecated.extend(self.quic.deprecated());
 		deprecated
@@ -203,8 +203,14 @@ impl Config {
 		// drops, so they are collected from the parse and reported with the file's
 		// own released keys in one message.
 		let mut deprecated = config.deprecated();
-		let (mut config, resolved) = moq_tokio::cli::merge(Settings::SETTINGS_REGISTRY, config, &cli_layer, &env, file)
-			.map_err(|err| anyhow::anyhow!("{err}"))?;
+		let (mut config, resolved) = moq_tokio::cli::Merge {
+			registry: Settings::SETTINGS_REGISTRY,
+			cli: &cli_layer,
+			env: &env,
+			file,
+		}
+		.apply(config)
+		.map_err(|err| anyhow::anyhow!("{err}"))?;
 		deprecated.extend(config.deprecated());
 		anyhow::ensure!(deprecated.is_empty(), "{deprecated}");
 		config.origins = Some(resolved);