diff --git a/src/client/conn/http2.rs b/src/client/conn/http2.rs index 10613b354a..856dc2de4c 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,32 @@ 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 +588,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 +608,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 +639,74 @@ 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/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/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..ed82ed9fdd 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,34 @@ 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 { + self.reuse_sleep = Some(self.timer.sleep_until(now + reuse_timeout)); + } + } } 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,