From 5e05bd54854e64e2bfec4e5d90c9d0fa3dc95bd9 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 26 Aug 2026 13:25:38 +0800 Subject: [PATCH] fix(rio): prevent h2 keepalive from aborting active streams (#6630) fix(rio): avoid false h2 keepalive stream aborts Co-authored-by: Henry Guo --- Cargo.lock | 2 + crates/rio/Cargo.toml | 2 + crates/rio/src/http_reader.rs | 136 ++++++++++++++++++++++++++-------- 3 files changed, 111 insertions(+), 29 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1731d9207..1768d6849 100644 --- a/Cargo.lock +++ b/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", diff --git a/crates/rio/Cargo.toml b/crates/rio/Cargo.toml index 72bb71af0..87ecc79cb 100644 --- a/crates/rio/Cargo.toml +++ b/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 } diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index 93f61a15c..8ab79bbb6 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -392,8 +392,8 @@ struct InternodeHttpClientTuning { http2_initial_connection_window_size: Option, http2_adaptive_window: bool, proxy_mode: InternodeHttpProxyMode, - /// True when the operator explicitly configured any HTTP/2 tuning knob - /// (a window size, the keepalive timeout, or a non-default tuning profile). + /// True when the operator explicitly configured an HTTP/2 window or a + /// 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) -> Option { /// 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 -/// `build_http_client`) is silently inert. Enable internode TLS to use them. +/// HTTP/1.1 and every `http2_*` knob here is silently inert. Enable internode +/// 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::) +} + async fn build_http_client( disable_proxy: bool, tuning: InternodeHttpClientTuning, outbound_tls: &rustfs_tls_runtime::GlobalPublishedOutboundTlsState, ) -> io::Result { - // Keep the data-plane HTTP/2 keepalive timeout consistent with the control-plane - // 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); + let mut builder = apply_http_data_plane_liveness(Client::builder().connect_timeout(std::time::Duration::from_secs(5))); 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| 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!(