Repository: seanmonstar/reqwest
Pull request #2434 (2024-09-26 08:29:53)

fix: pass pool_timer to hyper_util to enable the idle cleanup task
While testing an application I'm working on that makes requests to many different endpoints I found that reqwest wasn't cleaning up pooled connections even after the `pool_idle_timeout`.

After investigating it seems like the hyper_util connection pool is bailing on this condition: https://github.com/hyperium/hyper-util/blob/master/src/client/legacy/pool.rs#L429-L433. This seems to be because reqwest doesn't pass a `pool_timer`.

Reproduction:

`server.rs`:
```rust
use warp::Filter;

#[tokio::main]
async fn main() {
    simple_logger::init_with_level(log::Level::Info).unwrap();

    let basic = warp::filters::path::end()
        .map(|| "hello");

    let server = warp::serve(basic)
        .run(([127, 0, 0, 1], 8000));

    server.await;
}
```

`client.rs`:
```rust
#[tokio::main]
async fn main() {
    simple_logger::init_with_level(log::Level::Trace).unwrap();

    let client = reqwest::Client::builder()
        .pool_idle_timeout(std::time::Duration::from_secs(5))
        .build()
        .unwrap();

    client.get("http://localhost:8000").send().await.unwrap();

    tokio::time::sleep(std::time::Duration::from_secs(10)).await;

    client.get("http://localhost:8000").send().await.unwrap();
}
```

Output, using `master`:
```
2024-09-26T08:27:19.715Z DEBUG [reqwest::connect] starting new connection: http://localhost:8000/
...
2024-09-26T08:27:29.719Z TRACE [hyper_util::client::legacy::pool] removing expired connection for ("http", localhost:8000)
...
2024-09-26T08:27:29.719Z DEBUG [reqwest::connect] starting new connection: http://localhost:8000/
```
(connection is only cleaned up when creating a new connection)

Output, after this commit:
```
2024-09-26T08:26:10.285Z DEBUG [reqwest::connect] starting new connection: http://localhost:8000/
...
2024-09-26T08:26:15.289Z TRACE [hyper_util::client::legacy::pool] idle interval evicting expired for ("http", localhost:8000)
...
2024-09-26T08:26:20.289Z DEBUG [reqwest::connect] starting new connection: http://localhost:8000/
```
(connection is actively cleaned up after the idle timeout by the `IdleTask`)

This is important for applications that make requests to many different endpoints. If these old connections don't get cleaned up we run into memory and file descriptor issues.

RobMor: I'm not an expert in the history here but it seems like this was just missed during [the hyper v1 upgrade](https://github.com/seanmonstar/reqwest/commit/eb94f26919e881177ad3d9cd172f47d1a8263799). It looks like at that time the separate `pool_timer` builder option was [already in place](https://github.com/hyperium/hyper-util/commit/784109db021d076b0822ea5344a315e530831973).

Any ideas on how one could add some kind of automated test case around this behavior?
RobMor: Also, was wrapping both in the wasm32 check the right thing to do? Not really sure what that was there for.
seanmonstar: Oh wow, thanks for fixing this!

> Any ideas on how one could add some kind of automated test case around this behavior? 

Coming up with an integration test in reqwest is likely hard, since there's no way for you access the details outside the library. However, I'm thinking the mistake is in hyper-util: it should notice when the timeouts are set but not timer is set (hyper does the same for its timeouts).

> Also, was wrapping both in the wasm32 check the right thing to do? 

Um, that seems like it previously wasn't doing anything. None of that code is ever configured on for WASM, so that extra `cfg` doesn't matter. We can probably just remove it!
RobMor: > there's no way for you access the details outside the library

I was thinking maybe we could spin up a dummy server for the test. That would give us access to the server side connection state where we should be able to see that the connection gets closed after the idle timeout.

> I'm thinking the mistake is in hyper-util: it should notice when the timeouts are set but not timer is set

Curious on how one would implement a check like this? It seems like it would be a breaking change to start panicking or erroring on previously "working" code.

> We can probably just remove it!

Will do
RobMor: I've added a test! I confirmed that this test fails without this commit, and passes after adding the pool_timer.

Let me know what you think of it.
RobMor: Realized I didn't need to use tokio's `unbounded_channel`, and also had to fix some formatting. Should be good now...
RobMor: Agh, tests should be passing with the `http3` feature now too.

