From 4ab2f08deb2ed082e3cfebda06fe172d0584d4f1 Mon Sep 17 00:00:00 2001 From: imabdulbasit Date: Wed, 18 Feb 2026 15:12:39 +0500 Subject: [PATCH 1/4] drop handler if request is canceled --- src/listener.rs | 117 +++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 105 insertions(+), 12 deletions(-) diff --git a/src/listener.rs b/src/listener.rs index da63e6b6..cffbaca6 100644 --- a/src/listener.rs +++ b/src/listener.rs @@ -7,13 +7,13 @@ use crate::StatusCode; use async_lock::Semaphore; use async_std::{ - net::TcpListener, + net::{TcpListener, TcpStream}, sync::Arc, task::{sleep, spawn}, }; use async_trait::async_trait; use derivative::Derivative; -use futures::stream::StreamExt; +use futures::{select, stream::StreamExt, FutureExt}; use std::{ fmt::{self, Display, Formatter}, io::{self, ErrorKind}, @@ -26,6 +26,8 @@ use tide::{ Server, }; + + /// TCP listener which accepts only a limited number of connections at a time. /// /// This listener is based on `tide::listener::TcpListener` and should match the semantics of that @@ -107,17 +109,26 @@ where spawn(async move { let local_addr = stream.local_addr().ok(); let peer_addr = stream.peer_addr().ok(); + let detect_stream = stream.clone(); - let fut = async_h1::accept(stream, |mut req| async { - // Handle the request if we can get a permit. - if let Some(_guard) = permit.try_acquire() { - req.set_local_addr(local_addr); - req.set_peer_addr(peer_addr); - app.respond(req).await - } else { - // Otherwise, we are rate limited. Respond immediately with an - // error. - Ok(http::Response::new(StatusCode::TOO_MANY_REQUESTS)) + let fut = async_h1::accept(stream, |mut req| { + let detect_stream = detect_stream.clone(); + let permit = permit.clone(); + let app = app.clone(); + async move { + if let Some(_guard) = permit.try_acquire() { + req.set_local_addr(local_addr); + req.set_peer_addr(peer_addr); + select! { + result = app.respond(req).fuse() => result, + _ = wait_for_disconnect(detect_stream).fuse() => { + tracing::debug!("handler cancelled due to client disconnect"); + Err(tide::Error::from(io::Error::from(ErrorKind::ConnectionAborted))) + } + } + } else { + Ok(http::Response::new(StatusCode::TOO_MANY_REQUESTS)) + } } }); @@ -139,6 +150,24 @@ where } } +/// Resolves when the client disconnects by polling the stream with `peek`. +async fn wait_for_disconnect(stream: TcpStream) { + let mut buf = [0u8; 1]; + loop { + match stream.peek(&mut buf).await { + Ok(0) => { + tracing::debug!("client disconnected (EOF on peek)"); + return; + } + Err(e) => { + tracing::debug!(%e, "client disconnected (error on peek)"); + return; + } + Ok(_) => sleep(Duration::from_millis(100)).await, + } + } +} + impl ToListener for RateLimitListener where State: Clone + Send + Sync + 'static, @@ -179,6 +208,10 @@ mod test { }; use futures::future::{try_join_all, FutureExt}; use portpicker::pick_unused_port; + use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }; use toml::toml; use vbs::version::{StaticVersion, StaticVersionType}; @@ -234,4 +267,64 @@ mod test { assert_eq!(StatusCode::OK, res.status()); } } + + /// When a client disconnects , the handler future should be dropped + /// rather than running to completion. + #[async_std::test] + async fn test_handler_dropped_on_client_disconnect() { + + let handler_completed = Arc::new(AtomicBool::new(false)); + + let mut app = App::<_, ServerError>::with_state(()); + let api_toml = toml! { + [route.slow] + PATH = ["/slow"] + METHOD = "GET" + }; + { + let flag = handler_completed.clone(); + let mut api = app + .module::("mod", api_toml) + .unwrap(); + api.get("slow", move |_req, _state| { + let flag = flag.clone(); + async move { + sleep(Duration::from_secs(5)).await; + flag.store(true, Ordering::SeqCst); + Ok(()) + } + .boxed() + }) + .unwrap(); + } + + let port = pick_unused_port().unwrap(); + spawn(app.serve( + RateLimitListener::with_port(port, 10), + StaticVer01::instance(), + )); + + + sleep(Duration::from_secs(1)).await; + + + let req_task = spawn(async move { + reqwest::Client::new() + .get(format!("http://localhost:{port}/mod/slow")) + .send() + .await + }); + + + sleep(Duration::from_millis(200)).await; + + + req_task.cancel().await; + + sleep(Duration::from_secs(6)).await; + + assert!( + !handler_completed.load(Ordering::SeqCst), + ); + } } From 16980ffd7d6f8825a2e0ac86258a2b1aa3f7265f Mon Sep 17 00:00:00 2001 From: imabdulbasit Date: Wed, 18 Feb 2026 15:24:13 +0500 Subject: [PATCH 2/4] lint --- Cargo.toml | 2 +- src/listener.rs | 23 +++++++---------------- 2 files changed, 8 insertions(+), 17 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 34d89209..fb848fdd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tide-disco" -version = "0.9.6" +version = "0.9.7" edition = "2021" authors = ["Espresso Systems "] description = "Discoverability for Tide" diff --git a/src/listener.rs b/src/listener.rs index cffbaca6..ac1c9f49 100644 --- a/src/listener.rs +++ b/src/listener.rs @@ -26,8 +26,6 @@ use tide::{ Server, }; - - /// TCP listener which accepts only a limited number of connections at a time. /// /// This listener is based on `tide::listener::TcpListener` and should match the semantics of that @@ -102,7 +100,7 @@ where tracing::warn!(%err, "TCP error"); sleep(Duration::from_millis(500)).await; continue; - } + }, Ok(stream) => { let app = server.clone(); let permit = self.permit.clone(); @@ -136,7 +134,7 @@ where tracing::error!(%error, "HTTP error"); } }); - } + }, }; } Ok(()) @@ -158,11 +156,11 @@ async fn wait_for_disconnect(stream: TcpStream) { Ok(0) => { tracing::debug!("client disconnected (EOF on peek)"); return; - } + }, Err(e) => { tracing::debug!(%e, "client disconnected (error on peek)"); return; - } + }, Ok(_) => sleep(Duration::from_millis(100)).await, } } @@ -185,7 +183,7 @@ impl Display for RateLimitListener { Some(listener) => { let addr = listener.local_addr().expect("Could not get local addr"); write!(f, "http://{}", addr) - } + }, None => write!(f, "http://{}", self.addr), } } @@ -272,7 +270,6 @@ mod test { /// rather than running to completion. #[async_std::test] async fn test_handler_dropped_on_client_disconnect() { - let handler_completed = Arc::new(AtomicBool::new(false)); let mut app = App::<_, ServerError>::with_state(()); @@ -304,10 +301,8 @@ mod test { StaticVer01::instance(), )); - sleep(Duration::from_secs(1)).await; - let req_task = spawn(async move { reqwest::Client::new() .get(format!("http://localhost:{port}/mod/slow")) @@ -315,16 +310,12 @@ mod test { .await }); - sleep(Duration::from_millis(200)).await; - req_task.cancel().await; - sleep(Duration::from_secs(6)).await; + sleep(Duration::from_secs(6)).await; - assert!( - !handler_completed.load(Ordering::SeqCst), - ); + assert!(!handler_completed.load(Ordering::SeqCst),); } } From d426cc7bed0c91a90217609f712d60c53414383f Mon Sep 17 00:00:00 2001 From: imabdulbasit Date: Wed, 18 Feb 2026 15:31:16 +0500 Subject: [PATCH 3/4] lint --- src/listener.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/listener.rs b/src/listener.rs index ac1c9f49..ab97e104 100644 --- a/src/listener.rs +++ b/src/listener.rs @@ -100,7 +100,7 @@ where tracing::warn!(%err, "TCP error"); sleep(Duration::from_millis(500)).await; continue; - }, + } Ok(stream) => { let app = server.clone(); let permit = self.permit.clone(); @@ -134,7 +134,7 @@ where tracing::error!(%error, "HTTP error"); } }); - }, + } }; } Ok(()) @@ -156,11 +156,11 @@ async fn wait_for_disconnect(stream: TcpStream) { Ok(0) => { tracing::debug!("client disconnected (EOF on peek)"); return; - }, + } Err(e) => { tracing::debug!(%e, "client disconnected (error on peek)"); return; - }, + } Ok(_) => sleep(Duration::from_millis(100)).await, } } @@ -183,7 +183,7 @@ impl Display for RateLimitListener { Some(listener) => { let addr = listener.local_addr().expect("Could not get local addr"); write!(f, "http://{}", addr) - }, + } None => write!(f, "http://{}", self.addr), } } From f71e49107ad604b4432f44f2cd6232888da85ed2 Mon Sep 17 00:00:00 2001 From: imabdulbasit Date: Wed, 18 Feb 2026 15:32:16 +0500 Subject: [PATCH 4/4] lockfile --- Cargo.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 5162512a..ab62f83f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3618,7 +3618,7 @@ dependencies = [ [[package]] name = "tide-disco" -version = "0.9.6" +version = "0.9.7" dependencies = [ "anyhow", "ark-serialize",