fix(http1): evict pooled conn on request-side Connection: close

src/client/tests.rs

@@ -23,6 +23,136 @@ async fn client_connect_uri_argument() {
         .expect_err("response should fail");
 }
 
+// A client request carrying `Connection: close` must NOT be pooled or reused,
+// even when the backend's response omits `Connection: close` (i.e. the backend keeps the
+// socket alive). A keep-alive request, by contrast, reuses a single pooled connection.
+#[tokio::test]
+async fn conn_close_request_is_not_pooled_even_if_backend_keeps_alive() {
+    // Control: keep-alive requests reuse a single pooled connection, so the connection
+    // count stays at 1 across requests.
+    assert_eq!(
+        keepalive_backend_conn_counts(&[], 2).await,
+        vec![1, 1],
+        "keep-alive requests should reuse a single pooled connection"
+    );
+    // A `Connection: close` request must open a fresh connection every time: the count
+    // increments per request rather than staying at 1.
+    assert_eq!(
+        keepalive_backend_conn_counts(&["close"], 2).await,
+        vec![1, 2],
+        "each Connection: close request must open a new connection (no pool reuse)"
+    );
+    // A `close` token in a comma-separated value is honored.
+    assert_eq!(
+        keepalive_backend_conn_counts(&["keep-alive, close"], 2).await,
+        vec![1, 2],
+        "a `close` token in a comma-separated Connection value must not be pooled"
+    );
+    // A `close` in ANY of multiple Connection header lines is honored (get_all,
+    // not just the first line).
+    assert_eq!(
+        keepalive_backend_conn_counts(&["keep-alive", "close"], 2).await,
+        vec![1, 2],
+        "a `close` in any Connection header line must not be pooled"
+    );
+}
+
+// Sends `num_requests` sequential requests (with the given `Connection` header lines,
+// one `.header()` append each) through a client whose backend always replies keep-alive,
+// and returns the cumulative number of connections opened observed after each request.
+async fn keepalive_backend_conn_counts(
+    connection_values: &[&'static str],
+    num_requests: usize,
+) -> Vec<usize> {
+    use crate::client::connect::{Connected, Connection};
+    use std::pin::Pin;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::sync::Arc;
+    use std::task::{Context, Poll};
+    use std::time::Duration;
+    use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
+
+    struct MockIo(tokio::io::DuplexStream);
+    impl AsyncRead for MockIo {
+        fn poll_read(
+            mut self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            b: &mut ReadBuf<'_>,
+        ) -> Poll<std::io::Result<()>> {
+            Pin::new(&mut self.0).poll_read(cx, b)
+        }
+    }
+    impl AsyncWrite for MockIo {
+        fn poll_write(
+            mut self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            b: &[u8],
+        ) -> Poll<std::io::Result<usize>> {
+            Pin::new(&mut self.0).poll_write(cx, b)
+        }
+        fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Pin::new(&mut self.0).poll_flush(cx)
+        }
+        fn poll_shutdown(
+            mut self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<std::io::Result<()>> {
+            Pin::new(&mut self.0).poll_shutdown(cx)
+        }
+    }
+    impl Connection for MockIo {
+        fn connected(&self) -> Connected {
+            Connected::new()
+        }
+    }
+
+    let conn_count = Arc::new(AtomicUsize::new(0));
+    let cc = conn_count.clone();
+    let connector = tower::service_fn(move |_dst: http::Uri| {
+        cc.fetch_add(1, Ordering::SeqCst);
+        let (client_io, mut server_io) = tokio::io::duplex(1024);
+        tokio::spawn(async move {
+            let mut buf = [0u8; 1024];
+            // Always reply keep-alive (NO Connection: close) to every request on this connection.
+            while let Ok(read) = server_io.read(&mut buf).await {
+                if read == 0 {
+                    break;
+                }
+                if server_io
+                    .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
+                    .await
+                    .is_err()
+                {
+                    break;
+                }
+            }
+        });
+        future::ok::<_, std::io::Error>(MockIo(client_io))
+    });
+
+    let client = Client::builder().build::<_, crate::Body>(connector);
+
+    let mut counts = Vec::with_capacity(num_requests);
+    for i in 0..num_requests {
+        let mut builder = http::Request::builder().uri(format!("http://mock.local/{}", i));
+        for value in connection_values {
+            builder = builder.header(http::header::CONNECTION, *value);
+        }
+        let res = client
+            .request(builder.body(crate::Body::empty()).unwrap())
+            .await
+            .expect("request ok");
+        assert_eq!(res.status(), 200);
+        crate::body::to_bytes(res.into_body())
+            .await
+            .expect("drain body");
+        // Let any pool insertion settle so a reusable connection would be reused next.
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        counts.push(conn_count.load(Ordering::SeqCst));
+    }
+    counts
+}
+
 /*
 // FIXME: re-implement tests with `async/await`
 #[test]

src/headers.rs

@@ -16,6 +16,17 @@ pub(super) fn connection_close(value: &HeaderValue) -> bool {
     connection_has(value, "close")
 }
 
+// Returns true if any `Connection` header field carries a `close` token.
+// A message may have more than one `Connection` header line, so all of them
+// must be inspected (`get`/`connection_close` alone only sees the first).
+#[cfg(feature = "http1")]
+pub(super) fn connection_any_close(headers: &HeaderMap) -> bool {
+    headers
+        .get_all(http::header::CONNECTION)
+        .into_iter()
+        .any(connection_close)
+}
+
 #[cfg(feature = "http1")]
 fn connection_has(value: &HeaderValue, needle: &str) -> bool {
     if let Ok(s) = value.to_str() {

src/proto/h1/conn.rs

@@ -19,7 +19,7 @@ use tracing::{debug, error, trace};
 use super::io::Buffered;
 use super::{Decoder, Encode, EncodedBuf, Encoder, Http1Transaction, ParseContext, Wants};
 use crate::body::DecodedLength;
-use crate::headers::connection_keep_alive;
+use crate::headers::{connection_any_close, connection_keep_alive};
 use crate::proto::{BodyLength, MessageHead};
 
 const H2_PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
@@ -548,6 +548,15 @@ where
 
         if !T::should_read_first() {
             self.state.busy();
+            // A client request carrying `Connection: close` must not be pooled or
+            // reused. hyper otherwise derives connection reuse from the response
+            // alone, so a backend that ignores the request-side close (omits
+            // `Connection: close` in its response) would leave the connection in
+            // the pool. Disable keep-alive up front so the connection is evicted
+            // regardless of the response.
+            if connection_any_close(&head.headers) {
+                self.state.disable_keep_alive();
+            }
         }
 
         self.enforce_version(&mut head);