fix(tokio): WebSocket upgrade edge cases

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
 

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

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

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<Output = crate::Result<S>> + 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",

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<bool>,
+		gate: std::sync::Arc<tokio::sync::watch::Sender<bool>>,
 	}
 
 	#[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();