diff --git a/quest/m1/transport-upgrade/README.md b/quest/m1/transport-upgrade/README.md index f3bd6c0174..610a6b1942 100644 --- a/quest/m1/transport-upgrade/README.md +++ b/quest/m1/transport-upgrade/README.md @@ -58,8 +58,9 @@ Shared decisions: ## Quests -- [Rust edge cases](/quest/m1/transport-upgrade/polish.md) - no false GOAWAY warning on upgrade, and a failed WebSocket handshake falls back to the pending QUIC dial - [JavaScript](/quest/m1/transport-upgrade/js.md) - js/net keeps the WebTransport dial after WebSocket wins and migrates through the client-goaway handover +- [Closed fallback](/quest/m1/transport-upgrade/closed-fallback.md) - a WebSocket session that closes right after connecting falls back to the pending QUIC dial instead of redialing +- [JS qmux finish](/quest/m1/transport-upgrade/js-qmux-finish.md) - `@moq/qmux` reports a cleanly finished send stream as closed without error ## Related diff --git a/quest/m1/transport-upgrade/closed-fallback.md b/quest/m1/transport-upgrade/closed-fallback.md new file mode 100644 index 0000000000..8df94079c1 --- /dev/null +++ b/quest/m1/transport-upgrade/closed-fallback.md @@ -0,0 +1,30 @@ +# [S] Fall back to QUIC when the WebSocket session closes at once + +## Goal + +When WebSocket wins the dial race and its session closes right after +connecting, the `Connection` moves onto the QUIC dial that is still pending +instead of dropping it and redialing. On moq-lite and IETF draft 17+ the +client handshake does not wait for the server, so a server refusing the +WebSocket session looks like this rather than like a handshake failure, which +already falls back. + +## Plan + +- Today `run_session` in `moq-tokio`'s `connection.rs` returns when the + WebSocket session closes, and the pending upgrade is dropped. The redial + starts a fresh QUIC dial and has lost QUIC's head start. +- Decide what counts as "at once". The unhealthy-session threshold that + already feeds the backoff may be the right line. Keep the connect deadline + that bounds the pending dial as the only deadline. +- An auth close on the WebSocket session is probably still terminal, the way + it is for a lone session. Check how the race treats a mixed auth pair and + match it. +- Test with the held-QUIC forwarder in the connection tests, as the handshake + fallback test does, with a default moq-lite client. + +Public API: none. Wire: none. + +## Related + +- [#4296](https://github.com/moq-dev/moq/pull/4296) - the handshake-failure fallback this extends diff --git a/quest/m1/transport-upgrade/js-qmux-finish.md b/quest/m1/transport-upgrade/js-qmux-finish.md new file mode 100644 index 0000000000..1ce689a9bd --- /dev/null +++ b/quest/m1/transport-upgrade/js-qmux-finish.md @@ -0,0 +1,22 @@ +# [XS] Clean finish in @moq/qmux + +## Goal + +`@moq/qmux` reports a send stream that finished cleanly as closed without +error, so `@moq/net` does not treat a delivered GOAWAY or SETUP as a failure +over the WebSocket fallback. + +## Plan + +The Rust qmux crate reported every finished stream as `connection closed` +(fixed in moq-dev/web-transport#402). Check whether the TypeScript peer in +`moq-dev/web-transport` (`js/qmux`) has the same bug, and whether `@moq/net` +waits on a finished writable anywhere it would notice. If it does, fix it +there with a regression test, release, and bump `@moq/qmux` in `js/net`. If +not, note that in the PR that deletes this quest. + +Public API: none. Wire: none. + +## Related + +- [JavaScript upgrade](/quest/m1/transport-upgrade/js.md) - sends a GOAWAY over the WebSocket session on every upgrade diff --git a/quest/m1/transport-upgrade/polish.md b/quest/m1/transport-upgrade/polish.md deleted file mode 100644 index 9a3c1b602f..0000000000 --- a/quest/m1/transport-upgrade/polish.md +++ /dev/null @@ -1,24 +0,0 @@ -# [S] WebSocket upgrade edge cases - -## Goal - -The Rust WebSocket-to-QUIC upgrade leaves no false warnings and no avoidable -failures: a GOAWAY the peer received is not logged as a send failure, and a -WebSocket handshake that fails after WebSocket won the dial race falls back to -the still-pending QUIC dial instead of failing the connection attempt. - -## Plan - -- Every upgrade logs `failed to send goaway: transport: connection closed` - although the server receives it. Find why the send reports failure (likely - the close racing the flush) and fix it at the source. -- When WebSocket wins the race but its MoQ handshake fails, the pending QUIC - dial is still alive; use it rather than failing, bounded by the existing - connect timeout. -- Tests for both, including the deterministic QUIC-held forwarder from #4189. - -Public API: none. Wire: none. - -## Related - -- [#4189](https://github.com/moq-dev/moq/pull/4189) - the WebSocket-to-QUIC upgrade this polish quest follows diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index 300765d61b..bd373053d4 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -443,7 +443,8 @@ impl Client { /// 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. + /// [`Dialed::upgrade`] rather than being dropped, and the attempt falls back to it + /// if the MoQ handshake over WebSocket fails. /// /// `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 @@ -462,7 +463,8 @@ impl Client { Q: Future> + Unpin + Send + 'static, S: moq_net::transport::poll::Boxable, { - let transport = quic_transport(addr.url()); + let url = addr.url().clone(); + let transport = quic_transport(&url); let alpns = self.versions.alpns(); let ws_config = self.websocket.clone(); let ws_tls = self.tls.clone(); @@ -476,7 +478,27 @@ impl Client { match race_transport_connect(quic, 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 session = match connect_session(&self.moq, crate::transport::Session::new(session)).await { + Ok(session) => session, + Err(err) => { + // The fallback got through but its MoQ handshake did not. A QUIC dial still + // pending may yet connect, bounded by the same deadline as the race. + let err = Error::from(err); + let Some(quic) = quic else { return Err(err) }; + tracing::warn!(%err, "WebSocket handshake failed; waiting on QUIC"); + let quic = match quic.await { + Ok(quic) => quic, + Err(quic) => return Err(race_error(quic, err)), + }; + // UDP gets through after all, so the next dial gives QUIC its head start. + crate::websocket::forget(&url); + // Both handshakes failing is still a two-arm loss: a mixed auth pair stays retryable. + let session = connect_session(moq, quic) + .await + .map_err(|quic| race_error(quic.into(), err))?; + return Ok(Dialed::new(session, transport)); + } + }; let mut dialed = Dialed::new(session, crate::Transport::WebSocket); dialed.upgrade = quic.map(|quic| { let moq = moq.clone(); @@ -719,23 +741,30 @@ where } } - // Auth is terminal only when both arms refused. A WebTransport-only endpoint - // answers the fallback with 403 while QUIC is still in flight, and reconnect - // treats is_auth() as terminal, so a mixed pair reports the retryable error. match (quic_err, websocket_err) { - (Some(quic), Some(websocket)) => Err(match (quic.is_auth(), websocket.is_auth()) { - (false, false) => Error::TransportRace { - quic: std::sync::Arc::new(quic), - websocket: std::sync::Arc::new(websocket), - }, - (true, false) => websocket, - _ => quic, - }), + (Some(quic), Some(websocket)) => Err(race_error(quic, websocket)), (Some(err), None) | (None, Some(err)) => Err(err), (None, None) => Err(Error::ConnectFailed), } } +/// The error for a race both arms lost. +/// +/// Auth is terminal only when both arms refused. A WebTransport-only endpoint +/// answers the fallback with 403 while QUIC is still in flight, and reconnect +/// treats is_auth() as terminal, so a mixed pair reports the retryable error. +#[cfg(all(feature = "websocket", feature = "noq"))] +fn race_error(quic: Error, websocket: Error) -> Error { + match (quic.is_auth(), websocket.is_auth()) { + (false, false) => Error::TransportRace { + quic: std::sync::Arc::new(quic), + websocket: std::sync::Arc::new(websocket), + }, + (true, false) => websocket, + _ => quic, + } +} + #[cfg(any( feature = "noq", feature = "iroh", diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs index 3e49d87628..e24217411a 100644 --- a/rs/moq-tokio/src/connection.rs +++ b/rs/moq-tokio/src/connection.rs @@ -2066,9 +2066,22 @@ mod tests { accepted: tokio::sync::mpsc::UnboundedReceiver<(crate::Transport, moq_net::Session)>, } + /// When a [`Fallback`] lets QUIC through. + #[cfg(all(feature = "websocket", feature = "noq"))] + #[derive(Clone, Copy, PartialEq)] + enum Quic { + /// From the start. + Open, + /// Once the test calls [`Forwarder::open`]. + Held, + /// Once the WebSocket fallback has closed its first session mid-handshake. No + /// MoQ server sits behind the fallback in this mode. + AfterWebSocketFails, + } + #[cfg(all(feature = "websocket", feature = "noq"))] impl Fallback { - async fn start(origin: &moq_net::origin::Producer, open: bool) -> Self { + async fn start(origin: &moq_net::origin::Producer, quic: Quic) -> Self { let mut listen = crate::listen::Config { bind: Some("127.0.0.1:0".parse().unwrap()), ..Default::default() @@ -2090,16 +2103,36 @@ mod tests { panic!("could not bind a matching TCP and UDP port after 20 attempts"); }; + // With no MoQ server behind it, the fallback's session gets as far as the MoQ + // handshake and no further. + let (websocket, failing) = match quic { + Quic::AfterWebSocketFails => (None, Some(websocket)), + _ => (Some(websocket), None), + }; + let config = crate::server::Config { listen, - websocket: Some(websocket), + websocket, publisher: Some(origin.consume()), ..Default::default() }; let mut server = config.init().unwrap().listen().await.unwrap(); - let quic = server.local_addr().unwrap(); + let server_addr = server.local_addr().unwrap(); let port = forwarder.local_addr().unwrap().port(); - let forwarder = Forwarder::start(forwarder, quic, open).await; + let forwarder = Forwarder::start(forwarder, server_addr, quic == Quic::Open).await; + + if let Some(websocket) = failing { + let gate = forwarder.clone(); + tokio::spawn(async move { + use web_transport_trait::Session as _; + let session = websocket.accept().await.unwrap().unwrap(); + // The client's SETUP stream: WebSocket has won the race. + let _setup = session.accept_bi().await.unwrap(); + session.close(1, "no MoQ here"); + gate.open(); + session.closed().await; + }); + } let (tx, accepted) = tokio::sync::mpsc::unbounded_channel(); tokio::spawn(async move { @@ -2131,8 +2164,9 @@ mod tests { /// 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"))] + #[derive(Clone)] struct Forwarder { - gate: tokio::sync::watch::Sender, + gate: std::sync::Arc>, } #[cfg(all(feature = "websocket", feature = "noq"))] @@ -2184,7 +2218,7 @@ mod tests { } }); - Self { gate: tx } + Self { gate: tx.into() } } /// Start forwarding, releasing whatever was held. @@ -2256,7 +2290,7 @@ mod tests { group.finish().unwrap(); }; - let mut fallback = Fallback::start(&origin, false).await; + let mut fallback = Fallback::start(&origin, Quic::Held).await; let subscriber = crate::origin::spawn(); let mut config = crate::connect::Config::default(); @@ -2339,6 +2373,37 @@ mod tests { assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); } + /// When WebSocket wins the race but its MoQ handshake fails, the attempt falls back + /// to the QUIC dial still pending instead of failing. + /// + /// One-shot, so a redial cannot stand in for the fallback. Draft 16, since it is the + /// newest version whose client handshake waits on the server's SETUP. + #[cfg(all(feature = "websocket", feature = "noq"))] + #[tokio::test] + async fn failed_websocket_handshake_falls_back_to_quic() { + let origin = crate::origin::spawn(); + let mut fallback = Fallback::start(&origin, Quic::AfterWebSocketFails).await; + + let mut config = crate::connect::Config::default(); + config.tls.insecure = Some(true); + config.version = vec!["moq-transport-16".parse().unwrap()]; + let client = config.init(Default::default()).unwrap().with_reconnect(false); + let connection = tokio::time::timeout(UPGRADE_WAIT, client.connect(fallback.url.clone()).established()) + .await + .expect("never connected") + .expect("a failed WebSocket handshake must fall back to the pending QUIC dial"); + + assert_eq!(connection.transport(), Some(crate::Transport::WebTransport)); + assert_eq!(connection.epoch(), 1); + let (transport, _session) = 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 fallback kept WebSocket's win" + ); + } + /// When QUIC wins the race nothing changes: one session, over QUIC, and no /// WebSocket session is ever opened. #[cfg(all(feature = "websocket", feature = "noq"))] @@ -2349,7 +2414,7 @@ mod tests { broadcast.announce(Default::default()).unwrap(); let track = broadcast.create_track("video", None).unwrap(); - let mut fallback = Fallback::start(&origin, true).await; + let mut fallback = Fallback::start(&origin, Quic::Open).await; let subscriber = crate::origin::spawn(); let mut config = crate::connect::Config::default();