diff --git a/src/client/legacy/client.rs b/src/client/legacy/client.rs index d4f77be4..3d69b510 100644 --- a/src/client/legacy/client.rs +++ b/src/client/legacy/client.rs @@ -543,6 +543,9 @@ where Either::Left(Box::pin(async move { let tx = if is_h2 { #[cfg(feature = "http2")] { + // This builder and observer belong only to this connection. + let mut h2_builder = h2_builder; + h2_builder.keep_alive_observer(connected.poisoned.clone()); let (mut tx, conn) = h2_builder.handshake(io).await.map_err(Error::tx)?; @@ -1477,6 +1480,26 @@ impl Builder { self } + /// Stops reusing an HTTP/2 connection when its keep-alive PING remains + /// unacknowledged for this duration. + /// + /// The connection is permanently retired from the pool, even if a late ACK + /// arrives. Existing requests and response bodies continue under the original + /// `http2_keep_alive_timeout`. Subsequent requests can establish a replacement; + /// no connection is opened proactively and no request is replayed by this option. + /// Requests that already checked out the old connection may still use it. + /// + /// Defaults to `None` (disabled). Does nothing if keep-alive is disabled. + /// When enabled, the duration must be greater than zero and less than the + /// final hard timeout; otherwise connection handshake panics. Requires a timer. + #[cfg(feature = "tokio")] + #[cfg(feature = "http2")] + #[cfg_attr(docsrs, doc(cfg(feature = "http2")))] + pub fn http2_keep_alive_reuse_timeout(&mut self, timeout: Option) -> &mut Self { + self.h2_builder.keep_alive_reuse_timeout(timeout); + self + } + /// Sets whether HTTP2 keep-alive should apply while the connection is idle. /// /// If disabled, keep-alive pings are only sent while there are open diff --git a/src/client/legacy/connect/mod.rs b/src/client/legacy/connect/mod.rs index fff8262f..45384d2b 100644 --- a/src/client/legacy/connect/mod.rs +++ b/src/client/legacy/connect/mod.rs @@ -137,6 +137,14 @@ impl PoisonPill { } } +#[cfg(feature = "http2")] +impl hyper::client::conn::http2::KeepAliveObserver for PoisonPill { + fn on_reuse_timeout(&self) { + self.poison(); + tracing::debug!(poison_pill = ?self, "HTTP/2 keep-alive reuse timeout; connection retired"); + } +} + pub(super) struct Extra(Box); #[derive(Clone, Copy, Debug, PartialEq)] diff --git a/tests/keepalive_reuse.rs b/tests/keepalive_reuse.rs new file mode 100644 index 00000000..0798c0be --- /dev/null +++ b/tests/keepalive_reuse.rs @@ -0,0 +1,344 @@ +#![cfg(all( + feature = "client-legacy", + feature = "server", + feature = "http2", + feature = "tokio" +))] + +use std::convert::Infallible; +use std::future::{Ready, ready}; +use std::io; +use std::pin::Pin; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::task::{Context, Poll}; +use std::time::Duration; + +use bytes::Bytes; +use futures_util::task::AtomicWaker; +use http_body_util::{BodyExt, Empty, Full, StreamBody}; +use hyper::body::Frame; +use hyper::service::service_fn; +use hyper::{Request, Response, Uri}; +use hyper_util::client::legacy::Client; +use hyper_util::client::legacy::connect::{Connected, Connection}; +use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer}; +use tokio::io::{AsyncRead, AsyncWrite, DuplexStream, ReadBuf}; +use tokio::sync::{Notify, mpsc}; +use tower_service::Service; + +// Pause both directions without EOF, reset, or discarding buffered bytes. +#[derive(Default)] +struct Gate { + paused: AtomicBool, + read: AtomicWaker, + write: AtomicWaker, +} + +impl Gate { + fn blocked(&self, cx: &Context<'_>, waker: &AtomicWaker) -> bool { + waker.register(cx.waker()); + self.paused.load(Ordering::SeqCst) + } + + fn pause(&self) { + self.paused.store(true, Ordering::SeqCst); + } + + fn resume(&self) { + self.paused.store(false, Ordering::SeqCst); + self.read.wake(); + self.write.wake(); + } +} + +struct GatedIo { + io: DuplexStream, + gate: Arc, +} + +impl AsyncRead for GatedIo { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + if self.gate.blocked(cx, &self.gate.read) { + return Poll::Pending; + } + Pin::new(&mut self.io).poll_read(cx, buf) + } +} + +impl AsyncWrite for GatedIo { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + if self.gate.blocked(cx, &self.gate.write) { + return Poll::Pending; + } + Pin::new(&mut self.io).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + if self.gate.blocked(cx, &self.gate.write) { + return Poll::Pending; + } + Pin::new(&mut self.io).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.io).poll_shutdown(cx) + } +} + +impl Connection for GatedIo { + fn connected(&self) -> Connected { + Connected::new() + } +} + +#[derive(Clone)] +struct Connector { + count: Arc, + gates: Arc>>>, + seen: mpsc::UnboundedSender<(usize, String)>, + headers: Arc, + body: Arc, +} + +impl Service for Connector { + type Response = TokioIo; + type Error = io::Error; + type Future = Ready>; + + fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, _: Uri) -> Self::Future { + let id = self.count.fetch_add(1, Ordering::SeqCst); + let gate = Arc::new(Gate::default()); + self.gates.lock().unwrap().push(gate.clone()); + let (client, server) = tokio::io::duplex(64 * 1024); + let seen = self.seen.clone(); + let headers = self.headers.clone(); + let body = self.body.clone(); + tokio::spawn(async move { + let service = service_fn(move |req: Request| { + let seen = seen.clone(); + let headers = headers.clone(); + let body = body.clone(); + async move { + let path = req.uri().path().to_owned(); + let _ = seen.send((id, path.clone())); + if path == "/pending" { + headers.notified().await; + } + let payload = if path == "/stream" { + StreamBody::new(futures_util::stream::once(async move { + body.notified().await; + Ok::<_, Infallible>(Frame::data(Bytes::from_static(b"complete"))) + })) + .boxed_unsync() + } else { + Full::new(Bytes::from_static(b"complete")) + .map_err(|never| match never {}) + .boxed_unsync() + }; + Ok::<_, Infallible>( + Response::builder() + .header("connection-id", id.to_string()) + .body(payload) + .unwrap(), + ) + } + }); + let _ = hyper::server::conn::http2::Builder::new(TokioExecutor::new()) + .serve_connection(TokioIo::new(GatedIo { io: server, gate }), service) + .await; + }); + ready(Ok(TokioIo::new(GatedIo { + io: client, + gate: Arc::new(Gate::default()), + }))) + } +} + +type TestClient = Client>; + +fn setup( + reuse: Option, + adaptive: bool, +) -> ( + TestClient, + Connector, + mpsc::UnboundedReceiver<(usize, String)>, +) { + let (seen, rx) = mpsc::unbounded_channel(); + let connector = Connector { + count: Arc::new(AtomicUsize::new(0)), + gates: Arc::new(Mutex::new(Vec::new())), + seen, + headers: Arc::new(Notify::new()), + body: Arc::new(Notify::new()), + }; + let client = Client::builder(TokioExecutor::new()) + .http2_only(true) + .timer(TokioTimer::new()) + .http2_adaptive_window(adaptive) + .http2_keep_alive_interval(Some(Duration::from_secs(10))) + .http2_keep_alive_reuse_timeout(reuse) + .http2_keep_alive_timeout(Duration::from_secs(60)) + .http2_keep_alive_while_idle(true) + .build(connector.clone()); + (client, connector, rx) +} + +fn request(path: &str) -> Request> { + Request::builder() + .uri(format!("http://test{path}")) + .body(Empty::new()) + .unwrap() +} + +async fn settle() { + // Keep this task runnable so paused time cannot auto-advance while flushing IO. + for _ in 0..64 { + tokio::task::yield_now().await; + } +} + +async fn advance(seconds: u64) { + tokio::time::advance(Duration::from_secs(seconds)).await; + settle().await; +} + +async fn get(client: &TestClient) -> usize { + let response = client.request(request("/ok")).await.unwrap(); + let id = response.headers()["connection-id"] + .to_str() + .unwrap() + .parse() + .unwrap(); + assert_eq!(response.collect().await.unwrap().to_bytes(), "complete"); + id +} + +async fn recovery(adaptive: bool) { + let (client, connector, mut seen) = setup(Some(Duration::from_secs(5)), adaptive); + assert_eq!(get(&client).await, 0); + seen.recv().await.unwrap(); + let stream = client.request(request("/stream")).await.unwrap(); + assert_eq!(seen.recv().await.unwrap(), (0, "/stream".into())); + let stream = tokio::spawn(async move { stream.collect().await }); + let pending = tokio::spawn(client.request(request("/pending"))); + assert_eq!(seen.recv().await.unwrap(), (0, "/pending".into())); + settle().await; + let old = connector.gates.lock().unwrap()[0].clone(); + old.pause(); + advance(10).await; // The original keep-alive PING enters its timeout phase. + advance(5).await; // No IO event is needed to retire the connection. + assert!(!pending.is_finished()); + assert!(!stream.is_finished()); + + let ids = futures_util::future::join_all((0..12).map(|_| get(&client))).await; + assert!(ids.iter().all(|id| *id == 1)); + assert_eq!(connector.count.load(Ordering::SeqCst), 2); + assert!(!pending.is_finished()); + assert!(!stream.is_finished()); + + connector.headers.notify_one(); + connector.body.notify_one(); + old.resume(); + settle().await; + let response = pending.await.unwrap().unwrap(); + assert_eq!(response.headers()["connection-id"], "0"); + assert_eq!(response.collect().await.unwrap().to_bytes(), "complete"); + assert_eq!(stream.await.unwrap().unwrap().to_bytes(), "complete"); + assert_eq!(get(&client).await, 1); // Late ACK never restores the old pool entry. + assert_eq!(connector.count.load(Ordering::SeqCst), 2); +} + +#[tokio::test(start_paused = true)] +async fn soft_timeout_replaces_connection_and_preserves_headers_and_body() { + recovery(false).await; +} + +#[tokio::test(start_paused = true)] +async fn adaptive_window_preserves_recovery_behavior() { + recovery(true).await; +} + +#[tokio::test(start_paused = true)] +async fn original_sixty_second_hard_deadline_survives_retirement() { + let (client, connector, mut seen) = setup(Some(Duration::from_secs(5)), false); + assert_eq!(get(&client).await, 0); + seen.recv().await.unwrap(); + let pending = tokio::spawn(client.request(request("/pending"))); + seen.recv().await.unwrap(); + settle().await; + connector.gates.lock().unwrap()[0].pause(); + advance(10).await; + advance(5).await; + assert_eq!(get(&client).await, 1); + advance(54).await; + assert!( + !pending.is_finished(), + "soft timeout must not shorten the hard deadline" + ); + advance(1).await; + assert!( + pending.is_finished(), + "hard timer must still wake at PING + 60s" + ); + let error = pending.await.unwrap().unwrap_err(); + let mut cause: &(dyn std::error::Error + 'static) = &error; + let mut timed_out = false; + loop { + if let Some(error) = cause.downcast_ref::() { + timed_out |= error.is_timeout(); + } + match cause.source() { + Some(source) => cause = source, + None => break, + } + } + assert!(timed_out, "expected original keep-alive timeout: {error:?}"); +} + +#[tokio::test(start_paused = true)] +async fn ack_before_soft_deadline_keeps_connection_reusable() { + let (client, connector, _) = setup(Some(Duration::from_secs(5)), false); + assert_eq!(get(&client).await, 0); + settle().await; + let old = connector.gates.lock().unwrap()[0].clone(); + old.pause(); + advance(10).await; + advance(4).await; + old.resume(); + settle().await; + advance(1).await; + assert_eq!(get(&client).await, 0); + assert_eq!(connector.count.load(Ordering::SeqCst), 1); +} + +#[tokio::test(start_paused = true)] +async fn disabled_soft_timeout_preserves_pool_behavior() { + let (client, connector, _) = setup(None, false); + assert_eq!(get(&client).await, 0); + settle().await; + let old = connector.gates.lock().unwrap()[0].clone(); + old.pause(); + advance(10).await; + advance(5).await; + let request = tokio::spawn(client.request(request("/ok"))); + settle().await; + assert!(!request.is_finished()); + assert_eq!(connector.count.load(Ordering::SeqCst), 1); + old.resume(); + let response = request.await.unwrap().unwrap(); + assert_eq!(response.headers()["connection-id"], "0"); +}