feat(relay): serve moq-transport from the io_uring workers

Cargo.toml

@@ -221,6 +221,7 @@ rcgen = { version = "0.14", default-features = false }
 reqwest = { version = "0.13", default-features = false }
 rustc-hash = "2.1.3"
 sd-notify = "0.5"
+send_wrapper = "0.6"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 serde_with = "3"

doc/bin/relay/config.md

@@ -145,8 +145,8 @@ first cores, so two relays on one host, or one limited to fewer CPUs than the
 host has, would crowd onto them. Turn it on for a host the relay has to itself.
 
 `io_uring` additionally needs Linux 6.12+ and the `io-uring` cargo feature; it
-shares the same reloadable TLS identity, serves moq-lite only, and refuses
-`mtu_discovery`. A setting the build or host cannot deliver refuses to
+shares the same reloadable TLS identity, serves every configured version
+like the shared runtime, and refuses `mtu_discovery`. A setting the build or host cannot deliver refuses to
 start rather than being ignored. If io\_uring workers fail to start naming
 `RLIMIT_MEMLOCK`, raise it (`LimitMEMLOCK=` under systemd).
 

quest/m2/README.md

@@ -129,7 +129,6 @@ ranked just below deterministic TS legs, which it requires.
 - [Link quality](/quest/m2/link-quality.md) - a radio link's cost follows its measured quality without flapping routes
 - [One transport adapter](/quest/m2/transport-adapter-dedup.md) - the poll transport adapter exists once, and the 64 KiB cap in moq-net's default `poll_read_buf` (not in the adapter) has a test
 - [Archive recovery listing](/quest/m2/archive-recovery-listing.md) - a resumed DVR lists what changed since its checkpoint, not every stored group
-- [IETF on the ring](/quest/m2/uring-ietf.md) - the io_uring workers serve moq-transport sessions too, so a uring relay drops no client protocol
 - [One connection lifecycle](/quest/m2/uring-connection-log.md) - every accept path, io_uring included, logs a connection's span and close from one owner, so Interop's accounting works on the ring
 - [Relay io_uring packages](/quest/m2/relay-io-uring-package.md) - Linux relay packages ship io_uring once the ring is on par with tokio
 - [Opus implementation](/quest/m2/audio-opus-backend.md) - compare Opus codec quality, CPU, build cost, and the loss recovery each backend offers

quest/m2/relay-io-uring-package.md

@@ -34,6 +34,4 @@ Public API: none. Wire: none.
 
 ## Required
 
-- [moq-transport on io_uring](/quest/m2/uring-ietf.md) - a packaged relay
-  must not drop protocols when the ring is on
 - [One port on the io_uring workers](/quest/m2/uring-demux.md) - a one-port config is served on the ring too

quest/m2/uring-connection-log.md

@@ -52,5 +52,4 @@ Public API: none expected. Wire: none.
 
 ## Related
 
-- [IETF on the ring](/quest/m2/uring-ietf.md) - also touches the ring's accept path
 - [Relay io_uring packages](/quest/m2/relay-io-uring-package.md) - ships the ring, whose operators read these logs

rs/moq-relay/src/runtime.rs

