From 3c56757e5a6c793ad80cd82a42e9f4eb0c3f942f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 18:29:16 -0700 Subject: [PATCH 1/5] quest(m2): claim uring-ietf From 4268d5c7fc4d64c5fe1c38cfd0afda3648aa6c20 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 18:42:26 -0700 Subject: [PATCH 2/5] feat(relay): serve moq-transport from the io_uring workers The ring advertised only the moq-lite ALPNs, refused SETUP-negotiated lite versions named in the config, and warned that moq-transport was not served. It now serves every configured version, like the shared runtime. moq-lite on its own ALPN still runs unboxed on the worker. moq-transport and the SETUP-negotiated versions box their futures with a native `Send` bound, so their session goes through the new `moq_uring::transport::Pinned`, which checks the thread on every use and panics anywhere else. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + doc/bin/relay/config.md | 4 +- quest/m2/README.md | 1 - quest/m2/relay-io-uring-package.md | 2 - quest/m2/uring-connection-log.md | 1 - quest/m2/uring-ietf.md | 37 ---- rs/moq-relay/src/runtime.rs | 2 +- rs/moq-relay/src/uring.rs | 126 +++++++----- .../tests/integration/runtime_uring.rs | 94 ++++++++- rs/moq-uring/Cargo.toml | 2 + rs/moq-uring/README.md | 5 +- rs/moq-uring/src/lib.rs | 3 +- rs/moq-uring/src/transport.rs | 2 + rs/moq-uring/src/transport/pinned.rs | 185 ++++++++++++++++++ 14 files changed, 373 insertions(+), 92 deletions(-) delete mode 100644 quest/m2/uring-ietf.md create mode 100644 rs/moq-uring/src/transport/pinned.rs diff --git a/Cargo.lock b/Cargo.lock index bee4d449786..a52c7a9f882 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5036,6 +5036,7 @@ dependencies = [ "rustc-hash 2.1.3", "rustls", "rustls-platform-verifier", + "send_wrapper", "serde_json", "slab", "tempfile", diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 5faf7df0a11..9f58763ffe3 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -132,8 +132,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). diff --git a/quest/m2/README.md b/quest/m2/README.md index fbd839f065b..83373b3775f 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -139,7 +139,6 @@ ranked just below deterministic TS legs, which it requires. - [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 S3 wire proof](/quest/m2/archive-s3.md) - the archive proof also runs through the S3 client against an in-process S3-compatible server - [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 - [Dropped uring session closes](/quest/m2/uring-drop-close.md) - a moq-uring session dropped without close() closes its connection - [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 diff --git a/quest/m2/relay-io-uring-package.md b/quest/m2/relay-io-uring-package.md index 21b7a153d7d..c6ec48de9fb 100644 --- a/quest/m2/relay-io-uring-package.md +++ b/quest/m2/relay-io-uring-package.md @@ -34,7 +34,5 @@ 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 - [Dropped session close](/quest/m2/uring-drop-close.md) - closes are delivered when a session is dropped on the ring - [One port on the io_uring workers](/quest/m2/uring-demux.md) - a one-port config is served on the ring too diff --git a/quest/m2/uring-connection-log.md b/quest/m2/uring-connection-log.md index 8fdf63601f1..597439a6d9e 100644 --- a/quest/m2/uring-connection-log.md +++ b/quest/m2/uring-connection-log.md @@ -52,6 +52,5 @@ 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 - [Session outcomes](/quest/m1/session-outcomes.md) - counts refusals and ends on both paths, from the same duplicated admission diff --git a/quest/m2/uring-ietf.md b/quest/m2/uring-ietf.md deleted file mode 100644 index da02102b919..00000000000 --- a/quest/m2/uring-ietf.md +++ /dev/null @@ -1,37 +0,0 @@ -# [M] The io_uring workers serve moq-transport sessions - -## Goal - -A relay started with `--runtime-io-uring` serves every version its -configuration lists, IETF moq-transport included, instead of filtering to -moq-lite and warning that the rest are not served. A fleet that moves to the -ring then drops no client protocol over raw QUIC. - -## Plan - -The io_uring listener's bind in `rs/moq-relay/src/uring.rs` advertises only -the lite ALPNs (`is_lite`), refuses a lite version named in the config that -negotiates in SETUP, and only warns when the config lists moq-transport -versions. -The tokio path already drives both wires through one session type, so the -gap is in the worker's accept path, not in moq-net. - -- Accept the moq-transport ALPNs on the worker and hand the connection to - the same session constructor the tokio workers use for them; the `!Send` - local task set is the constraint to satisfy, as it was for lite. -- Serve the SETUP-negotiated lite versions the same way, or record why they - stay refused; the operator-facing rule must be "the ring serves what the - config lists". -- `tests/integration/runtime_uring.rs` gains an IETF session and a SETUP-negotiated - session; both skip loudly below the kernel floor like the rest. -- `doc/bin/relay/config.md` and the `moq-uring` README drop the lite-only - caveat. - -Additive, so it lands on main. moq.pro's fleet deploy of the -ring requires the release carrying it. Decided 2026-10-08: moved to m2 with -that fleet deploy, which is m2 work. - -## Related - -- [Stream sessions](/quest/m3/uring-tcp/README.md) - the other protocol gap - on the ring, WebSocket and HTTP diff --git a/rs/moq-relay/src/runtime.rs b/rs/moq-relay/src/runtime.rs index 625e7d89aef..83f612b7df4 100644 --- a/rs/moq-relay/src/runtime.rs +++ b/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. diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index c0f9b57a60f..582aa7209f8 100644 --- a/rs/moq-relay/src/uring.rs +++ b/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::Pinned`], 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>, /// The negotiated-version restriction, when the operator set one. versions: moq_net::Versions, @@ -180,39 +182,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 = 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 = versions.alpns().into_iter().map(str::to_owned).collect(); let alpn = alpns .iter() @@ -647,11 +624,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() { @@ -686,13 +658,77 @@ async fn serve_connection( Some(_) => moq_net::transport::Binding::Uri, None => moq_net::transport::Binding::InBand, }; - let request = deadline - .bound(server.accept_request_lite( - std::time::Instant::now(), - moq_uring::transport::Session::new(transport, binding), - )) - .await - .context("moq handshake failed")?; + let session = moq_uring::transport::Session::new(transport, binding); + let now = std::time::Instant::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 `Pinned` 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, + 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::Pinned::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, + deadline: Deadline, + serve: Serve, + id: u64, + remote: SocketAddr, + sni: Option, + identity: Option, + alpn: Option, + url: Option, +} + +/// Authenticate on the shared runtime, then start the session on the worker. +async fn serve_request( + accepted: Accepted, + request: moq_net::server::Handshake, +) -> anyhow::Result<()> { + let Accepted { + handle, + 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 diff --git a/rs/moq-relay/tests/integration/runtime_uring.rs b/rs/moq-relay/tests/integration/runtime_uring.rs index 5d66675801d..d6fb33363ab 100644 --- a/rs/moq-relay/tests/integration/runtime_uring.rs +++ b/rs/moq-relay/tests/integration/runtime_uring.rs @@ -190,6 +190,98 @@ 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 +335,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() { diff --git a/rs/moq-uring/Cargo.toml b/rs/moq-uring/Cargo.toml index ef486a6000a..050464348ad 100644 --- a/rs/moq-uring/Cargo.toml +++ b/rs/moq-uring/Cargo.toml @@ -34,6 +34,8 @@ moq-quic = { workspace = true, features = ["bloom", "rustls-aws-lc-rs"], optiona moq-sock = { workspace = true } rustc-hash = { workspace = true } rustls = { version = "0.23", optional = true } +# Runtime thread checks for the session moq-transport's `Send` drivers capture. +send_wrapper = "0.6" # 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 } diff --git a/rs/moq-uring/README.md b/rs/moq-uring/README.md index 0529f27982b..88fe22b8e68 100644 --- a/rs/moq-uring/README.md +++ b/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::Pinned`, 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 diff --git a/rs/moq-uring/src/lib.rs b/rs/moq-uring/src/lib.rs index 09f75226f62..d116bd17df7 100644 --- a/rs/moq-uring/src/lib.rs +++ b/rs/moq-uring/src/lib.rs @@ -16,7 +16,8 @@ //! serves many connections on one socket (demuxed by connection id, dials //! included), each a [`quic::Connection`] wrapped by [`transport::Session`], //! so `moq_net::Client::connect_lite` and `Server::accept_lite` run real -//! moq-lite sessions on the worker. The socket is the identity: whatever is +//! moq-lite sessions on the worker, and [`transport::Pinned`] lets the +//! `Send`-bounded moq-transport paths run there too. The socket is the identity: whatever is //! built on it runs on the worker that adopted it, and a socket adopted as a //! member of a steered reuseport group ([`udp::Bound`]) issues connection ids //! that steer back to it. [`Handle::run`] supplies time and schedules driver diff --git a/rs/moq-uring/src/transport.rs b/rs/moq-uring/src/transport.rs index f4492304b33..17f39b6549f 100644 --- a/rs/moq-uring/src/transport.rs +++ b/rs/moq-uring/src/transport.rs @@ -1,6 +1,8 @@ //! Adapt backend poll sessions to moq-net without changing thread affinity. mod adapter; +mod pinned; pub use adapter::{Error, RecvStream, SendStream, Session}; +pub use pinned::Pinned; use adapter::{Grouped, GroupedSend}; use crate::quic; diff --git a/rs/moq-uring/src/transport/pinned.rs b/rs/moq-uring/src/transport/pinned.rs new file mode 100644 index 00000000000..a5e9b71d72d --- /dev/null +++ b/rs/moq-uring/src/transport/pinned.rs @@ -0,0 +1,185 @@ +use bytes::{Buf, BufMut, Bytes}; +use moq_net::transport::{self as moq, poll}; +use send_wrapper::SendWrapper; +use std::task::{Context, Poll}; + +/// A worker's transport, usable where moq-net wants a `Send` one. +/// +/// moq-transport boxes its futures with a native `Send` bound, which a worker's +/// `!Send` [`Session`](super::Session) cannot meet, so wrap the session in this +/// before `Server::accept_request` or `Client::connect`. The session, its +/// streams, and its send groups never leave the worker that polls them, so +/// this checks the thread on every use and panics anywhere else instead of +/// proving it at compile time. moq-lite needs none of it: hand the session to +/// `accept_request_lite` or `connect_lite` directly. +#[derive(Clone)] +pub struct Pinned(SendWrapper); + +impl Pinned { + /// Pin `inner` to the current thread. + pub fn new(inner: T) -> Self { + Self(SendWrapper::new(inner)) + } +} + +impl poll::Session for Pinned { + type SendStream = Pinned; + type RecvStream = Pinned; + type Error = S::Error; + + fn poll_accept_uni(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_accept_uni(cx).map_ok(Pinned::new) + } + + fn poll_accept_bi(&mut self, cx: &mut Context<'_>) -> Poll, Self::Error>> { + self.0 + .poll_accept_bi(cx) + .map_ok(|(send, recv)| (Pinned::new(send), Pinned::new(recv))) + } + + fn poll_open_uni(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_open_uni(cx).map_ok(Pinned::new) + } + + fn poll_open_bi(&mut self, cx: &mut Context<'_>) -> Poll, Self::Error>> { + self.0 + .poll_open_bi(cx) + .map_ok(|(send, recv)| (Pinned::new(send), Pinned::new(recv))) + } + + fn send_group(&mut self, priority: i64) -> Option::Group>> { + self.0.send_group(priority).map(Pinned::new) + } + + fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll> { + self.0.poll_send_datagram(cx, payload) + } + + fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_recv_datagram(cx) + } + + fn max_datagram_size(&self) -> usize { + self.0.max_datagram_size() + } + + fn protocol(&self) -> Option<&str> { + self.0.protocol() + } + + fn binding(&self) -> moq::Binding { + self.0.binding() + } + + fn close(&mut self, code: u32, reason: &str) { + self.0.close(code, reason) + } + + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll { + self.0.poll_closed(cx) + } + + fn stats(&self) -> impl moq::Stats { + self.0.stats() + } +} + +impl poll::SendStream for Pinned { + type Error = T::Error; + type Group = Pinned; + + fn poll_write(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + self.0.poll_write(cx, buf) + } + + fn poll_write_buf(&mut self, cx: &mut Context<'_>, buf: &mut B) -> Poll> { + self.0.poll_write_buf(cx, buf) + } + + fn set_priority(&mut self, priority: i64) { + self.0.set_priority(priority) + } + + fn set_group(&mut self, group: &Self::Group) { + self.0.set_group(&group.0) + } + + fn finish(&mut self) -> Result<(), Self::Error> { + self.0.finish() + } + + fn reset(&mut self, code: u32) { + self.0.reset(code) + } + + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_closed(cx) + } +} + +impl poll::RecvStream for Pinned { + type Error = T::Error; + + fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { + self.0.poll_read(cx, dst) + } + + fn poll_read_buf( + &mut self, + cx: &mut Context<'_>, + buf: &mut B, + ) -> Poll, Self::Error>> { + self.0.poll_read_buf(cx, buf) + } + + fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Self::Error>> { + self.0.poll_read_chunk(cx, max) + } + + fn stop(&mut self, code: u32) { + self.0.stop(code) + } + + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_closed(cx) + } +} + +impl poll::SendGroup for Pinned { + fn set_priority(&mut self, priority: i64) { + self.0.set_priority(priority) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use moq_net::transport::memory; + + /// The point of the wrapper: a worker's `!Send` session meets moq-transport's bounds. + #[cfg(feature = "quic")] + #[test] + fn meets_the_boxed_driver_bounds() { + fn boxable() {} + boxable::>>(); + } + + /// Touching it off its thread fails loud rather than racing the worker. + #[test] + fn panics_off_its_thread() { + let (session, _peer) = memory::pair(Default::default()); + let session = Pinned::new(session); + let other = session.clone(); + let (panicked, other) = std::thread::spawn(move || { + let used = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + poll::Session::max_datagram_size(&other); + })); + // Dropping it here would panic too; hand it back instead. + (used.is_err(), other) + }) + .join() + .expect("the thread returns its outcome"); + assert!(panicked, "an off-thread use must panic"); + drop(other); + } +} From 84409b1504177ef10f0b5d6fe1a776af9eb73c21 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 18:47:29 -0700 Subject: [PATCH 3/5] style: format Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/tests/integration/runtime_uring.rs | 7 ++++++- rs/moq-uring/Cargo.toml | 4 ++-- rs/moq-uring/src/transport.rs | 2 +- 3 files changed, 9 insertions(+), 4 deletions(-) diff --git a/rs/moq-relay/tests/integration/runtime_uring.rs b/rs/moq-relay/tests/integration/runtime_uring.rs index d6fb33363ab..c2e6eb192b6 100644 --- a/rs/moq-relay/tests/integration/runtime_uring.rs +++ b/rs/moq-relay/tests/integration/runtime_uring.rs @@ -235,7 +235,12 @@ async fn uring_workers_serve_every_configured_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 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); diff --git a/rs/moq-uring/Cargo.toml b/rs/moq-uring/Cargo.toml index 050464348ad..df2f6a7df00 100644 --- a/rs/moq-uring/Cargo.toml +++ b/rs/moq-uring/Cargo.toml @@ -34,11 +34,11 @@ moq-quic = { workspace = true, features = ["bloom", "rustls-aws-lc-rs"], optiona moq-sock = { workspace = true } rustc-hash = { workspace = true } rustls = { version = "0.23", optional = true } -# Runtime thread checks for the session moq-transport's `Send` drivers capture. -send_wrapper = "0.6" # 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 = "0.6" slab = "0.4" thiserror = "2" tracing = "0.1" diff --git a/rs/moq-uring/src/transport.rs b/rs/moq-uring/src/transport.rs index 17f39b6549f..22c54ead0b7 100644 --- a/rs/moq-uring/src/transport.rs +++ b/rs/moq-uring/src/transport.rs @@ -2,8 +2,8 @@ mod adapter; mod pinned; pub use adapter::{Error, RecvStream, SendStream, Session}; -pub use pinned::Pinned; use adapter::{Grouped, GroupedSend}; +pub use pinned::Pinned; use crate::quic; From ed8f3abd815f06a6e48d1bb865601644adb2ff4a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 19:12:46 -0700 Subject: [PATCH 4/5] refactor(uring): rename transport::Pinned to Local Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/src/uring.rs | 6 ++-- rs/moq-uring/README.md | 2 +- rs/moq-uring/src/lib.rs | 2 +- rs/moq-uring/src/transport.rs | 4 +-- .../src/transport/{pinned.rs => local.rs} | 36 +++++++++---------- 5 files changed, 25 insertions(+), 25 deletions(-) rename rs/moq-uring/src/transport/{pinned.rs => local.rs} (82%) diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 582aa7209f8..ba68213a16e 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -7,7 +7,7 @@ //! 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::Pinned`], since their drivers need `Send`. +//! [`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 @@ -663,7 +663,7 @@ async fn serve_connection( // 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 `Pinned` session meets. + // native `Send` bound, which only a `Local` session meets. let unboxed = alpn .as_deref() .and_then(moq_net::Version::from_alpn) @@ -691,7 +691,7 @@ async fn serve_connection( false => { let request = accepted .deadline - .bound(server.accept_request(now, moq_uring::transport::Pinned::new(session))) + .bound(server.accept_request(now, moq_uring::transport::Local::new(session))) .await .context("moq handshake failed")?; serve_request(accepted, request).await diff --git a/rs/moq-uring/README.md b/rs/moq-uring/README.md index 88fe22b8e68..8344d70c04b 100644 --- a/rs/moq-uring/README.md +++ b/rs/moq-uring/README.md @@ -35,7 +35,7 @@ task set, and the UDP sockets bound through it. the session and its driver; poll the driver or await it inside a `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::Pinned`, which checks + 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 diff --git a/rs/moq-uring/src/lib.rs b/rs/moq-uring/src/lib.rs index d116bd17df7..28e3a23bcbf 100644 --- a/rs/moq-uring/src/lib.rs +++ b/rs/moq-uring/src/lib.rs @@ -16,7 +16,7 @@ //! serves many connections on one socket (demuxed by connection id, dials //! included), each a [`quic::Connection`] wrapped by [`transport::Session`], //! so `moq_net::Client::connect_lite` and `Server::accept_lite` run real -//! moq-lite sessions on the worker, and [`transport::Pinned`] lets the +//! moq-lite sessions on the worker, and [`transport::Local`] lets the //! `Send`-bounded moq-transport paths run there too. The socket is the identity: whatever is //! built on it runs on the worker that adopted it, and a socket adopted as a //! member of a steered reuseport group ([`udp::Bound`]) issues connection ids diff --git a/rs/moq-uring/src/transport.rs b/rs/moq-uring/src/transport.rs index 22c54ead0b7..bd46683fd7f 100644 --- a/rs/moq-uring/src/transport.rs +++ b/rs/moq-uring/src/transport.rs @@ -1,9 +1,9 @@ //! Adapt backend poll sessions to moq-net without changing thread affinity. mod adapter; -mod pinned; +mod local; pub use adapter::{Error, RecvStream, SendStream, Session}; use adapter::{Grouped, GroupedSend}; -pub use pinned::Pinned; +pub use local::Local; use crate::quic; diff --git a/rs/moq-uring/src/transport/pinned.rs b/rs/moq-uring/src/transport/local.rs similarity index 82% rename from rs/moq-uring/src/transport/pinned.rs rename to rs/moq-uring/src/transport/local.rs index a5e9b71d72d..ef461373c0c 100644 --- a/rs/moq-uring/src/transport/pinned.rs +++ b/rs/moq-uring/src/transport/local.rs @@ -13,42 +13,42 @@ use std::task::{Context, Poll}; /// proving it at compile time. moq-lite needs none of it: hand the session to /// `accept_request_lite` or `connect_lite` directly. #[derive(Clone)] -pub struct Pinned(SendWrapper); +pub struct Local(SendWrapper); -impl Pinned { - /// Pin `inner` to the current thread. +impl Local { + /// Bind `inner` to the current thread. pub fn new(inner: T) -> Self { Self(SendWrapper::new(inner)) } } -impl poll::Session for Pinned { - type SendStream = Pinned; - type RecvStream = Pinned; +impl poll::Session for Local { + type SendStream = Local; + type RecvStream = Local; type Error = S::Error; fn poll_accept_uni(&mut self, cx: &mut Context<'_>) -> Poll> { - self.0.poll_accept_uni(cx).map_ok(Pinned::new) + self.0.poll_accept_uni(cx).map_ok(Local::new) } fn poll_accept_bi(&mut self, cx: &mut Context<'_>) -> Poll, Self::Error>> { self.0 .poll_accept_bi(cx) - .map_ok(|(send, recv)| (Pinned::new(send), Pinned::new(recv))) + .map_ok(|(send, recv)| (Local::new(send), Local::new(recv))) } fn poll_open_uni(&mut self, cx: &mut Context<'_>) -> Poll> { - self.0.poll_open_uni(cx).map_ok(Pinned::new) + self.0.poll_open_uni(cx).map_ok(Local::new) } fn poll_open_bi(&mut self, cx: &mut Context<'_>) -> Poll, Self::Error>> { self.0 .poll_open_bi(cx) - .map_ok(|(send, recv)| (Pinned::new(send), Pinned::new(recv))) + .map_ok(|(send, recv)| (Local::new(send), Local::new(recv))) } - fn send_group(&mut self, priority: i64) -> Option::Group>> { - self.0.send_group(priority).map(Pinned::new) + fn send_group(&mut self, priority: i64) -> Option::Group>> { + self.0.send_group(priority).map(Local::new) } fn poll_send_datagram(&mut self, cx: &mut Context<'_>, payload: &[u8]) -> Poll> { @@ -84,9 +84,9 @@ impl poll::Session for Pinned { } } -impl poll::SendStream for Pinned { +impl poll::SendStream for Local { type Error = T::Error; - type Group = Pinned; + type Group = Local; fn poll_write(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { self.0.poll_write(cx, buf) @@ -117,7 +117,7 @@ impl poll::SendStream for Pinned { } } -impl poll::RecvStream for Pinned { +impl poll::RecvStream for Local { type Error = T::Error; fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { @@ -145,7 +145,7 @@ impl poll::RecvStream for Pinned { } } -impl poll::SendGroup for Pinned { +impl poll::SendGroup for Local { fn set_priority(&mut self, priority: i64) { self.0.set_priority(priority) } @@ -161,14 +161,14 @@ mod tests { #[test] fn meets_the_boxed_driver_bounds() { fn boxable() {} - boxable::>>(); + boxable::>>(); } /// Touching it off its thread fails loud rather than racing the worker. #[test] fn panics_off_its_thread() { let (session, _peer) = memory::pair(Default::default()); - let session = Pinned::new(session); + let session = Local::new(session); let other = session.clone(); let (panicked, other) = std::thread::spawn(move || { let used = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { From b94040e2e090ab0bcbddd80da71db1478cef91cd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 19:47:15 -0700 Subject: [PATCH 5/5] build: declare send_wrapper as a workspace dependency Co-Authored-By: Claude Opus 5.5 --- Cargo.toml | 1 + rs/moq-uring/Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 5ede5997b9d..ab4da06788b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -219,6 +219,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" diff --git a/rs/moq-uring/Cargo.toml b/rs/moq-uring/Cargo.toml index df2f6a7df00..871894d6ffd 100644 --- a/rs/moq-uring/Cargo.toml +++ b/rs/moq-uring/Cargo.toml @@ -38,7 +38,7 @@ rustls = { version = "0.23", optional = true } # 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 = "0.6" +send_wrapper = { workspace = true } slab = "0.4" thiserror = "2" tracing = "0.1"