fix(rio): prevent h2 keepalive from aborting active streams (#6630)

fix(rio): avoid false h2 keepalive stream aborts

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>

Cargo.lock

@@ -10273,6 +10273,8 @@ dependencies = [
  "hotpath",
  "http 1.5.0",
  "http-body-util",
+ "hyper",
+ "hyper-util",
  "md-5 0.11.0",
  "pin-project-lite",
  "rand 0.10.2",

crates/rio/Cargo.toml

@@ -91,4 +91,6 @@ hex-simd.workspace = true
 [dev-dependencies]
 tokio-test = { workspace = true }
 axum = { workspace = true }
+hyper = { workspace = true, features = ["http2", "server"] }
+hyper-util = { workspace = true, features = ["tokio"] }
 http-body-util = { workspace = true }

crates/rio/src/http_reader.rs

@@ -392,8 +392,8 @@ struct InternodeHttpClientTuning {
     http2_initial_connection_window_size: Option<u32>,
     http2_adaptive_window: bool,
     proxy_mode: InternodeHttpProxyMode,
-    /// True when the operator explicitly configured any HTTP/2 tuning knob
+    /// True when the operator explicitly configured an HTTP/2 window or a
-    /// (a window size, the keepalive timeout, or a non-default tuning profile).
+    /// non-default tuning profile.
     /// Used to warn once when that tuning is inert because the internode
     /// connection negotiated HTTP/1.1 over plaintext (backlog#805-C3).
     h2_tuning_explicit: bool,
@@ -410,7 +410,7 @@ impl InternodeHttpClientTuning {
     fn from_env() -> Self {
         let profile =
             parse_internode_http_tuning_profile(get_env_opt_str(rustfs_config::ENV_INTERNODE_HTTP_TUNING_PROFILE).as_deref());
-        let mut tuning = Self::from_values(
+        Self::from_values(
             profile,
             get_env_opt_usize(rustfs_config::ENV_INTERNODE_HTTP_POOL_MAX_IDLE_PER_HOST),
             get_env_opt_u64(rustfs_config::ENV_INTERNODE_HTTP_POOL_IDLE_TIMEOUT_SECS),
@@ -421,14 +421,7 @@ impl InternodeHttpClientTuning {
                 profile.default_http2_adaptive_window(),
             ),
             get_env_opt_str(rustfs_config::ENV_INTERNODE_HTTP_PROXY).as_deref(),
-        );
+        )
-        // The keepalive timeout env is read in build_http_client, not passed to
-        // from_values; fold it into the explicit-tuning signal here so the
-        // inert-h2 warning also fires when only the keepalive knob is set.
-        if get_env_opt_u64(rustfs_config::ENV_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS).is_some() {
-            tuning.h2_tuning_explicit = true;
-        }
-        tuning
     }
 
     fn from_values(
@@ -442,7 +435,6 @@ impl InternodeHttpClientTuning {
     ) -> Self {
         // Explicit h2 tuning = an operator-set window size (raw arg, before the
         // profile fallback) or a non-default (non-Legacy) tuning profile. The
-        // keepalive-env contribution is OR-ed in by from_env.
         let h2_tuning_explicit =
             stream_window_size.is_some() || connection_window_size.is_some() || profile != InternodeHttpTuningProfile::Legacy;
         Self {
@@ -530,11 +522,11 @@ fn clamp_http2_window(value: Option<u64>) -> Option<u32> {
 /// Applies internode HTTP client tuning (connection pool + HTTP/2 window
 /// sizes) to a reqwest client builder.
 ///
-/// IMPORTANT (backlog#805-C3): the HTTP/2 window and keepalive settings only
+/// IMPORTANT (backlog#805-C3): the HTTP/2 window settings only
 /// take effect when the internode connection actually negotiates HTTP/2, which
 /// happens via TLS-ALPN. Over plaintext internode transport the connection is
-/// HTTP/1.1 and every `http2_*` knob here (and the keepalive settings in
+/// HTTP/1.1 and every `http2_*` knob here is silently inert. Enable internode
-/// `build_http_client`) is silently inert. Enable internode TLS to use them.
+/// TLS to use them.
 /// `should_warn_h2_inert` emits a one-time warning when explicitly-configured
 /// h2 tuning is observed to be inert on a live connection.
 fn apply_http_client_tuning(mut builder: reqwest::ClientBuilder, tuning: InternodeHttpClientTuning) -> reqwest::ClientBuilder {
@@ -557,23 +549,26 @@ fn apply_http_client_tuning(mut builder: reqwest::ClientBuilder, tuning: Interno
     builder
 }
 
+/// Apply liveness detection for the internode HTTP data plane.
+///
+/// Hyper does not cancel a pending HTTP/2 keepalive deadline when DATA resumes.
+/// Under connection-level backpressure, a PING ACK can therefore remain queued
+/// behind healthy stream traffic until the timeout closes every multiplexed
+/// stream. Long-running data-plane paths already enforce operation-specific
+/// stall and total deadlines, while TCP keepalive covers a dead peer, so
+/// HTTP/2 PING keepalive is deliberately disabled for this client.
+fn apply_http_data_plane_liveness(builder: reqwest::ClientBuilder) -> reqwest::ClientBuilder {
+    builder
+        .tcp_keepalive(std::time::Duration::from_secs(rustfs_config::DEFAULT_INTERNODE_TCP_KEEPALIVE_SECS))
+        .http2_keep_alive_interval(None::<std::time::Duration>)
+}
+
 async fn build_http_client(
     disable_proxy: bool,
     tuning: InternodeHttpClientTuning,
     outbound_tls: &rustfs_tls_runtime::GlobalPublishedOutboundTlsState,
 ) -> io::Result<Client> {
-    // Keep the data-plane HTTP/2 keepalive timeout consistent with the control-plane
+    let mut builder = apply_http_data_plane_liveness(Client::builder().connect_timeout(std::time::Duration::from_secs(5)));
-    // gRPC channel and env-configurable. A too-aggressive value (e.g. 3s) tears down a
-    // busy data connection when a PING ACK is delayed under load, aborting in-flight
-    // shard read streams and truncating large-object GETs mid-stream (backlog#832).
-    let http2_keepalive_timeout_secs = get_env_opt_u64(rustfs_config::ENV_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS)
-        .unwrap_or(rustfs_config::DEFAULT_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS);
-    let mut builder = Client::builder()
-        .connect_timeout(std::time::Duration::from_secs(5))
-        .tcp_keepalive(std::time::Duration::from_secs(10))
-        .http2_keep_alive_interval(std::time::Duration::from_secs(5))
-        .http2_keep_alive_timeout(std::time::Duration::from_secs(http2_keepalive_timeout_secs))
-        .http2_keep_alive_while_idle(true);
     builder = apply_http_client_tuning(builder, tuning);
 
     if disable_proxy {
@@ -1592,7 +1587,7 @@ static H2_INERT_WARNED: AtomicBool = AtomicBool::new(false);
 /// inert. Warn only when the negotiated protocol is not HTTP/2, the operator
 /// explicitly configured h2 tuning, and we have not warned before. HTTP/2 for
 /// internode is negotiated via TLS-ALPN; over plaintext the connection is
-/// HTTP/1.1 and the h2 window/keepalive knobs are silently ignored.
+/// HTTP/1.1 and the h2 window knobs are silently ignored.
 fn should_warn_h2_inert(negotiated_is_http2: bool, h2_tuning_explicit: bool, already_warned: bool) -> bool {
     !negotiated_is_http2 && h2_tuning_explicit && !already_warned
 }
@@ -1609,7 +1604,7 @@ fn maybe_warn_h2_inert(negotiated_version: Version) {
             .is_ok()
     {
         warn!(
-            "internode connection negotiated HTTP/1.1 while HTTP/2 tuning is configured; the HTTP/2 window/keepalive settings apply only over internode TLS (h2 is negotiated via ALPN) — enable internode TLS to use them."
+            "internode connection negotiated HTTP/1.1 while HTTP/2 tuning is configured; the HTTP/2 window settings apply only over internode TLS (h2 is negotiated via ALPN) — enable internode TLS to use them."
         );
     }
 }
@@ -1866,7 +1861,10 @@ mod tests {
     use axum::{Router, body::Body, extract::State, http::StatusCode, response::IntoResponse, routing::get};
     use futures::stream::{self, StreamExt as _};
     use http_body_util::BodyExt as _;
+    use hyper::{Request, Response, body::Incoming, server::conn::http2, service::service_fn};
+    use hyper_util::rt::{TokioExecutor, TokioIo};
     use rustfs_io_metrics::internode_metrics::global_internode_metrics;
+    use std::convert::Infallible;
     use std::io::{self, IoSlice};
     use std::sync::{
         Arc,
@@ -2174,6 +2172,86 @@ mod tests {
         Some((format!("http://{addr}/stream"), handle))
     }
 
+    async fn start_temporarily_blocked_h2_server(blocked_for: Duration) -> Option<(String, tokio::task::JoinHandle<()>)> {
+        let listener = match TcpListener::bind("127.0.0.1:0").await {
+            Ok(listener) => listener,
+            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
+            Err(err) => panic!("HTTP/2 test listener should bind: {err}"),
+        };
+        let addr = listener
+            .local_addr()
+            .expect("HTTP/2 test listener address should be available");
+        let handle = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.expect("HTTP/2 test server should accept client");
+            let service = service_fn(move |_request: Request<Incoming>| async move {
+                let body_stream =
+                    stream::once(async { Ok::<_, Infallible>(Bytes::from_static(b"partial")) }).chain(stream::once(async move {
+                        tokio::time::sleep(Duration::from_millis(10)).await;
+                        // Model a saturated peer whose connection task cannot
+                        // process a PING ACK even though an active response body
+                        // remains valid. Blocking this one connection task is
+                        // intentional; the multi-thread test runtime keeps the
+                        // client timer and transport progressing.
+                        tokio::task::block_in_place(|| std::thread::sleep(blocked_for));
+                        Ok::<_, Infallible>(Bytes::from_static(b"complete"))
+                    }));
+                Ok::<_, Infallible>(Response::new(Body::from_stream(body_stream)))
+            });
+            let _ = http2::Builder::new(TokioExecutor::new())
+                .serve_connection(TokioIo::new(stream), service)
+                .await;
+        });
+
+        Some((format!("http://{addr}"), handle))
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn data_plane_liveness_does_not_abort_a_valid_h2_stream_while_peer_is_busy() {
+        let blocked_for = Duration::from_millis(500);
+        let Some((baseline_url, baseline_server)) = start_temporarily_blocked_h2_server(blocked_for).await else {
+            return;
+        };
+        let aggressive_keepalive_client = Client::builder()
+            .no_proxy()
+            .http2_prior_knowledge()
+            .http2_keep_alive_interval(Duration::from_millis(20))
+            .http2_keep_alive_timeout(Duration::from_millis(40))
+            .http2_keep_alive_while_idle(true)
+            .build()
+            .expect("aggressive HTTP/2 keepalive client should build");
+
+        let baseline_result = tokio::time::timeout(Duration::from_secs(2), async {
+            aggressive_keepalive_client.get(&baseline_url).send().await?.bytes().await
+        })
+        .await
+        .expect("aggressive keepalive body should finish before the outer test deadline");
+        assert!(
+            baseline_result.is_err(),
+            "fixture must reproduce a PING timeout while the server connection task is blocked"
+        );
+        baseline_server.abort();
+
+        let Some((candidate_url, candidate_server)) = start_temporarily_blocked_h2_server(blocked_for).await else {
+            return;
+        };
+        let candidate_client = apply_http_data_plane_liveness(Client::builder().no_proxy().http2_prior_knowledge())
+            .build()
+            .expect("internode data-plane HTTP/2 client should build");
+        let response = tokio::time::timeout(Duration::from_secs(2), async {
+            let response = candidate_client.get(&candidate_url).send().await?;
+            let version = response.version();
+            let body = response.bytes().await?;
+            Ok::<_, reqwest::Error>((version, body))
+        })
+        .await
+        .expect("data-plane request should finish before the outer test deadline")
+        .expect("disabled HTTP/2 PING keepalive should preserve the valid stream");
+
+        assert_eq!(response.0, Version::HTTP_2);
+        assert_eq!(response.1, Bytes::from_static(b"partialcomplete"));
+        candidate_server.abort();
+    }
+
     #[test]
     fn internode_rpc_operation_maps_known_routes() {
         assert_eq!(