@@ -38,7 +38,7 @@ pub struct Config {
 	///
 	/// Each worker owns a `SINGLE_ISSUER` ring, batched UDP (multishot receive
 	/// with `UDP_GRO`, `UDP_SEGMENT` send), userspace timers, and a local task
-	/// set; sessions are moq-lite only (native raw QUIC and browser
+	/// set; sessions get every configured version (native raw QUIC and browser
 	/// WebTransport alike), and everything else (HTTP, WebSocket, cluster
 	/// dials, stats) stays on the shared runtime. Requires `runtime-workers`,
 	/// Linux 6.12+, and refuses to start anywhere it cannot deliver.

rs/moq-relay/src/uring.rs

@@ -4,8 +4,10 @@
 //! per worker, each with its own ring, its own `SO_REUSEPORT` socket steered
 //! by connection id ([`moq_sock::shard`]), and its own
 //! [`moq_uring::quic::Endpoint`] serving browsers (WebTransport over `h3`)
-//! and native peers (raw QUIC, moq-lite ALPNs) alike. Sessions are moq-lite
-//! only.
+//! and native peers (raw QUIC) alike, on every version the listener is
+//! configured for. moq-lite on its own ALPN runs unboxed on the worker;
+//! moq-transport and the SETUP-negotiated versions run through
+//! [`moq_uring::transport::Local`], since their drivers need `Send`.
 //!
 //! The split of work mirrors the relay's topology: a worker owns everything
 //! transport-shaped (the handshake, the session's protocol machine), while
@@ -64,8 +66,8 @@ struct Serve {
 	/// The shared runtime, which owns authentication (the auth API's HTTP
 	/// client needs its reactor) and session supervision.
 	tokio: tokio::runtime::Handle,
-	/// The moq-lite ALPNs this listener speaks, for the WebTransport
-	/// subprotocol pick.
+	/// The moq ALPNs this listener speaks, for the WebTransport subprotocol
+	/// pick.
 	alpns: Arc<Vec<String>>,
 	/// The negotiated-version restriction, when the operator set one.
 	versions: moq_net::Versions,
@@ -181,39 +183,14 @@ impl Workers {
 		// The requested address, with an ephemeral port resolved.
 		let addr = group.addr();
 
-		// The moq-lite ALPNs this listener speaks: the operator's version
-		// restriction when one is set, minus anything that is not lite (the
-		// io_uring path runs unboxed lite machines only). `h3` rides along
-		// for browsers, which negotiate the moq version as a WebTransport
-		// subprotocol instead.
+		// Every configured version's ALPN, as the tokio listener advertises.
+		// `h3` rides along for browsers, which negotiate the moq version as a
+		// WebTransport subprotocol instead.
 		let versions: moq_net::Versions = match listen.version.is_empty() {
 			true => Default::default(),
 			false => listen.version.clone().into(),
 		};
-		let alpns: Vec<String> = versions
-			.iter()
-			.filter(|version| version.is_lite())
-			.map(|version| version.alpn().to_string())
-			.collect();
-		if alpns.is_empty() {
-			anyhow::bail!("io_uring workers speak moq-lite only, and the version restriction leaves none");
-		}
-		// lite 01/02 have no ALPN of their own: they negotiate over the shared
-		// `moql` one through the bidi SETUP exchange, which
-		// `accept_request_lite` does not implement. The default set carries
-		// them and simply does not advertise them, like the moq-transport
-		// versions below; asking for one *by name* is the case that would
-		// start cleanly and then fail every handshake.
-		if !listen.version.is_empty()
-			&& let Some(version) = versions.iter().find(|version| version.uses_setup_negotiation())
-		{
-			anyhow::bail!("io_uring workers cannot serve {version:?}, which negotiates its version in the SETUP");
-		}
-		if versions.iter().any(|version| !version.is_lite()) {
-			tracing::warn!(
-				"io_uring workers speak moq-lite only; the configured moq-transport versions are not served"
-			);
-		}
+		let alpns: Vec<String> = versions.alpns().into_iter().map(str::to_owned).collect();
 
 		let alpn = alpns
 			.iter()
@@ -648,11 +625,6 @@ async fn serve_connection(
 
 	// Browsers speak WebTransport; everyone else raw QUIC. The moq version
 	// rides the ALPN for raw QUIC and the WebTransport subprotocol for `h3`.
-	// A browser offering no subprotocol we speak is refused with
-	// `Error::Version` rather than falling back to the pre-lite-05 bidi SETUP
-	// the tokio listener still accepts: that path is moq-transport-shaped, and
-	// reaching it would drag the thread-affinity bounds `accept_request_lite`
-	// exists to avoid back onto this transport.
 	// The negotiated moq protocol: the WebTransport sub-protocol for `h3`, else the ALPN.
 	let mut alpn = conn.protocol().map(str::to_owned);
 	let (transport, url) = match conn.protocol() {
@@ -688,13 +660,80 @@ async fn serve_connection(
 		Some(_) => moq_net::transport::Binding::Uri,
 		None => moq_net::transport::Binding::InBand,
 	};
-	let request = deadline
-		.bound(server.accept_request_lite(
-			moq_net::time::now(),
-			moq_uring::transport::Session::new(transport, binding),
-		))
-		.await
-		.context("moq handshake failed")?;
+	let session = moq_uring::transport::Session::new(transport, binding);
+	let now = moq_net::time::now();
+	// moq-lite on its own ALPN runs unboxed on this worker. Everything else
+	// (moq-transport, and the versions that negotiate in a bidi SETUP, which a
+	// browser offering no subprotocol falls back to) boxes its futures with a
+	// native `Send` bound, which only a `Local` session meets.
+	let unboxed = alpn
+		.as_deref()
+		.and_then(moq_net::Version::from_alpn)
+		.is_some_and(|version| version.is_lite());
+	let mut accepted = Accepted {
+		handle,
+		conn,
+		deadline,
+		serve,
+		id,
+		remote,
+		sni,
+		identity,
+		alpn,
+		url,
+	};
+	match unboxed {
+		true => {
+			let request = accepted
+				.deadline
+				.bound(server.accept_request_lite(now, session))
+				.await
+				.context("moq handshake failed")?;
+			serve_request(accepted, request).await
+		}
+		false => {
+			let request = accepted
+				.deadline
+				.bound(server.accept_request(now, moq_uring::transport::Local::new(session)))
+				.await
+				.context("moq handshake failed")?;
+			serve_request(accepted, request).await
+		}
+	}
+}
+
+/// A connection past its transport handshake, with the facts read off it
+/// before the moq handshake consumed it.
+struct Accepted {
+	handle: moq_uring::Handle,
+	conn: moq_uring::quic::web::Session,
+	deadline: Deadline,
+	serve: Serve,
+	id: u64,
+	remote: SocketAddr,
+	sni: Option<String>,
+	identity: Option<moq_tokio::tls::PeerIdentity>,
+	alpn: Option<String>,
+	url: Option<moq_uring::url::Url>,
+}
+
+/// Authenticate on the shared runtime, then start the session on the worker.
+async fn serve_request<S: moq_net::transport::poll::Session>(
+	accepted: Accepted,
+	request: moq_net::server::Handshake<S>,
+) -> anyhow::Result<()> {
+	let Accepted {
+		handle,
+		conn,
+		mut deadline,
+		serve,
+		id,
+		remote,
+		sni,
+		identity,
+		alpn,
+		url,
+	} = accepted;
 
 	// The path + `?jwt=` ride the URL for WebTransport and the SETUP for raw
 	// QUIC; either way the grant comes from the shared runtime, which owns the

rs/moq-relay/tests/integration/runtime_uring.rs

@@ -190,6 +190,103 @@ async fn uring_workers_serve_webtransport_and_raw_quic() {
 	let _ = running.await;
 }
 
+/// The workers serve every version the listener lists, not only moq-lite on
+/// its own ALPN: moq-transport, both on its own ALPN and over the bidi SETUP
+/// exchange, and the SETUP-negotiated moq-lite versions. A moq-transport
+/// publisher feeds a subscriber per version over raw QUIC and WebTransport,
+/// each of which must negotiate exactly the version it was pinned to.
+#[tokio::test]
+async fn uring_workers_serve_every_configured_version() {
+	let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+	if !supported() {
+		return;
+	}
+
+	let pinned = |version: &str| {
+		let version: moq_net::Version = version.parse().expect("known version");
+		let mut config = moq_tokio::connect::Config::default();
+		config.tls.insecure = Some(true);
+		config.once = Some(true);
+		config.bind = Some("127.0.0.1:0".parse().expect("parse bind"));
+		config.version = vec![version];
+		(version, config.init(Default::default()).expect("client init"))
+	};
+
+	let dir = tempfile::tempdir().expect("tempdir");
+	let (cert, key) = certificate(dir.path());
+	let relay = Relay::load(uring_config(&cert, &key)).await.expect("load relay");
+	let port = relay.addr().expect("workers bound an address").port();
+	let running = tokio::spawn(relay.run());
+
+	let origin = moq_tokio::origin::spawn();
+	let broadcast = origin.create_broadcast("test").expect("create broadcast");
+	broadcast.announce(Default::default()).expect("announce broadcast");
+	let track = broadcast.create_track("video", timed()).expect("create track");
+	let mut group = track.append_group().expect("append group");
+	group
+		.write_frame(moq_net::Timestamp::ZERO, b"hello".as_ref())
+		.expect("write frame");
+	group.finish().expect("finish group");
+
+	let raw = format!("moql://127.0.0.1:{port}/versions");
+	let (version, client) = pinned("moq-transport-22");
+	let publisher = connect(client.with_publisher(&origin), raw.parse().unwrap()).await;
+	assert_eq!(publisher.version(), Some(version), "publisher version");
+
+	// moq-transport-14 and moq-lite-02 negotiate in the SETUP; 16 has its own
+	// ALPN but still a bidi SETUP; 22 reads its SETUP off a uni stream.
+	for name in [
+		"moq-transport-22",
+		"moq-transport-16",
+		"moq-transport-14",
+		"moq-lite-02",
+	] {
+		for scheme in ["moql", "https"] {
+			let url: url::Url = format!("{scheme}://127.0.0.1:{port}/versions").parse().unwrap();
+			let (version, client) = pinned(name);
+			let origin = moq_tokio::origin::spawn();
+			let consumer = origin.consume();
+			let mut announced = consumer.announced();
+			let connection = connect(client.with_subscriber(origin), url).await;
+			assert_eq!(connection.version(), Some(version), "{scheme} {name}");
+
+			let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announced))
+				.await
+				.unwrap_or_else(|_| panic!("{scheme} {name}: announcement timeout"))
+				.expect("origin closed");
+			// moq-lite-02 has no in-band path, so a raw QUIC session sits at the root.
+			let path = update.prefix.as_str().to_owned();
+			assert!(active && path.ends_with("test"), "{scheme} {name}: announced {path}");
+			let broadcast = tokio::time::timeout(TIMEOUT, consumer.request_broadcast(&path, None))
+				.await
+				.unwrap_or_else(|_| panic!("{scheme} {name}: request timeout"))
+				.expect("announced broadcast resolves");
+			let mut subscription = broadcast
+				.track("video")
+				.unwrap()
+				.subscribe(None)
+				.await
+				.expect("subscribe");
+			let mut group = tokio::time::timeout(TIMEOUT, subscription.recv_group())
+				.await
+				.unwrap_or_else(|_| panic!("{scheme} {name}: recv_group timeout"))
+				.expect("recv_group failed")
+				.expect("track closed prematurely");
+			let frame = tokio::time::timeout(TIMEOUT, group.read_frame())
+				.await
+				.unwrap_or_else(|_| panic!("{scheme} {name}: read_frame timeout"))
+				.expect("read_frame failed")
+				.expect("group closed prematurely");
+			assert_eq!(&frame.payload[..], b"hello", "{scheme} {name}");
+		}
+	}
+
+	assert!(!running.is_finished(), "the relay stopped while serving");
+	drop(publisher);
+	running.abort();
+	let _ = running.await;
+}
+
 /// A session the workers accept reports its peer address, the listener's
 /// address, and the name the client dialed, as the tokio listener does: the
 /// SNI on raw QUIC and WebTransport alike.
@@ -243,7 +340,7 @@ async fn uring_workers_report_link_facts() {
 /// threads joined and the port free.
 ///
 /// What an arrival is told is left of the window only reaches the wire on
-/// moq-transport-17+, which the workers do not speak; `shutdown_signal.rs`
+/// moq-transport-17+, which this test does not dial; `shutdown_signal.rs`
 /// reads it through the shared runtime, whose supervision this path shares.
 #[tokio::test]
 async fn uring_workers_drain_on_the_trigger() {

rs/moq-uring/Cargo.toml

@@ -37,6 +37,8 @@ rustls = { version = "0.23", optional = true }
 # OS-native server-cert verifier for the rustls backends. Version matched to
 # the copy moq-tokio pulls in, to avoid duplication.
 rustls-platform-verifier = { version = "0.7", optional = true }
+# Runtime thread checks for the session moq-transport's `Send` drivers capture.
+send_wrapper = { workspace = true }
 slab = "0.4"
 thiserror = "2"
 tracing = "0.1"

rs/moq-uring/README.md

@@ -33,7 +33,10 @@ task set, and the UDP sockets bound through it.
   `transport::Session::new(session, binding)`, using `moq_net::transport::Binding::Uri`
   for WebTransport or `InBand` for raw QUIC, before `connect_lite`/`accept_lite`, which return
   the session and its driver; poll the driver or await it inside a
-  `Handle::spawn` task to run it on the worker. Web mode maps stream and close codes through the
+  `Handle::spawn` task to run it on the worker. moq-transport and the
+  SETUP-negotiated moq-lite versions box their futures with a native `Send`
+  bound, so wrap the session once more in `transport::Local`, which checks
+  the thread on every use, before `accept_request`/`connect`. Web mode maps stream and close codes through the
   HTTP/3 error space.
 - **qlog**: `quic::qlog::Sink` points a group of workers at a directory and
   `quic::Transport::qlog` turns capture on. The pinned worker never writes to