From 7fe39e5d0afac08aa1f089767ce4f65ced0ec270 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 20 Oct 2026 23:11:35 -0700 Subject: [PATCH 01/12] quest: claim quest/m1/net-erase-transport.md Co-Authored-By: Claude Opus 5.5 From f80321ecf72f33a22250b39e0273b18995a1ec79 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 20 Oct 2026 23:36:53 -0700 Subject: [PATCH 02/12] perf(net)!: box the transport so the session code compiles once Co-Authored-By: Claude Opus 5.5 --- rs/moq-audio/src/decode/consumer.rs | 10 +- rs/moq-mux/src/container/test_util.rs | 10 +- rs/moq-net/src/client.rs | 277 +---------- rs/moq-net/src/driver.rs | 49 +- rs/moq-net/src/ietf/session.rs | 73 +-- rs/moq-net/src/server.rs | 558 +++++++++------------- rs/moq-net/src/session.rs | 20 +- rs/moq-net/src/transport.rs | 26 +- rs/moq-net/src/transport/boxed.rs | 495 +++++++++++++++++++ rs/moq-net/src/transport/memory.rs | 2 +- rs/moq-relay/src/auth.rs | 4 +- rs/moq-relay/src/incoming.rs | 17 +- rs/moq-relay/src/uring.rs | 42 +- rs/moq-room/src/chat.rs | 10 +- rs/moq-tokio/src/server.rs | 145 ++---- rs/moq-uring/README.md | 11 +- rs/moq-uring/benches/session_lite.rs | 16 +- rs/moq-uring/src/lib.rs | 8 +- rs/moq-uring/src/quic/client.rs | 4 +- rs/moq-uring/src/quic/mod.rs | 6 +- rs/moq-uring/src/transport/local.rs | 7 +- rs/moq-uring/tests/integration/session.rs | 35 +- rs/moq-uring/tests/integration/web.rs | 13 +- 23 files changed, 943 insertions(+), 895 deletions(-) create mode 100644 rs/moq-net/src/transport/boxed.rs diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index a77ad17ebe..1998716ab4 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -789,18 +789,12 @@ mod tests { .with_subscriber(to.clone()); tokio::join!( async { - let (session, driver) = client - .connect_lite(moq_net::time::now(), client_transport) - .await - .unwrap(); + let (session, driver) = client.connect(moq_net::time::now(), client_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session }, async { - let (session, driver) = server - .accept_lite(moq_net::time::now(), server_transport) - .await - .unwrap(); + let (session, driver) = server.accept(moq_net::time::now(), server_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session } diff --git a/rs/moq-mux/src/container/test_util.rs b/rs/moq-mux/src/container/test_util.rs index b69610165d..6807fc8ff4 100644 --- a/rs/moq-mux/src/container/test_util.rs +++ b/rs/moq-mux/src/container/test_util.rs @@ -258,18 +258,12 @@ impl Link { // Each driver runs as soon as its handshake resolves: the accept waits on the client's SETUP. let sessions = tokio::join!( async { - let (session, driver) = client - .connect_lite(moq_net::time::now(), client_transport) - .await - .unwrap(); + let (session, driver) = client.connect(moq_net::time::now(), client_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session }, async { - let (session, driver) = server - .accept_lite(moq_net::time::now(), server_transport) - .await - .unwrap(); + let (session, driver) = server.accept(moq_net::time::now(), server_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session } diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index fa06045ed8..f132131bab 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -1,5 +1,7 @@ +use crate::driver::Transport; use crate::origin; use crate::time::{Deadlines, Instant}; +use crate::transport::poll::Session as _; use crate::{ ALPN_14, ALPN_15, ALPN_16, ALPN_17, ALPN_18, ALPN_19, ALPN_20, ALPN_21, ALPN_22, ALPN_LITE, ALPN_LITE_03, ALPN_LITE_04, ALPN_LITE_05, ALPN_LITE_06, ALPN_LITE_07_WIP, Consume, Error, NEGOTIATED, Session, Version, Versions, @@ -176,16 +178,13 @@ impl Client { /// Start a lite session on an already-negotiated version: build our SETUP, /// wire the origins, and return the session and its driver. - fn start_lite( + fn start_lite( &self, runtime: Deadlines, - session: S, + session: Transport, version: lite::Version, peer_hop: crate::Hop, - ) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Session, - { + ) -> Result<(Session, crate::Driver), Error> { let (publish, subscribe) = self.origins(peer_hop); // Advertise our capabilities (we report what the transport measures; we @@ -237,39 +236,19 @@ impl Client { )) } - /// Perform the MoQ handshake for moq-lite only, over any transport. - /// - /// Unlike [`connect`](Self::connect) this puts no thread-affinity bound on - /// the transport, so a pinned `!Send` transport works and yields a `!Send` - /// machine that stays on its thread. The trade is protocol scope: only a - /// moq-lite ALPN is accepted, since the moq-transport driver still needs a - /// [`Boxable`](crate::transport::poll::Boxable) transport. An ietf ALPN, an - /// unknown one, or the legacy no-ALPN SETUP negotiation is refused with - /// [`Error::Version`]. - pub async fn connect_lite(&self, now: Instant, session: S) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Session, - { - let runtime = Deadlines::new(now); - let version = match session.protocol() { - Some(ALPN_LITE_07_WIP) => lite::Version::Lite07, - Some(ALPN_LITE_06) => lite::Version::Lite06, - Some(ALPN_LITE_05) => lite::Version::Lite05, - Some(ALPN_LITE_04) => lite::Version::Lite04, - Some(ALPN_LITE_03) => lite::Version::Lite03, - _ => return Err(Error::Version), - }; - self.versions.select(Version::Lite(version)).ok_or(Error::Version)?; - self.start_lite(runtime, session, version, self.assigned_hop()) - } - /// Perform the MoQ handshake, returning the [`Session`] and its [`Driver`](crate::Driver). /// /// Poll the returned driver with nondecreasing time, starting at `now`. - pub async fn connect(&self, now: Instant, mut session: S) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Boxable, - { + pub fn connect( + &self, + now: Instant, + session: impl crate::transport::poll::Boxable, + ) -> impl Future> { + // Box before any protocol code runs, so it is compiled once rather than per transport. + self.connect_boxed(now, Transport::new(session)) + } + + async fn connect_boxed(&self, now: Instant, mut session: Transport) -> Result<(Session, crate::Driver), Error> { let runtime = Deadlines::new(now); let peer_hop = self.assigned_hop(); let (publish, subscribe) = self.origins(peer_hop); @@ -961,232 +940,6 @@ mod tests { assert!(matches!(futures::executor::block_on(session.closed()), Error::Cancel)); } - /// A transport made deliberately `!Send` by an `Rc` marker on the session and - /// both stream types: compiling at all is the point, proving the lite path - /// never demands thread mobility of any transport piece. - #[derive(Clone)] - struct LocalSession { - inner: FakeSession, - _local: std::rc::Rc<()>, - } - - struct LocalSend { - inner: FakeSendStream, - _local: std::rc::Rc<()>, - } - - struct LocalRecv { - inner: FakeRecvStream, - _local: std::rc::Rc<()>, - } - - impl crate::transport::poll::Session for LocalSession { - type SendStream = LocalSend; - type RecvStream = LocalRecv; - type Error = FakeError; - - fn poll_accept_uni(&mut self, cx: &mut Context<'_>) -> Poll> { - self.inner.poll_accept_uni(cx).map_ok(|stream| LocalRecv { - inner: stream, - _local: self._local.clone(), - }) - } - - fn poll_accept_bi( - &mut self, - cx: &mut Context<'_>, - ) -> Poll> { - self.inner.poll_accept_bi(cx).map_ok(|(send, recv)| { - ( - LocalSend { - inner: send, - _local: self._local.clone(), - }, - LocalRecv { - inner: recv, - _local: self._local.clone(), - }, - ) - }) - } - - fn poll_open_bi( - &mut self, - cx: &mut Context<'_>, - options: crate::transport::poll::SendOptions<'_, crate::transport::poll::Ungrouped>, - ) -> Poll> { - self.inner.poll_open_bi(cx, options).map_ok(|(send, recv)| { - ( - LocalSend { - inner: send, - _local: self._local.clone(), - }, - LocalRecv { - inner: recv, - _local: self._local.clone(), - }, - ) - }) - } - - fn poll_open_uni( - &mut self, - cx: &mut Context<'_>, - options: crate::transport::poll::SendOptions<'_, crate::transport::poll::Ungrouped>, - ) -> Poll> { - self.inner.poll_open_uni(cx, options).map_ok(|stream| LocalSend { - inner: stream, - _local: self._local.clone(), - }) - } - - fn send_group(&mut self, _: i64) -> Option { - None - } - - fn send_datagram( - &mut self, - payload: Bytes, - options: crate::transport::poll::SendOptions<'_, crate::transport::poll::Ungrouped>, - ) -> Result<(), Self::Error> { - self.inner.send_datagram(payload, options) - } - - fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { - self.inner.poll_recv_datagram(cx) - } - - fn max_datagram_size(&self) -> usize { - self.inner.max_datagram_size() - } - - fn binding(&self) -> crate::transport::Binding { - crate::transport::Binding::InBand - } - fn protocol(&self) -> Option<&str> { - self.inner.protocol() - } - - fn close(&mut self, code: u32, reason: &str) { - self.inner.close(code, reason); - } - - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll { - self.inner.poll_closed(cx) - } - - fn stats(&self) -> impl crate::transport::Stats { - self.inner.stats() - } - } - - impl crate::transport::poll::SendStream for LocalSend { - type Error = FakeError; - type Group = crate::transport::poll::Ungrouped; - - fn poll_write(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { - self.inner.poll_write(cx, buf) - } - - fn set_priority(&mut self, order: i64) { - self.inner.set_priority(order); - } - - fn set_group(&mut self, group: &Self::Group) { - match *group {} - } - - fn finish(&mut self) -> Result<(), Self::Error> { - self.inner.finish() - } - - fn reset(&mut self, code: u32) { - self.inner.reset(code); - } - - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { - crate::transport::poll::SendStream::poll_closed(&mut self.inner, cx) - } - } - - impl crate::transport::poll::RecvStream for LocalRecv { - type Error = FakeError; - - fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Self::Error>> { - self.inner.poll_read(cx, dst) - } - - fn stop(&mut self, code: u32) { - self.inner.stop(code); - } - - fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { - crate::transport::poll::RecvStream::poll_closed(&mut self.inner, cx) - } - } - - // The point of the lite-only entry: a `!Send` transport yields a `!Send` - // driver polled by its caller, while the severed Session handle stays - // Send + Sync. Compiling is most of the assertion; the rest checks the - // machine still relays the close. - #[test] - fn connect_lite_over_a_send_less_transport() { - let fake = FakeSession::new(Some(ALPN_LITE_04), Vec::new()); - let local = LocalSession { - inner: fake.clone(), - _local: std::rc::Rc::new(()), - }; - let client = Client::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - - let runtime = crate::time::Deadlines::new(crate::time::Instant::now()); - let (session, mut driver) = futures::executor::block_on(client.connect_lite(runtime.now(), local)).unwrap(); - assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_ok()); - - fn assert_send_sync(_: &T) {} - assert_send_sync(&session); - - session.abort(Error::Cancel); - let _ = driver.poll(runtime.now(), &kio::Waiter::noop()); - assert_eq!(fake.state.close_events.read()[0].0, SessionError::Cancel.to_code()); - } - - // The server-side twin: a `!Send` transport accepts a lite session whose - // driver the caller polls directly. - #[test] - fn accept_lite_over_a_send_less_transport() { - let fake = FakeSession::new(Some(ALPN_LITE_04), Vec::new()); - let local = LocalSession { - inner: fake.clone(), - _local: std::rc::Rc::new(()), - }; - let server = crate::Server::new().with_versions(Version::Lite(lite::Version::Lite04).into()); - - let runtime = crate::time::Deadlines::new(crate::time::Instant::now()); - let (session, mut driver) = futures::executor::block_on(server.accept_lite(runtime.now(), local)).unwrap(); - assert_eq!(session.version(), Version::Lite(lite::Version::Lite04)); - assert!(driver.poll(runtime.now(), &kio::Waiter::noop()).is_ok()); - - drop(session); - assert!(fake.state.close_events.read().is_empty()); - let _ = driver.poll(runtime.now(), &kio::Waiter::noop()); - assert_eq!(fake.state.close_events.read()[0].0, SessionError::Cancel.to_code()); - } - - // The lite-only entry refuses everything that still needs the boxed ietf - // driver, instead of silently negotiating it. - #[test] - fn connect_lite_refuses_ietf_alpns() { - let fake = FakeSession::new(Some(ALPN_19), Vec::new()); - let local = LocalSession { - inner: fake, - _local: std::rc::Rc::new(()), - }; - let client = Client::new(); - let runtime = crate::time::Deadlines::new(crate::time::Instant::now()); - let result = futures::executor::block_on(client.connect_lite(runtime.now(), local)); - assert!(matches!(result, Err(Error::Version))); - } - // `stats()` reads the machine's latest sample and primes the sampler, so a // periodic poller observes fresh counters without consuming the bandwidth // channel. diff --git a/rs/moq-net/src/driver.rs b/rs/moq-net/src/driver.rs index d560f799e8..f32f4c3ae9 100644 --- a/rs/moq-net/src/driver.rs +++ b/rs/moq-net/src/driver.rs @@ -15,31 +15,27 @@ use crate::time::{Deadlines, Instant}; /// /// It holds no session handle: dropping the last [`crate::Session`] requests /// closure on the next poll. Dropping the driver cancels the session, and -/// [`crate::Session::closed`] resolves with [`Error::Cancel`]. Its `Send`-ness -/// follows its transport. +/// [`crate::Session::closed`] resolves with [`Error::Cancel`]. It is `Send` on +/// native, like the [`Boxable`](crate::transport::poll::Boxable) transport it drives. #[must_use = "the session makes no progress unless its driver is polled"] -pub struct Driver { - state: State, +pub struct Driver { + state: State, clock: Deadlines, } /// The protocol half of a machine, one variant per negotiated wire protocol. -/// -/// The lite driver is a named machine, so the machine's `Send`-ness follows -/// the transport (a pinned `!Send` transport yields a `!Send` machine that -/// stays on its thread). The ietf driver is still a boxed future; the box -/// demands `Send` on native, which is why the ietf path requires a -/// [`Boxable`](crate::transport::poll::Boxable) transport until it too becomes -/// a named machine. -pub(crate) enum Protocol { - /// Boxed for size only: a concrete box, so `Send` stays inferred. - Lite(Box>), +pub(crate) enum Protocol { + /// Boxed for size only. + Lite(Box>), Ietf(crate::ietf::Driver), } +/// The transport every session runs on, boxed once by `Client` or `Server`. +pub(crate) type Transport = crate::transport::boxed::Session; + /// Protocol and lifecycle work owned by the driver. -pub(crate) struct State { - pub(crate) protocol: Protocol, +pub(crate) struct State { + pub(crate) protocol: Protocol, // The session supervisor, polled alongside the protocol: it executes the // handles' close requests, publishes the transport's terminal error, and // samples stats. It finishes once the transport reports closed, and the @@ -47,13 +43,13 @@ pub(crate) struct State { // is what `Session::closed` observes, so resolving before it is published // would leave waiters parked on a machine nobody polls again. `None` once // finished, since a completed machine must not be polled again. - pub(crate) supervisor: Option>, + pub(crate) supervisor: Option, // Cached so a poll after completion doesn't re-poll a finished protocol. pub(crate) result: Option>, } -impl Driver { - pub(crate) fn new(clock: Deadlines, state: State) -> Self { +impl Driver { + pub(crate) fn new(clock: Deadlines, state: State) -> Self { Self { state, clock } } @@ -72,7 +68,7 @@ impl Driver { } } -impl Protocol { +impl Protocol { pub(crate) fn local_close(&self) -> std::sync::Arc { match self { Self::Lite(driver) => driver.local_close.clone(), @@ -104,7 +100,7 @@ impl Protocol { } } -impl State { +impl State { fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { if let Some(supervisor) = &mut self.supervisor && supervisor.poll(waiter).is_ready() @@ -145,16 +141,23 @@ impl State { } } -impl crate::time::Driver for Driver { +impl crate::time::Driver for Driver { fn poll(&mut self, now: Instant, waiter: &kio::Waiter) -> Result, Error> { self.poll(now, waiter) } } -impl std::fmt::Debug for Driver { +impl std::fmt::Debug for Driver { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Driver") .field("done", &self.state.result.is_some()) .finish() } } + +// The relay and the bindings spawn drivers on multi-threaded runtimes. +#[cfg(not(target_family = "wasm"))] +const _: () = { + const fn assert_send() {} + assert_send::(); +}; diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 8c7b01bd03..9f05733778 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -1,4 +1,5 @@ use crate::origin; +use crate::transport::poll::Session as _; use crate::{ Error, Hop, SessionError, StreamError, coding::{Decode, DecodeError, Encode, Reader, Stream, Writer}, @@ -109,10 +110,7 @@ impl std::future::Future for Driver { } } -pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle), Error> -where - S: crate::transport::poll::Boxable, -{ +pub fn start(config: Config) -> Result<(Driver, crate::goaway::Handle), Error> { let Config { runtime, mut session, @@ -225,11 +223,14 @@ where }; let control = Control::new(request_id_max, client).with_window(limits.requests(), client, version); let adapter = ControlStreamAdapter::new(session.clone(), control.clone(), version); + // Boxed like a draft-17+ transport, so the publisher and subscriber below + // are the same instantiation as theirs. + let virtual_session = crate::driver::Transport::new(adapter.clone()); // No AUTH on these drafts, but a limit still reaches both halves. let mut publisher = Publisher::new( runtime.clone(), - adapter.clone(), + virtual_session.clone(), publish, control.clone(), peer_hop, @@ -245,7 +246,7 @@ where let mut subscriber = Subscriber::new( runtime.clone(), - adapter.clone(), + virtual_session.clone(), subscribe, control, peer_hop, @@ -287,7 +288,7 @@ where } drop(tasks); - let dispatch_session = adapter.clone(); + let dispatch_session = virtual_session.clone(); let sub_ns = subscriber.clone(); let sub_ns_adapter = adapter.clone(); @@ -295,7 +296,7 @@ where // completion); the task set draining is the one clean exit. let mut adapter_run = std::pin::pin!(err_only(adapter.run(setup.reader, setup.writer, goaway.clone()))); let mut unis = std::pin::pin!(err_only(run_unis( - adapter.clone(), + virtual_session.clone(), subscriber.clone(), UniSetup { peer: None, @@ -315,7 +316,8 @@ where None, version ))); - let mut datagrams = std::pin::pin!(err_only(run_datagrams(adapter.clone(), subscriber.clone()))); + let mut datagrams = + std::pin::pin!(err_only(run_datagrams(virtual_session.clone(), subscriber.clone()))); // Unsolicited PUBLISH_NAMESPACE unless the peer requires solicitation; // see `Publisher::run_publish_namespaces`. let mut pub_ns_run = std::pin::pin!(err_only(publisher.clone().run_publish_namespaces())); @@ -330,11 +332,17 @@ where let mut sub_ns_adapter = sub_ns_adapter; let (send, recv) = sub_ns_adapter.open_native_bi().await?; Stream { - writer: crate::coding::Writer::new(send, version), - reader: crate::coding::Reader::new(recv, version), + writer: crate::coding::Writer::new( + crate::transport::boxed::SendStream::new(send), + version, + ), + reader: crate::coding::Reader::new( + crate::transport::boxed::RecvStream::new(recv), + version, + ), } } - _ => Stream::open(&mut sub_ns_adapter.clone(), version).await?, + _ => Stream::open(&mut crate::driver::Transport::new(sub_ns_adapter), version).await?, }; if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix, snapshot).await { // The peer breaking the protocol is fatal, and the driver @@ -1566,7 +1574,7 @@ mod tests { let log = session.log.clone(); let (driver, _) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -1621,7 +1629,7 @@ mod tests { let log = session.log.clone(); let (driver, _) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -1667,7 +1675,7 @@ mod tests { let log = session.log.clone(); let (driver, _) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -1761,7 +1769,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -1824,7 +1832,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -1889,8 +1897,9 @@ mod tests { }; let gate = kio::Producer::new(true); - let mut session = crate::lite::test_transport::SinkSession::gated_bi(gate.consume()); + let session = crate::lite::test_transport::SinkSession::gated_bi(gate.consume()); let log = session.log.clone(); + let mut session = crate::driver::Transport::new(session); let setup = Stream::open(&mut session, version) .await .expect("open the control stream"); @@ -1985,7 +1994,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2088,7 +2097,7 @@ mod tests { let handle = crate::auth::Handle::ietf(true); let (driver, goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2172,7 +2181,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2268,7 +2277,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2357,7 +2366,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2400,7 +2409,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2667,7 +2676,7 @@ mod tests { fn bidi_driver(session: crate::lite::test_transport::ScriptedSession, publish: Option) -> Driver { start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2810,7 +2819,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -2878,7 +2887,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, client: false, @@ -2971,9 +2980,10 @@ mod tests { let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); let consumer = origin.consume(); - let mut session = + let session = crate::lite::test_transport::ScriptedSession::new(publish_namespace_then_two_withdrawals(VERSION).await); let log = session.log.clone(); + let mut session = crate::driver::Transport::new(session); let setup = Stream::open(&mut session, VERSION) .await .expect("open the control stream"); @@ -3088,13 +3098,14 @@ mod tests { ), ] { let payload = subscribe_with_token(version, token); - let mut session = match version { + let session = match version { Version::Draft14 => crate::lite::test_transport::ScriptedSession::new(payload), _ => { crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_bidis(vec![payload]) } }; let log = session.log.clone(); + let mut session = crate::driver::Transport::new(session); let setup = match version { Version::Draft14 => Some(Stream::open(&mut session, version).await.unwrap()), _ => None, @@ -3333,7 +3344,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), @@ -3435,7 +3446,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), - session, + session: crate::driver::Transport::new(session), setup: None, request_id_max: None, limits: Default::default(), diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 27f5c9fc70..5c97ededf1 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -1,10 +1,10 @@ //! Accepting a MoQ session, including the paused handshake that inspects the //! peer's SETUP before granting origins. -use crate::transport::{MaybeSend, MaybeSync}; - +use crate::driver::Transport; use crate::origin; use crate::time::{Deadlines, Instant}; +use crate::transport::poll::Session as _; use crate::{ ALPN_14, ALPN_15, ALPN_16, ALPN_17, ALPN_18, ALPN_19, ALPN_20, ALPN_21, ALPN_22, ALPN_LITE, ALPN_LITE_03, ALPN_LITE_04, ALPN_LITE_05, ALPN_LITE_06, ALPN_LITE_07_WIP, Consume, Error, NEGOTIATED, Role, Session, @@ -143,18 +143,15 @@ impl Server { /// Start a lite session on an accepted transport: wire the origins, answer /// with our SETUP, and return the session and its driver. - fn start_lite( + fn start_lite( &self, runtime: Deadlines, - session: S, + session: Transport, version: lite::Version, - client_setup: Option>, + client_setup: Option>, peer_hop: Option, auth: crate::auth::Handle, - ) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Session, - { + ) -> Result<(Session, crate::Driver), Error> { let (publish, subscribe) = self.stat_tagged_origins(); // We report what the transport actually measures; a server never @@ -204,41 +201,7 @@ impl Server { )) } - /// Perform the MoQ handshake for moq-lite only, over any transport. - /// - /// Same trade as [`Client::connect_lite`](crate::Client::connect_lite): no - /// thread-affinity bound on the transport, so a pinned `!Send` transport - /// works, and only a moq-lite ALPN is accepted (anything else is refused - /// with [`Error::Version`]). Completes the handshake immediately; a caller - /// gating on the advertised path uses - /// [`accept_request_lite`](Self::accept_request_lite) instead. - pub async fn accept_lite(&self, now: Instant, session: S) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Session, - { - self.accept_request_lite(now, session).await?.ok().await - } - - /// Begin the moq-lite handshake, pausing like - /// [`accept_request`](Self::accept_request) but for moq-lite ALPNs only, - /// which is what drops the thread-affinity bounds: a pinned `!Send` - /// transport can gate on the advertised path too. Anything but a moq-lite - /// ALPN is refused with [`Error::Version`]. - pub async fn accept_request_lite(&self, now: Instant, session: S) -> Result, Error> - where - S: crate::transport::poll::Session, - { - let mut refused = session.clone(); - self.handshake_lite(now, session) - .await - .inspect_err(|err| close(&mut refused, err, None)) - } - - async fn handshake_lite(&self, now: Instant, mut session: S) -> Result, Error> - where - S: crate::transport::poll::Session, - { - let runtime = Deadlines::new(now); + async fn handshake_lite(&self, runtime: Deadlines, mut session: Transport) -> Result { let (path, role, origin, handshake) = match session.protocol() { Some(alpn @ (ALPN_LITE_05 | ALPN_LITE_06 | ALPN_LITE_07_WIP)) => { let version = match alpn { @@ -298,7 +261,7 @@ impl Server { PausedHandshake::LiteBare { version, .. } | PausedHandshake::LiteSetup { version, .. } => { version.has_auth() } - PausedHandshake::Boxed(_) => false, + PausedHandshake::IetfModern(_) | PausedHandshake::Legacy(_) => false, }); Ok(Handshake { path, @@ -323,12 +286,11 @@ impl Server { /// Convenience wrapper over [`accept_request`](Self::accept_request) that /// completes the handshake immediately. Use `accept_request` when you need to /// inspect the client's advertised path before deciding what to serve. - pub async fn accept(&self, now: Instant, session: S) -> Result<(Session, crate::Driver), Error> - where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, - { + pub async fn accept( + &self, + now: Instant, + session: impl crate::transport::poll::Boxable, + ) -> Result<(Session, crate::Driver), Error> { self.accept_request(now, session).await?.ok().await } @@ -345,25 +307,23 @@ impl Server { /// /// A SETUP that fails to parse or negotiate closes the session with the matching code, /// so the peer learns why instead of seeing a bare disconnect. - pub async fn accept_request(&self, now: Instant, session: S) -> Result, Error> - where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, - { + pub fn accept_request( + &self, + now: Instant, + session: impl crate::transport::poll::Boxable, + ) -> impl Future> { + // Box before any protocol code runs, so it is compiled once rather than per transport. + self.accept_boxed(Deadlines::new(now), Transport::new(session)) + } + + async fn accept_boxed(&self, runtime: Deadlines, session: Transport) -> Result { let mut refused = session.clone(); - self.handshake(now, session) + self.handshake(runtime, session) .await .inspect_err(|err| close(&mut refused, err, None)) } - async fn handshake(&self, now: Instant, mut session: S) -> Result, Error> - where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, - { - let runtime = Deadlines::new(now); + async fn handshake(&self, runtime: Deadlines, mut session: Transport) -> Result { let (encoding, supported) = match session.protocol() { Some(alpn @ (ALPN_22 | ALPN_21 | ALPN_20 | ALPN_19 | ALPN_18 | ALPN_17)) => { let draft = match alpn { @@ -399,10 +359,8 @@ impl Server { .ok_or(Error::Version)?; (v, v.into()) } - // Every lite ALPN goes through the same entry point, which is also - // what a `!Send` transport calls directly. Some(ALPN_LITE_07_WIP | ALPN_LITE_06 | ALPN_LITE_05 | ALPN_LITE_04 | ALPN_LITE_03) => { - return self.handshake_lite(now, session).await; + return self.handshake_lite(runtime, session).await; } Some(ALPN_LITE) | None => { let supported = self.versions.filter(&NEGOTIATED.into()).ok_or(Error::Version)?; @@ -468,30 +426,25 @@ impl Server { inner: Some(RequestInner { server: self.clone(), runtime, - handshake: PausedHandshake::Boxed(Box::new(PausedLegacy { + handshake: PausedHandshake::Legacy(PausedLegacy { session, stream, version, request_id_max, peer_declared, - })), + }), }), }) } /// Read a draft-17/18 client's SETUP (with its request path) off its uni stream, /// then pause. `ok()` starts the session and hands the stream back for GOAWAY. - async fn accept_ietf_modern( + async fn accept_ietf_modern( &self, runtime: Deadlines, - mut session: S, + mut session: Transport, version: ietf::Version, - ) -> Result, Error> - where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, - { + ) -> Result { let peer_setup = ietf::accept_setup(&mut session, version).await?; Ok(Handshake { path: peer_setup.path.clone(), @@ -507,11 +460,11 @@ impl Server { inner: Some(RequestInner { server: self.clone(), runtime, - handshake: PausedHandshake::Boxed(Box::new(PausedIetfModern { + handshake: PausedHandshake::IetfModern(PausedIetfModern { session, version, peer_setup, - })), + }), }), }) } @@ -524,7 +477,7 @@ impl Server { /// the origins to serve, then call [`ok`](Self::ok) to complete the handshake, or /// [`close`](Self::close) to reject it. Modeled on the WebTransport `Request` in /// moq-tokio. -pub struct Handshake { +pub struct Handshake { path: Option, role: Option, origin: Option, @@ -537,279 +490,228 @@ pub struct Handshake { /// take the peer's token requests before any arrive. auth: crate::auth::Handle, // Taken by `ok`/`close`; `Drop` rejects the handshake if neither ran. - inner: Option>, + inner: Option, } /// The parts of a [`Handshake`] consumed by [`Handshake::ok`] / [`Handshake::close`]. -struct RequestInner { +struct RequestInner { server: Server, /// Supplies the clock and timers for the accepted session. runtime: Deadlines, - handshake: PausedHandshake, + handshake: PausedHandshake, } /// The handshake state captured at the pause point. Every variant defers its /// session start to [`Handshake::ok`] so origins set on the handshake still apply. -enum PausedHandshake { +enum PausedHandshake { /// moq-lite 03/04: no Setup Stream. - LiteBare { session: S, version: lite::Version }, + LiteBare { + session: Transport, + version: lite::Version, + }, /// moq-lite 05+: the client's Setup Stream has been read. `ok()` starts the /// session, seeding the SETUP back so PROBE gating resolves. LiteSetup { - session: S, + session: Transport, version: lite::Version, - client_setup: lite::AcceptedSetup, + client_setup: lite::AcceptedSetup, }, - /// An IETF (or legacy bidi-SETUP) handshake, boxed where its - /// thread-affinity bounds held. The boxing is what keeps [`Handshake`] and - /// its lite path free of those bounds: the ietf machinery erases its - /// futures, which forces a per-target `Send` choice a pinned `!Send` - /// transport cannot satisfy, so the choice is made here, at construction, - /// where the caller proved the bounds. - Boxed(Box>), + IetfModern(PausedIetfModern), + Legacy(PausedLegacy), } -type Accept = crate::util::MaybeSendBox<'static, Result<(Session, crate::Driver), Error>>; - -/// A paused non-lite handshake. See [`PausedHandshake::Boxed`] for why this is a -/// trait object. -/// -/// `MaybeSync` is not decoration: a caller holding a [`Handshake`] across an -/// await behind `&self` (moq-relay authenticates that way) needs -/// `&Handshake: Send`, which is `Handshake: Sync`, which is this. -trait Paused: MaybeSend + MaybeSync { - /// Complete the handshake with the final server config. - fn ok( - self: Box, - server: Server, - runtime: Deadlines, - peer_hop: Option, - auth: crate::auth::Handle, - ) -> Accept; - - /// Reject the handshake, closing the transport with `err`'s wire code. - fn close(self: Box, err: Error); -} - -/// Modern IETF (17/18): the client's SETUP (with its request path) has been +/// Modern IETF (17+): the client's SETUP (with its request path) has been /// read off its uni stream; `ok` starts the session, handing that stream back /// for GOAWAY monitoring. -struct PausedIetfModern { - session: S, +struct PausedIetfModern { + session: Transport, version: ietf::Version, - peer_setup: ietf::PeerSetup, + peer_setup: ietf::PeerSetup, } -impl Paused for PausedIetfModern -where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, -{ +impl PausedIetfModern { + /// Complete the handshake with the final server config. fn ok( - self: Box, + self, server: Server, runtime: Deadlines, peer_hop: Option, auth: crate::auth::Handle, - ) -> Accept { - use crate::util::MaybeBoxedExt as _; - async move { - let Self { - session, - version, - peer_setup, - } = *self; - let (publish, subscribe) = server.stat_tagged_origins(); - - // The client's SETUP was read at the pause; hand the stream back - // for GOAWAY. A server never advertises a path, hence `None`. - let (protocol, goaway) = ietf::start(ietf::Config { - runtime: runtime.clone(), - limits: server.limits, - session: session.clone(), - setup: None, - request_id_max: None, - client: false, - publish, - subscribe, - peer_hop, - // Only the dialing side prices a link. - cost: None, - version, - path: None, - authority: None, - peer_setup_stream: Some(peer_setup.stream), - peer_declared: Some(peer_setup.declared), - auth: auth.clone(), - early_unis: peer_setup.early, - extensions: server.extensions, - })?; - tracing::debug!(?version, "connected"); - Ok(Session::new( - runtime, - session, - version.into(), - None, - crate::driver::Protocol::Ietf(protocol), - crate::session::Handles { - stats: server.stats.clone(), - goaway, - auth, - estimate: None, - }, - )) - } - .maybe_boxed() - } + ) -> Result<(Session, crate::Driver), Error> { + let Self { + session, + version, + peer_setup, + } = self; + let (publish, subscribe) = server.stat_tagged_origins(); - fn close(mut self: Box, err: Error) { - close(&mut self.session, &err, Some(self.version.into())); + // The client's SETUP was read at the pause; hand the stream back + // for GOAWAY. A server never advertises a path, hence `None`. + let (protocol, goaway) = ietf::start(ietf::Config { + runtime: runtime.clone(), + limits: server.limits, + session: session.clone(), + setup: None, + request_id_max: None, + client: false, + publish, + subscribe, + peer_hop, + // Only the dialing side prices a link. + cost: None, + version, + path: None, + authority: None, + peer_setup_stream: Some(peer_setup.stream), + peer_declared: Some(peer_setup.declared), + auth: auth.clone(), + early_unis: peer_setup.early, + extensions: server.extensions, + })?; + tracing::debug!(?version, "connected"); + Ok(Session::new( + runtime, + session, + version.into(), + None, + crate::driver::Protocol::Ietf(protocol), + crate::session::Handles { + stats: server.stats.clone(), + goaway, + auth, + estimate: None, + }, + )) } } /// Legacy IETF (draft 14-16) and lite 01/02: the client SETUP has been read /// off the bidi stream (including its request path) but the server SETUP /// hasn't been sent; `ok` finishes it. -struct PausedLegacy { - session: S, - stream: Stream, +struct PausedLegacy { + session: Transport, + stream: Stream, version: Version, request_id_max: Option, /// What the client's SETUP declared, for the options `ok` acts on. peer_declared: ietf::peer::Peer, } -impl Paused for PausedLegacy -where - S: crate::transport::poll::Boxable, - S::SendStream: MaybeSync, - S::RecvStream: MaybeSync, -{ - fn ok( - self: Box, +impl PausedLegacy { + /// Send the server SETUP and start the session. + async fn ok( + self, server: Server, runtime: Deadlines, peer_hop: Option, auth: crate::auth::Handle, - ) -> Accept { - use crate::util::MaybeBoxedExt as _; - async move { - let Self { - session, - mut stream, - version, - request_id_max, - peer_declared, - } = *self; - let (publish, subscribe) = server.stat_tagged_origins(); - - // Encode parameters using the version-appropriate type. - let parameters = match version { - Version::Ietf(v) => { - let mut parameters = ietf::Parameters::default(); - // The client's requests, admitted up to our limits and granted back as they close. - parameters.set_varint( - ietf::ParameterVarInt::MaxRequestId, - ietf::initial_max_request_id(server.limits.requests(), true), - ); - parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec()); - if server.extensions.solicit { - ietf::solicit::into_setup(&mut parameters, v); - } - ietf::hidden::into_setup(&mut parameters, v); - ietf::max_age::into_setup(&mut parameters, v); - ietf::active_count::into_setup(&mut parameters, v); - parameters.encode_bytes(v)? - } - Version::Lite(v) => lite::Parameters::default().encode_bytes(v)?, - }; - - let server_setup = setup::Server { - version: version.into(), - parameters, - }; - stream.writer.encode(&server_setup).await?; - - let (recv_bw, protocol, goaway, auth, estimate) = match version { - Version::Lite(v) => { - let stream = stream.with_version(v); - // Pre-lite-05: no Setup Stream, so nothing to advertise or seed. - let start = lite::start(lite::Config { - runtime: runtime.clone(), - client: false, - limits: server.limits, - session: session.clone(), - setup_stream: Some(stream), - publish, - subscribe, - peer_hop, - version: v, - our_setup: lite::Setup::default(), - peer_setup: None, - auth, - })?; - ( - start.recv_bandwidth, - crate::driver::Protocol::Lite(Box::new(start.driver)), - start.goaway, - start.auth, - Some(start.estimate), - ) - } - Version::Ietf(v) => { - let stream = stream.with_version(v); - // Draft 14-16: path came in the bidi SETUP, no uni SETUP to hand back. - let (protocol, goaway) = ietf::start(ietf::Config { - runtime: runtime.clone(), - limits: server.limits, - session: session.clone(), - setup: Some(stream), - request_id_max, - client: false, - publish, - subscribe, - peer_hop, - cost: None, - version: v, - path: None, - authority: None, - peer_setup_stream: None, - peer_declared: Some(peer_declared), - auth: auth.clone(), - early_unis: Vec::new(), - extensions: server.extensions, - })?; - (None, crate::driver::Protocol::Ietf(protocol), goaway, auth, None) + ) -> Result<(Session, crate::Driver), Error> { + let Self { + session, + mut stream, + version, + request_id_max, + peer_declared, + } = self; + let (publish, subscribe) = server.stat_tagged_origins(); + + // Encode parameters using the version-appropriate type. + let parameters = match version { + Version::Ietf(v) => { + let mut parameters = ietf::Parameters::default(); + // The client's requests, admitted up to our limits and granted back as they close. + parameters.set_varint( + ietf::ParameterVarInt::MaxRequestId, + ietf::initial_max_request_id(server.limits.requests(), true), + ); + parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec()); + if server.extensions.solicit { + ietf::solicit::into_setup(&mut parameters, v); } - }; + ietf::hidden::into_setup(&mut parameters, v); + ietf::max_age::into_setup(&mut parameters, v); + ietf::active_count::into_setup(&mut parameters, v); + parameters.encode_bytes(v)? + } + Version::Lite(v) => lite::Parameters::default().encode_bytes(v)?, + }; - Ok(Session::new( - runtime, - session, - version, - recv_bw, - protocol, - crate::session::Handles { - stats: server.stats.clone(), - goaway, + let server_setup = setup::Server { + version: version.into(), + parameters, + }; + stream.writer.encode(&server_setup).await?; + + let (recv_bw, protocol, goaway, auth, estimate) = match version { + Version::Lite(v) => { + let stream = stream.with_version(v); + // Pre-lite-05: no Setup Stream, so nothing to advertise or seed. + let start = lite::start(lite::Config { + runtime: runtime.clone(), + client: false, + limits: server.limits, + session: session.clone(), + setup_stream: Some(stream), + publish, + subscribe, + peer_hop, + version: v, + our_setup: lite::Setup::default(), + peer_setup: None, auth, - estimate, - }, - )) - } - .maybe_boxed() - } + })?; + ( + start.recv_bandwidth, + crate::driver::Protocol::Lite(Box::new(start.driver)), + start.goaway, + start.auth, + Some(start.estimate), + ) + } + Version::Ietf(v) => { + let stream = stream.with_version(v); + // Draft 14-16: path came in the bidi SETUP, no uni SETUP to hand back. + let (protocol, goaway) = ietf::start(ietf::Config { + runtime: runtime.clone(), + limits: server.limits, + session: session.clone(), + setup: Some(stream), + request_id_max, + client: false, + publish, + subscribe, + peer_hop, + cost: None, + version: v, + path: None, + authority: None, + peer_setup_stream: None, + peer_declared: Some(peer_declared), + auth: auth.clone(), + early_unis: Vec::new(), + extensions: server.extensions, + })?; + (None, crate::driver::Protocol::Ietf(protocol), goaway, auth, None) + } + }; - fn close(mut self: Box, err: Error) { - close(&mut self.session, &err, Some(self.version)); + Ok(Session::new( + runtime, + session, + version, + recv_bw, + protocol, + crate::session::Handles { + stats: server.stats.clone(), + goaway, + auth, + estimate, + }, + )) } } -impl Handshake -where - S: crate::transport::poll::Session, -{ +impl Handshake { /// The request path the client advertised in its SETUP. /// /// Empty when the client advertised none: either it sent an empty path, or the @@ -903,14 +805,14 @@ where self } - fn inner_mut(&mut self) -> &mut RequestInner { + fn inner_mut(&mut self) -> &mut RequestInner { self.inner.as_mut().expect("request already responded") } /// Accept the session, returning the [`Session`] and its [`Driver`](crate::Driver). /// /// Poll or spawn the returned driver to run the session. - pub async fn ok(mut self) -> Result<(Session, crate::Driver), Error> { + pub async fn ok(mut self) -> Result<(Session, crate::Driver), Error> { let peer_hop = Some(self.assigned_hop); let auth = self.auth.clone(); let RequestInner { @@ -928,7 +830,8 @@ where version, client_setup, } => server.start_lite(runtime, session, version, Some(client_setup), peer_hop, auth), - PausedHandshake::Boxed(paused) => paused.ok(server, runtime, peer_hop, auth).await, + PausedHandshake::IetfModern(paused) => paused.ok(server, runtime, peer_hop, auth), + PausedHandshake::Legacy(paused) => paused.ok(server, runtime, peer_hop, auth).await, } } @@ -939,25 +842,26 @@ where } } -impl RequestInner { +impl RequestInner { fn close(self, err: Error) { let (mut session, version) = match self.handshake { - PausedHandshake::LiteBare { session, version } => (session, version), - PausedHandshake::LiteSetup { session, version, .. } => (session, version), - PausedHandshake::Boxed(paused) => return paused.close(err), + PausedHandshake::LiteBare { session, version } => (session, version.into()), + PausedHandshake::LiteSetup { session, version, .. } => (session, version.into()), + PausedHandshake::IetfModern(paused) => (paused.session, paused.version.into()), + PausedHandshake::Legacy(paused) => (paused.session, paused.version), }; - close(&mut session, &err, Some(version.into())); + close(&mut session, &err, Some(version)); } } /// Close `session` with `err`'s wire code. -fn close(session: &mut S, err: &Error, version: Option) { +fn close(session: &mut Transport, err: &Error, version: Option) { let error = SessionError::from(err); let code = version.map_or_else(|| error.to_code(), |version| version.session_code(&error)); session.close(code, &err.to_string()); } -impl Drop for Handshake { +impl Drop for Handshake { // A dropped request would otherwise leave the client hanging until its idle // timeout: it already sent SETUP and is waiting on a response. Reject loudly. fn drop(&mut self) { @@ -1417,7 +1321,7 @@ mod tests { let observed = session.clone(); let result = Server::new() .with_versions(alpn.parse::().unwrap().into()) - .accept_request_lite(moq_clock::now(), session) + .accept_request(moq_clock::now(), session) .await; assert!( matches!(result, Err(Error::ProtocolViolation)), @@ -1432,10 +1336,7 @@ mod tests { async fn uri_binding_accepts_setup_without_path() { let mut session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None)]); session.binding = crate::transport::Binding::Uri; - let request = Server::new() - .accept_request_lite(moq_clock::now(), session) - .await - .unwrap(); + let request = Server::new().accept_request(moq_clock::now(), session).await.unwrap(); assert_eq!(request.path(), ""); } @@ -1456,10 +1357,7 @@ mod tests { [lite05_group(), lite05_setup(Some("/team/room"), None, None)], ); let stops = session.stops.clone(); - let request = Server::new() - .accept_request_lite(moq_clock::now(), session) - .await - .unwrap(); + let request = Server::new().accept_request(moq_clock::now(), session).await.unwrap(); assert_eq!(request.path(), "/team/room"); assert!(stops.lock().unwrap().is_empty(), "the early stream must stay open"); } @@ -1468,10 +1366,7 @@ mod tests { async fn accept_request_reads_buffered_setup_after_transport_close() { let mut session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some("/closed"), None, None)]); crate::transport::poll::Session::close(&mut session, SessionError::Cancel.to_code(), "closed"); - let request = Server::new() - .accept_request_lite(moq_clock::now(), session) - .await - .unwrap(); + let request = Server::new().accept_request(moq_clock::now(), session).await.unwrap(); assert_eq!(request.path(), "/closed"); } @@ -1479,10 +1374,7 @@ mod tests { async fn accepted_lite_setup_refuses_a_second_setup_stream() { let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None)]); let transport = session.clone(); - let request = Server::new() - .accept_request_lite(moq_clock::now(), session) - .await - .unwrap(); + let request = Server::new().accept_request(moq_clock::now(), session).await.unwrap(); let (_session, mut driver) = request.ok().await.unwrap(); transport.uni.lock().unwrap().push_back(vec![1]); let _ = driver.poll(moq_clock::now(), &kio::Waiter::noop()); @@ -1516,12 +1408,12 @@ mod tests { inner: Some(RequestInner { server: Server::new().with_publisher(&origin), runtime: Deadlines::new(moq_clock::now()), - handshake: PausedHandshake::Boxed(Box::new(PausedIetfModern { - session: transport, + handshake: PausedHandshake::IetfModern(PausedIetfModern { + session: Transport::new(transport), version, peer_setup: ietf::PeerSetup { stream: crate::coding::Reader::new( - crate::lite::test_transport::PendingRecv, + crate::transport::boxed::RecvStream::new(crate::lite::test_transport::PendingRecv), Version::Ietf(version), ), path: None, @@ -1529,7 +1421,7 @@ mod tests { declared: ietf::peer::Peer::default(), early: Vec::new(), }, - })), + }), }), }; let assigned = request.assigned_hop; diff --git a/rs/moq-net/src/session.rs b/rs/moq-net/src/session.rs index 09a0d80bac..969b41286d 100644 --- a/rs/moq-net/src/session.rs +++ b/rs/moq-net/src/session.rs @@ -3,6 +3,7 @@ use std::{sync::Arc, task::Poll, time::Duration}; use crate::transport::Stats as _; +use crate::transport::poll::Session as _; use crate::{Error, SessionError, Version, auth, bandwidth, coding::StreamCodes, goaway}; @@ -290,17 +291,14 @@ pub(super) struct Handles { } impl Session { - pub(super) fn new( + pub(super) fn new( runtime: crate::time::Deadlines, - session: S, + session: crate::driver::Transport, version: Version, recv_bandwidth: Option, - protocol: crate::driver::Protocol, + protocol: crate::driver::Protocol, handles: Handles, - ) -> (Self, crate::Driver) - where - S: crate::transport::poll::Session, - { + ) -> (Self, crate::Driver) { let Handles { goaway, auth, @@ -377,14 +375,14 @@ impl Session { /// anyone is consuming them. /// /// Finishes once the transport reports closed; everything else is moot then. -pub(crate) struct Supervisor { +pub(crate) struct Supervisor { version: Version, local_close: Arc, runtime: crate::time::Deadlines, - session: S, + session: crate::driver::Transport, // A dedicated clone for the close watch, since each pending poll operation // needs its own handle. - closed_watch: S, + closed_watch: crate::driver::Transport, /// Handle-side close requests; `None` once one was executed (only the /// first close matters, and the channel closing is the last handle /// dropping). @@ -426,7 +424,7 @@ enum SamplerMode { }, } -impl Supervisor { +impl Supervisor { const POLL_INTERVAL: Duration = Duration::from_millis(100); pub(crate) fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { diff --git a/rs/moq-net/src/transport.rs b/rs/moq-net/src/transport.rs index ddde6f5065..db36000abc 100644 --- a/rs/moq-net/src/transport.rs +++ b/rs/moq-net/src/transport.rs @@ -3,11 +3,14 @@ //! Implement these traits for a transport, or wrap a backend at the application //! boundary. `moq-tokio`, `moq-wasm`, and `moq-uring` provide their own adapters, and //! [`memory`] links two peers in one process. -//! The async helpers only drive the required poll methods. Thread-local transports -//! stay supported; [`poll::Boxable`] adds native thread bounds for boxed drivers. +//! The async helpers only drive the required poll methods. Sessions take a +//! [`poll::Boxable`] transport, which adds native thread bounds; wrap a thread-local +//! transport in a type that checks the thread on use (`moq-uring` has one). use std::time::Duration; +#[cfg(feature = "async")] +pub(crate) mod boxed; pub mod memory; /// A trait that is Send on native targets and empty on WASM. @@ -393,14 +396,16 @@ pub mod poll { /// A pending [`RecvStream::closed`]. RecvClosed, RecvStream, poll_closed, Result<(), S::Error>); - /// A transport whose session, streams, and errors can be captured by the - /// boxed drivers (`Send` boxes on native): what the moq-transport path - /// requires until it too becomes named machines. Implemented automatically. + /// A transport that [`Client`](crate::Client) and [`Server`](crate::Server) can box: + /// the session, its streams, send groups, and errors are thread-safe on native. + /// Implemented automatically. + /// + /// Boxing is what compiles the session code once per binary rather than once per + /// transport. pub trait Boxable: Session< - SendStream: MaybeSend + SendStream, - RecvStream: MaybeSend, - Error: MaybeSend, + SendStream: MaybeSend + MaybeSync + SendStream, + RecvStream: MaybeSend + MaybeSync, > + MaybeSend + MaybeSync { @@ -408,9 +413,8 @@ pub mod poll { impl Boxable for S where S: Session< - SendStream: MaybeSend + SendStream, - RecvStream: MaybeSend, - Error: MaybeSend, + SendStream: MaybeSend + MaybeSync + SendStream, + RecvStream: MaybeSend + MaybeSync, > + MaybeSend + MaybeSync { diff --git a/rs/moq-net/src/transport/boxed.rs b/rs/moq-net/src/transport/boxed.rs new file mode 100644 index 0000000000..dd9546c1c3 --- /dev/null +++ b/rs/moq-net/src/transport/boxed.rs @@ -0,0 +1,495 @@ +//! The one transport type the session code is compiled for. +//! +//! [`Client`](crate::Client) and [`Server`](crate::Server) box the caller's transport +//! here before running any protocol code, so the lite and ietf drivers are +//! monomorphized once per binary instead of once per transport and per crate that +//! awaits them. Each transport only instantiates the thin forwarding impls below. + +use std::any::Any; +use std::task::{Context, Poll}; + +use bytes::{Buf, BufMut, Bytes}; + +use super::poll::{self, SendOptions}; +use super::{Binding, CloseKind, MaybeSend, MaybeSync, Wire}; + +/// A boxed [`poll::Boxable`] session. +pub(crate) struct Session(Box); + +impl Session { + pub(crate) fn new(session: S) -> Self { + Self(Box::new(session)) + } +} + +impl Clone for Session { + fn clone(&self) -> Self { + Self(self.0.clone_box()) + } +} + +/// A boxed transport error, keeping both code registries and the message. +#[derive(Debug)] +pub(crate) struct Error(Box); + +impl Error { + fn new(err: impl super::Error) -> Self { + Self(Box::new(err)) + } +} + +impl std::fmt::Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.fmt(f) + } +} + +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + self.0.source() + } +} + +impl super::Error for Error { + fn close_kind(&self) -> CloseKind { + self.0.close_kind() + } + + fn session_error(&self) -> Option<(u32, String)> { + self.0.session_error() + } + + fn stream_error(&self) -> Option { + self.0.stream_error() + } +} + +/// A boxed send group. It only ever goes back to the session that opened it. +pub(crate) struct Group(Box); + +impl Clone for Group { + fn clone(&self) -> Self { + Self(self.0.clone_box()) + } +} + +impl poll::SendGroup for Group { + fn set_priority(&mut self, priority: i64) { + self.0.set_priority(priority) + } +} + +/// A boxed outgoing stream. +pub(crate) struct SendStream(Box); + +impl SendStream { + /// Box a stream whose send group, if any, is a [`Group`]. + pub(crate) fn new(stream: impl poll::SendStream + MaybeSend + MaybeSync) -> Self { + Self(Box::new(stream)) + } +} + +/// A boxed incoming stream. +pub(crate) struct RecvStream(Box); + +impl RecvStream { + pub(crate) fn new(stream: impl poll::RecvStream + MaybeSend + MaybeSync) -> Self { + Self(Box::new(stream)) + } +} + +/// The transport's connection stats, read once. +#[derive(Clone, Copy, Debug, Default)] +pub(crate) struct Stats { + bytes_sent: Option, + bytes_received: Option, + bytes_lost: Option, + packets_sent: Option, + packets_received: Option, + packets_lost: Option, + rtt: Option, + estimated_send_rate: Option, +} + +impl Stats { + fn read(stats: impl super::Stats) -> Self { + Self { + bytes_sent: stats.bytes_sent(), + bytes_received: stats.bytes_received(), + bytes_lost: stats.bytes_lost(), + packets_sent: stats.packets_sent(), + packets_received: stats.packets_received(), + packets_lost: stats.packets_lost(), + rtt: stats.rtt(), + estimated_send_rate: stats.estimated_send_rate(), + } + } +} + +impl super::Stats for Stats { + fn bytes_sent(&self) -> Option { + self.bytes_sent + } + fn bytes_received(&self) -> Option { + self.bytes_received + } + fn bytes_lost(&self) -> Option { + self.bytes_lost + } + fn packets_sent(&self) -> Option { + self.packets_sent + } + fn packets_received(&self) -> Option { + self.packets_received + } + fn packets_lost(&self) -> Option { + self.packets_lost + } + fn rtt(&self) -> Option { + self.rtt + } + fn estimated_send_rate(&self) -> Option { + self.estimated_send_rate + } +} + +trait ErasedSession: MaybeSend + MaybeSync { + fn clone_box(&self) -> Box; + fn poll_accept_uni(&mut self, cx: &mut Context<'_>) -> Poll>; + fn poll_accept_bi(&mut self, cx: &mut Context<'_>) -> Poll>; + fn poll_open_uni( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll>; + fn poll_open_bi( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll>; + fn send_group(&mut self, priority: i64) -> Option; + fn send_datagram(&mut self, payload: Bytes, options: SendOptions<'_, Group>) -> Result<(), Error>; + fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll>; + fn max_datagram_size(&self) -> usize; + fn protocol(&self) -> Option<&str>; + fn binding(&self) -> Binding; + fn close(&mut self, code: u32, reason: &str); + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll; + fn stats(&self) -> Stats; + fn wire(&self) -> Option; +} + +trait ErasedGroup: MaybeSend + MaybeSync { + fn clone_box(&self) -> Box; + fn set_priority(&mut self, priority: i64); + fn as_any(&self) -> &dyn Any; +} + +trait ErasedSend: MaybeSend + MaybeSync { + fn poll_write(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll>; + fn poll_write_buf(&mut self, cx: &mut Context<'_>, buf: &mut dyn Buf) -> Poll>; + fn set_priority(&mut self, priority: i64); + fn set_group(&mut self, group: &Group); + fn finish(&mut self) -> Result<(), Error>; + fn reset(&mut self, code: u32); + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll>; +} + +trait ErasedRecv: MaybeSend + MaybeSync { + fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Error>>; + fn poll_read_buf(&mut self, cx: &mut Context<'_>, buf: &mut dyn BufMut) -> Poll, Error>>; + fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Error>>; + fn stop(&mut self, code: u32); + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll>; +} + +type GroupOf = <::SendStream as poll::SendStream>::Group; + +/// Recover the transport's own group from a boxed one. +fn unbox<'a, G: 'static>(options: SendOptions<'a, Group>) -> SendOptions<'a, G> { + SendOptions { + group: options.group.map(|group| { + group + .0 + .as_any() + .downcast_ref::() + .expect("send group belongs to another transport") + }), + priority: options.priority, + } +} + +fn bi((send, recv): poll::BiStreams) -> (SendStream, RecvStream) { + (SendStream(Box::new(send)), RecvStream(Box::new(recv))) +} + +impl ErasedSession for S { + fn clone_box(&self) -> Box { + Box::new(self.clone()) + } + + fn poll_accept_uni(&mut self, cx: &mut Context<'_>) -> Poll> { + poll::Session::poll_accept_uni(self, cx) + .map(|res| res.map(|recv| RecvStream(Box::new(recv))).map_err(Error::new)) + } + + fn poll_accept_bi(&mut self, cx: &mut Context<'_>) -> Poll> { + poll::Session::poll_accept_bi(self, cx).map(|res| res.map(bi::).map_err(Error::new)) + } + + fn poll_open_uni( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll> { + poll::Session::poll_open_uni(self, cx, unbox(options)) + .map(|res| res.map(|send| SendStream(Box::new(send))).map_err(Error::new)) + } + + fn poll_open_bi( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll> { + poll::Session::poll_open_bi(self, cx, unbox(options)).map(|res| res.map(bi::).map_err(Error::new)) + } + + fn send_group(&mut self, priority: i64) -> Option { + poll::Session::send_group(self, priority).map(|group| Group(Box::new(group))) + } + + fn send_datagram(&mut self, payload: Bytes, options: SendOptions<'_, Group>) -> Result<(), Error> { + poll::Session::send_datagram(self, payload, unbox::>(options)).map_err(Error::new) + } + + fn poll_recv_datagram(&mut self, cx: &mut Context<'_>) -> Poll> { + poll::Session::poll_recv_datagram(self, cx).map_err(Error::new) + } + + fn max_datagram_size(&self) -> usize { + poll::Session::max_datagram_size(self) + } + + fn protocol(&self) -> Option<&str> { + poll::Session::protocol(self) + } + + fn binding(&self) -> Binding { + poll::Session::binding(self) + } + + fn close(&mut self, code: u32, reason: &str) { + poll::Session::close(self, code, reason) + } + + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll { + poll::Session::poll_closed(self, cx).map(Error::new) + } + + fn stats(&self) -> Stats { + Stats::read(poll::Session::stats(self)) + } + + fn wire(&self) -> Option { + poll::Session::wire(self) + } +} + +impl ErasedGroup for G { + fn clone_box(&self) -> Box { + Box::new(self.clone()) + } + + fn set_priority(&mut self, priority: i64) { + poll::SendGroup::set_priority(self, priority) + } + + fn as_any(&self) -> &dyn Any { + self + } +} + +impl + MaybeSend + MaybeSync> ErasedSend for T { + fn poll_write(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + poll::SendStream::poll_write(self, cx, buf).map_err(Error::new) + } + + fn poll_write_buf(&mut self, cx: &mut Context<'_>, mut buf: &mut dyn Buf) -> Poll> { + // `&mut dyn Buf` forwards `copy_to_bytes`, so a `Bytes` source stays zero-copy. + poll::SendStream::poll_write_buf(self, cx, &mut buf).map_err(Error::new) + } + + fn set_priority(&mut self, priority: i64) { + poll::SendStream::set_priority(self, priority) + } + + fn set_group(&mut self, group: &Group) { + let group = group + .0 + .as_any() + .downcast_ref::() + .expect("send group belongs to another transport"); + poll::SendStream::set_group(self, group) + } + + fn finish(&mut self) -> Result<(), Error> { + poll::SendStream::finish(self).map_err(Error::new) + } + + fn reset(&mut self, code: u32) { + poll::SendStream::reset(self, code) + } + + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { + poll::SendStream::poll_closed(self, cx).map_err(Error::new) + } +} + +impl ErasedRecv for T { + fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Error>> { + poll::RecvStream::poll_read(self, cx, dst).map_err(Error::new) + } + + fn poll_read_buf(&mut self, cx: &mut Context<'_>, mut buf: &mut dyn BufMut) -> Poll, Error>> { + poll::RecvStream::poll_read_buf(self, cx, &mut buf).map_err(Error::new) + } + + fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, Error>> { + poll::RecvStream::poll_read_chunk(self, cx, max).map_err(Error::new) + } + + fn stop(&mut self, code: u32) { + poll::RecvStream::stop(self, code) + } + + fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { + poll::RecvStream::poll_closed(self, cx).map_err(Error::new) + } +} + +impl poll::Session for Session { + type SendStream = SendStream; + type RecvStream = RecvStream; + type Error = Error; + + fn poll_accept_uni(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_accept_uni(cx) + } + + fn poll_accept_bi(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_accept_bi(cx) + } + + fn poll_open_uni( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll> { + self.0.poll_open_uni(cx, options) + } + + fn poll_open_bi( + &mut self, + cx: &mut Context<'_>, + options: SendOptions<'_, Group>, + ) -> Poll> { + self.0.poll_open_bi(cx, options) + } + + fn send_group(&mut self, priority: i64) -> Option { + self.0.send_group(priority) + } + + fn send_datagram(&mut self, payload: Bytes, options: SendOptions<'_, Group>) -> Result<(), Error> { + self.0.send_datagram(payload, options) + } + + 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) -> 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 super::Stats { + self.0.stats() + } + + fn wire(&self) -> Option { + self.0.wire() + } +} + +impl poll::SendStream for SendStream { + type Error = Error; + type Group = Group; + + 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: &Group) { + self.0.set_group(group) + } + + fn finish(&mut self) -> Result<(), 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 RecvStream { + type Error = Error; + + fn poll_read(&mut self, cx: &mut Context<'_>, dst: &mut [u8]) -> Poll, Error>> { + self.0.poll_read(cx, dst) + } + + fn poll_read_buf(&mut self, cx: &mut Context<'_>, buf: &mut B) -> Poll, Error>> { + self.0.poll_read_buf(cx, buf) + } + + fn poll_read_chunk(&mut self, cx: &mut Context<'_>, max: usize) -> Poll, 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) + } +} diff --git a/rs/moq-net/src/transport/memory.rs b/rs/moq-net/src/transport/memory.rs index db944fd01f..458c662baf 100644 --- a/rs/moq-net/src/transport/memory.rs +++ b/rs/moq-net/src/transport/memory.rs @@ -665,7 +665,7 @@ mod tests { #[test] fn bounds() { - fn boxable>() {} + fn boxable() {} boxable::(); } diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index 61d611a5c8..ea20cfbbc1 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -1065,9 +1065,9 @@ pub fn request_for(auth: &Auth, request: &moq_tokio::server::Request) -> Request } /// Build an auth request from the path, credential, and role carried in SETUP. -pub(crate) fn request_for_setup( +pub(crate) fn request_for_setup( auth: &Auth, - setup: &moq_net::server::Handshake, + setup: &moq_net::server::Handshake, transport: moq_auth::Transport, ) -> Request { let (path, query) = match setup.path().split_once('?') { diff --git a/rs/moq-relay/src/incoming.rs b/rs/moq-relay/src/incoming.rs index 86ea3e3d49..2761d7c70e 100644 --- a/rs/moq-relay/src/incoming.rs +++ b/rs/moq-relay/src/incoming.rs @@ -2,7 +2,6 @@ use crate::auth; use axum::http; -use moq_net::transport::poll; /// A session whose SETUP awaits the relay's reply: what a [`Connection`](crate::Connection) serves. /// @@ -11,20 +10,20 @@ pub trait Incoming: Sealed {} impl Incoming for moq_tokio::server::Request {} -impl Incoming for Accepted {} +impl Incoming for Accepted {} /// A session accepted on any moq-net transport, its SETUP awaiting the relay's reply. /// /// The auth decider sees the path, query, token, and role the SETUP carried, so the /// client must send its path in-band. -pub struct Accepted { - handshake: moq_net::server::Handshake, +pub struct Accepted { + handshake: moq_net::server::Handshake, transport: moq_auth::Transport, } -impl Accepted { +impl Accepted { /// Wrap `handshake`, reported to the auth decider as arriving over `transport`. - pub fn new(handshake: moq_net::server::Handshake, transport: moq_auth::Transport) -> Self { + pub fn new(handshake: moq_net::server::Handshake, transport: moq_auth::Transport) -> Self { Self { handshake, transport } } } @@ -145,15 +144,15 @@ impl Sealed for moq_tokio::server::Request { } } -impl Sealed for Accepted { +impl Sealed for Accepted { /// Polled on the connection's own task, so it needs no runtime to spawn on. - type Driver = moq_net::Driver; + type Driver = moq_net::Driver; fn remote(&self) -> Option { None } - fn transport(&self) -> impl std::fmt::Display + use { + fn transport(&self) -> impl std::fmt::Display + use<> { self.transport.as_str() } diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index ba1a3f1d95..620e8909ff 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -692,16 +692,10 @@ async fn serve_connection( Some(_) => moq_net::transport::Binding::Uri, None => moq_net::transport::Binding::InBand, }; - let session = moq_uring::transport::Session::new(transport, binding); + // moq-net boxes the transport with a native `Send` bound, which only a `Local` + // session meets; it still never leaves this worker. + let session = moq_uring::transport::Local::new(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, @@ -713,24 +707,12 @@ async fn serve_connection( 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, lifecycle).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, lifecycle).await - } - } + let request = accepted + .deadline + .bound(server.accept_request(now, session)) + .await + .context("moq handshake failed")?; + serve_request(accepted, request, lifecycle).await } /// A connection past its transport handshake, with the facts read off it @@ -747,14 +729,14 @@ struct Accepted { url: Option, } -/// A started session's supervision, boxed so both handshake paths return one type. +/// A started session's supervision. type Supervision = futures::future::BoxFuture<'static, anyhow::Result<()>>; /// Authenticate on the shared runtime, then start the session on the worker. /// Returns the session's supervision, which runs on the shared runtime too. -async fn serve_request( +async fn serve_request( accepted: Accepted, - request: moq_net::server::Handshake, + request: moq_net::server::Handshake, lifecycle: &Lifecycle, ) -> anyhow::Result { let Accepted { diff --git a/rs/moq-room/src/chat.rs b/rs/moq-room/src/chat.rs index 22ebb790c6..8d7ab2c530 100644 --- a/rs/moq-room/src/chat.rs +++ b/rs/moq-room/src/chat.rs @@ -247,18 +247,12 @@ mod tests { .with_subscriber(to.clone()); tokio::join!( async { - let (session, driver) = client - .connect_lite(moq_net::time::now(), client_transport) - .await - .unwrap(); + let (session, driver) = client.connect(moq_net::time::now(), client_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session }, async { - let (session, driver) = server - .accept_lite(moq_net::time::now(), server_transport) - .await - .unwrap(); + let (session, driver) = server.accept(moq_net::time::now(), server_transport).await.unwrap(); tokio::spawn(moq_net::time::run(driver)); session } diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 0cb916e091..88c8bd928e 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -740,7 +740,7 @@ impl Server { let (request, deadline) = accept_request(&setup.server, session, binding, deadline).await?; // Only WebTransport carries a request URL; raw QUIC puts the path in the SETUP. let transport = match url { Some(_) => Transport::WebTransport, None => Transport::Quic }; - Ok(Request { transport, url, identity, authority, link, kind: RequestKind::Quic(Box::new(request)), quic, deadline, active }) + Ok(Request { transport, url, identity, authority, link, handshake: request, quic, deadline, active }) }.boxed()); } } @@ -763,7 +763,7 @@ impl Server { identity, authority, link, - kind: RequestKind::Iroh(Box::new(request)), + handshake: request, #[cfg(feature = "quic")] quic: None, deadline, @@ -936,7 +936,7 @@ impl Listener { alpn: Some(protocol), ..Default::default() }, - kind: RequestKind::Qmux(Box::new(request)), + handshake: request, #[cfg(feature = "quic")] quic: None, deadline, @@ -1055,12 +1055,10 @@ async fn accept_request( session: S, binding: moq_net::transport::Binding, deadline: Option, -) -> crate::Result<(PendingRequest>, Option)> +) -> crate::Result<(moq_net::server::Handshake, Option)> where S: crate::transport::Grouped, crate::transport::Session: moq_net::transport::poll::Boxable, - as moq_net::transport::poll::Session>::SendStream: moq_net::transport::MaybeSync, - as moq_net::transport::poll::Session>::RecvStream: moq_net::transport::MaybeSync, { let session = crate::transport::Session::new(session, binding); let deadline = deadline.map(|at| Deadline::new(at, session.clone())); @@ -1084,7 +1082,7 @@ pub(crate) async fn accept_websocket( authority, identity: options.identity, link: options.link, - kind: RequestKind::Qmux(Box::new(request)), + handshake: request, #[cfg(feature = "quic")] quic: None, deadline, @@ -1575,7 +1573,7 @@ async fn stream_request( authority: None, identity: None, link, - kind: RequestKind::Qmux(Box::new(request)), + handshake: request, #[cfg(feature = "quic")] quic: None, deadline, @@ -1587,29 +1585,6 @@ async fn stream_request( } } -/// An accepted connection whose MoQ SETUP has already been exchanged. -/// -/// Every backend drives the transport connect *and* the MoQ handshake up front, so the -/// [`path`](Request::path)/[`role`](Request::role) a client advertised are available on -/// every transport before the caller authorizes. The variant only distinguishes the -/// underlying session type; all of them delegate identically. -/// A pending moq-net request over transport `S`, driven by our tokio runtime. -type PendingRequest = moq_net::server::Handshake; - -pub(crate) enum RequestKind { - #[cfg(feature = "quic")] - Quic(Box>>), - #[cfg(feature = "iroh")] - Iroh(Box>>), - #[cfg(any( - feature = "tcp", - all(feature = "uds", unix), - feature = "websocket", - feature = "webrtc" - ))] - Qmux(Box>>), -} - /// The transport-level facts a backend captures while accepting a connection, before the /// MoQ SETUP. Grouped so the shared accept loop builds a [`Request`] from named fields /// rather than a wide tuple. @@ -1661,7 +1636,8 @@ pub struct Request { identity: Option, /// The link facts the transport could see. link: Link, - kind: RequestKind, + /// The paused moq-net handshake; the same type whatever the transport. + handshake: moq_net::server::Handshake, /// The connection under a QUIC or WebTransport request, for [`Request::quic`]. #[cfg(feature = "quic")] quic: Option, @@ -1681,63 +1657,6 @@ pub enum Reject { App(u16), } -/// Delegate a read-only call to the inner [`moq_net::server::Handshake`], whatever the transport. -macro_rules! request_ref { - ($self:expr, $r:ident => $body:expr) => { - match &$self.kind { - #[cfg(feature = "quic")] - RequestKind::Quic($r) => $body, - #[cfg(feature = "iroh")] - RequestKind::Iroh($r) => $body, - #[cfg(any( - feature = "tcp", - all(feature = "uds", unix), - feature = "websocket", - feature = "webrtc" - ))] - RequestKind::Qmux($r) => $body, - } - }; -} - -/// Delegate a consuming call whose arms all yield the same type (e.g. `ok`, `close`). -macro_rules! request_into { - ($kind:expr, $r:ident => $body:expr) => { - match $kind { - #[cfg(feature = "quic")] - RequestKind::Quic($r) => $body, - #[cfg(feature = "iroh")] - RequestKind::Iroh($r) => $body, - #[cfg(any( - feature = "tcp", - all(feature = "uds", unix), - feature = "websocket", - feature = "webrtc" - ))] - RequestKind::Qmux($r) => $body, - } - }; -} - -/// Delegate a consuming builder call, rebuilding the same variant from the returned request. -macro_rules! request_map { - ($kind:expr, $r:ident => $body:expr) => { - match $kind { - #[cfg(feature = "quic")] - RequestKind::Quic($r) => RequestKind::Quic(Box::new($body)), - #[cfg(feature = "iroh")] - RequestKind::Iroh($r) => RequestKind::Iroh(Box::new($body)), - #[cfg(any( - feature = "tcp", - all(feature = "uds", unix), - feature = "websocket", - feature = "webrtc" - ))] - RequestKind::Qmux($r) => RequestKind::Qmux(Box::new($body)), - } - }; -} - impl Request { /// Reject the session. The transport is already accepted, so this closes the /// just-established MoQ session rather than answering the transport handshake: @@ -1747,19 +1666,19 @@ impl Request { Reject::Unauthorized | Reject::Forbidden => moq_net::Error::Unauthorized, Reject::App(code) => moq_net::Error::App(code), }; - request_into!(self.kind, request => request.close(err)); + self.handshake.close(err); Ok(()) } /// Publish the given origin to the session. pub fn with_publisher(mut self, publish: impl moq_net::Consume) -> Self { - self.kind = request_map!(self.kind, request => request.with_publisher(publish)); + self.handshake = self.handshake.with_publisher(publish); self } /// Subscribe to the given origin from the session. pub fn with_subscriber(mut self, subscribe: moq_net::origin::Producer) -> Self { - self.kind = request_map!(self.kind, request => request.with_subscriber(subscribe)); + self.handshake = self.handshake.with_subscriber(subscribe); self } @@ -1767,19 +1686,19 @@ impl Request { /// [`moq_net::server::Handshake::with_peer_hop`]. Derive it from [`Self::peer_identity`], /// never from something coarser. pub fn with_peer_hop(mut self, hop: moq_net::Hop) -> Self { - self.kind = request_map!(self.kind, request => request.with_peer_hop(hop)); + self.handshake = self.handshake.with_peer_hop(hop); self } /// Attach a per-connection [`moq_net::stats::Session`] context to this session. pub fn with_stats(mut self, stats: moq_net::stats::Session) -> Self { - self.kind = request_map!(self.kind, request => request.with_stats(stats)); + self.handshake = self.handshake.with_stats(stats); self } /// Set request caps after authorizing the peer and before answering SETUP. pub fn with_limits(mut self, limits: moq_net::session::Limits) -> Self { - self.kind = request_map!(self.kind, request => request.with_limits(limits)); + self.handshake = self.handshake.with_limits(limits); self } @@ -1810,21 +1729,27 @@ impl Request { /// otherwise hold open. pub async fn ok(self) -> crate::Result<(Session, tokio::task::JoinHandle>)> { let Request { - kind, deadline, active, .. + handshake, + deadline, + active, + .. } = self; let ok = async { - Ok(request_into!(kind, request => { - let (session, driver) = request.ok().await?; - use tracing::Instrument; - let driver = tokio::spawn(async move { + let (session, driver) = handshake.ok().await?; + use tracing::Instrument; + let driver = tokio::spawn( + async move { crate::panic::isolate(async move { let _active = active; moq_net::time::run(driver).await; Ok(()) - }).await.unwrap_or(Err(Error::Panic)) - }.instrument(tracing::Span::current())); - (session, driver) - })) + }) + .await + .unwrap_or(Err(Error::Panic)) + } + .instrument(tracing::Span::current()), + ); + Ok((session, driver)) }; Deadline::bound(deadline.as_ref(), ok).await } @@ -1864,7 +1789,7 @@ impl Request { // An empty SETUP path means the client advertised none, so fall back to the // request URL. URL-carrying bindings are the ones that must not send a path at // all, so this never discards a path the client meant us to use. - let setup = request_ref!(self, r => r.path()); + let setup = self.handshake.path(); let path = if setup.is_empty() { self.url.as_ref().map(Url::path).unwrap_or("") } else { @@ -1877,7 +1802,7 @@ impl Request { /// /// Query values can contain credentials. Avoid logging this value. pub fn query(&self) -> Option<&str> { - let setup = request_ref!(self, r => r.path()); + let setup = self.handshake.path(); if setup.is_empty() { self.url.as_ref().and_then(Url::query) } else { @@ -1911,12 +1836,12 @@ impl Request { /// Available on every transport. Use it to reject a token that lacks the scope for /// the client's intended direction. pub fn role(&self) -> Option { - request_ref!(self, r => r.role()) + self.handshake.role() } /// Answer the peer's tokens before accepting and starting its protocol driver. pub fn auth(&self) -> moq_net::auth::Handle { - request_ref!(self, r => r.auth()) + self.handshake.auth() } /// The Hop ID the peer declared in its SETUP (moq-lite-05+). @@ -1927,7 +1852,7 @@ impl Request { /// Self-declared, so treat it as a correlation hint rather than an /// authenticated identity: authorize on the token or client certificate. pub fn peer_hop(&self) -> Option { - request_ref!(self, r => r.peer_hop()) + self.handshake.peer_hop() } /// The credential a moq-transport client presented in its SETUP's `AUTHORIZATION @@ -1935,7 +1860,7 @@ impl Request { /// /// Like [`query`](Self::query), it can hold a credential. Avoid logging it. pub fn token(&self) -> Option<&moq_net::setup::Token> { - request_ref!(self, r => r.token()) + self.handshake.token() } /// The client certificate chain the peer presented, if any, validated diff --git a/rs/moq-uring/README.md b/rs/moq-uring/README.md index 676249bc4b..ec459a5356 100644 --- a/rs/moq-uring/README.md +++ b/rs/moq-uring/README.md @@ -32,12 +32,11 @@ local (`!Send`) task set, and the UDP sockets bound through it. over the same adapter via `web-transport-proto`. `quic::web::Session` is a raw or web transport (`Session::raw`). Wrap it with `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. 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 + for WebTransport or `InBand` for raw QUIC. moq-net boxes the transport with a + native `Send` bound, so wrap the session once more in `transport::Local`, which + checks the thread on every use, before `connect`/`accept`. They 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 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/benches/session_lite.rs b/rs/moq-uring/benches/session_lite.rs index e59725e755..005a71fd3a 100644 --- a/rs/moq-uring/benches/session_lite.rs +++ b/rs/moq-uring/benches/session_lite.rs @@ -140,15 +140,15 @@ mod linux { .expect("quic accept"); let (session, driver) = moq_net::Server::new() .with_publisher(&pub_origin) - .accept_lite( + .accept( moq_net::time::now(), - moq_uring::transport::Session::new( + moq_uring::transport::Local::new(moq_uring::transport::Session::new( quic::web::Session::raw(conn), moq_net::transport::Binding::InBand, - ), + )), ) .await - .expect("accept_lite"); + .expect("accept"); let _ = server_handle.run(driver).await; session.closed().await; }); @@ -159,15 +159,15 @@ mod linux { let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect"); let (session, driver) = moq_net::Client::new() .with_subscriber(sub_origin.clone()) - .connect_lite( + .connect( moq_net::time::now(), - moq_uring::transport::Session::new( + moq_uring::transport::Local::new(moq_uring::transport::Session::new( quic::web::Session::raw(conn), moq_net::transport::Binding::InBand, - ), + )), ) .await - .expect("connect_lite"); + .expect("connect"); let task_handle = handle.clone(); handle.spawn(async move { let _ = task_handle.run(driver).await; diff --git a/rs/moq-uring/src/lib.rs b/rs/moq-uring/src/lib.rs index 9ba152d82b..b4a3f96065 100644 --- a/rs/moq-uring/src/lib.rs +++ b/rs/moq-uring/src/lib.rs @@ -14,10 +14,10 @@ //! //! [`quic`] stacks a sans-IO QUIC stack on that path: a [`quic::Endpoint`] //! 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::Local`] lets the -//! `Send`-bounded moq-transport paths run there too. The socket is the identity: whatever is +//! included), each a [`quic::Connection`] wrapped by [`transport::Session`] +//! and [`transport::Local`], which meets moq-net's `Send` bound by checking the +//! thread on every use, so `moq_net::Client::connect` and `Server::accept` run +//! real MoQ sessions on the worker. 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/quic/client.rs b/rs/moq-uring/src/quic/client.rs index af87e4597f..0417298723 100644 --- a/rs/moq-uring/src/quic/client.rs +++ b/rs/moq-uring/src/quic/client.rs @@ -54,8 +54,8 @@ impl Config { /// [`connect`](super::Endpoint::connect) through it. The connection's driver /// runs as a task on the worker that adopted `socket`. Wrap the returned /// [`Connection`] with [`crate::transport::Session::new`] and -/// [`moq_net::transport::Binding::InBand`] to hand it to -/// `moq_net::Client::connect_lite`, or use the stream API directly. +/// [`moq_net::transport::Binding::InBand`], then [`crate::transport::Local`], to +/// hand it to `moq_net::Client::connect`, or use the stream API directly. pub async fn connect(socket: udp::Socket, config: &Config) -> Result { let endpoint = super::Endpoint::new(socket, super::endpoint::Config::default())?; endpoint.connect(config).await diff --git a/rs/moq-uring/src/quic/mod.rs b/rs/moq-uring/src/quic/mod.rs index 0b918d1492..68f2b178ce 100644 --- a/rs/moq-uring/src/quic/mod.rs +++ b/rs/moq-uring/src/quic/mod.rs @@ -7,9 +7,9 @@ //! connection's timeout, and wakes stream waiters. The returned //! [`Connection`] implements [`web_transport_trait::poll`], so //! wrapping it with `crate::transport::Session::new` and -//! [`moq_net::transport::Binding::InBand`] lets -//! `moq_net::Client::connect_lite` / `Server::accept_lite` run real moq-lite -//! sessions on the worker; everything is `Rc`-shared and `!Send` by design. +//! [`moq_net::transport::Binding::InBand`], then `crate::transport::Local`, lets +//! `moq_net::Client::connect` / `Server::accept` run real MoQ sessions on the +//! worker; everything is `Rc`-shared and `!Send` by design. //! //! An [`Endpoint`] serves many connections on one socket, demuxed by //! connection id; [`client::connect`] and [`server::accept`] are its diff --git a/rs/moq-uring/src/transport/local.rs b/rs/moq-uring/src/transport/local.rs index 7c42bcb9b7..fb6b615889 100644 --- a/rs/moq-uring/src/transport/local.rs +++ b/rs/moq-uring/src/transport/local.rs @@ -5,13 +5,12 @@ 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 +/// moq-net boxes every transport 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. +/// proving it at compile time. #[derive(Clone)] pub struct Local(SendWrapper); @@ -176,7 +175,7 @@ mod tests { use super::*; use moq_net::transport::memory; - /// The point of the wrapper: a worker's `!Send` session meets moq-transport's bounds. + /// The point of the wrapper: a worker's `!Send` session meets moq-net's bounds. #[cfg(feature = "quic")] #[test] fn meets_the_boxed_driver_bounds() { diff --git a/rs/moq-uring/tests/integration/session.rs b/rs/moq-uring/tests/integration/session.rs index eceb3bd7ca..79f1f9b0f1 100644 --- a/rs/moq-uring/tests/integration/session.rs +++ b/rs/moq-uring/tests/integration/session.rs @@ -2,7 +2,7 @@ //! //! A publisher and a subscriber session run on one worker (two sockets, two //! [`moq_uring::quic::Connection`]s over loopback), negotiated end to end by -//! `moq_net::Server::accept_lite` / `Client::connect_lite`, with a broadcast, +//! `moq_net::Server::accept` / `Client::connect`, with a broadcast, //! a track, and a frame flowing through the model. //! //! Origin drivers run through Tokio's explicit-time adapter while session @@ -93,12 +93,15 @@ fn lite_session_over_the_worker() { .expect("quic accept"); let (session, driver) = moq_net::Server::new() .with_publisher(&pub_origin) - .accept_lite( + .accept( moq_net::time::now(), - moq_uring::transport::Session::new(quic::web::Session::raw(conn), moq_net::transport::Binding::InBand), + moq_uring::transport::Local::new(moq_uring::transport::Session::new( + quic::web::Session::raw(conn), + moq_net::transport::Binding::InBand, + )), ) .await - .expect("accept_lite"); + .expect("accept"); let _ = server_handle.run(driver).await; // Serve until the client walks away. session.closed().await; @@ -116,15 +119,15 @@ fn lite_session_over_the_worker() { let (session, driver) = moq_net::Client::new() .with_subscriber(sub.clone()) - .connect_lite( + .connect( moq_net::time::now(), - moq_uring::transport::Session::new( + moq_uring::transport::Local::new(moq_uring::transport::Session::new( quic::web::Session::raw(conn), moq_net::transport::Binding::InBand, - ), + )), ) .await - .expect("connect_lite"); + .expect("connect"); let task_handle = handle.clone(); handle.spawn(async move { let _ = task_handle.run(driver).await; @@ -228,15 +231,15 @@ fn two_lite_sessions_share_the_server_socket() { server_handle.spawn(async move { let (session, driver) = moq_net::Server::new() .with_publisher(&pub_origin) - .accept_lite( + .accept( moq_net::time::now(), - moq_uring::transport::Session::new( + moq_uring::transport::Local::new(moq_uring::transport::Session::new( quic::web::Session::raw(conn), moq_net::transport::Binding::InBand, - ), + )), ) .await - .expect("accept_lite"); + .expect("accept"); let _ = session_handle.run(driver).await; session.closed().await; }); @@ -257,15 +260,15 @@ fn two_lite_sessions_share_the_server_socket() { let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect"); let (session, driver) = moq_net::Client::new() .with_subscriber(sub.clone()) - .connect_lite( + .connect( moq_net::time::now(), - moq_uring::transport::Session::new( + moq_uring::transport::Local::new(moq_uring::transport::Session::new( quic::web::Session::raw(conn), moq_net::transport::Binding::InBand, - ), + )), ) .await - .expect("connect_lite"); + .expect("connect"); let task_handle = handle.clone(); handle.spawn(async move { let _ = task_handle.run(driver).await; diff --git a/rs/moq-uring/tests/integration/web.rs b/rs/moq-uring/tests/integration/web.rs index b07faed896..23e225c333 100644 --- a/rs/moq-uring/tests/integration/web.rs +++ b/rs/moq-uring/tests/integration/web.rs @@ -248,12 +248,12 @@ fn lite_session_over_webtransport() { let (moq, session_driver) = moq_net::Client::new() .with_subscriber(sub_origin.clone()) - .connect_lite( + .connect( moq_net::time::now(), moq_tokio::transport::Session::new(session, moq_net::transport::Binding::Uri), ) .await - .expect("connect_lite"); + .expect("connect"); tokio::spawn(moq_net::time::run(session_driver)); let bc = { @@ -299,12 +299,15 @@ fn lite_session_over_webtransport() { let (session, driver) = moq_net::Server::new() .with_publisher(&serve_origin) - .accept_lite( + .accept( moq_net::time::now(), - moq_uring::transport::Session::new(session, moq_net::transport::Binding::Uri), + moq_uring::transport::Local::new(moq_uring::transport::Session::new( + session, + moq_net::transport::Binding::Uri, + )), ) .await - .expect("accept_lite"); + .expect("accept"); let task_handle = handle.clone(); handle.spawn(async move { let _ = task_handle.run(driver).await; From 3875c15d4ed13e834c1ee6eadc0bf661b09a6049 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 20 Oct 2026 23:39:07 -0700 Subject: [PATCH 03/12] docs: migrate notes for the boxed transport Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/MIGRATE.md | 8 ++++++++ rs/moq-net/src/transport.rs | 6 ++---- rs/moq-relay/MIGRATE.md | 4 ++++ rs/moq-uring/MIGRATE.md | 5 +++++ 4 files changed, 19 insertions(+), 4 deletions(-) diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index 8009db2c31..a9c8275361 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,14 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Sessions box their transport + +`Driver` and `server::Handshake` lose their transport type parameter: write `moq_net::Driver` and `moq_net::server::Handshake`. +`Client::connect_lite`, `Server::accept_lite`, and `Server::accept_request_lite` are gone; call `connect`, `accept`, or `accept_request`, which accept every ALPN. +Every session now takes a `transport::poll::Boxable` transport, whose streams must also be `Sync` on native. +A `!Send` transport wraps itself in a type that checks the thread on use, such as `moq_uring::transport::Local`. +`Client::connect` and `Server::accept_request` are plain functions returning `impl Future`, so call them as before with `.await`. + ### Anchors can name a time (#6074) `track::Anchor` gains `Time(Timestamp)`, the group containing a presentation time, so exhaustive matches need an arm for it. diff --git a/rs/moq-net/src/transport.rs b/rs/moq-net/src/transport.rs index db36000abc..9a057f7ffb 100644 --- a/rs/moq-net/src/transport.rs +++ b/rs/moq-net/src/transport.rs @@ -190,12 +190,10 @@ pub mod poll { /// operations release retained resources; shared resources must not be held /// across a wait. Implementations register the supplied waker before Pending. pub trait Session: Clone + 'static { - /// The outgoing stream type. Only the poll half is required, so a `!Send` - /// session can hang `!Send` streams off it. + /// The outgoing stream type. type SendStream: SendStream; - /// The incoming stream type. Only the poll half is required, so a `!Send` - /// session can hang `!Send` streams off it. + /// The incoming stream type. type RecvStream: RecvStream; /// The error type for every operation on this session. diff --git a/rs/moq-relay/MIGRATE.md b/rs/moq-relay/MIGRATE.md index 12d6ecc094..cb7c594f3e 100644 --- a/rs/moq-relay/MIGRATE.md +++ b/rs/moq-relay/MIGRATE.md @@ -5,6 +5,10 @@ Rolling a cluster across a release also follows the [upgrade guide](https://moq. ## Unreleased +### `Accepted` takes any transport's handshake + +`Accepted` loses its transport type parameter, since `moq_net::server::Handshake` has none: write `Accepted` instead of `Accepted`. + ### `cluster.lan.secret` is a file path only (#5910) `--cluster-lan-secret`, `MOQ_CLUSTER_LAN_SECRET`, and `[cluster.lan] secret` no longer accept 64 hex characters inline. diff --git a/rs/moq-uring/MIGRATE.md b/rs/moq-uring/MIGRATE.md index 8ff58a08bc..b5a151957c 100644 --- a/rs/moq-uring/MIGRATE.md +++ b/rs/moq-uring/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Wrap sessions in `transport::Local` + +moq-net dropped `connect_lite` and `accept_lite`, and every session now needs a `Send` transport. +Wrap the worker's session as `transport::Local::new(transport::Session::new(conn, binding))` and call `connect` or `accept`. + ### Timers run on moq-clock The worker is a moq-clock backend, so its deadlines are `moq_clock::Instant`s instead of `std::time::Instant`s. From 17c4f750960dcb7b48c7c76ccd23667ab8898342 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 20 Oct 2026 23:56:30 -0700 Subject: [PATCH 04/12] fix(net): keep the accept futures small The paused handshake used to box its moq-transport continuation. Box the legacy bidi SETUP path and moq-tokio's held handshake again, so a debug build's accept task fits its thread stack. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/server.rs | 3 ++- rs/moq-tokio/src/server.rs | 22 +++++++++++----------- 2 files changed, 13 insertions(+), 12 deletions(-) diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 5c97ededf1..52aa5ce0dd 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -831,7 +831,8 @@ impl Handshake { client_setup, } => server.start_lite(runtime, session, version, Some(client_setup), peer_hop, auth), PausedHandshake::IetfModern(paused) => paused.ok(server, runtime, peer_hop, auth), - PausedHandshake::Legacy(paused) => paused.ok(server, runtime, peer_hop, auth).await, + // Boxed so the bidi SETUP write of these old versions doesn't size every caller's future. + PausedHandshake::Legacy(paused) => Box::pin(paused.ok(server, runtime, peer_hop, auth)).await, } } diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 88c8bd928e..3de7d5b557 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -740,7 +740,7 @@ impl Server { let (request, deadline) = accept_request(&setup.server, session, binding, deadline).await?; // Only WebTransport carries a request URL; raw QUIC puts the path in the SETUP. let transport = match url { Some(_) => Transport::WebTransport, None => Transport::Quic }; - Ok(Request { transport, url, identity, authority, link, handshake: request, quic, deadline, active }) + Ok(Request { transport, url, identity, authority, link, handshake: Box::new(request), quic, deadline, active }) }.boxed()); } } @@ -763,7 +763,7 @@ impl Server { identity, authority, link, - handshake: request, + handshake: Box::new(request), #[cfg(feature = "quic")] quic: None, deadline, @@ -936,7 +936,7 @@ impl Listener { alpn: Some(protocol), ..Default::default() }, - handshake: request, + handshake: Box::new(request), #[cfg(feature = "quic")] quic: None, deadline, @@ -1082,7 +1082,7 @@ pub(crate) async fn accept_websocket( authority, identity: options.identity, link: options.link, - handshake: request, + handshake: Box::new(request), #[cfg(feature = "quic")] quic: None, deadline, @@ -1573,7 +1573,7 @@ async fn stream_request( authority: None, identity: None, link, - handshake: request, + handshake: Box::new(request), #[cfg(feature = "quic")] quic: None, deadline, @@ -1637,7 +1637,7 @@ pub struct Request { /// The link facts the transport could see. link: Link, /// The paused moq-net handshake; the same type whatever the transport. - handshake: moq_net::server::Handshake, + handshake: Box, /// The connection under a QUIC or WebTransport request, for [`Request::quic`]. #[cfg(feature = "quic")] quic: Option, @@ -1672,13 +1672,13 @@ impl Request { /// Publish the given origin to the session. pub fn with_publisher(mut self, publish: impl moq_net::Consume) -> Self { - self.handshake = self.handshake.with_publisher(publish); + *self.handshake = (*self.handshake).with_publisher(publish); self } /// Subscribe to the given origin from the session. pub fn with_subscriber(mut self, subscribe: moq_net::origin::Producer) -> Self { - self.handshake = self.handshake.with_subscriber(subscribe); + *self.handshake = (*self.handshake).with_subscriber(subscribe); self } @@ -1686,19 +1686,19 @@ impl Request { /// [`moq_net::server::Handshake::with_peer_hop`]. Derive it from [`Self::peer_identity`], /// never from something coarser. pub fn with_peer_hop(mut self, hop: moq_net::Hop) -> Self { - self.handshake = self.handshake.with_peer_hop(hop); + *self.handshake = (*self.handshake).with_peer_hop(hop); self } /// Attach a per-connection [`moq_net::stats::Session`] context to this session. pub fn with_stats(mut self, stats: moq_net::stats::Session) -> Self { - self.handshake = self.handshake.with_stats(stats); + *self.handshake = (*self.handshake).with_stats(stats); self } /// Set request caps after authorizing the peer and before answering SETUP. pub fn with_limits(mut self, limits: moq_net::session::Limits) -> Self { - self.handshake = self.handshake.with_limits(limits); + *self.handshake = (*self.handshake).with_limits(limits); self } From 5118d1ff7337a32ce3da3c480844aaa8572e9a18 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 00:40:07 -0700 Subject: [PATCH 05/12] quest: complete net-erase-transport Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 2 -- quest/m1/audit-size.md | 13 ++++++------ quest/m1/net-erase-transport.md | 36 --------------------------------- 3 files changed, 7 insertions(+), 44 deletions(-) delete mode 100644 quest/m1/net-erase-transport.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 0cec85655e..25415864f4 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -118,8 +118,6 @@ lands before the cut. - [Nightly media players](/quest/m1/media-engines-nightly.md) - the Firefox and WebKit player rows pass, or an engine that cannot play leaves the nightly matrix -- [moq-net compiles once](/quest/m1/net-erase-transport.md) - the session drivers stop monomorphizing per transport and per crate, saving several MiB of the relay, FFI library, `libmoq` and CLI - - [Dead code and API surface audit](/quest/m1/audit-dead-code.md) - repeat the removal audit before the next major release - [Dependency and alternatives audit](/quest/m1/audit-deps.md) - dependencies are current, deduplicated, needed, and the best maintained option - [Test coverage audit](/quest/m1/audit-test-coverage.md) - repeat the coverage audit against the m0 baselines before the next major release diff --git a/quest/m1/audit-size.md b/quest/m1/audit-size.md index 0d4c7f3c43..06cb4efbec 100644 --- a/quest/m1/audit-size.md +++ b/quest/m1/audit-size.md @@ -22,8 +22,7 @@ follow-up through `/quest-plan`. What to check first: whether the nightly `sizes` job (`sh/size/`) has run. It needs the Linux release builds, skipped since the 2026-10-11 nixpkgs bump broke their link. If it has, its history replaces the local method below. -Then whether [moq-net compiles once](/quest/m1/net-erase-transport.md) and -[the catalog publish path compiles once](/quest/m2/catalog-publish-once.md) +Then whether [the catalog publish path compiles once](/quest/m2/catalog-publish-once.md) landed, and re-attribute the native rows if so. ### Baselines (m0 run, 2026-10-19 and 2026-10-20) @@ -54,10 +53,12 @@ Native and WASM main numbers are from 2026-10-19 (2b28f5eac); JS from (watch), 199.8 to 173.8 KB (watch-ui) and 176.8 to 151.2 KB (publish). `js/net/src/ietf/index.test.ts` fails if the entry statically reaches it again. -- Planned: [moq-net compiles once](/quest/m1/net-erase-transport.md), most of - the native growth since July (the drivers are monomorphized per transport and - per instantiating crate), and - [the catalog publish path compiles once](/quest/m2/catalog-publish-once.md), +- Fixed: moq-net boxes the transport, so its session drivers compile once + instead of per transport and per instantiating crate (PR_LINK). + Stripped profiling builds fell 31.8 to 22.5 MiB (relay), 49.1 to 42.5 MiB + (CLI), 28.0 to 21.8 MiB (`libmoq_ffi.so`), 254.6 to 207.8 MiB (unstripped + `libmoq.a`) and 862 to 786 KB gzipped (`@moq/wasm`). +- Planned: [the catalog publish path compiles once](/quest/m2/catalog-publish-once.md), about 1 MiB of the CLI. - Explained, no action: `moq-quic` replacing quinn, `kio`, `unsafe-libopus`, `moq-video`/`moq-vaapi` in the FFI library and `fast_image_resize` in diff --git a/quest/m1/net-erase-transport.md b/quest/m1/net-erase-transport.md deleted file mode 100644 index c66f52879d..0000000000 --- a/quest/m1/net-erase-transport.md +++ /dev/null @@ -1,36 +0,0 @@ -# [L] moq-net compiles its protocol code once - -## Goal - -`moq-net`'s session drivers compile once per binary, not once per transport -session type and per crate that drives them. The relay, `libmoq_ffi.so`, -`libmoq`, the CLI and `@moq/wasm` shrink by the duplicated copies, and the -relay fan-out benchmark shows no regression beyond its noise. - -## Plan - -Found by the 2026-10 size audit (main at -2b28f5eac). The drivers are generic over the session, so each backend -(QUIC/WebTransport, qmux) gets a full copy, the IETF `ControlStreamAdapter` -doubles the IETF half, and every crate that awaits a `moq-tokio` future or -calls `moq_net::Server`/`Client` with a concrete session codegens its own copy: -release builds share no generics across crates. `moq-net` is 9.1 of 23.5 MiB of -Rust text in the relay (eight copies), 6.4 of 19.2 MiB in the FFI library and -7.4 of 36.5 MiB in the CLI. Estimated saving: 5 to 6 MiB of the relay, 4 to -5 MiB of the FFI library, similar for `libmoq` and the CLI. - -Open decision (public API): `Driver`, `server::Handshake` and friends -carry the session type. - -1. Drop the type parameter: the public types take a boxed session trait object - at their constructor. Breaking, and the API gets simpler. -2. Keep the generic signatures as thin entry points that erase at once. - Non-breaking, but it is a compatibility shim the repo rules refuse. -3. Leave it, accepting the size. - -Recommendation: 1, if the dynamic dispatch prices out. Measure it first with -`bench/relay.sh` on `bench/workloads/fanout-heavy.toml` and the `moq-net` -session benchmark, before and after. The async bodies must end up -non-generic inside `moq-net`, or each awaiting crate still instantiates them. -Re-measure with the audit's method: `nm` symbol attribution by crate path and -by instantiating crate on an unstripped `profiling` build. From 988657d1b5c931823d3ccd1d07dd9c065dc54a28 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 00:51:20 -0700 Subject: [PATCH 06/12] docs: link the boxed transport PR Co-Authored-By: Claude Opus 5.5 --- quest/m1/audit-size.md | 2 +- rs/moq-net/MIGRATE.md | 2 +- rs/moq-relay/MIGRATE.md | 2 +- rs/moq-uring/MIGRATE.md | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/quest/m1/audit-size.md b/quest/m1/audit-size.md index 06cb4efbec..d47280bebe 100644 --- a/quest/m1/audit-size.md +++ b/quest/m1/audit-size.md @@ -54,7 +54,7 @@ Native and WASM main numbers are from 2026-10-19 (2b28f5eac); JS from `js/net/src/ietf/index.test.ts` fails if the entry statically reaches it again. - Fixed: moq-net boxes the transport, so its session drivers compile once - instead of per transport and per instantiating crate (PR_LINK). + instead of per transport and per instantiating crate ([#6176](https://github.com/moq-dev/moq/pull/6176)). Stripped profiling builds fell 31.8 to 22.5 MiB (relay), 49.1 to 42.5 MiB (CLI), 28.0 to 21.8 MiB (`libmoq_ffi.so`), 254.6 to 207.8 MiB (unstripped `libmoq.a`) and 862 to 786 KB gzipped (`@moq/wasm`). diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index a9c8275361..099e2cd33d 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,7 +4,7 @@ How to update your code for each breaking change, newest first. ## Unreleased -### Sessions box their transport +### Sessions box their transport (#6176) `Driver` and `server::Handshake` lose their transport type parameter: write `moq_net::Driver` and `moq_net::server::Handshake`. `Client::connect_lite`, `Server::accept_lite`, and `Server::accept_request_lite` are gone; call `connect`, `accept`, or `accept_request`, which accept every ALPN. diff --git a/rs/moq-relay/MIGRATE.md b/rs/moq-relay/MIGRATE.md index cb7c594f3e..48d55b426f 100644 --- a/rs/moq-relay/MIGRATE.md +++ b/rs/moq-relay/MIGRATE.md @@ -5,7 +5,7 @@ Rolling a cluster across a release also follows the [upgrade guide](https://moq. ## Unreleased -### `Accepted` takes any transport's handshake +### `Accepted` takes any transport's handshake (#6176) `Accepted` loses its transport type parameter, since `moq_net::server::Handshake` has none: write `Accepted` instead of `Accepted`. diff --git a/rs/moq-uring/MIGRATE.md b/rs/moq-uring/MIGRATE.md index b5a151957c..25d6764f99 100644 --- a/rs/moq-uring/MIGRATE.md +++ b/rs/moq-uring/MIGRATE.md @@ -4,7 +4,7 @@ How to update your code for each breaking change, newest first. ## Unreleased -### Wrap sessions in `transport::Local` +### Wrap sessions in `transport::Local` (#6176) moq-net dropped `connect_lite` and `accept_lite`, and every session now needs a `Send` transport. Wrap the worker's session as `transport::Local::new(transport::Session::new(conn, binding))` and call `connect` or `accept`. From 1efb8e8644705cb7706264624160024d97e65e11 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 10:43:16 -0700 Subject: [PATCH 07/12] docs(uring): name every removed _lite entry point in MIGRATE.md Co-Authored-By: Claude Opus 5.5 --- rs/moq-uring/MIGRATE.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rs/moq-uring/MIGRATE.md b/rs/moq-uring/MIGRATE.md index 25d6764f99..a7a01f45e6 100644 --- a/rs/moq-uring/MIGRATE.md +++ b/rs/moq-uring/MIGRATE.md @@ -6,8 +6,8 @@ How to update your code for each breaking change, newest first. ### Wrap sessions in `transport::Local` (#6176) -moq-net dropped `connect_lite` and `accept_lite`, and every session now needs a `Send` transport. -Wrap the worker's session as `transport::Local::new(transport::Session::new(conn, binding))` and call `connect` or `accept`. +moq-net dropped `connect_lite`, `accept_lite`, and `accept_request_lite`, and every session now needs a `Send` transport. +Wrap the worker's session as `transport::Local::new(transport::Session::new(conn, binding))` and call `connect`, `accept`, or `accept_request`. ### Timers run on moq-clock From 2bba60cbfea018486d977236f4cb6a44867495e1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 16:53:09 -0700 Subject: [PATCH 08/12] quest: point the compile-time audit at the merged moq-net change Co-Authored-By: Claude Opus 5.5 --- quest/m1/audit-compile-time.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/quest/m1/audit-compile-time.md b/quest/m1/audit-compile-time.md index adfd07acbe..471f20cf96 100644 --- a/quest/m1/audit-compile-time.md +++ b/quest/m1/audit-compile-time.md @@ -24,8 +24,8 @@ incremental). The [CI workflow audit](/quest/m1/audit-ci.md) owns job-level CI cost and caching; the [dependency audit](/quest/m1/audit-deps.md) owns deduplicating and upgrading dependencies. -What to check first: whether [moq-net compiles once](/quest/m1/net-erase-transport.md) -landed. If so, rerun `cargo llvm-lines` on the moq-tokio lib and compare the +What to check first: [moq-net compiles once](https://github.com/moq-dev/moq/pull/6176) +landed, so rerun `cargo llvm-lines` on the moq-tokio lib and compare the transport share below, then rerun the thin-LTO size comparison against the smaller archive. Then whether `cargo-llvm-lines` joined the pinned shell. @@ -136,7 +136,7 @@ Not measured: thin sizes for linux_arm64, darwin_arm64 and windows_amd64; build time above two jobs; wall-clock runtime. Decision (maintainer, 2026-10-20): keep fat LTO, and revisit once -[moq-net compiles once](/quest/m1/net-erase-transport.md) shrinks the archive. +[moq-net compiles once](https://github.com/moq-dev/moq/pull/6176) shrinks the archive. Thin saves about 40% of the final unit's wall time on two jobs, but release builds run only nightly and on release PRs, off the merge-queue gate. It costs 10 MiB of the Go archive's headroom under a hard publish limit while that @@ -145,7 +145,7 @@ builds join the PR gate. ### Results of the m0 run -- Planned: [moq-net compiles once](/quest/m1/net-erase-transport.md) removes +- Planned: [moq-net compiles once](https://github.com/moq-dev/moq/pull/6176) removes the per-transport copies, the largest attributable dev IR (43% of moq-tokio) and a share of the release fat-LTO module. - Explained, no action: the release `moq` unit is all LLVM backend work on one From f053257889e0bfd6491703c7ead42899e399be9d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 17:01:54 -0700 Subject: [PATCH 09/12] test(net): drop the stale timescale argument from the SUBSCRIBE_DROP fixture Same one-line fix as #6223; main's moq-net lib tests fail to compile without it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/subscriber.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 1ae7f9305b..8fa8173f84 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -3233,7 +3233,7 @@ mod tests { }; let broadcast = crate::broadcast::Info::new().produce(); let request = broadcast.reserve_track("video").unwrap(); - let serving = ServeLoop::new(&serve, request, crate::track::Info::timed(), Some(Timescale::MILLI)); + let serving = ServeLoop::new(&serve, request, crate::track::Info::timed()); let mut arriving = serving.serving.create_group(group::Info { sequence: 2 }).unwrap(); arriving.write_frame(crate::Timestamp::ZERO, b"2".as_slice()).unwrap(); let mut reader = broadcast From fe1293c12d6f3caf48540db0d89ff7f7cfe552cd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 02:47:40 -0700 Subject: [PATCH 10/12] fix(uring): forward wire counters through Local Every io_uring session now reaches moq-net through Local, which dropped poll::Session::wire(), so relay sessions lost their wire byte counters. uring_sessions_supply_direct_wire_counters fails without this. Co-Authored-By: Claude Opus 5.5 --- rs/moq-uring/src/transport/local.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/rs/moq-uring/src/transport/local.rs b/rs/moq-uring/src/transport/local.rs index fb6b615889..946b862435 100644 --- a/rs/moq-uring/src/transport/local.rs +++ b/rs/moq-uring/src/transport/local.rs @@ -101,6 +101,10 @@ impl poll::Session for Local { fn stats(&self) -> impl moq::Stats { self.0.stats() } + + fn wire(&self) -> Option { + self.0.wire() + } } impl poll::SendStream for Local { From 08e41c8770e22e7d528c5f54c203178edd531f1e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 13:39:46 -0700 Subject: [PATCH 11/12] quest: drop the link to the finished net-erase-transport quest Co-Authored-By: Claude Opus 5.5 --- quest/m2/ietf-adapter-session-errors.md | 4 ---- 1 file changed, 4 deletions(-) diff --git a/quest/m2/ietf-adapter-session-errors.md b/quest/m2/ietf-adapter-session-errors.md index 597ca50a20..971be36943 100644 --- a/quest/m2/ietf-adapter-session-errors.md +++ b/quest/m2/ietf-adapter-session-errors.md @@ -29,7 +29,3 @@ ProtocolViolation while a uni or datagram poll is pending, and the client must see a terminal error and not redial. Public API: none. Wire: none. - -## Related - -- [moq-net compiles once](/quest/m1/net-erase-transport.md) - its open PR [moq-dev/moq#6176](https://github.com/moq-dev/moq/pull/6176) touches the same adapter From fe3816e9155e671497159b0029c5f28e0d47cadf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 22 Oct 2026 14:32:11 -0700 Subject: [PATCH 12/12] fix(net): carry streams_evicted through the boxed transport stats Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/transport/boxed.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/rs/moq-net/src/transport/boxed.rs b/rs/moq-net/src/transport/boxed.rs index 1051e8851b..e4272edab1 100644 --- a/rs/moq-net/src/transport/boxed.rs +++ b/rs/moq-net/src/transport/boxed.rs @@ -112,6 +112,7 @@ pub(crate) struct Stats { packets_lost: Option, rtt: Option, estimated_send_rate: Option, + streams_evicted: Option, } impl Stats { @@ -125,6 +126,7 @@ impl Stats { packets_lost: stats.packets_lost(), rtt: stats.rtt(), estimated_send_rate: stats.estimated_send_rate(), + streams_evicted: stats.streams_evicted(), } } } @@ -154,6 +156,9 @@ impl super::Stats for Stats { fn estimated_send_rate(&self) -> Option { self.estimated_send_rate } + fn streams_evicted(&self) -> Option { + self.streams_evicted + } } trait ErasedSession: MaybeSend + MaybeSync {