From c056e32596592e2479cf58623282cc7ff9abdfbc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 12:24:34 -0700 Subject: [PATCH 1/4] quest: claim transport-upgrade/rust Co-Authored-By: Claude Opus 5.5 From 07f41036eabc98101258917d1465a77f94206d18 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 13:15:04 -0700 Subject: [PATCH 2/4] feat(tokio): upgrade a WebSocket fallback session to QUIC When WebSocket wins the race, the QUIC dial keeps going. If it lands, the Connection runs the MoQ handshake on it, swaps it in as the live session, sends an empty-URI GOAWAY on the WebSocket session, and drains it within the handover cap. The "WebSocket won" memo forgets the URL so the next dial gives QUIC its head start again. Adds Connection::transport(), reusing server::Transport as moq_tokio::Transport. Co-Authored-By: Claude Opus 5.5 --- doc/concept/transport.md | 3 +- doc/lib/rs/index.md | 4 +- quest/m1/transport-upgrade/README.md | 6 +- quest/m1/transport-upgrade/rust.md | 49 --- rs/moq-tokio/src/client.rs | 272 +++++++++++++--- rs/moq-tokio/src/connection.rs | 467 +++++++++++++++++++++++++-- rs/moq-tokio/src/lib.rs | 1 + rs/moq-tokio/src/server.rs | 36 +-- rs/moq-tokio/src/transport.rs | 37 +++ rs/moq-tokio/src/websocket.rs | 38 ++- 10 files changed, 746 insertions(+), 167 deletions(-) delete mode 100644 quest/m1/transport-upgrade/rust.md 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/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-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index e9133a8e89..65c8fbe585 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, crate::Transport::Quic)); } } @@ -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,9 +457,9 @@ 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 alpns = self.versions.alpns(); @@ -452,9 +473,18 @@ 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?, crate::Transport::Quic)), + 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(); + Upgrade::new(Box::pin(async move { + let quic = quic.await?; + Ok(Box::pin(async move { Ok(connect_session(&moq, quic).await?) }) as Handshake) + })) + }); + Ok(dialed) } } } @@ -516,20 +546,120 @@ fn setup_path(url: &Url) -> Option { } } +/// 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, + /// 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. + Done(moq_net::Session), +} + +#[cfg_attr(not(all(feature = "websocket", feature = "noq")), allow(dead_code))] +impl Upgrade { + pub(crate) fn new(dial: BoxFuture<'static, crate::Result>) -> Self { + Self { + stage: Stage::Dialing(dial), + 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) => { + Poll::Ready(Ok(Step::Done(ready!(waiter.poll_future(handshake.as_mut()))?))) + } + } + } +} + #[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 +681,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 +1274,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 +1290,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 +1308,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 +1343,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 +1364,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::Quic); + drop(dialed); accepted.await.unwrap().expect("server handshake failed"); } @@ -1237,7 +1375,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 +1390,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 +1403,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 +1441,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..550f3ffaeb 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) => { + tracing::info!(peer = %Endpoint(&url), "upgraded from WebSocket to QUIC"); + // 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, crate::Transport::Quic); + // 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 replaces the + /// WebSocket one. + Upgraded(moq_net::Session), +} + /// 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(Next::Upgraded) }) .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 { + 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)) => { + *upgrade = None; + return Poll::Ready(session); + } + 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,329 @@ 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) }))); + + 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::Quic)); + assert_eq!(connection.epoch(), 2); + let (transport, _quic) = fallback.accept().await; + assert_eq!(transport, crate::Transport::Quic); + // 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::Quic)); + } + + /// 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::Quic)); + let (transport, _session) = fallback.accept().await; + assert_eq!(transport, crate::Transport::Quic); + + 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..75a522294d 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -1098,40 +1098,8 @@ 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()) - } -} +/// Re-exported here too, where the accept side first named it. +pub use crate::Transport; /// An incoming MoQ session that can be accepted or rejected. /// diff --git a/rs/moq-tokio/src/transport.rs b/rs/moq-tokio/src/transport.rs index ffc046c658..839eedf8ad 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,41 @@ 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 { + /// 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()) + } +} + /// 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. From a2476af00171934c7352ae7303f4dcaa56ae19bf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 14:26:22 -0700 Subject: [PATCH 3/4] feat(tokio)!: moq_tokio::Transport at the root, with a WebTransport variant Moves server::Transport to moq_tokio::Transport with no re-export, and splits WebTransport from raw QUIC on both the accept and dial sides. moq-ffi gains MoqTransport::WebTransport; the relay maps it to moq_auth::Transport::Quic. Co-Authored-By: Claude Opus 5.5 --- dart/moq_ffi/lib/src/moq.dart | 4 +++- rs/moq-cli/src/main.rs | 12 +++++------ rs/moq-ffi/src/server.rs | 25 ++++++++++++++--------- rs/moq-relay/src/auth.rs | 11 ++++++----- rs/moq-relay/src/uring.rs | 2 +- rs/moq-tokio/src/client.rs | 34 +++++++++++++++++++++++--------- rs/moq-tokio/src/connection.rs | 35 ++++++++++++++++++--------------- rs/moq-tokio/src/server.rs | 10 +++++----- rs/moq-tokio/src/transport.rs | 5 ++++- rs/moq-tokio/tests/backend.rs | 2 +- rs/moq-tokio/tests/broadcast.rs | 8 ++++---- 11 files changed, 90 insertions(+), 58 deletions(-) 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/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 65c8fbe585..300765d61b 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -418,7 +418,7 @@ impl Client { { let session = quic_handle.await?; let session = connect_session(&moq, session).await?; - return Ok(Dialed::new(session, crate::Transport::Quic)); + return Ok(Dialed::new(session, quic_transport(&url))); } } @@ -462,6 +462,7 @@ impl Client { 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(); @@ -473,16 +474,17 @@ impl Client { }; match race_transport_connect(quic, websocket).await? { - TransportRace::Quic(quic) => Ok(Dialed::new(connect_session(moq, quic).await?, crate::Transport::Quic)), + 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(); - Upgrade::new(Box::pin(async move { + 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) } @@ -546,6 +548,16 @@ 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, @@ -584,6 +596,8 @@ pub(crate) type Handshake = BoxFuture<'static, crate::Result>; /// 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)>, @@ -601,15 +615,16 @@ enum Stage { 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. - Done(moq_net::Session), + /// 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>) -> Self { + pub(crate) fn new(dial: BoxFuture<'static, crate::Result>, transport: crate::Transport) -> Self { Self { stage: Stage::Dialing(dial), + transport, deadline: None, } } @@ -637,7 +652,8 @@ impl Upgrade { Poll::Ready(Ok(Step::Handshaking)) } Stage::Handshaking(handshake) => { - Poll::Ready(Ok(Step::Done(ready!(waiter.poll_future(handshake.as_mut()))?))) + let session = ready!(waiter.poll_future(handshake.as_mut()))?; + Poll::Ready(Ok(Step::Done(session, self.transport))) } } } @@ -1364,7 +1380,7 @@ mod tests { .expect("client connect timed out") .expect("a fallback refused on auth must not end a connect whose QUIC arm succeeds"); - assert_eq!(dialed.transport, crate::Transport::Quic); + assert_eq!(dialed.transport, crate::Transport::WebTransport); drop(dialed); accepted.await.unwrap().expect("server handshake failed"); } diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs index 550f3ffaeb..3e49d87628 100644 --- a/rs/moq-tokio/src/connection.rs +++ b/rs/moq-tokio/src/connection.rs @@ -781,8 +781,8 @@ impl Connection { let ended = loop { match run_session(shared, &session, &mut draining, &mut upgrade).await { Next::Ended(ended) => break ended, - Next::Upgraded(next) => { - tracing::info!(peer = %Endpoint(&url), "upgraded from WebSocket to QUIC"); + 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); @@ -791,7 +791,7 @@ impl Connection { // 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, crate::Transport::Quic); + 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); @@ -1215,9 +1215,9 @@ fn retry_wait(delay: Duration, retry_start: tokio::time::Instant, timeout: Durat enum Next { /// The session stopped being the live one. Ended(Ended), - /// The pending QUIC dial finished its handshake; this session replaces the - /// WebSocket one. - Upgraded(moq_net::Session), + /// 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. @@ -1360,7 +1360,7 @@ async fn run_session( return Poll::Ready(Next::Ended(Ended::Closed(Err(err)))); } - poll_upgrade(shared, upgrade, waiter).map(Next::Upgraded) + poll_upgrade(shared, upgrade, waiter).map(|(session, transport)| Next::Upgraded(session, transport)) }) .await } @@ -1370,13 +1370,13 @@ fn poll_upgrade( shared: &Shared, upgrade: &mut Option, waiter: &kio::Waiter, -) -> Poll { +) -> 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)) => { + Ok(crate::client::Step::Done(session, transport)) => { *upgrade = None; - return Poll::Ready(session); + return Poll::Ready((session, transport)); } Err(err) => { // Only the Handshaking stage moved the status, but restoring it is harmless @@ -2224,7 +2224,10 @@ mod tests { let _ = failed.await; Err(Error::ConnectFailed) }); - let mut upgrade = Some(crate::client::Upgrade::new(Box::pin(async move { Ok(handshake) }))); + 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)); @@ -2306,10 +2309,10 @@ mod tests { "the WebSocket session was not counted as ended" ); assert!(connection.connected()); - assert_eq!(connection.transport(), Some(crate::Transport::Quic)); + assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); assert_eq!(connection.epoch(), 2); let (transport, _quic) = fallback.accept().await; - assert_eq!(transport, crate::Transport::Quic); + 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), @@ -2333,7 +2336,7 @@ mod tests { tokio::time::timeout(UPGRADE_WAIT, websocket.closed()) .await .expect("the WebSocket session outlived the handover cap"); - assert_eq!(connection.transport(), Some(crate::Transport::Quic)); + assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); } /// When QUIC wins the race nothing changes: one session, over QUIC, and no @@ -2359,9 +2362,9 @@ mod tests { .await .expect("never connected") .unwrap(); - assert_eq!(connection.transport(), Some(crate::Transport::Quic)); + assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); let (transport, _session) = fallback.accept().await; - assert_eq!(transport, crate::Transport::Quic); + assert_eq!(transport, crate::Transport::WebTransport); let cam = tokio::time::timeout(UPGRADE_WAIT, subscriber.consume().routed_broadcast("cam")) .await diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 75a522294d..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,9 +1100,6 @@ pub struct Link { pub alpn: Option, } -/// Re-exported here too, where the accept side first named it. -pub use crate::Transport; - /// An incoming MoQ session that can be accepted or rejected. /// /// The transport connection and the MoQ SETUP are already complete, so [`path`](Self::path), @@ -1691,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 839eedf8ad..0331ff6322 100644 --- a/rs/moq-tokio/src/transport.rs +++ b/rs/moq-tokio/src/transport.rs @@ -20,7 +20,7 @@ use web_transport_trait::poll as wt_poll; #[non_exhaustive] #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub enum Transport { - /// QUIC, either directly or through WebTransport over HTTP/3. + /// Raw QUIC, negotiating a MoQ ALPN directly. Quic, /// An Iroh QUIC connection. Iroh, @@ -30,6 +30,8 @@ pub enum Transport { Tcp, /// A Unix domain socket using qmux framing. Unix, + /// WebTransport over HTTP/3 on QUIC. + WebTransport, } impl Transport { @@ -41,6 +43,7 @@ impl Transport { Self::WebSocket => "websocket", Self::Tcp => "tcp", Self::Unix => "unix", + Self::WebTransport => "webtransport", } } } 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?; From 6bd6ee85b90cabe513d8fda2227d4bf38321e8b1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 15:30:16 -0700 Subject: [PATCH 4/4] fix(go): export TransportWebTransport The FFI enum gained WebTransport, but the Go wrapper only re-exported constants through TransportUnix. Co-Authored-By: Grok 4.7 --- go/wrapper/server.go | 2 ++ 1 file changed, 2 insertions(+) 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).