Patch:
diff --git a/src/async_impl/client.rs b/src/async_impl/client.rs
index 9a34f3f..095adf4 100644
--- a/src/async_impl/client.rs
+++ b/src/async_impl/client.rs
@@ -726,8 +726,8 @@ impl ClientBuilder {
             }
         }
 
-        #[cfg(not(target_arch = "wasm32"))]
         builder.timer(hyper_util::rt::TokioTimer::new());
+        builder.pool_timer(hyper_util::rt::TokioTimer::new());
         builder.pool_idle_timeout(config.pool_idle_timeout);
         builder.pool_max_idle_per_host(config.pool_max_idle_per_host);
         connector.set_keepalive(config.tcp_keepalive);


Test patch:
diff --git a/tests/client.rs b/tests/client.rs
index 18aaf4e..51fb9df 100644
--- a/tests/client.rs
+++ b/tests/client.rs
@@ -572,3 +572,24 @@ async fn highly_concurrent_requests_to_slow_http2_server_with_low_max_concurrent
 
     server.shutdown().await;
 }
+
+#[tokio::test]
+async fn close_connection_after_idle_timeout() {
+    let mut server = server::http(move |_| async move { http::Response::default() });
+
+    let client = reqwest::Client::builder()
+        .pool_idle_timeout(std::time::Duration::from_secs(1))
+        .build()
+        .unwrap();
+
+    let url = format!("http://{}", server.addr());
+
+    client.get(&url).send().await.unwrap();
+
+    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+
+    assert!(server
+        .events()
+        .iter()
+        .any(|e| matches!(e, server::Event::ConnectionClosed)));
+}
diff --git a/tests/support/server.rs b/tests/support/server.rs
index 43742b6..29835ea 100644
--- a/tests/support/server.rs
+++ b/tests/support/server.rs
@@ -12,13 +12,27 @@ use tokio::sync::oneshot;
 pub struct Server {
     addr: net::SocketAddr,
     panic_rx: std_mpsc::Receiver<()>,
+    events_rx: std_mpsc::Receiver<Event>,
     shutdown_tx: Option<oneshot::Sender<()>>,
 }
 
+#[non_exhaustive]
+pub enum Event {
+    ConnectionClosed,
+}
+
 impl Server {
     pub fn addr(&self) -> net::SocketAddr {
         self.addr
     }
+
+    pub fn events(&mut self) -> Vec<Event> {
+        let mut events = Vec::new();
+        while let Ok(event) = self.events_rx.try_recv() {
+            events.push(event);
+        }
+        events
+    }
 }
 
 impl Drop for Server {
@@ -67,6 +81,7 @@ where
 
         let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
         let (panic_tx, panic_rx) = std_mpsc::channel();
+        let (events_tx, events_rx) = std_mpsc::channel();
         let tname = format!(
             "test({})-support-server",
             test_name,
@@ -92,8 +107,10 @@ where
                                     async move { Ok::<_, Infallible>(fut.await) }
                                 });
                                 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 _ = events_tx.send(Event::ConnectionClosed);
                                 });
                             }
                         }
@@ -105,6 +122,7 @@ where
         Server {
             addr,
             panic_rx,
+            events_rx,
             shutdown_tx: Some(shutdown_tx),
         }
     })
@@ -152,6 +170,7 @@ where
 
         let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
         let (panic_tx, panic_rx) = std_mpsc::channel();
+        let (events_tx, events_rx) = std_mpsc::channel();
         let tname = format!(
             "test({})-support-server",
             test_name,
@@ -169,9 +188,11 @@ where
                             Some(accepted) = endpoint.accept() => {
                                 let conn = accepted.await.expect("accepted");
                                 let mut h3_conn = h3::server::Connection::new(h3_quinn::Connection::new(conn)).await.unwrap();
+                                let events_tx = events_tx.clone();
                                 let func = func.clone();
                                 tokio::spawn(async move {
                                     while let Ok(Some((req, stream))) = h3_conn.accept().await {
+                                        let events_tx = events_tx.clone();
                                         let func = func.clone();
                                         tokio::spawn(async move {
                                             let (mut tx, rx) = stream.split();
@@ -198,6 +219,7 @@ where
                                                 }
                                             }
                                             tx.finish().await.unwrap();
+                                            events_tx.send(Event::ConnectionClosed).unwrap();
                                         });
                                     }
                                 });
@@ -211,6 +233,7 @@ where
         Server {
             addr,
             panic_rx,
+            events_rx,
             shutdown_tx: Some(shutdown_tx),
         }
     })