From 41b145dd04b2655796b9e55ff6e2043dec090fd1 Mon Sep 17 00:00:00 2001 From: Tin Lai Date: Fri, 9 Oct 2026 13:11:10 +1100 Subject: [PATCH] fix(webtransport): cancel pending handshakes when frontend closes --- Cargo.lock | 1 + crates/aeronet_webtransport/Cargo.toml | 3 + .../src/client/backend.rs | 75 +++++++++++++++++-- 3 files changed, 73 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bd24ba31..470d0604 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -298,6 +298,7 @@ dependencies = [ "futures", "js-sys", "spki", + "tokio", "tracing", "wasm-bindgen", "wtransport", diff --git a/crates/aeronet_webtransport/Cargo.toml b/crates/aeronet_webtransport/Cargo.toml index 1031b81b..6cfb9e33 100644 --- a/crates/aeronet_webtransport/Cargo.toml +++ b/crates/aeronet_webtransport/Cargo.toml @@ -59,6 +59,9 @@ wtransport = { workspace = true } x509-cert = { workspace = true } xwt-wtransport = { workspace = true } +[target.'cfg(not(target_family = "wasm"))'.dev-dependencies] +tokio = { workspace = true, features = ["rt", "net", "time"] } + [target.'cfg(target_family = "wasm")'.dependencies] js-sys = { workspace = true } wasm-bindgen = { workspace = true } diff --git a/crates/aeronet_webtransport/src/client/backend.rs b/crates/aeronet_webtransport/src/client/backend.rs index 97e7117a..0538b049 100644 --- a/crates/aeronet_webtransport/src/client/backend.rs +++ b/crates/aeronet_webtransport/src/client/backend.rs @@ -11,10 +11,24 @@ use { xwt_core::prelude::*, }; +// Dropping a Rust future does not cancel a JavaScript `ready` promise. +// Explicitly close the browser transport unless ownership reaches Session. +#[cfg(target_family = "wasm")] +struct PendingWebTransport(Option); + +#[cfg(target_family = "wasm")] +impl Drop for PendingWebTransport { + fn drop(&mut self) { + if let Some(transport) = self.0.take() { + transport.close(); + } + } +} + pub async fn start( config: ClientConfig, target: ConnectTarget, - tx_next: oneshot::Sender, + mut tx_next: oneshot::Sender, ) -> Result { let endpoint = { #[cfg(target_family = "wasm")] @@ -35,16 +49,20 @@ pub async fn start( }; debug!("Created endpoint"); - let conn = { + let connecting = async { #[cfg(target_family = "wasm")] { - endpoint + let connecting = endpoint .connect(&target) .await - .map_err(|err| ClientError::Connect(err.into()))? + .map_err(|err| ClientError::Connect(err.into()))?; + let mut pending = PendingWebTransport(Some(connecting.transport.clone())); + let session = connecting .wait_connect() .await - .map_err(|err| ClientError::AwaitConnect(err.into()))? + .map_err(|err| ClientError::AwaitConnect(err.into()))?; + pending.0 = None; + Ok::<_, ClientError>(session) } #[cfg(not(target_family = "wasm"))] @@ -54,7 +72,18 @@ pub async fn start( .connect(target) .await .map(xwt_wtransport::Connection) - .map_err(ClientError::Connect)? + .map_err(ClientError::Connect) + } + }; + // Cancellation must also cover the handshake, before SessionBackend has + // a disconnect channel. Dropping the frontend's receiver cancels this + // future and drops the native endpoint rather than leaving a retired task. + let conn = { + let cancelled = tx_next.cancellation(); + futures::pin_mut!(connecting, cancelled); + match futures::future::select(cancelled, connecting).await { + futures::future::Either::Left(_) => return Err(SessionError::FrontendClosed.into()), + futures::future::Either::Right((result, _)) => result?, } }; debug!("Connected"); @@ -92,3 +121,37 @@ pub async fn start( debug!("Starting session loop"); Err(backend.start().await) } + +#[cfg(all(test, not(target_family = "wasm")))] +mod tests { + use {super::*, core::time::Duration, wtransport::endpoint::IntoConnectOptions}; + + #[test] + fn dropping_frontend_cancels_pending_handshake_without_waiting_for_network_timeout() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + // Bound but silent UDP peer: no connection-refused shortcut. + let peer = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let config = ClientConfig::builder() + .with_bind_default() + .with_native_certs() + .max_idle_timeout(Some(Duration::from_secs(5))) + .unwrap() + .build(); + let (tx, rx) = oneshot::channel(); + let target = format!("https://{}/", peer.local_addr().unwrap()).into_options(); + let task = tokio::spawn(start(config, target, tx)); + tokio::task::yield_now().await; + drop(rx); + let result = tokio::time::timeout(Duration::from_millis(500), task).await; + assert!( + result.is_ok(), + "abandoned handshake continued running after fallback" + ); + assert!(result.unwrap().unwrap().is_err()); + }); + } +}