diff --git a/Cargo.lock b/Cargo.lock index d7b3860a232..5d95f7e8bba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5284,6 +5284,7 @@ dependencies = [ "rustc-hash 2.1.3", "rustls", "rustls-platform-verifier", + "send_wrapper", "serde_json", "slab", "tempfile", diff --git a/Cargo.toml b/Cargo.toml index e6c06ec0ef1..a5ceedc962b 100644 --- a/Cargo.toml +++ b/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" diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index f0115c6632b..74b37f2669a 100644 --- a/doc/bin/relay/config.md +++ b/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). diff --git a/quest/m2/README.md b/quest/m2/README.md index 2c1512d6214..23feeacdf17 100644 --- a/quest/m2/README.md +++ b/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 diff --git a/quest/m2/relay-io-uring-package.md b/quest/m2/relay-io-uring-package.md index 5dac9d05146..ed006de0604 100644 --- a/quest/m2/relay-io-uring-package.md +++ b/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 diff --git a/quest/m2/uring-connection-log.md b/quest/m2/uring-connection-log.md index b3dcbc5f501..ce5f7ded54c 100644 --- a/quest/m2/uring-connection-log.md +++ b/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 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 a03f38f28f7..393a87d43b5 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::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>, /// 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 = 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() @@ -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, + 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, + 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 diff --git a/rs/moq-relay/tests/integration/runtime_uring.rs b/rs/moq-relay/tests/integration/runtime_uring.rs index 14a648d194f..c9e0be41b52 100644 --- a/rs/moq-relay/tests/integration/runtime_uring.rs +++ b/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() { diff --git a/rs/moq-uring/Cargo.toml b/rs/moq-uring/Cargo.toml index ef486a6000a..871894d6ffd 100644 --- a/rs/moq-uring/Cargo.toml +++ b/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" diff --git a/rs/moq-uring/README.md b/rs/moq-uring/README.md index 0529f27982b..8344d70c04b 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::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 diff --git a/rs/moq-uring/src/lib.rs b/rs/moq-uring/src/lib.rs index 09f75226f62..28e3a23bcbf 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::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 //! 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 b77abb15754..e9e961bec77 100644 --- a/rs/moq-uring/src/transport.rs +++ b/rs/moq-uring/src/transport.rs @@ -1,8 +1,10 @@ //! Adapt backend poll sessions to moq-net without changing thread affinity. mod adapter; +mod local; pub use adapter::{Error, RecvStream, SendStream, Session}; #[cfg(feature = "quic")] use adapter::{Grouped, GroupedSend}; +pub use local::Local; #[cfg(feature = "quic")] use crate::quic; diff --git a/rs/moq-uring/src/transport/local.rs b/rs/moq-uring/src/transport/local.rs new file mode 100644 index 00000000000..9161d0100e1 --- /dev/null +++ b/rs/moq-uring/src/transport/local.rs @@ -0,0 +1,190 @@ +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 Local(SendWrapper); + +impl Local { + /// Bind `inner` to the current thread. + pub fn new(inner: T) -> Self { + Self(SendWrapper::new(inner)) + } +} + +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(Local::new) + } + + fn poll_accept_bi(&mut self, cx: &mut Context<'_>) -> Poll, Self::Error>> { + self.0 + .poll_accept_bi(cx) + .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(Local::new) + } + + fn poll_open_bi(&mut self, cx: &mut Context<'_>) -> Poll, Self::Error>> { + self.0 + .poll_open_bi(cx) + .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(Local::new) + } + + fn send_datagram( + &mut self, + payload: Bytes, + group: Option<&Local<::Group>>, + order: i64, + ) -> Result<(), Self::Error> { + self.0.send_datagram(payload, group.map(|group| &*group.0), order) + } + + 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 Local { + type Error = T::Error; + type Group = Local; + + 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 Local { + 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 Local { + 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 = Local::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); + } +}