From 9b0345db4cf40f4c8a743222ad4ceb5fb721da60 Mon Sep 17 00:00:00 2001 From: Jiahao Liang Date: Thu, 10 Sep 2026 10:51:56 +0800 Subject: [PATCH 1/4] feat(http2): add an optional keep-alive reuse timeout observer Notify connection pools without shortening the hard timeout for active streams. Keep the existing h2 dependency range and expose the observer on the H2 client API. Cover timer wakeups and final-configuration validation. --- src/client/conn/http2.rs | 101 +++++++++++++++++++++++++++++++++++ src/proto/h2/client.rs | 6 +++ src/proto/h2/ping.rs | 55 ++++++++++++++++++- src/proto/h2/ping/tests.rs | 105 +++++++++++++++++++++++++++++++++++++ src/proto/h2/server.rs | 2 + 5 files changed, 268 insertions(+), 1 deletion(-) create mode 100644 src/proto/h2/ping/tests.rs diff --git a/src/client/conn/http2.rs b/src/client/conn/http2.rs index 10613b354a..33ee61c573 100644 --- a/src/client/conn/http2.rs +++ b/src/client/conn/http2.rs @@ -20,6 +20,8 @@ use crate::proto; use crate::rt::bounds::Http2ClientConnExec; use crate::rt::Timer; +pub use crate::proto::h2::ping::KeepAliveObserver; + /// The sender side of an established connection. pub struct SendRequest { dispatch: dispatch::UnboundedSender, Response>, @@ -463,6 +465,35 @@ where self } + /// Sets the keep-alive PING acknowledgement timeout for stopping reuse. + /// + /// On expiry, the observer installed with [`Self::keep_alive_observer`] is + /// notified once. Hyper itself does not prevent new requests or close the + /// connection; the observer should retire it from the caller's connection + /// pool. Existing streams retain the original keep-alive timeout. A late + /// ACK does not undo the notification. + /// + /// Defaults to `None` (disabled). Does nothing when keep-alive is disabled. + /// When keep-alive is enabled, a configured duration must be greater than + /// zero and less than [`Self::keep_alive_timeout`]; otherwise `handshake` + /// panics. Validation uses the final configuration, regardless of setter order. + pub fn keep_alive_reuse_timeout(&mut self, timeout: Option) -> &mut Self { + self.h2_builder.keep_alive_reuse_timeout = timeout; + self + } + + /// Sets the observer for the keep-alive reuse timeout on new connections. + /// + /// Install a separate observer for each connection when retiring individual + /// pool entries. The observer must be nonblocking and must not panic. + pub fn keep_alive_observer( + &mut self, + observer: impl KeepAliveObserver + 'static, + ) -> &mut Self { + self.h2_builder.keep_alive_observer = Some(Arc::new(observer)); + 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 @@ -560,6 +591,12 @@ where /// Note, if [`Connection`] is not `await`-ed, [`SendRequest`] will /// do nothing. /// + /// # Panics + /// + /// Panics if keep-alive is enabled and the configured reuse timeout is zero + /// or is not less than the keep-alive timeout. Validation uses the final + /// builder configuration, regardless of setter order. + /// /// # Errors /// /// Returns an error if the HTTP/2 connection handshake fails. @@ -574,6 +611,14 @@ where B::Error: Into>, Ex: Http2ClientConnExec + Unpin, { + if self.h2_builder.keep_alive_interval.is_some() { + if let Some(timeout) = self.h2_builder.keep_alive_reuse_timeout { + assert!( + timeout > Duration::ZERO && timeout < self.h2_builder.keep_alive_timeout, + "keep_alive_reuse_timeout must be greater than zero and less than keep_alive_timeout" + ); + } + } let opts = self.clone(); async move { @@ -597,6 +642,62 @@ where #[cfg(test)] mod tests { + #[derive(Clone)] + struct UnusedExecutor; + + impl crate::rt::Executor for UnusedExecutor { + fn execute(&self, _: F) { + panic!("configuration validation must not spawn tasks"); + } + } + + fn check_reuse_config(soft: std::time::Duration, hard: std::time::Duration, enabled: bool) { + let mut builder = super::Builder::new(UnusedExecutor); + // Set soft before hard, even when soft exceeds the default hard timeout. + builder.keep_alive_reuse_timeout(Some(soft)); + builder.keep_alive_timeout(hard); + if enabled { + builder.keep_alive_interval(Some(std::time::Duration::from_secs(10))); + } + let (io, _peer) = tokio::io::duplex(64); + let handshake = builder.handshake::<_, http_body_util::Empty>( + crate::common::io::Compat::new(io), + ); + drop(handshake); + } + + #[test] + fn reuse_timeout_validates_final_config_independent_of_setter_order() { + check_reuse_config( + std::time::Duration::from_secs(30), + std::time::Duration::from_secs(60), + true, + ); + } + + #[test] + fn reuse_timeout_is_inactive_without_keepalive() { + check_reuse_config(std::time::Duration::ZERO, std::time::Duration::ZERO, false); + } + + #[test] + #[should_panic(expected = "keep_alive_reuse_timeout must be greater than zero")] + fn reuse_timeout_rejects_zero() { + check_reuse_config(std::time::Duration::ZERO, std::time::Duration::from_secs(60), true); + } + + #[test] + #[should_panic(expected = "keep_alive_reuse_timeout must be greater than zero")] + fn reuse_timeout_rejects_equal_hard_timeout() { + check_reuse_config(std::time::Duration::from_secs(60), std::time::Duration::from_secs(60), true); + } + + #[test] + #[should_panic(expected = "keep_alive_reuse_timeout must be greater than zero")] + fn reuse_timeout_rejects_larger_than_hard_timeout() { + check_reuse_config(std::time::Duration::from_secs(61), std::time::Duration::from_secs(60), true); + } + #[tokio::test] #[ignore] // only compilation is checked async fn send_sync_executor_of_non_send_futures() { diff --git a/src/proto/h2/client.rs b/src/proto/h2/client.rs index eb768baa76..f269cf3fb6 100644 --- a/src/proto/h2/client.rs +++ b/src/proto/h2/client.rs @@ -70,6 +70,8 @@ pub(crate) struct Config { pub(crate) max_header_list_size: u32, pub(crate) keep_alive_interval: Option, pub(crate) keep_alive_timeout: Duration, + pub(crate) keep_alive_reuse_timeout: Option, + pub(crate) keep_alive_observer: Option>, pub(crate) keep_alive_while_idle: bool, pub(crate) max_concurrent_reset_streams: Option, pub(crate) max_send_buffer_size: usize, @@ -91,6 +93,8 @@ impl Default for Config { max_header_list_size: DEFAULT_MAX_HEADER_LIST_SIZE, keep_alive_interval: None, keep_alive_timeout: Duration::from_secs(20), + keep_alive_reuse_timeout: None, + keep_alive_observer: None, keep_alive_while_idle: false, max_concurrent_reset_streams: None, max_send_buffer_size: DEFAULT_MAX_SEND_BUF_SIZE, @@ -143,6 +147,8 @@ fn new_ping_config(config: &Config) -> ping::Config { }, keep_alive_interval: config.keep_alive_interval, keep_alive_timeout: config.keep_alive_timeout, + keep_alive_reuse_timeout: config.keep_alive_reuse_timeout, + keep_alive_observer: config.keep_alive_observer.clone(), keep_alive_while_idle: config.keep_alive_while_idle, } } diff --git a/src/proto/h2/ping.rs b/src/proto/h2/ping.rs index 198bff465c..959031fd10 100644 --- a/src/proto/h2/ping.rs +++ b/src/proto/h2/ping.rs @@ -34,6 +34,20 @@ use crate::rt::Sleep; type WindowSize = u32; +/// Receives a one-way notification that an HTTP/2 connection should stop being reused. +/// +/// Implementations must return promptly and must not block or panic. Hyper calls +/// the observer at most once per connection, outside its internal PING lock. +/// The observer must not retain the connection or its request sender. +/// Notification does not close the connection or cancel existing streams. +pub trait KeepAliveObserver: std::fmt::Debug + Send + Sync { + /// The keep-alive PING exceeded the configured reuse timeout without an ACK. + fn on_reuse_timeout(&self); +} + +#[cfg(test)] +mod tests; + pub(super) fn disabled() -> Recorder { Recorder { shared: None } } @@ -63,6 +77,9 @@ pub(super) fn channel(ping_pong: PingPong, config: Config, timer: Time) -> (Reco let keep_alive = config.keep_alive_interval.map(|interval| KeepAlive { interval, timeout: config.keep_alive_timeout, + reuse_timeout: config.keep_alive_reuse_timeout, + reuse_sleep: None, + observer: config.keep_alive_observer, while_idle: config.keep_alive_while_idle, sleep: timer.sleep(interval), state: KeepAliveState::Init, @@ -101,6 +118,8 @@ pub(super) struct Config { /// After sending a keepalive PING, the connection will be closed if /// a pong is not received in this amount of time. pub(super) keep_alive_timeout: Duration, + pub(super) keep_alive_reuse_timeout: Option, + pub(super) keep_alive_observer: Option>, /// If true, sends pings even when there are no active streams. pub(super) keep_alive_while_idle: bool, } @@ -158,6 +177,10 @@ struct KeepAlive { /// After sending a keepalive PING, the connection will be closed if /// a pong is not received in this amount of time. timeout: Duration, + reuse_timeout: Option, + reuse_sleep: Option>>, + // Taking the observer makes retirement irreversible for this connection. + observer: Option>, /// If true, sends pings even when there are no active streams. while_idle: bool, state: KeepAliveState, @@ -315,6 +338,12 @@ impl Ponger { locked.is_keep_alive_timed_out = true; return Poll::Ready(Ponged::KeepAliveTimedOut); } + if let Some(observer) = ka.poll_reuse_timeout(cx) { + // Never invoke user code under the shared PING lock. + drop(locked); + observer.on_reuse_timeout(); + return Poll::Pending; + } } } } @@ -449,6 +478,7 @@ impl KeepAlive { } fn schedule(&mut self, shared: &Shared) { + self.reuse_sleep = None; let interval = shared.last_read_at() + self.interval; self.state = KeepAliveState::Scheduled(interval); self.timer.reset(&mut self.sleep, interval); @@ -473,13 +503,36 @@ impl KeepAlive { trace!("keep-alive interval ({:?}) reached", self.interval); shared.send_ping(); self.state = KeepAliveState::PingSent; - let timeout = self.timer.now() + self.timeout; + let now = self.timer.now(); + let timeout = now + self.timeout; self.timer.reset(&mut self.sleep, timeout); + if self.observer.is_some() { + if let Some(reuse_timeout) = self.reuse_timeout { + let mut sleep = self.timer.sleep(reuse_timeout); + self.timer.reset(&mut sleep, now + reuse_timeout); + self.reuse_sleep = Some(sleep); + } + } } KeepAliveState::Init | KeepAliveState::PingSent => (), } } + fn poll_reuse_timeout( + &mut self, + cx: &mut task::Context<'_>, + ) -> Option> { + if !matches!(self.state, KeepAliveState::PingSent) { + return None; + } + if self.reuse_sleep.as_mut()?.as_mut().poll(cx).is_pending() { + return None; + } + self.reuse_sleep = None; + trace!("keep-alive reuse timeout reached; retiring connection"); + self.observer.take() + } + fn maybe_timeout(&mut self, cx: &mut task::Context<'_>) -> Result<(), KeepAliveTimedOut> { match self.state { KeepAliveState::PingSent => { diff --git a/src/proto/h2/ping/tests.rs b/src/proto/h2/ping/tests.rs new file mode 100644 index 0000000000..198d27761b --- /dev/null +++ b/src/proto/h2/ping/tests.rs @@ -0,0 +1,105 @@ +use super::*; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use crate::rt::Timer; + +#[derive(Debug)] +struct TestTimer; + +struct TestSleep(Pin>); + +impl Future for TestSleep { + type Output = (); + + fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<()> { + self.0.as_mut().poll(cx) + } +} + +impl Sleep for TestSleep {} + +impl Timer for TestTimer { + fn now(&self) -> Instant { + tokio::time::Instant::now().into_std() + } + + fn sleep(&self, duration: Duration) -> Pin> { + Box::pin(TestSleep(Box::pin(tokio::time::sleep(duration)))) + } + + fn sleep_until(&self, deadline: Instant) -> Pin> { + Box::pin(TestSleep(Box::pin(tokio::time::sleep_until( + deadline.into(), + )))) + } +} + +#[derive(Debug, Default)] +struct Observer(AtomicUsize); + +impl KeepAliveObserver for Observer { + fn on_reuse_timeout(&self) { + self.0.fetch_add(1, Ordering::SeqCst); + } +} + +fn waiting() -> (KeepAlive, Arc) { + let timer = Time::Timer(Arc::new(TestTimer)); + let observer = Arc::new(Observer::default()); + let ka = KeepAlive { + interval: Duration::from_secs(10), + timeout: Duration::from_secs(60), + reuse_timeout: Some(Duration::from_secs(5)), + reuse_sleep: Some(timer.sleep(Duration::from_secs(5))), + observer: Some(observer.clone()), + while_idle: true, + state: KeepAliveState::PingSent, + sleep: timer.sleep(Duration::from_secs(60)), + timer, + }; + (ka, observer) +} + +#[tokio::test(start_paused = true)] +async fn reuse_timer_wakes_without_io_and_does_not_replace_hard_timer() { + let (mut ka, observer) = waiting(); + let start = tokio::time::Instant::now(); + let notification = std::future::poll_fn(|cx| { + assert!(ka.maybe_timeout(cx).is_ok()); + match ka.poll_reuse_timeout(cx) { + Some(observer) => Poll::Ready(observer), + None => Poll::Pending, + } + }) + .await; + assert_eq!(start.elapsed(), Duration::from_secs(5)); + notification.on_reuse_timeout(); + assert_eq!(observer.0.load(Ordering::SeqCst), 1); + + // Continue polling both paths; only the original hard deadline can complete. + std::future::poll_fn(|cx| { + assert!(ka.poll_reuse_timeout(cx).is_none()); + if ka.maybe_timeout(cx).is_err() { + Poll::Ready(()) + } else { + Poll::Pending + } + }) + .await; + assert_eq!(start.elapsed(), Duration::from_secs(60)); + assert_eq!(observer.0.load(Ordering::SeqCst), 1); +} + +#[tokio::test(start_paused = true)] +async fn only_keepalive_waiting_phase_can_report_reuse_timeout() { + let (mut ka, observer) = waiting(); + ka.state = KeepAliveState::Scheduled(ka.timer.now()); + tokio::time::advance(Duration::from_secs(5)).await; + std::future::poll_fn(|cx| { + assert!(ka.poll_reuse_timeout(cx).is_none()); + assert!(ka.maybe_timeout(cx).is_ok()); + Poll::Ready(()) + }) + .await; + assert_eq!(observer.0.load(Ordering::SeqCst), 0); +} diff --git a/src/proto/h2/server.rs b/src/proto/h2/server.rs index 92026276aa..ff379fdac0 100644 --- a/src/proto/h2/server.rs +++ b/src/proto/h2/server.rs @@ -164,6 +164,8 @@ where bdp_initial_window: bdp, keep_alive_interval: config.keep_alive_interval, keep_alive_timeout: config.keep_alive_timeout, + keep_alive_reuse_timeout: None, + keep_alive_observer: None, // If keep-alive is enabled for servers, always enabled while // idle, so it can more aggressively close dead connections. keep_alive_while_idle: true, From 91fe2a92b0f828256fffdfe6bf23283ab7a6bc23 Mon Sep 17 00:00:00 2001 From: Jiahao Liang Date: Thu, 10 Sep 2026 10:56:36 +0800 Subject: [PATCH 2/4] style(http2): format public observer and configuration tests --- src/client/conn/http2.rs | 23 ++++++++++++++++------- 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/src/client/conn/http2.rs b/src/client/conn/http2.rs index 33ee61c573..856dc2de4c 100644 --- a/src/client/conn/http2.rs +++ b/src/client/conn/http2.rs @@ -486,10 +486,7 @@ where /// /// Install a separate observer for each connection when retiring individual /// pool entries. The observer must be nonblocking and must not panic. - pub fn keep_alive_observer( - &mut self, - observer: impl KeepAliveObserver + 'static, - ) -> &mut Self { + pub fn keep_alive_observer(&mut self, observer: impl KeepAliveObserver + 'static) -> &mut Self { self.h2_builder.keep_alive_observer = Some(Arc::new(observer)); self } @@ -683,19 +680,31 @@ mod tests { #[test] #[should_panic(expected = "keep_alive_reuse_timeout must be greater than zero")] fn reuse_timeout_rejects_zero() { - check_reuse_config(std::time::Duration::ZERO, std::time::Duration::from_secs(60), true); + check_reuse_config( + std::time::Duration::ZERO, + std::time::Duration::from_secs(60), + true, + ); } #[test] #[should_panic(expected = "keep_alive_reuse_timeout must be greater than zero")] fn reuse_timeout_rejects_equal_hard_timeout() { - check_reuse_config(std::time::Duration::from_secs(60), std::time::Duration::from_secs(60), true); + check_reuse_config( + std::time::Duration::from_secs(60), + std::time::Duration::from_secs(60), + true, + ); } #[test] #[should_panic(expected = "keep_alive_reuse_timeout must be greater than zero")] fn reuse_timeout_rejects_larger_than_hard_timeout() { - check_reuse_config(std::time::Duration::from_secs(61), std::time::Duration::from_secs(60), true); + check_reuse_config( + std::time::Duration::from_secs(61), + std::time::Duration::from_secs(60), + true, + ); } #[tokio::test] From 40f17df10ba7bfb96f29270abfdd4c65eb160ad9 Mon Sep 17 00:00:00 2001 From: Jiahao Liang Date: Fri, 11 Sep 2026 16:32:23 +0800 Subject: [PATCH 3/4] refactor(http2): create reuse timeout with sleep_until --- src/common/time.rs | 5 ++++- src/proto/h2/ping.rs | 4 +--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/src/common/time.rs b/src/common/time.rs index b3534f1580..2ee73388d9 100644 --- a/src/common/time.rs +++ b/src/common/time.rs @@ -40,7 +40,10 @@ impl Time { } } - #[cfg(all(feature = "server", feature = "http1"))] + #[cfg(any( + all(any(feature = "client", feature = "server"), feature = "http2"), + all(feature = "server", feature = "http1"), + ))] pub(crate) fn sleep_until(&self, deadline: Instant) -> Pin> { match &self { Time::Empty => { diff --git a/src/proto/h2/ping.rs b/src/proto/h2/ping.rs index 959031fd10..ed82ed9fdd 100644 --- a/src/proto/h2/ping.rs +++ b/src/proto/h2/ping.rs @@ -508,9 +508,7 @@ impl KeepAlive { self.timer.reset(&mut self.sleep, timeout); if self.observer.is_some() { if let Some(reuse_timeout) = self.reuse_timeout { - let mut sleep = self.timer.sleep(reuse_timeout); - self.timer.reset(&mut sleep, now + reuse_timeout); - self.reuse_sleep = Some(sleep); + self.reuse_sleep = Some(self.timer.sleep_until(now + reuse_timeout)); } } } From 79fcd12d763a8f7f97c2ce473559183fbedac184 Mon Sep 17 00:00:00 2001 From: Jiahao Liang Date: Sun, 13 Sep 2026 16:15:36 +0800 Subject: [PATCH 4/4] chore: retrigger CI