diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 4ada2c2a1c..d0ffb4a7eb 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -3832,7 +3832,7 @@ class FfiConverterMoqVideoFormat { } } -enum MoqTransport { quic, iroh, webSocket, tcp, unix } +enum MoqTransport { quic, iroh, webSocket, tcp, unix, webTransport } class FfiConverterMoqTransport { static LiftRetVal read(Uint8List buf) { @@ -3848,6 +3848,8 @@ class FfiConverterMoqTransport { return LiftRetVal(MoqTransport.tcp, 4); case 5: return LiftRetVal(MoqTransport.unix, 4); + case 6: + return LiftRetVal(MoqTransport.webTransport, 4); default: throw UniffiInternalError( UniffiInternalError.unexpectedEnumCase, diff --git a/doc/concept/transport.md b/doc/concept/transport.md index 14c74e7005..8bb8e802f4 100644 --- a/doc/concept/transport.md +++ b/doc/concept/transport.md @@ -42,7 +42,8 @@ connection (`wss://`) and keep whichever wins. A small multiplexer, [qmux](/draft/qmux-websocket), carries MoQ streams over the socket. It works everywhere but cannot escape TCP head-of-line blocking, so priority and resets only help once bytes leave the TCP queue. The fallback is automatic in every -client; the relay enables it with `[web.https]`. +client; the relay enables it with `[web.https]`. A native Rust client that lands on +WebSocket keeps dialing QUIC and moves the session onto it if that dial completes. ## Raw QUIC (native) diff --git a/doc/lib/rs/index.md b/doc/lib/rs/index.md index 2a02438809..3dfbb96ef9 100644 --- a/doc/lib/rs/index.md +++ b/doc/lib/rs/index.md @@ -83,7 +83,9 @@ URLs may be `https://` (WebTransport, with raw QUIC preferred for native), `moql://`/`moqt://` (raw QUIC), or `iroh://`. A `?jwt=` query carries the token. `http://` is for a relay on localhost only: it fetches the certificate fingerprint unauthenticated before upgrading, so never send a token over it. Connections race -QUIC against WebSocket and remember which won. +QUIC against WebSocket and remember which won. When WebSocket wins, a `Connection` keeps +dialing QUIC and moves onto it once it lands, handing live tracks over at a group boundary +and draining the WebSocket session; `Connection::transport()` reports which is live. The `Default::default()` above is the QUIC transport section, and the same value serves a dial and a listener: diff --git a/go/wrapper/server.go b/go/wrapper/server.go index 02f668f116..108e51037d 100644 --- a/go/wrapper/server.go +++ b/go/wrapper/server.go @@ -22,6 +22,8 @@ const ( TransportTCP = ffi.MoqTransportTcp // TransportUnix is a session that arrived over a Unix domain socket. TransportUnix = ffi.MoqTransportUnix + // TransportWebTransport is a session that arrived over WebTransport (HTTP/3). + TransportWebTransport = ffi.MoqTransportWebTransport ) // Request is an incoming session that can be accepted (Accept) or rejected (Reject). diff --git a/quest/m1/transport-upgrade/README.md b/quest/m1/transport-upgrade/README.md index 7e88b12e6f..833fa9cd13 100644 --- a/quest/m1/transport-upgrade/README.md +++ b/quest/m1/transport-upgrade/README.md @@ -40,12 +40,13 @@ requires it. Shared decisions: - The race returns the winner plus the still-pending QUIC dial when WebSocket - wins. The QUIC handshake timeout bounds that dial; no extra deadline. + wins. The deadline that already bounds the race attempt (the connect + timeout in Rust) bounds that dial too; no extra deadline. - On a successful upgrade the "WebSocket won" memo (`WEBSOCKET_WON` in `moq-tokio`, `websocketWon` in `js/net`) forgets the URL: QUIC works on this network, so the head start comes back. Otherwise a network where WebSocket narrowly beats QUIC would open two connections on every reconnect. -- The old session gets `Goaway::same()` with the configured handover cap before +- The old session gets `Goaway::new()` with the configured handover cap before it enters draining. The relay refuses new requests on it from then on; the splice ends its subscriptions at the boundary. - One-shot `connect()` returns one session and never upgrades; every @@ -57,7 +58,6 @@ Shared decisions: ## Quests -- [Rust](/quest/m1/transport-upgrade/rust.md) - moq-tokio keeps the QUIC dial after WebSocket wins and migrates through the existing Draining path - [JavaScript](/quest/m1/transport-upgrade/js.md) - js/net keeps the WebTransport dial after WebSocket wins and migrates through the client-goaway handover ## Related diff --git a/quest/m1/transport-upgrade/rust.md b/quest/m1/transport-upgrade/rust.md deleted file mode 100644 index e0d05e29b5..0000000000 --- a/quest/m1/transport-upgrade/rust.md +++ /dev/null @@ -1,49 +0,0 @@ -# [M] Rust WebSocket to QUIC upgrade - -## Goal - -A `moq_tokio::Connection` that came up over the WebSocket fallback migrates to -QUIC once the QUIC dial completes: live tracks hand over at a group boundary, -the WebSocket session receives a GOAWAY and drains within the handover cap, -`Status::Migrating` is observable across the swap, and the "WebSocket won" -memo forgets the URL. When QUIC wins the race nothing changes. - -## Plan - -Lands in `rs/moq-tokio`. See the -[questline](/quest/m1/transport-upgrade/README.md) for the shared decisions. - -- `race_moq_connect` (`rs/moq-tokio/src/client.rs`) currently drops the losing - arm. When WebSocket wins, return the handshaken WebSocket session together - with the pending QUIC future; when QUIC wins or the QUIC arm has already - failed, return the session alone. `Client::dial` keeps its one-session shape - for the one-shot `connect()` path; give `Connection` the variant that carries - the pending upgrade. -- `Connection::run` (`rs/moq-tokio/src/connection.rs`) polls the pending QUIC - dial alongside `run_session`, the way it already polls `draining`. When it - completes, run the MoQ handshake on it with the same `moq_net::Client` (the - origins attach and the front reselects to the newest route), call - `shared.migrating()` then `shared.connected(&new)`, send - `old.drain().send(Goaway::same().timeout(handover))` with the configured - `connection::Goaway` cap, and move the old session into `Draining`. A QUIC dial - that fails after WebSocket won is logged at debug, leaves the memo untouched, - and the session carries on over WebSocket. The pending dial is dropped when the WebSocket session ends - first; the reconnect races again. -- On a successful upgrade remove the URL from `WEBSOCKET_WON` - (`rs/moq-tokio/src/websocket.rs`). -- Add `moq_tokio::Transport` (QUIC, WebTransport, WebSocket, TCP, Unix) and - `Connection::transport()` returning it for the live session, so telemetry can - see the swap. It lives in `moq-tokio`, not `moq_net`: the session is generic - over the transport trait and cannot know what it runs on, only the dial - does. Mirrors `js/net`'s `transportOf`. -- Regression test against the in-tree relay: the QUIC path goes through an - in-process UDP forwarder that delays packets past the WebSocket head start so - WebSocket wins deterministically; a subscribed track keeps every group across - the swap, `Status::Migrating` is observed, the WebSocket session closes - within the handover cap, and a second connect to the same URL gives QUIC the - head start again. A second test drops the delay to zero and asserts QUIC wins - and no second session is ever opened. -- Public API: additive (the transport accessor). Wire: none; the client GOAWAY - is already specified for either endpoint on moq-lite 04+ and an empty-URI - GOAWAY is legal for a moq-transport client. Update `doc/lib/rs` where the - fallback race is described. diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 27ee260451..5b771d9c5f 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -251,9 +251,9 @@ async fn serve_client( } /// Whether ordinary clients may use this transport on the shared LAN server. -fn is_public_transport(transport: moq_tokio::server::Transport, public_quic: bool) -> bool { +fn is_public_transport(transport: moq_tokio::Transport, public_quic: bool) -> bool { match transport { - moq_tokio::server::Transport::Tcp | moq_tokio::server::Transport::Unix => true, + moq_tokio::Transport::Tcp | moq_tokio::Transport::Unix => true, _ => public_quic, } } @@ -949,9 +949,9 @@ mod tests { #[test] fn explicit_stream_listeners_are_public_without_exposing_mesh_quic() { - assert!(is_public_transport(moq_tokio::server::Transport::Tcp, false)); - assert!(is_public_transport(moq_tokio::server::Transport::Unix, false)); - assert!(!is_public_transport(moq_tokio::server::Transport::Quic, false)); - assert!(is_public_transport(moq_tokio::server::Transport::Quic, true)); + assert!(is_public_transport(moq_tokio::Transport::Tcp, false)); + assert!(is_public_transport(moq_tokio::Transport::Unix, false)); + assert!(!is_public_transport(moq_tokio::Transport::Quic, false)); + assert!(is_public_transport(moq_tokio::Transport::Quic, true)); } } diff --git a/rs/moq-ffi/src/server.rs b/rs/moq-ffi/src/server.rs index daae44f026..c42937fb7a 100644 --- a/rs/moq-ffi/src/server.rs +++ b/rs/moq-ffi/src/server.rs @@ -202,7 +202,7 @@ struct RequestState { /// The network transport carrying an incoming session. #[derive(Clone, Copy, Debug, Eq, PartialEq, uniffi::Enum)] pub enum MoqTransport { - /// QUIC, either directly or through WebTransport over HTTP/3. + /// Raw QUIC, negotiating a MoQ ALPN directly. Quic, /// An Iroh QUIC connection. Iroh, @@ -212,18 +212,21 @@ pub enum MoqTransport { Tcp, /// A Unix domain socket using qmux framing. Unix, + /// WebTransport over HTTP/3 on QUIC. + WebTransport, } -impl TryFrom for MoqTransport { +impl TryFrom for MoqTransport { type Error = MoqError; - fn try_from(value: moq_tokio::server::Transport) -> Result { + fn try_from(value: moq_tokio::Transport) -> Result { Ok(match value { - moq_tokio::server::Transport::Quic => Self::Quic, - moq_tokio::server::Transport::Iroh => Self::Iroh, - moq_tokio::server::Transport::WebSocket => Self::WebSocket, - moq_tokio::server::Transport::Tcp => Self::Tcp, - moq_tokio::server::Transport::Unix => Self::Unix, + moq_tokio::Transport::Quic => Self::Quic, + moq_tokio::Transport::Iroh => Self::Iroh, + moq_tokio::Transport::WebSocket => Self::WebSocket, + moq_tokio::Transport::Tcp => Self::Tcp, + moq_tokio::Transport::Unix => Self::Unix, + moq_tokio::Transport::WebTransport => Self::WebTransport, _ => return Err(MoqError::Unsupported), }) } @@ -232,7 +235,7 @@ impl TryFrom for MoqTransport { #[cfg(test)] mod transport_tests { use super::MoqTransport; - use moq_tokio::server::Transport; + use moq_tokio::Transport; #[test] fn converts_supported_transports() { @@ -244,6 +247,10 @@ mod transport_tests { ); assert_eq!(MoqTransport::try_from(Transport::Tcp).unwrap(), MoqTransport::Tcp); assert_eq!(MoqTransport::try_from(Transport::Unix).unwrap(), MoqTransport::Unix); + assert_eq!( + MoqTransport::try_from(Transport::WebTransport).unwrap(), + MoqTransport::WebTransport + ); } } diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index 22ff0d7a5d..ff47f5e1db 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -488,11 +488,12 @@ impl Admission { /// transport knows, nothing parsed on the server's behalf. pub fn request_for(auth: &Auth, request: &moq_tokio::server::Request) -> Request { let transport = match request.transport() { - moq_tokio::server::Transport::Quic => moq_auth::Transport::Quic, - moq_tokio::server::Transport::Iroh => moq_auth::Transport::Iroh, - moq_tokio::server::Transport::WebSocket => moq_auth::Transport::WebSocket, - moq_tokio::server::Transport::Tcp => moq_auth::Transport::Tcp, - moq_tokio::server::Transport::Unix => moq_auth::Transport::Unix, + // The auth contract names QUIC either way; WebTransport is QUIC underneath. + moq_tokio::Transport::Quic | moq_tokio::Transport::WebTransport => moq_auth::Transport::Quic, + moq_tokio::Transport::Iroh => moq_auth::Transport::Iroh, + moq_tokio::Transport::WebSocket => moq_auth::Transport::WebSocket, + moq_tokio::Transport::Tcp => moq_auth::Transport::Tcp, + moq_tokio::Transport::Unix => moq_auth::Transport::Unix, // A transport this build does not know is still a session on the wire; the // server sees the same facts either way. other => unreachable!("unknown transport {other}"), diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index bb1d3e2559..0f12dfce4c 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -776,7 +776,7 @@ async fn serve_connection( }); let node_connection = peer_hop.map(|origin| serve.cluster.nodes.connect_inbound(id, origin)); - tracing::info!(id, version = %session.version(), transport = %moq_tokio::server::Transport::Quic, "negotiated"); + tracing::info!(id, version = %session.version(), transport = %moq_tokio::Transport::Quic, "negotiated"); // The session handle is Send + Sync however its transport is driven, so // its lifecycle (credential expiry, GOAWAY drain) lives with the timers diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index e9133a8e89..300765d61b 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -6,8 +6,10 @@ use crate::connection::Goaway; use crate::{Addrs, Backoff, Connection, Error}; +use futures::future::BoxFuture; #[cfg(all(feature = "websocket", feature = "noq"))] use std::future::Future; +use std::task::{Poll, ready}; use url::Url; /// Everything a [`Client`] is built from. @@ -270,7 +272,7 @@ impl Client { feature = "tcp", feature = "uds" )))] - pub(crate) async fn dial(&self, _addr: crate::connect::Addr) -> crate::Result { + pub(crate) async fn dial(&self, _addr: crate::connect::Addr) -> crate::Result { Err(Error::NoBackend( "no backend compiled; enable noq, iroh, websocket, tcp, or uds feature", )) @@ -279,7 +281,8 @@ impl Client { /// Dial the given URL and complete the MoQ handshake. /// /// The scheme picks the transport, and `https://` races QUIC against the - /// WebSocket fallback so a blocked UDP path still connects. The session's + /// WebSocket fallback so a blocked UDP path still connects. When the fallback + /// wins, the QUIC dial keeps going as [`Dialed::upgrade`]. The session's /// protocol driver is spawned on the current tokio runtime; the session /// closes once the last returned handle drops. #[cfg(any( @@ -289,7 +292,7 @@ impl Client { feature = "tcp", feature = "uds" ))] - pub(crate) async fn dial(&self, addr: crate::connect::Addr) -> crate::Result { + pub(crate) async fn dial(&self, addr: crate::connect::Addr) -> crate::Result { // Each compiled backend adds state to this dispatch future. Keep it off the // caller's stack so all-feature builds remain safe on standard 2 MiB threads. let attempt = Box::pin(self.connect_inner(addr)); @@ -297,16 +300,23 @@ impl Client { // The deadline covers the dial AND the handshake, for every transport: it is the // only bound some of them have. Dropping `attempt` on expiry cancels whichever // arm was still pending. - let session = match self.timeout.is_zero() { + let dialed = match self.timeout.is_zero() { true => attempt.await?, - false => match tokio::time::timeout(self.timeout, attempt).await { - Ok(res) => res?, - Err(_) => return Err(Error::ConnectTimeout(self.timeout)), - }, + false => { + let deadline = tokio::time::Instant::now() + self.timeout; + let mut dialed = match tokio::time::timeout_at(deadline, attempt).await { + Ok(res) => res?, + Err(_) => return Err(Error::ConnectTimeout(self.timeout)), + }; + // The QUIC arm that lost to WebSocket is still part of this attempt, so the + // same deadline bounds it. + dialed.upgrade = dialed.upgrade.map(|upgrade| upgrade.until(deadline, self.timeout)); + dialed + } }; - tracing::info!(version = %session.version(), "connected"); - Ok(session) + tracing::info!(version = %dialed.session.version(), transport = %dialed.transport, "connected"); + Ok(dialed) } /// The moq client builder, advertising `path` in the SETUP when there is one. @@ -331,7 +341,7 @@ impl Client { feature = "tcp", feature = "uds" ))] - async fn connect_inner(&self, addr: crate::connect::Addr) -> crate::Result { + async fn connect_inner(&self, addr: crate::connect::Addr) -> crate::Result { let url = addr.url().clone(); // Transports with no request URI of their own advertise the request target in the // SETUP instead; `setup_path` returns `None` for the ones that carry a URI, where @@ -346,7 +356,8 @@ impl Client { if url.scheme() == "tcp" { let session = crate::tcp::connect(url, &self.versions.alpns(), self.failover_delay, self.resolution_delay).await?; - return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?); + let session = connect_session(&moq, crate::transport::Session::new(session)).await?; + return Ok(Dialed::new(session, crate::Transport::Tcp)); } // Unix domain socket (qmux, no TLS). Same-host only; the server can @@ -354,7 +365,8 @@ impl Client { #[cfg(all(feature = "uds", unix))] if url.scheme() == "unix" { let session = crate::unix::connect(url, &self.versions.alpns()).await?; - return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?); + let session = connect_session(&moq, crate::transport::Session::new(session)).await?; + return Ok(Dialed::new(session, crate::Transport::Unix)); } // A WebSocket URL names its transport. No QUIC backend can dial it, so there is @@ -379,19 +391,23 @@ impl Client { crate::iroh::Binding::H3 => self.moq.clone(), }; - return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?); + let session = connect_session(&moq, crate::transport::Session::new(session)).await?; + return Ok(Dialed::new(session, crate::Transport::Iroh)); } #[cfg(feature = "noq")] - if let Some(noq) = self.noq.as_ref() { + if let Some(noq) = self.noq.clone() { + // Owned rather than borrowed from `self`: when WebSocket wins the race, this dial + // outlives the attempt as the pending upgrade. let tls = self.tls.clone(); + let versions = self.versions.clone(); let quic_addr = addr.clone(); - let quic_handle = async { - noq.connect(&tls, quic_addr, &self.versions) + let quic_handle = Box::pin(async move { + noq.connect(&tls, quic_addr, &versions) .await .map(crate::transport::Session::new) .map_err(Error::from) - }; + }); #[cfg(feature = "websocket")] { @@ -401,7 +417,8 @@ impl Client { #[cfg(not(feature = "websocket"))] { let session = quic_handle.await?; - return Ok(connect_session(&moq, session).await?); + let session = connect_session(&moq, session).await?; + return Ok(Dialed::new(session, quic_transport(&url))); } } @@ -415,15 +432,19 @@ impl Client { /// Connect over WebSocket alone. qmux over WebSocket carries the path in its request /// URI, so the plain builder is used: repeating it in the SETUP is a protocol violation. #[cfg(feature = "websocket")] - async fn connect_websocket(&self, addr: crate::connect::Addr) -> crate::Result { + async fn connect_websocket(&self, addr: crate::connect::Addr) -> crate::Result { let alpns = self.versions.alpns(); let session = crate::websocket::connect(&self.websocket, &self.tls, self.tls_host_name.as_deref(), addr, &alpns).await?; - Ok(connect_session(&self.moq, crate::transport::Session::new(session)).await?) + let session = connect_session(&self.moq, crate::transport::Session::new(session)).await?; + Ok(Dialed::new(session, crate::Transport::WebSocket)) } /// Race the QUIC dial against the WebSocket fallback, handshaking whichever wins. /// + /// When WebSocket wins while QUIC is still dialing, the QUIC dial carries on as + /// [`Dialed::upgrade`] rather than being dropped. + /// /// `moq` is the QUIC-side builder, which carries the SETUP path for a raw QUIC dial. /// The WebSocket fallback uses the plain builder: qmux over WebSocket carries the /// path in its request URI, so repeating it in the SETUP is a protocol violation. @@ -436,11 +457,12 @@ impl Client { moq: &moq_net::Client, addr: crate::connect::Addr, quic: Q, - ) -> crate::Result + ) -> crate::Result where - Q: Future>, + Q: Future> + Unpin + Send + 'static, S: moq_net::transport::poll::Boxable, { + let transport = quic_transport(addr.url()); let alpns = self.versions.alpns(); let ws_config = self.websocket.clone(); let ws_tls = self.tls.clone(); @@ -452,9 +474,19 @@ impl Client { }; match race_transport_connect(quic, websocket).await? { - TransportRace::Quic(quic) => Ok(connect_session(moq, quic).await?), - TransportRace::WebSocket(websocket) => { - Ok(connect_session(&self.moq, crate::transport::Session::new(websocket)).await?) + TransportRace::Quic(quic) => Ok(Dialed::new(connect_session(moq, quic).await?, transport)), + TransportRace::WebSocket { session, quic } => { + let session = connect_session(&self.moq, crate::transport::Session::new(session)).await?; + let mut dialed = Dialed::new(session, crate::Transport::WebSocket); + dialed.upgrade = quic.map(|quic| { + let moq = moq.clone(); + let dial = Box::pin(async move { + let quic = quic.await?; + Ok(Box::pin(async move { Ok(connect_session(&moq, quic).await?) }) as Handshake) + }); + Upgrade::new(dial, transport) + }); + Ok(dialed) } } } @@ -516,20 +548,134 @@ fn setup_path(url: &Url) -> Option { } } +/// What a noq dial to `url` runs on: WebTransport for `https://` (and the `http://` +/// bootstrap), raw QUIC for `moqt://` and `moql://`. +#[cfg(feature = "noq")] +fn quic_transport(url: &Url) -> crate::Transport { + match url.scheme() { + "moqt" | "moql" => crate::Transport::Quic, + _ => crate::Transport::WebTransport, + } +} + +/// A session [`Client::dial`] brought up. +pub(crate) struct Dialed { + pub session: moq_net::Session, + /// What `session` runs on. + pub transport: crate::Transport, + /// The QUIC dial still in flight when the WebSocket fallback won the race. + pub upgrade: Option, +} + +impl Dialed { + #[cfg_attr( + not(any( + feature = "noq", + feature = "iroh", + feature = "websocket", + feature = "tcp", + feature = "uds" + )), + allow(dead_code) + )] + fn new(session: moq_net::Session, transport: crate::Transport) -> Self { + Self { + session, + transport, + upgrade: None, + } + } +} + +/// The MoQ handshake running on an upgrade's QUIC session. +pub(crate) type Handshake = BoxFuture<'static, crate::Result>; + +/// A QUIC dial that lost the race to the WebSocket fallback but is still going. +/// +/// Polled by [`Connection`] alongside the WebSocket session, which it replaces +/// once both stages complete. Dropping it cancels the dial. +pub(crate) struct Upgrade { + stage: Stage, + /// What the upgraded session runs on. + transport: crate::Transport, + /// The connect deadline of the attempt this dial belongs to, and its length for + /// the error. + deadline: Option<(std::pin::Pin>, std::time::Duration)>, +} + +#[cfg_attr(not(all(feature = "websocket", feature = "noq")), allow(dead_code))] +enum Stage { + /// Waiting on the QUIC transport, and the WebTransport CONNECT for `https://`. + Dialing(BoxFuture<'static, crate::Result>), + /// The QUIC transport is up and the MoQ handshake is running on it. + Handshaking(Handshake), +} + +/// How far an [`Upgrade`] got on one poll. +pub(crate) enum Step { + /// The QUIC transport is up; the MoQ handshake has started on it. + Handshaking, + /// The MoQ session over QUIC is ready to take over, running on this transport. + Done(moq_net::Session, crate::Transport), +} + +#[cfg_attr(not(all(feature = "websocket", feature = "noq")), allow(dead_code))] +impl Upgrade { + pub(crate) fn new(dial: BoxFuture<'static, crate::Result>, transport: crate::Transport) -> Self { + Self { + stage: Stage::Dialing(dial), + transport, + deadline: None, + } + } + + /// Fail with [`Error::ConnectTimeout`] if still pending at `deadline`. + fn until(mut self, deadline: tokio::time::Instant, timeout: std::time::Duration) -> Self { + self.deadline = Some((Box::pin(tokio::time::sleep_until(deadline)), timeout)); + self + } + + /// Drive the dial: `Ready(Ok(Step::Handshaking))` once when the QUIC transport + /// comes up, then `Ready(Ok(Step::Done))` when the handshake on it completes. Not + /// polled again after `Done` or an error. + pub(crate) fn poll(&mut self, waiter: &moq_net::kio::Waiter) -> Poll> { + if let Some((sleep, timeout)) = &mut self.deadline + && waiter.poll_future(sleep.as_mut()).is_ready() + { + return Poll::Ready(Err(Error::ConnectTimeout(*timeout))); + } + + match &mut self.stage { + Stage::Dialing(dial) => { + let handshake = ready!(waiter.poll_future(dial.as_mut()))?; + self.stage = Stage::Handshaking(handshake); + Poll::Ready(Ok(Step::Handshaking)) + } + Stage::Handshaking(handshake) => { + let session = ready!(waiter.poll_future(handshake.as_mut()))?; + Poll::Ready(Ok(Step::Done(session, self.transport))) + } + } + } +} + #[cfg(all(feature = "websocket", feature = "noq"))] -#[derive(Debug, PartialEq, Eq)] -enum TransportRace { - Quic(Q), - WebSocket(W), +enum TransportRace { + Quic(QT), + /// The fallback won. `quic` is the QUIC dial if it was still pending, and `None` + /// when it had already failed. + WebSocket { + session: WT, + quic: Option, + }, } #[cfg(all(feature = "websocket", feature = "noq"))] -async fn race_transport_connect(quic: Q, websocket: W) -> crate::Result> +async fn race_transport_connect(mut quic: Q, websocket: W) -> crate::Result> where - Q: Future>, + Q: Future> + Unpin, W: Future>>, { - tokio::pin!(quic); tokio::pin!(websocket); let mut quic_err = None; @@ -551,7 +697,10 @@ where } res = &mut websocket, if !websocket_done => { match res { - Some(Ok(session)) => return Ok(TransportRace::WebSocket(session)), + Some(Ok(session)) => { + let quic = (!quic_done).then_some(quic); + return Ok(TransportRace::WebSocket { session, quic }); + } Some(Err(err)) => { tracing::warn!(%err, "WebSocket connection failed"); websocket_err = Some(err); @@ -1141,8 +1290,11 @@ mod tests { Some(Ok(1usize)) }; - let value = super::race_transport_connect(quic, websocket).await.unwrap(); - assert_eq!(value, super::TransportRace::WebSocket(1)); + let value = super::race_transport_connect(Box::pin(quic), websocket).await.unwrap(); + assert!(matches!( + value, + super::TransportRace::WebSocket { session: 1, quic: None } + )); } #[cfg(all(feature = "websocket", feature = "noq"))] @@ -1154,8 +1306,8 @@ mod tests { }; let websocket = async { Some(Err::(crate::ConnectError::Forbidden.into())) }; - let value = super::race_transport_connect(quic, websocket).await.unwrap(); - assert_eq!(value, super::TransportRace::Quic(3)); + let value = super::race_transport_connect(Box::pin(quic), websocket).await.unwrap(); + assert!(matches!(value, super::TransportRace::Quic(3))); } /// A WebTransport-only endpoint answers the WebSocket fallback with 403 while the @@ -1172,7 +1324,7 @@ mod tests { let listener = tokio::net::TcpListener::bind("[::]:0").await.unwrap(); let ws_port = listener.local_addr().unwrap().port(); - let mut forbid = tokio::spawn(async move { + let forbid = tokio::spawn(async move { let (mut stream, _) = listener.accept().await?; let mut buf = [0; 1024]; let _ = stream.read(&mut buf).await?; @@ -1207,19 +1359,20 @@ mod tests { // The same race `connect_inner` runs, except the fallback dials its own port, // as plain ws:// so the listener above can answer without TLS. - let noq = client.noq.as_ref().unwrap(); + let noq = client.noq.clone().unwrap(); + let (tls, versions) = (client.tls.clone(), client.versions.clone()); let quic_addr: crate::connect::Addr = Url::parse(&format!("https://localhost:{quic_port}")).unwrap().into(); let ws_addr: crate::connect::Addr = Url::parse(&format!("http://localhost:{ws_port}")).unwrap().into(); // Hold QUIC until the fallback has been refused, so the 403 is always exercised. - let quic = async { - (&mut forbid).await.unwrap().expect("fallback listener failed"); - noq.connect(&client.tls, quic_addr, &client.versions) + let quic = Box::pin(async move { + forbid.await.unwrap().expect("fallback listener failed"); + noq.connect(&tls, quic_addr, &versions) .await .map(crate::transport::Session::new) .map_err(Error::from) - }; + }); - let session = tokio::time::timeout( + let dialed = tokio::time::timeout( std::time::Duration::from_secs(10), client.race_moq_connect(&client.moq, ws_addr, quic), ) @@ -1227,7 +1380,8 @@ mod tests { .expect("client connect timed out") .expect("a fallback refused on auth must not end a connect whose QUIC arm succeeds"); - drop(session); + assert_eq!(dialed.transport, crate::Transport::WebTransport); + drop(dialed); accepted.await.unwrap().expect("server handshake failed"); } @@ -1237,7 +1391,9 @@ mod tests { let quic = async { Err::(crate::ConnectError::Unauthorized.into()) }; let websocket = async { Some(Err::(crate::ConnectError::Forbidden.into())) }; - let err = super::race_transport_connect(quic, websocket).await.unwrap_err(); + let Err(err) = super::race_transport_connect(Box::pin(quic), websocket).await else { + panic!("the race must fail"); + }; assert!(err.is_auth(), "unexpected error: {err}"); } @@ -1250,7 +1406,9 @@ mod tests { }; let websocket = async { Some(Err::(crate::ConnectError::Forbidden.into())) }; - let err = super::race_transport_connect(quic, websocket).await.unwrap_err(); + let Err(err) = super::race_transport_connect(Box::pin(quic), websocket).await else { + panic!("the race must fail"); + }; assert!(matches!(err, Error::ConnectFailed), "unexpected error: {err}"); assert!(!err.is_auth(), "mixed auth/non-auth must stay retryable: {err}"); } @@ -1261,8 +1419,34 @@ mod tests { let quic = async { Err::(Error::ConnectFailed) }; let websocket = async { Some(Ok(7usize)) }; - let value = super::race_transport_connect(quic, websocket).await.unwrap(); - assert_eq!(value, super::TransportRace::WebSocket(7)); + let value = super::race_transport_connect(Box::pin(quic), websocket).await.unwrap(); + // `select!` may poll either arm first, so the failed QUIC dial can come back unpolled. + assert!(matches!(value, super::TransportRace::WebSocket { session: 7, .. })); + } + + /// WebSocket winning hands back the QUIC dial it beat, still running, so the + /// session can move onto QUIC once it lands. + #[cfg(all(feature = "websocket", feature = "noq"))] + #[tokio::test] + async fn race_transport_connect_keeps_a_pending_quic_dial() { + let (land, landed) = tokio::sync::oneshot::channel::<()>(); + let quic = async move { + landed.await.unwrap(); + Ok("quic") + }; + let websocket = async { Some(Ok("websocket")) }; + + let race = super::race_transport_connect(Box::pin(quic), websocket).await; + let Ok(super::TransportRace::WebSocket { + session: "websocket", + quic: Some(quic), + }) = race + else { + panic!("WebSocket won, and the QUIC dial it beat must come back with it"); + }; + + land.send(()).unwrap(); + assert_eq!(quic.await.unwrap(), "quic"); } #[cfg(all(feature = "websocket", feature = "noq"))] @@ -1273,12 +1457,12 @@ mod tests { let value = tokio::time::timeout( std::time::Duration::from_secs(1), - super::race_transport_connect(quic, websocket), + super::race_transport_connect(Box::pin(quic), websocket), ) .await .expect("race waited for WebSocket after QUIC transport connected") .unwrap(); - assert_eq!(value, super::TransportRace::Quic("quic")); + assert!(matches!(value, super::TransportRace::Quic("quic"))); } /// The resolved default has to exist in every build, including ones that compile diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs index 13f3fb4e2a..3e49d87628 100644 --- a/rs/moq-tokio/src/connection.rs +++ b/rs/moq-tokio/src/connection.rs @@ -175,8 +175,8 @@ pub enum Status { Connected, /// An established session dropped; a reconnect attempt follows. Disconnected, - /// The peer sent a GOAWAY; the replacement is being dialed while the old - /// session keeps serving. + /// A replacement session is coming up while the old one keeps serving: the + /// peer sent a GOAWAY, or a QUIC dial landed after the WebSocket fallback won. Migrating, } @@ -456,6 +456,8 @@ struct State { presence: moq_net::stats::Presence, /// The negotiated MoQ version of the live session, or `None` when disconnected. version: Option, + /// What the live session runs on, or `None` when disconnected. + transport: Option, /// Set when the reconnect loop permanently gives up (reconnect timeout exceeded). error: Option, /// The currently-connected session, or `None` while reconnecting. Read by @@ -479,7 +481,7 @@ struct Shared { impl Shared { /// A session is live and serving. - fn connected(&self, session: &moq_net::Session) { + fn connected(&self, session: &moq_net::Session, transport: crate::Transport) { // Held across the publish: see [`CloseGuard`]. A close that already ran is // honored here rather than leaving this session parked in the final state. let closed = self.closed.lock().unwrap(); @@ -488,19 +490,35 @@ impl Shared { return; } if let Ok(mut state) = self.state.write() { + // A migration replaces a live session without a disconnect in between, so + // its end is counted here, keeping `started - ended` at 1 while connected. + if state.session.is_some() { + state.presence.sessions_ended += 1; + } state.status = Some(Status::Connected); state.epoch += 1; state.presence.sessions_started += 1; state.version = Some(session.version()); + state.transport = Some(transport); state.session = Some(session.clone()); } } - /// The peer sent a GOAWAY: the session in [`State`] keeps serving while its - /// replacement is dialed, so only the status moves. + /// The session in [`State`] keeps serving while its replacement comes up, so + /// only the status moves. fn migrating(&self) { + self.status(Status::Migrating); + } + + /// The replacement did not come up after all: the session in [`State`] was + /// serving throughout, so only the status moves back. + fn stayed(&self) { + self.status(Status::Connected); + } + + fn status(&self, status: Status) { if let Ok(mut state) = self.state.write() { - state.status = Some(Status::Migrating); + state.status = Some(status); } } @@ -516,6 +534,7 @@ impl Shared { } state.status = Some(Status::Disconnected); state.version = None; + state.transport = None; state.session = None; } let _ = self.send_bw.set(None); @@ -747,16 +766,46 @@ impl Connection { let budget = retry_budget(client.reconnect, retry_start, timeout); match Self::dial_any(shared, &client, &addrs, &mut draining, budget).await { - Ok((addr, session)) => { + Ok((addr, dialed)) => { let url = addr.url().clone(); - tracing::info!(peer = %Endpoint(&url), "connected"); - shared.connected(&session); + tracing::info!(peer = %Endpoint(&url), transport = %dialed.transport, "connected"); + shared.connected(&dialed.session, dialed.transport); + let mut session = dialed.session; + let mut upgrade = dialed.upgrade; - let connected = tokio::time::Instant::now(); + let mut connected = tokio::time::Instant::now(); // Wait for the session to end, forwarding its bandwidth estimates into the // persistent producers meanwhile so consumers track the live stats across the - // connection, and draining any predecessor left over from a migration. - let ended = run_session(shared, &session, &mut draining).await; + // connection, and draining any predecessor left over from a migration. A QUIC + // dial that lands after WebSocket won takes over here without a redial. + let ended = loop { + match run_session(shared, &session, &mut draining, &mut upgrade).await { + Next::Ended(ended) => break ended, + Next::Upgraded(next, transport) => { + tracing::info!(peer = %Endpoint(&url), %transport, "upgraded from WebSocket"); + // UDP gets through after all, so the next dial gives QUIC its head start. + #[cfg(feature = "websocket")] + crate::websocket::forget(&url); + + // The same handover as a peer's GOAWAY, initiated by us: the new session + // is live, and the old one serves until its routes splice over at a group + // boundary or the cap closes it. + let old = std::mem::replace(&mut session, next); + shared.connected(&session, transport); + // An empty URI is legal from either endpoint on every version; the old + // session's driver closes it at the deadline if the peer lingers. + let msg = moq_net::goaway::Goaway::new().with_timeout(goaway.handover); + if let Err(err) = old.drain().send(msg) { + tracing::debug!(%err, "failed to send GOAWAY on the WebSocket session"); + } + if let Some(mut old) = draining.take() { + old.retire(); + } + draining = Some(Draining::new(old, goaway.handover)); + connected = tokio::time::Instant::now(); + } + } + }; // A session that stayed up past the initial backoff is healthy; one that // ended sooner counts as a failed attempt however it ended. @@ -933,7 +982,7 @@ impl Connection { addrs: &Addrs, draining: &mut Option, budget: Option, - ) -> crate::Result<(crate::connect::Addr, moq_net::Session)> { + ) -> crate::Result<(crate::connect::Addr, crate::client::Dialed)> { let candidates = addrs.as_slice(); let mut last = None; @@ -965,7 +1014,7 @@ impl Connection { }; match dialed { - Ok(session) => return Ok((addr.clone(), session)), + Ok(dialed) => return Ok((addr.clone(), dialed)), // A status the peer actually sent is its answer, not this address's, so // unless it invites another attempt it settles the whole walk. Carrying // on would offer the same rejected credentials at the peer's other @@ -1076,6 +1125,14 @@ impl Connection { self.state.read().version } + /// What the live session runs on, or `None` while disconnected. + /// + /// Changes without a disconnect when a session that came up over the WebSocket + /// fallback moves onto QUIC. + pub fn transport(&self) -> Option { + self.state.read().transport + } + /// Observe a GOAWAY from the peer, or `None` while between sessions. /// /// The reconnect loop reads the same GOAWAY to drive migration, reported as @@ -1154,6 +1211,15 @@ fn retry_wait(delay: Duration, retry_start: tokio::time::Instant, timeout: Durat Some(wait.min(timeout.checked_sub(retry_start.elapsed())?)) } +/// What [`run_session`] returns on. +enum Next { + /// The session stopped being the live one. + Ended(Ended), + /// The pending QUIC dial finished its handshake; this session, on this + /// transport, replaces the WebSocket one. + Upgraded(moq_net::Session, crate::Transport), +} + /// Why a session stopped being the live one. enum Ended { /// The transport closed. @@ -1252,9 +1318,18 @@ async fn sleep_draining(delay: Duration, draining: &mut Option, shared /// /// One `poll_*` step drives it all: [`poll_forward`] mirrors each kio bandwidth estimate, the /// GOAWAY consumer is a kio channel, and the transport's close future (the one non-kio source) is -/// polled through the waiter's own waker. `migrate` is whether a GOAWAY ends this session's -/// tenure ([`Ended::Goaway`]); when false the arm isn't polled and only a close returns. -async fn run_session(shared: &Shared, session: &moq_net::Session, draining: &mut Option) -> Ended { +/// polled through the waiter's own waker. +/// +/// `upgrade` is a QUIC dial still in flight after the WebSocket fallback won. It reports +/// [`Status::Migrating`] once the QUIC transport is up and returns [`Next::Upgraded`] once +/// the handshake on it completes. A failure leaves this session serving. It is dropped with +/// the session otherwise: the redial races QUIC again. +async fn run_session( + shared: &Shared, + session: &moq_net::Session, + draining: &mut Option, + upgrade: &mut Option, +) -> Next { let mut send = session.send_bandwidth(); let mut recv = session.recv_bandwidth(); let goaway = session.draining(); @@ -1278,14 +1353,43 @@ async fn run_session(shared: &Shared, session: &moq_net::Session, draining: &mut // sides end up waiting on each other forever. The caller ends the connection // instead: asked to leave, with no way to migrate, leaving is the answer. if let Poll::Ready(Ok(msg)) = goaway.poll(waiter) { - return Poll::Ready(Ended::Goaway(msg)); + return Poll::Ready(Next::Ended(Ended::Goaway(msg))); + } + + if let Poll::Ready(err) = waiter.poll_future(closed.as_mut()) { + return Poll::Ready(Next::Ended(Ended::Closed(Err(err)))); } - waiter.poll_future(closed.as_mut()).map(|err| Ended::Closed(Err(err))) + poll_upgrade(shared, upgrade, waiter).map(|(session, transport)| Next::Upgraded(session, transport)) }) .await } +/// Drive a pending upgrade, `Ready` with the QUIC session once its handshake completes. +fn poll_upgrade( + shared: &Shared, + upgrade: &mut Option, + waiter: &kio::Waiter, +) -> Poll<(moq_net::Session, crate::Transport)> { + while let Some(pending) = upgrade.as_mut() { + match ready!(pending.poll(waiter)) { + Ok(crate::client::Step::Handshaking) => shared.migrating(), + Ok(crate::client::Step::Done(session, transport)) => { + *upgrade = None; + return Poll::Ready((session, transport)); + } + Err(err) => { + // Only the Handshaking stage moved the status, but restoring it is harmless + // either way: this session was serving throughout. + tracing::debug!(%err, "QUIC upgrade failed; staying on WebSocket"); + shared.stayed(); + *upgrade = None; + } + } + } + Poll::Pending +} + /// Mirror `bw`'s live estimate into `out` for as long as it changes, dropping the source handle once /// the session's producer is gone so we don't keep polling a dead arm. A `poll_*` step: on return, /// `waiter` is registered for the next change (unless the source is gone). Seeding is implicit @@ -1950,4 +2054,332 @@ mod tests { assert_eq!(attempt_timeout(1, 3), Some(CONNECT_ATTEMPT), "still one more"); assert_eq!(attempt_timeout(2, 3), None, "the last of three"); } + + /// A QUIC server with the WebSocket fallback on the same port number, the QUIC + /// half reached through a [`Forwarder`], publishing `origin`. + /// + /// Yields each accepted session with the transport it arrived on. + #[cfg(all(feature = "websocket", feature = "noq"))] + struct Fallback { + url: Url, + forwarder: Forwarder, + accepted: tokio::sync::mpsc::UnboundedReceiver<(crate::Transport, moq_net::Session)>, + } + + #[cfg(all(feature = "websocket", feature = "noq"))] + impl Fallback { + async fn start(origin: &moq_net::origin::Producer, open: bool) -> Self { + let mut listen = crate::listen::Config { + bind: Some("127.0.0.1:0".parse().unwrap()), + ..Default::default() + }; + listen.tls.generate = vec!["localhost".into()]; + + // The fallback dials the URL's port over TCP, so the forwarder takes the same + // number over UDP. Nothing reserves the pair, so retry on a collision. + let (websocket, forwarder) = 'bind: { + for _ in 0..20 { + let websocket = crate::websocket::Listener::bind("127.0.0.1:0".parse().unwrap()) + .await + .unwrap(); + let port = websocket.local_addr().unwrap().port(); + if let Ok(front) = tokio::net::UdpSocket::bind(("127.0.0.1", port)).await { + break 'bind (websocket, front); + } + } + panic!("could not bind a matching TCP and UDP port after 20 attempts"); + }; + + let config = crate::server::Config { + listen, + websocket: Some(websocket), + publisher: Some(origin.consume()), + ..Default::default() + }; + let mut server = config.init().unwrap().listen().await.unwrap(); + let quic = server.local_addr().unwrap(); + let port = forwarder.local_addr().unwrap().port(); + let forwarder = Forwarder::start(forwarder, quic, open).await; + + let (tx, accepted) = tokio::sync::mpsc::unbounded_channel(); + tokio::spawn(async move { + while let Some(request) = server.accept().await { + let transport = request.transport(); + if let Ok(session) = request.ok().await { + let _ = tx.send((transport, session)); + } + } + }); + + Self { + url: url(&format!("http://127.0.0.1:{port}/")), + forwarder, + accepted, + } + } + + async fn accept(&mut self) -> (crate::Transport, moq_net::Session) { + tokio::time::timeout(UPGRADE_WAIT, self.accepted.recv()) + .await + .expect("the server never accepted a session") + .expect("the server stopped") + } + } + + /// Relays QUIC datagrams to `server`, holding them all until [`Forwarder::open`]. + /// + /// Holding is what makes WebSocket win the race without depending on timing: the + /// QUIC dial cannot finish until the test says so. + #[cfg(all(feature = "websocket", feature = "noq"))] + struct Forwarder { + gate: tokio::sync::watch::Sender, + } + + #[cfg(all(feature = "websocket", feature = "noq"))] + impl Forwarder { + /// `open` is whether to forward from the start rather than hold. + async fn start(front: tokio::net::UdpSocket, server: std::net::SocketAddr, open: bool) -> Self { + use std::sync::Arc; + + let back = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap(); + back.connect(server).await.unwrap(); + let (front, back) = (Arc::new(front), Arc::new(back)); + let (tx, rx) = tokio::sync::watch::channel(open); + let (client_tx, client_rx) = tokio::sync::watch::channel(None); + + // Each datagram waits for the gate on its own task. + fn relay(gate: &tokio::sync::watch::Receiver, send: impl Future + Send + 'static) { + let mut gate = gate.clone(); + tokio::spawn(async move { + if gate.wait_for(|open| *open).await.is_ok() { + send.await; + } + }); + } + + { + let (front, back, gate) = (front.clone(), back.clone(), rx.clone()); + tokio::spawn(async move { + let mut buf = vec![0; 65536]; + while let Ok((len, from)) = front.recv_from(&mut buf).await { + client_tx.send_replace(Some(from)); + let packet = buf[..len].to_vec(); + let back = back.clone(); + relay(&gate, async move { + let _ = back.send(&packet).await; + }); + } + }); + } + + tokio::spawn(async move { + let mut buf = vec![0; 65536]; + while let Ok(len) = back.recv(&mut buf).await { + let Some(client) = *client_rx.borrow() else { continue }; + let packet = buf[..len].to_vec(); + let front = front.clone(); + relay(&rx, async move { + let _ = front.send_to(&packet, client).await; + }); + } + }); + + Self { gate: tx } + } + + /// Start forwarding, releasing whatever was held. + fn open(&self) { + self.gate.send_replace(true); + } + } + + #[cfg(all(feature = "websocket", feature = "noq"))] + const UPGRADE_WAIT: Duration = Duration::from_secs(10); + + /// The sequence of the next group `sub` receives. + #[cfg(all(feature = "websocket", feature = "noq"))] + async fn next_group(sub: &mut moq_net::track::Subscriber) -> u64 { + let group = tokio::time::timeout(UPGRADE_WAIT, sub.recv_group()) + .await + .expect("no group arrived") + .unwrap() + .expect("the track ended"); + group.sequence + } + + /// The upgrade reports [`Status::Migrating`] while the MoQ handshake runs on the + /// QUIC session, and a failure there puts the status back: the WebSocket session + /// served throughout. + /// + /// Pinned here rather than end to end, since a moq-lite client completes its + /// handshake without waiting on the server, leaving no window to observe. + #[test] + fn a_failed_upgrade_stays_connected() { + let shared = shared(); + shared.status(Status::Connected); + let waiter = kio::Waiter::noop(); + + let (fail, failed) = tokio::sync::oneshot::channel::<()>(); + let handshake: crate::client::Handshake = Box::pin(async move { + let _ = failed.await; + Err(Error::ConnectFailed) + }); + let mut upgrade = Some(crate::client::Upgrade::new( + Box::pin(async move { Ok(handshake) }), + crate::Transport::WebTransport, + )); + + assert!(poll_upgrade(&shared, &mut upgrade, &waiter).is_pending()); + assert_eq!(shared.state.consume().read().status, Some(Status::Migrating)); + assert!(upgrade.is_some()); + + fail.send(()).unwrap(); + assert!(poll_upgrade(&shared, &mut upgrade, &waiter).is_pending()); + assert_eq!(shared.state.consume().read().status, Some(Status::Connected)); + assert!(upgrade.is_none(), "a failed upgrade is not retried"); + } + + /// A session that came up over the WebSocket fallback moves onto QUIC once the + /// QUIC dial lands, without dropping a group, and forgets that WebSocket won. + #[cfg(all(feature = "websocket", feature = "noq"))] + #[tokio::test] + async fn websocket_upgrades_to_quic() { + const HANDOVER: Duration = Duration::from_millis(500); + + let origin = crate::origin::spawn(); + let broadcast = origin.create_broadcast("cam").unwrap(); + broadcast.announce(Default::default()).unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + let write = |payload: &'static [u8]| { + let mut group = track.append_group().unwrap(); + group.write_frame(moq_net::Timestamp::ZERO, payload).unwrap(); + group.finish().unwrap(); + }; + + let mut fallback = Fallback::start(&origin, false).await; + + let subscriber = crate::origin::spawn(); + let mut config = crate::connect::Config::default(); + config.tls.insecure = Some(true); + config.goaway.handover = HANDOVER; + let client = config + .init(Default::default()) + .unwrap() + .with_subscriber(subscriber.clone()); + let connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) + .await + .expect("never connected") + .unwrap(); + + // QUIC is held, so the fallback won. + assert_eq!(connection.transport(), Some(crate::Transport::WebSocket)); + assert!( + crate::websocket::won(&fallback.url), + "the fallback's win was not remembered" + ); + let mut monitor = connection.monitor(); + let (transport, websocket) = fallback.accept().await; + assert_eq!(transport, crate::Transport::WebSocket); + + let cam = tokio::time::timeout(UPGRADE_WAIT, subscriber.consume().routed_broadcast("cam")) + .await + .unwrap() + .unwrap(); + let mut sub = cam.track("video").unwrap().subscribe(None).await.unwrap(); + write(b"g0"); + assert_eq!(next_group(&mut sub).await, 0); + + // Let QUIC through, and publish while the replacement comes up. + fallback.forwarder.open(); + write(b"g1"); + + // The swap is a new session on the same handle, with no disconnect between. + let presence = tokio::time::timeout(UPGRADE_WAIT, async { + loop { + let presence = monitor.presence_changed().await.unwrap(); + if presence.sessions_started == 2 { + return presence; + } + } + }) + .await + .expect("never upgraded"); + assert_eq!( + presence.sessions_ended, 1, + "the WebSocket session was not counted as ended" + ); + assert!(connection.connected()); + assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); + assert_eq!(connection.epoch(), 2); + let (transport, _quic) = fallback.accept().await; + assert_eq!(transport, crate::Transport::WebTransport); + // QUIC works on this network, so the next dial gives it the head start again. + assert!( + !crate::websocket::won(&fallback.url), + "the upgrade kept WebSocket's win" + ); + + // Every group arrives exactly once across the swap: the one written while QUIC + // came up and the one after it. + write(b"g2"); + assert_eq!(next_group(&mut sub).await, 1); + assert_eq!(next_group(&mut sub).await, 2); + + // The WebSocket session is told to leave, and closes within the handover cap + // rather than lingering alongside QUIC. + let goaway = websocket.draining(); + let goaway = tokio::time::timeout(UPGRADE_WAIT, kio::wait(|waiter| goaway.poll(waiter))) + .await + .expect("the WebSocket session never received a GOAWAY") + .unwrap(); + assert_eq!(goaway.uri(), "", "a client may not redirect its server"); + tokio::time::timeout(UPGRADE_WAIT, websocket.closed()) + .await + .expect("the WebSocket session outlived the handover cap"); + assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); + } + + /// When QUIC wins the race nothing changes: one session, over QUIC, and no + /// WebSocket session is ever opened. + #[cfg(all(feature = "websocket", feature = "noq"))] + #[tokio::test] + async fn quic_winning_opens_one_session() { + let origin = crate::origin::spawn(); + let broadcast = origin.create_broadcast("cam").unwrap(); + broadcast.announce(Default::default()).unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + + let mut fallback = Fallback::start(&origin, true).await; + + let subscriber = crate::origin::spawn(); + let mut config = crate::connect::Config::default(); + config.tls.insecure = Some(true); + let client = config + .init(Default::default()) + .unwrap() + .with_subscriber(subscriber.clone()); + let connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) + .await + .expect("never connected") + .unwrap(); + assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); + let (transport, _session) = fallback.accept().await; + assert_eq!(transport, crate::Transport::WebTransport); + + let cam = tokio::time::timeout(UPGRADE_WAIT, subscriber.consume().routed_broadcast("cam")) + .await + .unwrap() + .unwrap(); + let mut sub = cam.track("video").unwrap().subscribe(None).await.unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(moq_net::Timestamp::ZERO, b"g0".as_ref()).unwrap(); + group.finish().unwrap(); + assert_eq!(next_group(&mut sub).await, 0); + + // QUIC's win dropped the fallback before it dialed, so there is nothing else to + // accept and nothing to upgrade from. + assert!(fallback.accepted.try_recv().is_err(), "a second session was opened"); + assert_eq!(connection.epoch(), 1); + assert!(!crate::websocket::won(&fallback.url)); + } } diff --git a/rs/moq-tokio/src/lib.rs b/rs/moq-tokio/src/lib.rs index f520b86565..7850224e75 100644 --- a/rs/moq-tokio/src/lib.rs +++ b/rs/moq-tokio/src/lib.rs @@ -75,6 +75,7 @@ pub use error::{Error, Result}; pub use log::{Log, RedactedUrl}; #[cfg(feature = "_transport")] pub use server::{Listener, Server}; +pub use transport::Transport; // Re-export these crates. pub use moq_net; diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index d71a05f658..b6967a38d8 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -8,9 +8,9 @@ use std::net; #[cfg(any(test, all(feature = "uds", unix)))] use std::path::PathBuf; -use crate::Error; #[cfg(feature = "iroh")] use crate::iroh; +use crate::{Error, Transport}; use moq_net::Session; use url::Url; @@ -587,7 +587,9 @@ impl Server { let Accepted { session, url, identity, authority, mut link } = super::noq::accept(_conn, alpns).await?; link.local = local; let request = server.accept_request(tokio::time::Instant::now().into_std(), crate::transport::Session::new(session)).await?; - Ok(Request { transport: Transport::Quic, url, identity, authority, link, kind: RequestKind::Noq(Box::new(request)) }) + // 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::Noq(Box::new(request)) }) }.boxed()); } } @@ -1098,41 +1100,6 @@ pub struct Link { pub alpn: Option, } -/// The network transport carrying an incoming MoQ session. -#[non_exhaustive] -#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] -pub enum Transport { - /// QUIC, either directly or through WebTransport over HTTP/3. - Quic, - /// An Iroh QUIC connection. - Iroh, - /// A WebSocket connection using qmux framing. - WebSocket, - /// A plaintext TCP connection using qmux framing. - Tcp, - /// A Unix domain socket using qmux framing. - Unix, -} - -impl Transport { - /// Returns the stable lowercase name used in logs and external metadata. - pub const fn as_str(self) -> &'static str { - match self { - Self::Quic => "quic", - Self::Iroh => "iroh", - Self::WebSocket => "websocket", - Self::Tcp => "tcp", - Self::Unix => "unix", - } - } -} - -impl std::fmt::Display for Transport { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str(self.as_str()) - } -} - /// An incoming MoQ session that can be accepted or rejected. /// /// The transport connection and the MoQ SETUP are already complete, so [`path`](Self::path), @@ -1723,6 +1690,7 @@ mod tests { assert_eq!(Transport::WebSocket.as_str(), "websocket"); assert_eq!(Transport::Tcp.as_str(), "tcp"); assert_eq!(Transport::Unix.as_str(), "unix"); + assert_eq!(Transport::WebTransport.as_str(), "webtransport"); } /// Building the endpoint needs a runtime, and `certificates()` must stay diff --git a/rs/moq-tokio/src/transport.rs b/rs/moq-tokio/src/transport.rs index ffc046c658..0331ff6322 100644 --- a/rs/moq-tokio/src/transport.rs +++ b/rs/moq-tokio/src/transport.rs @@ -7,6 +7,8 @@ //! closed-watch emulation is weaker than a native implementation (see //! [`Session`]), so a backend that implements the poll interface itself is //! handed to moq-net directly. +//! +//! It also names the [`Transport`] a session runs on, for the accept and dial sides alike. use std::task::{Context, Poll, ready}; @@ -14,6 +16,44 @@ use bytes::Bytes; use futures::FutureExt; use web_transport_trait::poll as wt_poll; +/// The network transport carrying a MoQ session, on either side of it. +#[non_exhaustive] +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum Transport { + /// Raw QUIC, negotiating a MoQ ALPN directly. + Quic, + /// An Iroh QUIC connection. + Iroh, + /// A WebSocket connection using qmux framing. + WebSocket, + /// A plaintext TCP connection using qmux framing. + Tcp, + /// A Unix domain socket using qmux framing. + Unix, + /// WebTransport over HTTP/3 on QUIC. + WebTransport, +} + +impl Transport { + /// Returns the stable lowercase name used in logs and external metadata. + pub const fn as_str(self) -> &'static str { + match self { + Self::Quic => "quic", + Self::Iroh => "iroh", + Self::WebSocket => "websocket", + Self::Tcp => "tcp", + Self::Unix => "unix", + Self::WebTransport => "webtransport", + } + } +} + +impl std::fmt::Display for Transport { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + /// A stored in-flight operation future. Native transports are Send, so the /// plain boxed flavor suffices. type OpBox = futures::future::BoxFuture<'static, T>; diff --git a/rs/moq-tokio/src/websocket.rs b/rs/moq-tokio/src/websocket.rs index ab6f911c78..2d0b2652d4 100644 --- a/rs/moq-tokio/src/websocket.rs +++ b/rs/moq-tokio/src/websocket.rs @@ -3,7 +3,8 @@ //! Used when QUIC is unreachable: UDP blocked by a firewall, a proxy in the way, a //! network that only passes TCP/443. The client races this against QUIC and gives QUIC //! a small head start ([`Config::delay`]), so WebSocket only wins when QUIC can't get -//! through. Servers accept it on a separate TCP port via [`Listener`]. +//! through. A [`crate::Connection`] that lands on WebSocket keeps the QUIC dial going and +//! moves onto it if it completes. Servers accept it on a separate TCP port via [`Listener`]. use qmux::ws::tokio_tungstenite; use qmux::ws::tokio_tungstenite::tungstenite::{self, client::IntoClientRequest, http}; @@ -106,6 +107,32 @@ type Result = std::result::Result; // Track servers (hostname:port) where WebSocket won the race, so we won't give QUIC a headstart next time static WEBSOCKET_WON: LazyLock>> = LazyLock::new(|| Mutex::new(HashSet::new())); +/// The [`WEBSOCKET_WON`] key for a dial URL: the host and port the fallback dials. +fn won_key(url: &Url) -> Result<(String, u16)> { + let host = url.host_str().ok_or(Error::MissingHostname)?.to_string(); + let port = url.port().unwrap_or_else(|| match url.scheme() { + "https" | "wss" | "moql" | "moqt" => 443, + "http" | "ws" => 80, + _ => 443, + }); + Ok((host, port)) +} + +/// Forget that WebSocket won for `url`, giving QUIC its head start again. +/// +/// Called once a QUIC dial to `url` lands after all, which proves UDP gets through. +pub(crate) fn forget(url: &Url) { + if let Ok(key) = won_key(url) { + WEBSOCKET_WON.lock().unwrap().remove(&key); + } +} + +/// Whether the next dial to `url` skips QUIC's head start. +#[cfg(all(test, feature = "noq"))] +pub(crate) fn won(url: &Url) -> bool { + won_key(url).is_ok_and(|key| WEBSOCKET_WON.lock().unwrap().contains(&key)) +} + /// WebSocket configuration for the client. #[derive(Clone, Debug, usage::Args, serde::Serialize, serde::Deserialize)] #[usage(unknown_flags = "error", args_override_self = false)] @@ -278,13 +305,8 @@ pub(crate) async fn connect( return Err(Error::Disabled); } - let host = url.host_str().ok_or(Error::MissingHostname)?.to_string(); - let port = url.port().unwrap_or_else(|| match url.scheme() { - "https" | "wss" | "moql" | "moqt" => 443, - "http" | "ws" => 80, - _ => 443, - }); - let key = (host.clone(), port); + let key = won_key(&url)?; + let (host, port) = key.clone(); // Apply a small penalty to WebSocket to improve odds for QUIC to connect first, // unless we've already had to fall back to WebSockets for this server. diff --git a/rs/moq-tokio/tests/backend.rs b/rs/moq-tokio/tests/backend.rs index 15d850c24b..15b1dce1b9 100644 --- a/rs/moq-tokio/tests/backend.rs +++ b/rs/moq-tokio/tests/backend.rs @@ -610,7 +610,7 @@ async fn iroh_connect() { assert_eq!(request.role(), Some(moq_tokio::moq_net::Role::Subscriber)); // iroh offers the moq ALPNs ahead of H3, so this lands on raw QUIC: no request // URL, leaving the SETUP as the only place for the request target. - assert_eq!(request.transport(), moq_tokio::server::Transport::Iroh); + assert_eq!(request.transport(), moq_tokio::Transport::Iroh); assert_eq!(request.url(), None); assert_eq!(request.path(), "/room"); assert_eq!(request.query(), Some("jwt=abc")); diff --git a/rs/moq-tokio/tests/broadcast.rs b/rs/moq-tokio/tests/broadcast.rs index d6e56e1723..221aa008d3 100644 --- a/rs/moq-tokio/tests/broadcast.rs +++ b/rs/moq-tokio/tests/broadcast.rs @@ -1928,7 +1928,7 @@ async fn broadcast_websocket() { // ── run server and client concurrently ────────────────────────── let server_handle = tokio::spawn(async move { let request = server.accept().await.expect("no incoming connection"); - assert_eq!(request.transport(), moq_tokio::server::Transport::WebSocket); + assert_eq!(request.transport(), moq_tokio::Transport::WebSocket); assert_eq!(request.path(), ""); // The dialed host reaches the server as the authority, like the QUIC transports. assert_eq!(request.authority(), Some("localhost")); @@ -2050,7 +2050,7 @@ async fn broadcast_websocket_fallback() { // ── run server and client concurrently ────────────────────────── let server_handle = tokio::spawn(async move { let request = server.accept().await.expect("no incoming connection"); - assert_eq!(request.transport(), moq_tokio::server::Transport::WebSocket); + assert_eq!(request.transport(), moq_tokio::Transport::WebSocket); assert_eq!(request.path(), "/admin"); assert_eq!(request.query(), Some("jwt=test")); assert_eq!(request.url().and_then(url::Url::query), Some("jwt=test")); @@ -2167,7 +2167,7 @@ async fn broadcast_websocket_uses_newest_version() { let server_handle = tokio::spawn(async move { let request = server.accept().await.expect("no incoming connection"); - assert_eq!(request.transport(), moq_tokio::server::Transport::WebSocket); + assert_eq!(request.transport(), moq_tokio::Transport::WebSocket); let session = request.with_publisher(&pub_origin).ok().await?; assert_eq!(session.version(), expected_version, "server negotiated stale version"); let _broadcast = broadcast; @@ -2245,7 +2245,7 @@ async fn broadcast_race_quic_wins() { let request = server.accept().await.expect("no incoming connection"); assert_eq!( request.transport(), - moq_tokio::server::Transport::Quic, + moq_tokio::Transport::WebTransport, "QUIC lost the race to WebSocket with both reachable", ); let session = request.with_publisher(&pub_origin).ok().await?;