feat(http2): add an optional keep-alive reuse timeout observer

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<B> {
     dispatch: dispatch::UnboundedSender<Request<B>, Response<IncomingBody>>,
@@ -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<Duration>) -> &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<Box<dyn Error + Send + Sync>>,
         Ex: Http2ClientConnExec<B, T> + 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<F> crate::rt::Executor<F> 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<bytes::Bytes>>(
+            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() {

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<Box<dyn Sleep>> {
         match &self {
             Time::Empty => {

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<Duration>,
     pub(crate) keep_alive_timeout: Duration,
+    pub(crate) keep_alive_reuse_timeout: Option<Duration>,
+    pub(crate) keep_alive_observer: Option<std::sync::Arc<dyn ping::KeepAliveObserver>>,
     pub(crate) keep_alive_while_idle: bool,
     pub(crate) max_concurrent_reset_streams: Option<usize>,
     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,
     }
 }

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<Duration>,
+    pub(super) keep_alive_observer: Option<Arc<dyn KeepAliveObserver>>,
     /// 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<Duration>,
+    reuse_sleep: Option<Pin<Box<dyn Sleep>>>,
+    // Taking the observer makes retirement irreversible for this connection.
+    observer: Option<Arc<dyn KeepAliveObserver>>,
     /// 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<Arc<dyn KeepAliveObserver>> {
+        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 => {

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<Box<tokio::time::Sleep>>);
+
+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<dyn Sleep>> {
+        Box::pin(TestSleep(Box::pin(tokio::time::sleep(duration))))
+    }
+
+    fn sleep_until(&self, deadline: Instant) -> Pin<Box<dyn Sleep>> {
+        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<Observer>) {
+    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);
+}