diff --git a/quest/m1/session-policy/README.md b/quest/m1/session-policy/README.md index e06bf901738..b632fb325f9 100644 --- a/quest/m1/session-policy/README.md +++ b/quest/m1/session-policy/README.md @@ -19,10 +19,10 @@ where moq-webrtc should plug its externally established transport in. The maintainer asked what that means for io_uring and how much of moq-tokio belongs in moq-net, and decided: -- ✅ **Move the policy first**, then give moq-tokio thin handoffs; the - [WebRTC rename](/quest/m2/webrtc-rename.md) waits on - [moq-tokio becomes a thin shell](/quest/m1/session-policy/tokio-handoff.md) - (#7804 closed unmerged). +- ✅ **Move the policy first**, then give moq-tokio thin handoffs + (`Listener::accept_with`, `Client::connect_with`), which the + [WebRTC rename](/quest/m2/webrtc-rename.md) builds on (#7804 closed + unmerged). Not chosen: a moq-tokio handoff now with the move later; moq-webrtc calling moq-net directly with its own policy. - ✅ **Reconnect is in scope**: the reconnect, backoff and GOAWAY candidate @@ -66,4 +66,3 @@ Research (2026-11-04, `origin/main` at e6be13d5f8): ## Required - [The io_uring relay path uses moq-net's policy](/quest/m1/session-policy/uring.md) - deletes `uring.rs`'s copy -- [moq-tokio becomes a thin shell](/quest/m1/session-policy/tokio-handoff.md) - sockets, TLS, accept loops and spawning, plus the transport handoffs WebRTC needs diff --git a/quest/m1/session-policy/tokio-handoff.md b/quest/m1/session-policy/tokio-handoff.md deleted file mode 100644 index b0b9946ea59..00000000000 --- a/quest/m1/session-policy/tokio-handoff.md +++ /dev/null @@ -1,36 +0,0 @@ -# [S] moq-tokio becomes a thin shell - -## Goal - -moq-tokio keeps only sockets, DNS, TLS, accept loops, the fallback runtime and -spawning, and calls moq-net for all session policy. It gains two public -handoffs for transports established elsewhere: `accept_with(transport, link)` -on the listener and `connect_with(transport)` on the client, which run the -moq-net accept or connect policy over an already-connected session and return -the usual request or session. moq-tokio's own `webrtc` module goes through -them, so the [WebRTC rename](/quest/m2/webrtc-rename.md) moves it into -moq-webrtc without reaching into moq-tokio's internals. - -## Plan - -Part of [Session policy lives in moq-net](/quest/m1/session-policy/README.md); -see it for the decisions and research. This is what the -[WebRTC rename](/quest/m2/webrtc-rename.md) waits on -([moq-dev/moq#7804](https://github.com/moq-dev/moq/pull/7804), which first -asked, closed unmerged): the maintainer decided on -2026-11-04 to move the policy first, so the handoff is a few lines over -moq-net rather than a copy of the policy. - -- The handoff's signature (transport plus link metadata in, request or session - out) is fixed by this quest and does not change if policy moves further. -- Delete moq-tokio's now-unused `Backoff` and the like (the accept `Deadline` already went), and its - per-transport `Request` construction that moq-net now owns. Bounding the - transport handshake (`within`) stays here, since moq-net has no session yet - at that point. -- Its consumer is the WebRTC transport: moq-tokio's `webrtc` module here, and - moq-webrtc once the rename moves it, so the - [API audit](/quest/m1/api-audit/README.md) rule (a new public surface needs a - consumer) is met. - -Public API: moq-tokio adds `accept_with` and `connect_with` and drops internals -it re-exported, with MIGRATE notes. Wire: none. diff --git a/quest/m2/webrtc-rename.md b/quest/m2/webrtc-rename.md index 42b0379b35d..f0504dd4bf3 100644 --- a/quest/m2/webrtc-rename.md +++ b/quest/m2/webrtc-rename.md @@ -11,17 +11,13 @@ Decided 2026-10-30 with the maintainer (planning interview); see the [epic](/que Milestone: ✅ m1 (moved from m2, maintainer, 2026-11-03). Not chosen: keep it in m2, or send [moq-rtc on rind](/quest/m1/rind/rtc.md) back to m2. [Vendor and trim the WebRTC stack](/quest/m1/webrtc-vendor.md) requires this, and m1's rind epic requires the vendoring, so this sits at the m2 top level. It stays a child of the m3 [WebRTC epic](/quest/m3/webrtc/README.md). - Move `rs/moq-rtc` to `rs/moq-webrtc` and merge `rs/moq-ice` in as a module. Keep the gateway's `server` feature; add a feature for the data-channel transport so WHIP/WHEP-only builds skip SCTP. -- Move moq-tokio's `webrtc` module (`rs/moq-tokio/src/webrtc.rs`, host gathering, `.local` resolution) into moq-webrtc, built on moq-tokio's public session types. Its connect and accept go through moq-tokio's `connect_with`/`accept_with` handoffs, which run moq-net's session policy over the established data channel. +- Move moq-tokio's `webrtc` module (`rs/moq-tokio/src/webrtc.rs`, host gathering, `.local` resolution) into moq-webrtc, built on moq-tokio's public session types. Its connect and accept go through moq-tokio's `connect_with`/`accept_with` handoffs, which run moq-net's session policy over the established data channel. The handoffs take the qmux session and bound only the MoQ handshake from the call; bounding ICE and the channel coming up (moq-tokio's wrappers use the listen and connect timeouts today) moves with the transport, so moq-webrtc needs its own establishment bound. - Boundary decided 2026-11-04 (maintainer, from [moq-dev/moq#7804](https://github.com/moq-dev/moq/pull/7804)): ✅ move session policy into moq-net first ([Session policy lives in moq-net](/quest/m1/session-policy/README.md)), then this rename uses the thin handoffs. Not chosen: a moq-tokio handoff over today's policy, or moq-webrtc calling moq-net with its own policy. Removing SCTP stays with [Vendor and trim the WebRTC stack](/quest/m1/webrtc-vendor.md). - Update moq-cli's `rtc` subcommand, the bindings or examples that use the transport, `test/p2p`, and every doc and quest naming the old crates. - Publishing: the last `moq-rtc`/`moq-ice` releases stay on crates.io; `doc/setup/upgrade.md` says where things moved. Version numbers are the maintainer's. Public API: breaking (crate rename; `moq_tokio::webrtc` moves). Wire: none. -## Required - -- [moq-tokio becomes a thin shell](/quest/m1/session-policy/tokio-handoff.md) - the `accept_with`/`connect_with` handoffs moq-webrtc uses - ## Related - [moq-rtc REMB and TIAS](/rs/moq-rtc/src/server/whip.rs) - #7373 added the capped answer in moq-rtc; carry it through the move diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index 4a6b1daaa83..de61056a724 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -322,40 +322,57 @@ impl Client { Some(self.with_subscriber(origin).connect(url)) } - /// Run a MoQ client session over the data channel an [`crate::webrtc::Offerer`] opened. + /// Run a MoQ client session over a qmux session established outside this client, such + /// as over a WebRTC data channel. /// - /// Waits for ICE and the channel to come up, bounded by the connect timeout. A - /// peer link cannot be redialed, so there is no reconnect: the session ends with - /// the peer connection. - #[cfg(feature = "webrtc")] - pub async fn connect_webrtc(&self, peer: &crate::webrtc::Offerer) -> crate::Result { + /// The connect timeout bounds the MoQ handshake from this call. A transport handed + /// over cannot be redialed, so there is no reconnect: the session ends with it. + #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] + pub async fn connect_with(&self, session: qmux::Session) -> crate::Result { + let timeout = self.policy.timeout; + let connect = connect_session(&self.moq, session); crate::runtime::scope(async move { - let protocol = peer.protocol(); - if !self.versions.alpns().contains(&protocol) { - return Err(crate::webrtc::Error::Protocol(protocol.to_string()).into()); + match timeout.is_zero() { + true => Ok(connect.await?), + false => Ok(kio::clock::timeout(timeout, connect) + .await + .map_err(|_| Error::ConnectTimeout(timeout))??), } + }) + .await + } - let attempt = async { - let opened = peer.channel().await?; - let config = opened.config; - let session = qmux::Session::connect(opened.channel, config) - .await - .map(crate::spawn_qmux) - .map_err(crate::webrtc::Error::from)?; - Ok::<_, Error>(connect_session(&self.moq, session).await?) - }; + /// Run a MoQ client session over the data channel an [`crate::webrtc::Offerer`] opened. + /// + /// The connect timeout bounds ICE and the channel coming up, then + /// [`Self::connect_with`] bounds the MoQ handshake. A peer link cannot be redialed, so + /// there is no reconnect: the session ends with the peer connection. + #[cfg(feature = "webrtc")] + pub async fn connect_webrtc(&self, peer: &crate::webrtc::Offerer) -> crate::Result { + let protocol = peer.protocol(); + if !self.versions.alpns().contains(&protocol) { + return Err(crate::webrtc::Error::Protocol(protocol.to_string()).into()); + } - let timeout = self.policy.timeout; - let session = match timeout.is_zero() { - true => attempt.await?, - false => kio::clock::timeout(timeout, attempt) + let establish = async { + let opened = peer.channel().await?; + let session = qmux::Session::connect(opened.channel, opened.config).await?; + Ok::<_, crate::webrtc::Error>(crate::spawn_qmux(session)) + }; + let timeout = self.policy.timeout; + let session = crate::runtime::scope(async move { + match timeout.is_zero() { + true => Ok(establish.await?), + false => kio::clock::timeout(timeout, establish) .await - .map_err(|_| Error::ConnectTimeout(timeout))??, - }; - tracing::info!(version = %session.version(), transport = %crate::Transport::WebRtc, "connected"); - Ok(session) + .map_err(|_| Error::ConnectTimeout(timeout))? + .map_err(Error::from), + } }) - .await + .await?; + let session = self.connect_with(session).await?; + tracing::info!(version = %session.version(), transport = %crate::Transport::WebRtc, "connected"); + Ok(session) } /// Dial one attempt of a [`Connection`]'s policy and complete the MoQ handshake. diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 891e99161d3..dd105872cab 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -727,11 +727,11 @@ impl Server { // (like the stream bindings). let Accepted { session, url, identity, authority, mut link } = within(deadline, crate::quic::accept(_conn, alpns, tls)).await?; link.local = local; - let quic = Some(session.connection().clone()); - let request = setup.server.accept_request(moq_net::time::Instant::now(), session).with_deadline(deadline).await?; + let under = Under::Quic(session.connection().clone()); // 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: Box::new(request), quic, quic_limits: Default::default(), #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] qmux: None, active }) + let incoming = Incoming { transport, url, link, authority, identity, deadline, active }; + Request::accept(&setup.server, session, under, incoming).await }.boxed()).boxed()); } } @@ -850,62 +850,59 @@ impl Listener { crate::runtime::scope(self.server.accept_next()).await } + /// Read the MoQ SETUP on a qmux session established outside this listener, such as + /// over a WebRTC data channel, leaving admission to the returned [`Request`]. + /// + /// `transport` and `link` are what the request reports about the connection. The + /// accept deadline counts from this call and bounds SETUP, admission, and the reply; + /// establishing the session is the caller's to bound. Returns [`Error::Stopped`] once + /// the listener is closing. + #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] + pub async fn accept_with( + &self, + transport: Transport, + session: qmux::Session, + link: Link, + ) -> crate::Result { + let server = &self.server; + let active = server.activity.enter().ok_or(Error::Stopped)?; + let mut incoming = Incoming::new(transport, server.moq.deadline(kio::clock::now()), active); + incoming.link = link; + crate::runtime::scope(Request::accept_qmux(&server.moq, session, incoming)).await + } + /// Read the MoQ SETUP from the data channel the remote peer opened on an /// [`crate::webrtc::Answerer`], leaving admission to the returned [`Request`]. /// - /// Waits for ICE and the channel to come up, bounded by the handshake timeout. - /// The channel must carry the answerer's protocol, a MoQ ALPN this server - /// speaks. Returns [`Error::Stopped`] once the listener is closing. + /// The handshake timeout bounds ICE and the channel coming up, then + /// [`Self::accept_with`] bounds the MoQ handshake. The channel must carry the + /// answerer's protocol, a MoQ ALPN this server speaks. Returns [`Error::Stopped`] once + /// the listener is closing. #[cfg(feature = "webrtc")] pub async fn accept_webrtc(&self, peer: &crate::webrtc::Answerer) -> crate::Result { - crate::runtime::scope(async move { - use crate::webrtc::Error as WebRtcError; - - let server = &self.server; - let active = server.activity.enter().ok_or(Error::Stopped)?; - if !server.versions.alpns().contains(&peer.protocol()) { - return Err(WebRtcError::Protocol(peer.protocol().to_string()).into()); + use crate::webrtc::Error as WebRtcError; + + // Counted toward the drain while establishing; a drain admits `accept_with`'s own + // guard while this one is held, so the connection is tracked throughout. + let _establishing = self.server.activity.enter().ok_or(Error::Stopped)?; + if !self.server.versions.alpns().contains(&peer.protocol()) { + return Err(WebRtcError::Protocol(peer.protocol().to_string()).into()); + } + let deadline = self.server.moq.deadline(kio::clock::now()); + let establish = async { + let opened = peer.channel().await?; + if opened.protocol != peer.protocol() { + return Err(WebRtcError::Protocol(opened.protocol)); } - let deadline = server.moq.deadline(kio::clock::now()); - - let handshake = async { - let opened = peer.channel().await?; - if opened.protocol != peer.protocol() { - return Err(WebRtcError::Protocol(opened.protocol)); - } - let config = opened.config; - let protocol = opened.protocol; - let session = crate::spawn_qmux(qmux::Session::accept(opened.channel, config).await?); - Ok((session, protocol)) - }; - let (session, protocol) = within(deadline, handshake).await?; - - let qmux = session.clone(); - let request = server - .moq - .accept_request(moq_net::time::Instant::now(), session) - .with_deadline(deadline) - .await?; - Ok(Request { - transport: Transport::WebRtc, - url: None, - authority: None, - identity: None, - link: Link { - alpn: Some(protocol), - ..Default::default() - }, - handshake: Box::new(request), - #[cfg(feature = "quic")] - quic: None, - #[cfg(feature = "_transport")] - quic_limits: Default::default(), - #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] - qmux: Some(qmux), - active, - }) - }) - .await + let session = crate::spawn_qmux(qmux::Session::accept(opened.channel, opened.config).await?); + Ok((session, opened.protocol)) + }; + let (session, protocol) = crate::runtime::scope(within(deadline, establish)).await?; + let link = Link { + alpn: Some(protocol), + ..Default::default() + }; + self.accept_with(Transport::WebRtc, session, link).await } /// Close every listener and connection, once each QUIC peer has been sent the @@ -988,28 +985,17 @@ pub(crate) async fn accept_websocket( options: crate::websocket::Options, deadline: Option, ) -> crate::Result { - let qmux = session.clone(); - let request = options - .server - .accept_request(moq_net::time::Instant::now(), session) - .with_deadline(deadline) - .await?; let authority = options.url.host_str().filter(|h| !h.is_empty()).map(str::to_owned); - Ok(Request { + let incoming = Incoming { transport: Transport::WebSocket, url: Some(options.url), + link: options.link, authority, identity: options.identity, - link: options.link, - handshake: Box::new(request), - #[cfg(feature = "quic")] - quic: None, - #[cfg(feature = "_transport")] - quic_limits: Default::default(), - #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] - qmux: Some(qmux), + deadline, active: options.active, - }) + }; + Request::accept_qmux(&options.server, session, incoming).await } /// Finish a transport handshake by `deadline`, or fail with a timeout. @@ -1377,20 +1363,7 @@ fn spawn_unix_loop( handshakes.spawn(crate::panic::isolate(async move { match within(deadline, pending).await { Ok((session, _cred)) => { - stream_request( - session, - &setup, - tx, - Incoming { - transport: Transport::Unix, - link: Link::default(), - authority: None, - identity: None, - deadline, - active, - }, - ) - .await; + stream_request(session, &setup, tx, Incoming::new(Transport::Unix, deadline, active)).await; } // Per-connection, as in `spawn_tcp_loop`. Err(err) => tracing::warn!(%err, "unix qmux handshake failed"), @@ -1400,14 +1373,54 @@ fn spawn_unix_loop( }) } -#[cfg(any(feature = "tcp", all(feature = "uds", unix)))] +/// What a transport learned while accepting a connection, before moq-net reads its SETUP: +/// everything a [`Request`] reports besides the handshake itself. +#[cfg(feature = "_transport")] pub(crate) struct Incoming { - transport: Transport, - link: Link, - authority: Option, - identity: Option, - deadline: Option, - active: moq_net::server::Active, + pub transport: Transport, + /// The request URL, on the transports that carry one (WebTransport, WebSocket). + pub url: Option, + pub link: Link, + pub authority: Option, + pub identity: Option, + /// When the whole accept must finish; see [`moq_net::Server::deadline`]. + pub deadline: Option, + pub active: moq_net::server::Active, +} + +#[cfg(any(test, feature = "websocket", feature = "tcp", feature = "webrtc"))] +impl Incoming { + /// A connection with no facts beyond its transport. + pub(crate) fn new( + transport: Transport, + deadline: Option, + active: moq_net::server::Active, + ) -> Self { + Self { + transport, + url: None, + link: Link::default(), + authority: None, + identity: None, + deadline, + active, + } + } +} + +/// The connection [`Request::ok`] applies [`Request::with_quic_limits`] to. +#[cfg(feature = "_transport")] +pub(crate) enum Under { + /// A transport moq-tokio cannot widen beyond what moq-net's `set_limits` reaches. + #[cfg(all(test, feature = "websocket"))] + Opaque, + /// A raw QUIC or WebTransport connection. + #[cfg(feature = "quic")] + Quic(crate::quic::driver::connection::Connection), + /// A qmux session: WebSocket, `tcp://`, `tls://`, `unix://`, or one handed over with + /// [`Listener::accept_with`]. + #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] + Qmux(qmux::Session), } #[cfg(feature = "tcp")] @@ -1428,6 +1441,7 @@ impl Incoming { .map(|(tls, chain)| tls.peer(chain.to_vec())); Self { transport: Transport::Tcp, + url: None, link: Link { remote, local, @@ -1458,7 +1472,7 @@ async fn stream_request( tx: tokio::sync::mpsc::Sender, incoming: Incoming, ) { - match accept_stream(&setup.server, session, incoming).await { + match Request::accept_qmux(&setup.server, session, incoming).await { Ok(request) => { let _ = tx.send(request).await; } @@ -1466,43 +1480,6 @@ async fn stream_request( } } -/// Read SETUP on a stream session, which carries its path in SETUP rather than a URL. -#[cfg(any(feature = "tcp", all(feature = "uds", unix)))] -pub(crate) async fn accept_stream( - server: &moq_net::Server, - session: qmux::Session, - incoming: Incoming, -) -> crate::Result { - let Incoming { - transport, - link, - authority, - identity, - deadline, - active, - } = incoming; - let qmux = session.clone(); - let request = server - .accept_request(moq_net::time::Instant::now(), session) - .with_deadline(deadline) - .await?; - Ok(Request { - transport, - url: None, - authority, - identity, - link, - handshake: Box::new(request), - #[cfg(feature = "quic")] - quic: None, - #[cfg(feature = "_transport")] - quic_limits: Default::default(), - #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] - qmux: Some(qmux), - active, - }) -} - /// 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. @@ -1556,19 +1533,58 @@ pub struct Request { link: Link, /// The paused moq-net handshake; the same type whatever the transport. handshake: Box, - /// The connection under a QUIC or WebTransport request, for [`Request::quic`]. - #[cfg(feature = "quic")] - quic: Option, - /// What [`Request::ok`] applies to the QUIC connection or qmux session before answering SETUP. - #[cfg(feature = "_transport")] + /// The connection [`Request::ok`] widens to [`Request::with_quic_limits`]. + under: Under, + /// What [`Request::ok`] applies to the connection before answering SETUP. quic_limits: moq_quic::Limits, - /// The session under a WebSocket, `tcp://`, `tls://`, `unix://`, or WebRTC request, which - /// [`Request::ok`] widens to [`Request::with_quic_limits`]. - #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] - qmux: Option, } impl Request { + /// Read SETUP on `session` by the accept deadline, the one path every transport's + /// request takes once its connection is established. + pub(crate) async fn accept( + server: &moq_net::Server, + session: S, + under: Under, + incoming: Incoming, + ) -> crate::Result { + let Incoming { + transport, + url, + link, + authority, + identity, + deadline, + active, + } = incoming; + let handshake = server + .accept_request(moq_net::time::Instant::now(), session) + .with_deadline(deadline) + .await?; + Ok(Self { + active, + transport, + url, + authority, + identity, + link, + handshake: Box::new(handshake), + under, + quic_limits: Default::default(), + }) + } + + /// [`Self::accept`] over a qmux session. + #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] + pub(crate) async fn accept_qmux( + server: &moq_net::Server, + session: qmux::Session, + incoming: Incoming, + ) -> crate::Result { + let under = Under::Qmux(session.clone()); + Self::accept(server, session, under, incoming).await + } + /// Reject the session. The transport is already accepted, so this closes the /// just-established MoQ session with `code` rather than answering the transport handshake. pub async fn reject(self, code: moq_net::SessionError) -> crate::Result<()> { @@ -1612,7 +1628,11 @@ impl Request { /// `None` on a transport moq-quic does not drive (qmux streams). #[cfg(feature = "quic")] pub fn quic(&self) -> Option<&crate::quic::driver::connection::Connection> { - self.quic.as_ref() + match &self.under { + Under::Quic(quic) => Some(quic), + #[allow(unreachable_patterns, reason = "a QUIC-only build has no other connection")] + _ => None, + } } /// Set the QUIC limits [`Self::ok`] applies before answering SETUP, replacing any set @@ -1624,7 +1644,6 @@ impl Request { /// window. A qmux session starts at the `[quic]` windows and takes the windows and stream /// count set here, except a WebRTC session's receive window, which its data channel fixes; /// it never paces, so `ok` refuses a rate cap on one. - #[cfg(feature = "_transport")] pub fn with_quic_limits(mut self, limits: moq_quic::Limits) -> Result { limits.validate()?; self.quic_limits = limits; @@ -1653,7 +1672,10 @@ impl Request { /// The receive window of the qmux session under this request. #[cfg(all(test, any(feature = "websocket", feature = "tcp")))] pub(crate) fn qmux_window(&self) -> moq_net::transport::ReceiveWindow { - let qmux = self.qmux.as_ref().expect("a qmux session"); + #[allow(irrefutable_let_patterns, reason = "a qmux-only build has no other connection")] + let Under::Qmux(qmux) = &self.under else { + panic!("a qmux session"); + }; moq_net::transport::poll::Session::receive_window(qmux).expect("qmux reports its window") } @@ -1665,27 +1687,23 @@ impl Request { /// flow-control credit, which keep-alives would otherwise hold open. pub async fn ok(self) -> crate::Result { // The caller's limits go first; moq-net opens the receive window after them. - #[cfg(feature = "_transport")] - let (limits, window) = { - let mut limits = self.quic_limits; - let window = limits.receive_window.take(); - (limits, window) - }; - #[cfg(feature = "quic")] - if let Some(quic) = &self.quic { - quic.set_limits(limits).expect("validated by with_quic_limits"); - } - #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] - if let Some(qmux) = &self.qmux - && let Err(err) = qmux.set_limits(limits) - { - self.handshake.close(moq_net::SessionError::Internal.into()); - return Err(crate::Error::QmuxLimits(crate::error::message(err))); + let mut limits = self.quic_limits; + let window = limits.receive_window.take(); + match &self.under { + #[cfg(all(test, feature = "websocket"))] + Under::Opaque => {} + #[cfg(feature = "quic")] + Under::Quic(quic) => quic.set_limits(limits).expect("validated by with_quic_limits"), + #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] + Under::Qmux(qmux) => { + if let Err(err) = qmux.set_limits(limits) { + self.handshake.close(moq_net::SessionError::Internal.into()); + return Err(crate::Error::QmuxLimits(crate::error::message(err))); + } + } } let Request { handshake, active, .. } = self; - #[allow(unused_mut, reason = "only a transport feature sets the window")] let mut handshake = *handshake; - #[cfg(feature = "_transport")] if let Some(window) = window { handshake = handshake.with_receive_window(window); } @@ -2508,29 +2526,14 @@ mod tests { server: moq_net::Server::new().with_timeout(Some(timeout)), }; let start = kio::clock::now(); - let accept = setup.server.accept_request(moq_net::time::Instant::now(), peer.clone()); - let Ok(request) = accept.with_deadline(setup.deadline()).await else { + let incoming = Incoming::new(Transport::WebSocket, setup.deadline(), setup.activity.enter().unwrap()); + let Ok(request) = Request::accept(&setup.server, peer.clone(), Under::Opaque, incoming).await else { panic!("a readable SETUP was refused"); }; assert_eq!(start.elapsed(), std::time::Duration::ZERO); let registry = moq_net::stats::Registry::new(Default::default()); - let request = Request { - active: setup.activity.enter().unwrap(), - transport: Transport::WebSocket, - url: None, - authority: None, - identity: None, - link: Link::default(), - handshake: Box::new(request), - #[cfg(feature = "quic")] - quic: None, - #[cfg(feature = "_transport")] - quic_limits: Default::default(), - #[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] - qmux: None, - } - .with_stats( + let request = request.with_stats( registry .tier(Default::default()) .session(moq_net::Path::from_static("room")), diff --git a/rs/moq-tokio/src/tcp.rs b/rs/moq-tokio/src/tcp.rs index fa1b9e981c6..edd72ed605a 100644 --- a/rs/moq-tokio/src/tcp.rs +++ b/rs/moq-tokio/src/tcp.rs @@ -319,7 +319,7 @@ impl TlsAcceptor { let session = crate::server::within(deadline, handshake).await?; let incoming = crate::server::Incoming::tcp(&session, Some(&self.tls), remote, local, deadline, active) .with_alpn(alpn); - crate::server::accept_stream(&self.server, session, incoming).await + crate::server::Request::accept_qmux(&self.server, session, incoming).await }) .await } diff --git a/rs/moq-tokio/src/webrtc/tests.rs b/rs/moq-tokio/src/webrtc/tests.rs index 9a96c99ab8b..ff488b79f5c 100644 --- a/rs/moq-tokio/src/webrtc/tests.rs +++ b/rs/moq-tokio/src/webrtc/tests.rs @@ -43,6 +43,7 @@ async fn transfer(offerer: &Offerer, answerer: &Answerer, payload: &[u8]) { let server = async { let request = listener.accept_webrtc(answerer).await.unwrap(); assert_eq!(request.transport(), crate::Transport::WebRtc); + assert_eq!(request.alpn(), Some(answerer.protocol())); request.with_publisher(&origin).ok().await.unwrap() }; @@ -386,3 +387,46 @@ async fn trickled_candidates_wait_for_the_driver() { second.await.unwrap(); assert!(matches!(driver.recv().await, Some(Command::Candidate(_)))); } + +/// A drain that starts while the channel is still coming up waits for the connection, +/// which is admitted rather than refused once its MoQ handshake starts. +#[tokio::test] +async fn a_drain_waits_for_an_establishing_peer() { + let config = Config::default(); + let (offerer, offer) = Offerer::with_hosts(&config, protocol(), vec![LOOPBACK]).await.unwrap(); + let (answerer, answer) = Answerer::with_hosts(&config, &offer, protocol(), vec![LOOPBACK]) + .await + .unwrap(); + + let activity = moq_net::server::Activity::default(); + let listener = crate::server::Config { + activity: activity.clone(), + ..Default::default() + } + .init_streams() + .unwrap() + .listen() + .await + .unwrap(); + let mut accept = std::pin::pin!(listener.accept_webrtc(&answerer)); + // The answer is not applied yet, so the channel cannot open. + assert!(futures::poll!(accept.as_mut()).is_pending()); + let mut closed = std::pin::pin!(activity.close()); + assert!( + futures::poll!(closed.as_mut()).is_pending(), + "the drain skipped the establishing peer" + ); + + offerer.accept(&answer).await.unwrap(); + let client = crate::client::Config::default().init().unwrap(); + let (request, session) = + kio::clock::timeout(TIMEOUT, async { tokio::join!(accept, client.connect_webrtc(&offerer)) }) + .await + .expect("handshake timed out"); + let request = request.expect("the drain refused an establishing peer"); + assert!(futures::poll!(closed.as_mut()).is_pending()); + drop((request, session)); + kio::clock::timeout(TIMEOUT, closed) + .await + .expect("the drain never finished"); +}