diff --git a/src/client/tests.rs b/src/client/tests.rs index 0a281a637d..2938d5cb15 100644 --- a/src/client/tests.rs +++ b/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 { + 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> { + 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> { + Pin::new(&mut self.0).poll_write(cx, b) + } + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.0).poll_flush(cx) + } + fn poll_shutdown( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + 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] diff --git a/src/headers.rs b/src/headers.rs index 2e5e5db0f2..0ee1c9a1dd 100644 --- a/src/headers.rs +++ b/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() { diff --git a/src/proto/h1/conn.rs b/src/proto/h1/conn.rs index 5ab72f264e..579252fcdf 100644 --- a/src/proto/h1/conn.rs +++ b/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);