Repository: seanmonstar/reqwest
Pull request #2684 (2025-05-19 20:34:01)

refactor: use hyper-util `Tunnel`
cc #2683 


Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 88f2afa..7868e43 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -173,7 +173,7 @@ futures-channel = { version = "0.3", optional = true }
 [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
 env_logger = "0.10"
 hyper = { version = "1.1.0", default-features = false, features = ["http1", "http2", "client", "server"] }
-hyper-util = { version = "0.1.12", features = ["http1", "http2", "client", "client-legacy", "server-auto", "tokio"] }
+hyper-util = { version = "0.1.12", features = ["http1", "http2", "client", "client-legacy", "server-auto", "server-graceful", "tokio"] }
 serde = { version = "1.0", features = ["derive"] }
 flate2 = "1.0.13"
 brotli_crate = { package = "brotli", version = "7.0.0" }
@@ -181,7 +181,6 @@ zstd_crate = { package = "zstd", version = "0.13" }
 doc-comment = "0.3"
 tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread"] }
 futures-util = { version = "0.3.28", default-features = false, features = ["std", "alloc"] }
-rustls = { version = "0.23", default-features = false, features = ["ring"] }
 
 # wasm
 
diff --git a/src/async_impl/client.rs b/src/async_impl/client.rs
index 58fe2f7..e00cdbb 100644
--- a/src/async_impl/client.rs
+++ b/src/async_impl/client.rs
@@ -1494,13 +1494,14 @@ impl ClientBuilder {
     /// # Example
     ///
     /// ```
-    /// # #[cfg(all(feature = "__rustls", not(feature = "__rustls-ring")))]
-    /// # let _ = rustls::crypto::ring::default_provider().install_default();
+    /// # fn doc() -> Result<(), reqwest::Error> {
     /// use std::net::IpAddr;
     /// let local_addr = IpAddr::from([12, 4, 1, 8]);
     /// let client = reqwest::Client::builder()
     ///     .local_address(local_addr)
-    ///     .build().unwrap();
+    ///     .build()?;
+    /// # Ok(())
+    /// # }
     /// ```
     pub fn local_address<T>(mut self, addr: T) -> ClientBuilder
     where
@@ -1531,12 +1532,13 @@ impl ClientBuilder {
     /// # Example
     ///
     /// ```
-    /// # #[cfg(all(feature = "__rustls", not(feature = "__rustls-ring")))]
-    /// # let _ = rustls::crypto::ring::default_provider().install_default();
+    /// # fn doc() -> Result<(), reqwest::Error> {
     /// let interface = "lo";
     /// let client = reqwest::Client::builder()
     ///     .interface(interface)
-    ///     .build().unwrap();
+    ///     .build()?;
+    /// # Ok(())
+    /// # }
     /// ```
     ///
     /// [man-7-socket]: https://man7.org/linux/man-pages/man7/socket.7.html
diff --git a/src/connect.rs b/src/connect.rs
index 40cd1eb..decf26d 100644
--- a/src/connect.rs
+++ b/src/connect.rs
@@ -632,24 +632,27 @@ impl ConnectorService {
             #[cfg(feature = "default-tls")]
             Inner::DefaultTls(http, tls) => {
                 if dst.scheme() == Some(&Scheme::HTTPS) {
-                    let host = dst.host().to_owned();
-                    let port = dst.port().map(|p| p.as_u16()).unwrap_or(443);
-                    let http = http.clone();
-                    let tls_connector = tokio_native_tls::TlsConnector::from(tls.clone());
-                    let mut http = hyper_tls::HttpsConnector::from((http, tls_connector));
-                    let conn = http.call(proxy_dst).await?;
                     log::trace!("tunneling HTTPS over proxy");
-                    let tunneled = tunnel(
-                        conn,
-                        host.ok_or("no host in url")?.to_string(),
-                        port,
-                        self.user_agent.clone(),
-                        auth,
-                    )
-                    .await?;
+                    let tls_connector = tokio_native_tls::TlsConnector::from(tls.clone());
+                    let inner =
+                        hyper_tls::HttpsConnector::from((http.clone(), tls_connector.clone()));
+                    // TODO: we could cache constructing this
+                    let mut tunnel =
+                        hyper_util::client::legacy::connect::proxy::Tunnel::new(proxy_dst, inner);
+                    if let Some(auth) = auth {
+                        tunnel = tunnel.with_auth(auth);
+                    }
+                    if let Some(ua) = self.user_agent {
+                        let mut headers = http::HeaderMap::new();
+                        headers.insert(http::header::USER_AGENT, ua);
+                        tunnel = tunnel.with_headers(headers);
+                    }
+                    // We don't wrap this again in an HttpsConnector since that uses Maybe,
+                    // and we know this is definitely HTTPS.
+                    let tunneled = tunnel.call(dst.clone()).await?;
                     let tls_connector = tokio_native_tls::TlsConnector::from(tls.clone());
                     let io = tls_connector
-                        .connect(host.ok_or("no host in url")?, TokioIo::new(tunneled))
+                        .connect(dst.host().ok_or("no host in url")?, TokioIo::new(tunneled))
                         .await?;
                     return Ok(Conn {
                         inner: self.verbose.wrap(NativeTlsConn {
@@ -671,18 +674,27 @@ impl ConnectorService {
                     use std::convert::TryFrom;
                     use tokio_rustls::TlsConnector as RustlsConnector;
 
-                    let host = dst.host().ok_or("no host in url")?.to_string();
-                    let port = dst.port().map(|r| r.as_u16()).unwrap_or(443);
-                    let http = http.clone();
-                    let mut http = hyper_rustls::HttpsConnector::from((http, tls_proxy.clone()));
-                    let tls = tls.clone();
-                    let conn = http.call(proxy_dst).await?;
                     log::trace!("tunneling HTTPS over proxy");
-                    let maybe_server_name = ServerName::try_from(host.as_str().to_owned())
-                        .map_err(|_| "Invalid Server Name");
-                    let tunneled = tunnel(conn, host, port, self.user_agent.clone(), auth).await?;
-                    let server_name = maybe_server_name?;
-                    let io = RustlsConnector::from(tls)
+                    let http = http.clone();
+                    let inner = hyper_rustls::HttpsConnector::from((http, tls_proxy.clone()));
+                    // TODO: we could cache constructing this
+                    let mut tunnel =
+                        hyper_util::client::legacy::connect::proxy::Tunnel::new(proxy_dst, inner);
+                    if let Some(auth) = auth {
+                        tunnel = tunnel.with_auth(auth);
+                    }
+                    if let Some(ua) = self.user_agent {
+                        let mut headers = http::HeaderMap::new();
+                        headers.insert(http::header::USER_AGENT, ua);
+                        tunnel = tunnel.with_headers(headers);
+                    }
+                    // We don't wrap this again in an HttpsConnector since that uses Maybe,
+                    // and we know this is definitely HTTPS.
+                    let tunneled = tunnel.call(dst.clone()).await?;
+                    let host = dst.host().ok_or("no host in url")?.to_string();
+                    let server_name = ServerName::try_from(host.as_str().to_owned())
+                        .map_err(|_| "Invalid Server Name")?;
+                    let io = RustlsConnector::from(tls.clone())
                         .connect(server_name, TokioIo::new(tunneled))
                         .await?;
 
@@ -950,83 +962,6 @@ pub(crate) mod sealed {
 
 pub(crate) type Connecting = Pin<Box<dyn Future<Output = Result<Conn, BoxError>> + Send>>;
 
-#[cfg(feature = "__tls")]
-async fn tunnel<T>(
-    mut conn: T,
-    host: String,
-    port: u16,
-    user_agent: Option<HeaderValue>,
-    auth: Option<HeaderValue>,
-) -> Result<T, BoxError>
-where
-    T: Read + Write + Unpin,
-{
-    use hyper_util::rt::TokioIo;
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
-
-    let mut buf = format!(
-        "\
-         CONNECT {host}:{port} HTTP/1.1\r\n\
-         Host: {host}:{port}\r\n\
-         "
-    )
-    .into_bytes();
-
-    // user-agent
-    if let Some(user_agent) = user_agent {
-        buf.extend_from_slice(b"User-Agent: ");
-        buf.extend_from_slice(user_agent.as_bytes());
-        buf.extend_from_slice(b"\r\n");
-    }
-
-    // proxy-authorization
-    if let Some(value) = auth {
-        log::debug!("tunnel to {host}:{port} using basic auth");
-        buf.extend_from_slice(b"Proxy-Authorization: ");
-        buf.extend_from_slice(value.as_bytes());
-        buf.extend_from_slice(b"\r\n");
-    }
-
-    // headers end
-    buf.extend_from_slice(b"\r\n");
-
-    let mut tokio_conn = TokioIo::new(&mut conn);
-
-    tokio_conn.write_all(&buf).await?;
-
-    let mut buf = [0; 8192];
-    let mut pos = 0;
-
-    loop {
-        let n = tokio_conn.read(&mut buf[pos..]).await?;
-
-        if n == 0 {
-            return Err(tunnel_eof());
-        }
-        pos += n;
-
-        let recvd = &buf[..pos];
-        if recvd.starts_with(b"HTTP/1.1 200") || recvd.starts_with(b"HTTP/1.0 200") {
-            if recvd.ends_with(b"\r\n\r\n") {
-                return Ok(conn);
-            }
-            if pos == buf.len() {
-                return Err("proxy headers too long for tunnel".into());
-            }
-        // else read more
-        } else if recvd.starts_with(b"HTTP/1.1 407") {
-            return Err("proxy authentication required".into());
-        } else {
-            return Err("unsuccessful tunnel".into());
-        }
-    }
-}
-
-#[cfg(feature = "__tls")]
-fn tunnel_eof() -> BoxError {
-    "unexpected eof while tunneling".into()
-}
-
 #[cfg(feature = "default-tls")]
 mod native_tls_conn {
     use super::TlsInfoFactory;
@@ -1518,170 +1453,3 @@ mod verbose {
         }
     }
 }
-
-#[cfg(feature = "__tls")]
-#[cfg(test)]
-mod tests {
-    use super::tunnel;
-    use crate::proxy;
-    use hyper_util::rt::TokioIo;
-    use std::io::{Read, Write};
-    use std::net::TcpListener;
-    use std::thread;
-    use tokio::net::TcpStream;
-    use tokio::runtime;
-
-    static TUNNEL_UA: &str = "tunnel-test/x.y";
-    static TUNNEL_OK: &[u8] = b"\
-        HTTP/1.1 200 OK\r\n\
-        \r\n\
-    ";
-
-    macro_rules! mock_tunnel {
-        () => {{
-            mock_tunnel!(TUNNEL_OK)
-        }};
-        ($write:expr) => {{
-            mock_tunnel!($write, "")
-        }};
-        ($write:expr, $auth:expr) => {{
-            let listener = TcpListener::bind("127.0.0.1:0").unwrap();
-            let addr = listener.local_addr().unwrap();
-            let connect_expected = format!(
-                "\
-                 CONNECT {0}:{1} HTTP/1.1\r\n\
-                 Host: {0}:{1}\r\n\
-                 User-Agent: {2}\r\n\
-                 {3}\
-                 \r\n\
-                 ",
-                addr.ip(),
-                addr.port(),
-                TUNNEL_UA,
-                $auth
-            )
-            .into_bytes();
-
-            thread::spawn(move || {
-                let (mut sock, _) = listener.accept().unwrap();
-                let mut buf = [0u8; 4096];
-                let n = sock.read(&mut buf).unwrap();
-                assert_eq!(&buf[..n], &connect_expected[..]);
-
-                sock.write_all($write).unwrap();
-            });
-            addr
-        }};
-    }
-
-    fn ua() -> Option<http::header::HeaderValue> {
-        Some(http::header::HeaderValue::from_static(TUNNEL_UA))
-    }
-
-    #[test]
-    fn test_tunnel() {
-        let addr = mock_tunnel!();
-
-        let rt = runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("new rt");
-        let f = async move {
-            let tcp = TokioIo::new(TcpStream::connect(&addr).await?);
-            let host = addr.ip().to_string();
-            let port = addr.port();
-            tunnel(tcp, host, port, ua(), None).await
-        };
-
-        rt.block_on(f).unwrap();
-    }
-
-    #[test]
-    fn test_tunnel_eof() {
-        let addr = mock_tunnel!(b"HTTP/1.1 200 OK");
-
-        let rt = runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("new rt");
-        let f = async move {
-            let tcp = TokioIo::new(TcpStream::connect(&addr).await?);
-            let host = addr.ip().to_string();
-            let port = addr.port();
-            tunnel(tcp, host, port, ua(), None).await
-        };
-
-        rt.block_on(f).unwrap_err();
-    }
-
-    #[test]
-    fn test_tunnel_non_http_response() {
-        let addr = mock_tunnel!(b"foo bar baz hallo");
-
-        let rt = runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("new rt");
-        let f = async move {
-            let tcp = TokioIo::new(TcpStream::connect(&addr).await?);
-            let host = addr.ip().to_string();
-            let port = addr.port();
-            tunnel(tcp, host, port, ua(), None).await
-        };
-
-        rt.block_on(f).unwrap_err();
-    }
-
-    #[test]
-    fn test_tunnel_proxy_unauthorized() {
-        let addr = mock_tunnel!(
-            b"\
-            HTTP/1.1 407 Proxy Authentication Required\r\n\
-            Proxy-Authenticate: Basic realm=\"nope\"\r\n\
-            \r\n\
-        "
-        );
-
-        let rt = runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("new rt");
-        let f = async move {
-            let tcp = TokioIo::new(TcpStream::connect(&addr).await?);
-            let host = addr.ip().to_string();
-            let port = addr.port();
-            tunnel(tcp, host, port, ua(), None).await
-        };
-
-        let error = rt.block_on(f).unwrap_err();
-        assert_eq!(error.to_string(), "proxy authentication required");
-    }
-
-    #[test]
-    fn test_tunnel_basic_auth() {
-        let addr = mock_tunnel!(
-            TUNNEL_OK,
-            "Proxy-Authorization: Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==\r\n"
-        );
-
-        let rt = runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("new rt");
-        let f = async move {
-            let tcp = TokioIo::new(TcpStream::connect(&addr).await?);
-            let host = addr.ip().to_string();
-            let port = addr.port();
-            tunnel(
-                tcp,
-                host,
-                port,
-                ua(),
-                Some(proxy::encode_basic_auth("Aladdin", "open sesame")),
-            )
-            .await
-        };
-
-        rt.block_on(f).unwrap();
-    }
-}


Test patch:
diff --git a/tests/ci.rs b/tests/ci.rs
new file mode 100644
index 0000000..a33e4d2
--- /dev/null
+++ b/tests/ci.rs
@@ -0,0 +1,14 @@
+#![cfg(not(target_arch = "wasm32"))]
+#![cfg(not(feature = "rustls-tls-manual-roots-no-provider"))]
+mod support;
+use support::server;
+
+#[tokio::test]
+#[should_panic(expected = "test server should not panic")]
+async fn server_panics_should_propagate() {
+    let server = server::http(|_| async {
+        panic!("kaboom");
+    });
+
+    let _ = reqwest::get(format!("http://{}/ci", server.addr())).await;
+}
diff --git a/tests/proxy.rs b/tests/proxy.rs
index ce23c71..d0cb1b4 100644
--- a/tests/proxy.rs
+++ b/tests/proxy.rs
@@ -13,11 +13,11 @@ static HTTP_PROXY_ENV_MUTEX: Lazy<Mutex<()>> = Lazy::new(|| Mutex::new(()));
 
 #[tokio::test]
 async fn http_proxy() {
-    let url = "http://hyper.rs/prox";
+    let url = "http://hyper.rs.local/prox";
     let server = server::http(move |req| {
         assert_eq!(req.method(), "GET");
         assert_eq!(req.uri(), url);
-        assert_eq!(req.headers()["host"], "hyper.rs");
+        assert_eq!(req.headers()["host"], "hyper.rs.local");
 
         async { http::Response::default() }
     });
@@ -39,11 +39,11 @@ async fn http_proxy() {
 
 #[tokio::test]
 async fn http_proxy_basic_auth() {
-    let url = "http://hyper.rs/prox";
+    let url = "http://hyper.rs.local/prox";
     let server = server::http(move |req| {
         assert_eq!(req.method(), "GET");
         assert_eq!(req.uri(), url);
-        assert_eq!(req.headers()["host"], "hyper.rs");
+        assert_eq!(req.headers()["host"], "hyper.rs.local");
         assert_eq!(
             req.headers()["proxy-authorization"],
             "Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ=="
@@ -73,11 +73,11 @@ async fn http_proxy_basic_auth() {
 
 #[tokio::test]
 async fn http_proxy_basic_auth_parsed() {
-    let url = "http://hyper.rs/prox";
+    let url = "http://hyper.rs.local/prox";
     let server = server::http(move |req| {
         assert_eq!(req.method(), "GET");
         assert_eq!(req.uri(), url);
-        assert_eq!(req.headers()["host"], "hyper.rs");
+        assert_eq!(req.headers()["host"], "hyper.rs.local");
         assert_eq!(
             req.headers()["proxy-authorization"],
             "Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ=="
@@ -103,11 +103,11 @@ async fn http_proxy_basic_auth_parsed() {
 
 #[tokio::test]
 async fn system_http_proxy_basic_auth_parsed() {
-    let url = "http://hyper.rs/prox";
+    let url = "http://hyper.rs.local/prox";
     let server = server::http(move |req| {
         assert_eq!(req.method(), "GET");
         assert_eq!(req.uri(), url);
-        assert_eq!(req.headers()["host"], "hyper.rs");
+        assert_eq!(req.headers()["host"], "hyper.rs.local");
         assert_eq!(
             req.headers()["proxy-authorization"],
             "Basic QWxhZGRpbjpvcGVuc2VzYW1l"
@@ -174,11 +174,11 @@ async fn test_no_proxy() {
 
 #[tokio::test]
 async fn test_using_system_proxy() {
-    let url = "http://not.a.real.sub.hyper.rs/prox";
+    let url = "http://not.a.real.sub.hyper.rs.local/prox";
     let server = server::http(move |req| {
         assert_eq!(req.method(), "GET");
         assert_eq!(req.uri(), url);
-        assert_eq!(req.headers()["host"], "not.a.real.sub.hyper.rs");
+        assert_eq!(req.headers()["host"], "not.a.real.sub.hyper.rs.local");
 
         async { http::Response::default() }
     });
@@ -206,12 +206,12 @@ async fn test_using_system_proxy() {
 
 #[tokio::test]
 async fn http_over_http() {
-    let url = "http://hyper.rs/prox";
+    let url = "http://hyper.rs.local/prox";
 
     let server = server::http(move |req| {
         assert_eq!(req.method(), "GET");
         assert_eq!(req.uri(), url);
-        assert_eq!(req.headers()["host"], "hyper.rs");
+        assert_eq!(req.headers()["host"], "hyper.rs.local");
 
         async { http::Response::default() }
     });
@@ -230,3 +230,119 @@ async fn http_over_http() {
     assert_eq!(res.url().as_str(), url);
     assert_eq!(res.status(), reqwest::StatusCode::OK);
 }
+
+#[cfg(feature = "__tls")]
+#[tokio::test]
+async fn tunnel_detects_auth_required() {
+    let url = "https://hyper.rs.local/prox";
+
+    let server = server::http(move |req| {
+        assert_eq!(req.method(), "CONNECT");
+        assert_eq!(req.uri(), "hyper.rs.local:443");
+        assert!(!req
+            .headers()
+            .contains_key(http::header::PROXY_AUTHORIZATION));
+
+        async {
+            let mut res = http::Response::default();
+            *res.status_mut() = http::StatusCode::PROXY_AUTHENTICATION_REQUIRED;
+            res
+        }
+    });
+
+    let proxy = format!("http://{}", server.addr());
+
+    let err = reqwest::Client::builder()
+        .proxy(reqwest::Proxy::https(&proxy).unwrap())
+        .build()
+        .unwrap()
+        .get(url)
+        .send()
+        .await
+        .unwrap_err();
+
+    let err = support::error::inspect(err).pop().unwrap();
+    assert!(
+        err.contains("auth"),
+        "proxy auth err expected, got: {:?}",
+        err
+    );
+}
+
+#[cfg(feature = "__tls")]
+#[tokio::test]
+async fn tunnel_includes_proxy_auth() {
+    let url = "https://hyper.rs.local/prox";
+
+    let server = server::http(move |req| {
+        assert_eq!(req.method(), "CONNECT");
+        assert_eq!(req.uri(), "hyper.rs.local:443");
+        assert_eq!(
+            req.headers()["proxy-authorization"],
+            "Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ=="
+        );
+
+        async {
+            // return 400 to not actually deal with TLS tunneling
+            let mut res = http::Response::default();
+            *res.status_mut() = http::StatusCode::BAD_REQUEST;
+            res
+        }
+    });
+
+    let proxy = format!("http://Aladdin:open%20sesame@{}", server.addr());
+
+    let err = reqwest::Client::builder()
+        .proxy(reqwest::Proxy::https(&proxy).unwrap())
+        .build()
+        .unwrap()
+        .get(url)
+        .send()
+        .await
+        .unwrap_err();
+
+    let err = support::error::inspect(err).pop().unwrap();
+    assert!(
+        err.contains("unsuccessful"),
+        "tunnel unsuccessful expected, got: {:?}",
+        err
+    );
+}
+
+#[cfg(feature = "__tls")]
+#[tokio::test]
+async fn tunnel_includes_user_agent() {
+    let url = "https://hyper.rs.local/prox";
+
+    let server = server::http(move |req| {
+        assert_eq!(req.method(), "CONNECT");
+        assert_eq!(req.uri(), "hyper.rs.local:443");
+        assert_eq!(req.headers()["user-agent"], "reqwest-test");
+
+        async {
+            // return 400 to not actually deal with TLS tunneling
+            let mut res = http::Response::default();
+            *res.status_mut() = http::StatusCode::BAD_REQUEST;
+            res
+        }
+    });
+
+    let proxy = format!("http://{}", server.addr());
+
+    let err = reqwest::Client::builder()
+        .proxy(reqwest::Proxy::https(&proxy).unwrap())
+        .user_agent("reqwest-test")
+        .build()
+        .unwrap()
+        .get(url)
+        .send()
+        .await
+        .unwrap_err();
+
+    let err = support::error::inspect(err).pop().unwrap();
+    assert!(
+        err.contains("unsuccessful"),
+        "tunnel unsuccessful expected, got: {:?}",
+        err
+    );
+}
diff --git a/tests/support/error.rs b/tests/support/error.rs
new file mode 100644
index 0000000..e38c674
--- /dev/null
+++ b/tests/support/error.rs
@@ -0,0 +1,15 @@
+use std::error::Error as StdError;
+
+pub fn inspect<E>(err: E) -> Vec<String>
+where
+    E: Into<Box<dyn StdError + Send + Sync>>,
+{
+    let berr = err.into();
+    let mut err = Some(&*berr as &(dyn StdError + 'static));
+    let mut errs = Vec::new();
+    while let Some(e) = err {
+        errs.push(e.to_string());
+        err = e.source();
+    }
+    errs
+}
diff --git a/tests/support/mod.rs b/tests/support/mod.rs
index 9d4ce7b..1d3f1f3 100644
--- a/tests/support/mod.rs
+++ b/tests/support/mod.rs
@@ -1,5 +1,6 @@
 pub mod delay_layer;
 pub mod delay_server;
+pub mod error;
 pub mod server;
 
 // TODO: remove once done converting to new support server?
diff --git a/tests/support/server.rs b/tests/support/server.rs
index b3e11b1..9d79d5a 100644
--- a/tests/support/server.rs
+++ b/tests/support/server.rs
@@ -95,10 +95,13 @@ where
                     let mut builder =
                         hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new());
                     apply_config(&mut builder);
+                    let mut tasks = tokio::task::JoinSet::new();
+                    let graceful = hyper_util::server::graceful::GracefulShutdown::new();
 
                     loop {
                         tokio::select! {
                             _ = &mut shutdown_rx => {
+                                graceful.shutdown().await;
                                 break;
                             }
                             accepted = listener.accept() => {
@@ -110,13 +113,25 @@ where
                                 });
                                 let builder = builder.clone();
                                 let events_tx = events_tx.clone();
-                                tokio::spawn(async move {
-                                    let _ = builder.serve_connection_with_upgrades(hyper_util::rt::TokioIo::new(io), svc).await;
+                                let watcher = graceful.watcher();
+
+                                tasks.spawn(async move {
+                                    let conn = builder.serve_connection_with_upgrades(hyper_util::rt::TokioIo::new(io), svc);
+                                    let _ = watcher.watch(conn).await;
                                     let _ = events_tx.send(Event::ConnectionClosed);
                                 });
                             }
                         }
                     }
+
+                    // try to drain
+                    while let Some(result) = tasks.join_next().await {
+                        if let Err(e) = result {
+                            if e.is_panic() {
+                                std::panic::resume_unwind(e.into_panic());
+                            }
+                        }
+                    }
                     let _ = panic_tx.send(());
                 });
             })