From 89e94bb08668a06d6208e49b914e2c7d92980b66 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 4 Sep 2026 16:24:40 +0800 Subject: [PATCH] fix(tier): bound remote transition requests --- crates/config/README.md | 15 + crates/config/src/constants/object.rs | 32 ++ .../ecstore/src/services/tier/warm_backend.rs | 26 +- .../src/services/tier/warm_backend_s3.rs | 5 +- crates/s3-client/src/api_get_object.rs | 252 +++++++++--- crates/s3-client/src/api_list.rs | 92 ++++- .../s3-client/src/api_put_object_multipart.rs | 12 +- crates/s3-client/src/api_remove.rs | 12 +- crates/s3-client/src/api_stat.rs | 83 +++- crates/s3-client/src/bucket_cache.rs | 15 +- crates/s3-client/src/transition_api.rs | 369 ++++++++++++++++-- .../replication-outbound-transport.md | 12 + 12 files changed, 783 insertions(+), 142 deletions(-) diff --git a/crates/config/README.md b/crates/config/README.md index c450c05b0..2313a41ec 100644 --- a/crates/config/README.md +++ b/crates/config/README.md @@ -130,6 +130,21 @@ Scanner cycle budget controls: - timeout returns S3 `SlowDown`, so clients should use normal SDK retry handling. - this is not a fdatasync or group-commit switch. Track fdatasync batching separately with `rustfs_s3_put_object_rename_fdatasync_batch_files`. +## Remote tier timeout environment variables + +- `RUSTFS_TIER_REMOTE_CONNECT_TIMEOUT_SECS` + - remote tier TCP connect timeout. + - default is `10`. + - must be positive; zero fails tier client initialization, while an invalid integer is logged and falls back to the default. +- `RUSTFS_TIER_REMOTE_REQUEST_TIMEOUT_SECS` + - remote tier request timeout through response headers. + - default is `86400` so large transition uploads keep a production-safe budget. + - must be positive; zero fails tier client initialization, while an invalid integer is logged and falls back to the default. Very large values are accepted and act as a correspondingly long budget. +- `RUSTFS_TIER_REMOTE_RESPONSE_BODY_IDLE_TIMEOUT_SECS` + - maximum idle time between remote tier response-body chunks. + - default is `60`; the timer resets only when non-empty body data keeps progressing. + - must be positive; zero fails tier client initialization, while an invalid integer is logged and falls back to the default. + ## Drive timeout environment variables - `RUSTFS_DRIVE_METADATA_TIMEOUT_SECS` diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 9af74f2a3..7ec459afd 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -137,6 +137,28 @@ pub const DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED: bool = false; const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE); const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED); +/// Environment variable for remote tier TCP connect timeout in seconds. +pub const ENV_TIER_REMOTE_CONNECT_TIMEOUT_SECS: &str = "RUSTFS_TIER_REMOTE_CONNECT_TIMEOUT_SECS"; +/// Default remote tier TCP connect timeout in seconds. +pub const DEFAULT_TIER_REMOTE_CONNECT_TIMEOUT_SECS: u64 = 10; + +/// Environment variable for the remote tier request timeout in seconds. +/// +/// This bounds upload/download request progress through response headers. The +/// default is intentionally large so multi-TiB transition uploads keep their +/// previous production budget while black-hole remotes no longer wait forever. +pub const ENV_TIER_REMOTE_REQUEST_TIMEOUT_SECS: &str = "RUSTFS_TIER_REMOTE_REQUEST_TIMEOUT_SECS"; +/// Default remote tier request timeout in seconds. +pub const DEFAULT_TIER_REMOTE_REQUEST_TIMEOUT_SECS: u64 = 24 * 60 * 60; + +/// Environment variable for remote tier response-body idle timeout in seconds. +/// +/// The timer is re-armed on every non-empty response-body chunk, so slow but +/// progressing remotes can continue while silent response bodies are cancelled. +pub const ENV_TIER_REMOTE_RESPONSE_BODY_IDLE_TIMEOUT_SECS: &str = "RUSTFS_TIER_REMOTE_RESPONSE_BODY_IDLE_TIMEOUT_SECS"; +/// Default remote tier response-body idle timeout in seconds. +pub const DEFAULT_TIER_REMOTE_RESPONSE_BODY_IDLE_TIMEOUT_SECS: u64 = 60; + /// Request the object-transaction fencing contract used by storage-owned /// cleanup receipts and lock-window optimizations. /// @@ -812,6 +834,16 @@ mod remote_version_state_tests { ); } + #[test] + fn remote_tier_timeout_env_names_are_stable() { + assert_eq!(super::ENV_TIER_REMOTE_CONNECT_TIMEOUT_SECS, "RUSTFS_TIER_REMOTE_CONNECT_TIMEOUT_SECS"); + assert_eq!(super::ENV_TIER_REMOTE_REQUEST_TIMEOUT_SECS, "RUSTFS_TIER_REMOTE_REQUEST_TIMEOUT_SECS"); + assert_eq!( + super::ENV_TIER_REMOTE_RESPONSE_BODY_IDLE_TIMEOUT_SECS, + "RUSTFS_TIER_REMOTE_RESPONSE_BODY_IDLE_TIMEOUT_SECS" + ); + } + #[test] fn data_movement_part_checksum_gate_uses_stable_environment_names() { assert_eq!(super::ENV_DATA_MOVEMENT_PART_CHECKSUMS_WRITE, "RUSTFS_DATA_MOVEMENT_PART_CHECKSUMS_WRITE"); diff --git a/crates/ecstore/src/services/tier/warm_backend.rs b/crates/ecstore/src/services/tier/warm_backend.rs index ee48c116c..6528daa5b 100644 --- a/crates/ecstore/src/services/tier/warm_backend.rs +++ b/crates/ecstore/src/services/tier/warm_backend.rs @@ -37,7 +37,7 @@ use crate::services::tier::{ use bytes::Bytes; use http::StatusCode; use rustfs_s3_client::credentials::{Credentials, SignatureType, Static, Value}; -use rustfs_s3_client::transition_api::{BucketLookupType, Options, TransitionClient, TransitionCore}; +use rustfs_s3_client::transition_api::{BucketLookupType, Options, TransitionClient, TransitionClientTimeouts, TransitionCore}; use rustfs_s3_client::{ admin_handler_utils::AdminError, api_put_object::{AdvancedPutOptions, PutObjectOptions}, @@ -280,6 +280,27 @@ pub(crate) fn endpoint_authority(url: &url::Url) -> Result Duration { + Duration::from_secs(rustfs_utils::get_env_u64(env_key, default_secs)) +} + +pub(crate) fn transition_client_timeouts_from_env() -> TransitionClientTimeouts { + TransitionClientTimeouts::new( + transition_timeout_from_env( + rustfs_config::ENV_TIER_REMOTE_CONNECT_TIMEOUT_SECS, + rustfs_config::DEFAULT_TIER_REMOTE_CONNECT_TIMEOUT_SECS, + ), + transition_timeout_from_env( + rustfs_config::ENV_TIER_REMOTE_REQUEST_TIMEOUT_SECS, + rustfs_config::DEFAULT_TIER_REMOTE_REQUEST_TIMEOUT_SECS, + ), + transition_timeout_from_env( + rustfs_config::ENV_TIER_REMOTE_RESPONSE_BODY_IDLE_TIMEOUT_SECS, + rustfs_config::DEFAULT_TIER_REMOTE_RESPONSE_BODY_IDLE_TIMEOUT_SECS, + ), + ) +} + /// Build the [`WarmBackendS3`] shared by the S3-compatible warm backend providers. /// /// Credential, bucket, and endpoint validation run in this order because the @@ -310,6 +331,7 @@ pub(crate) async fn new_s3_compatible_warm_backend( signer_type: SignatureType::SignatureV4, ..Default::default() })); + let timeouts = transition_client_timeouts_from_env(); let opts = Options { creds, secure: u.scheme() == "https", @@ -322,7 +344,7 @@ pub(crate) async fn new_s3_compatible_warm_backend( // Run the SSRF guard after the host-presence check so a host-less endpoint // keeps this constructor's stable error text. (params.validate_endpoint)(&u).map_err(|err| std::io::Error::other(format!("tier endpoint is not allowed: {err}")))?; - let client = TransitionClient::new(&endpoint, opts, params.provider_tag).await?; + let client = TransitionClient::new_with_timeouts(&endpoint, opts, params.provider_tag, timeouts).await?; let client = Arc::new(client); let core = TransitionCore(Arc::clone(&client)); diff --git a/crates/ecstore/src/services/tier/warm_backend_s3.rs b/crates/ecstore/src/services/tier/warm_backend_s3.rs index 5462fc52c..48da4f312 100644 --- a/crates/ecstore/src/services/tier/warm_backend_s3.rs +++ b/crates/ecstore/src/services/tier/warm_backend_s3.rs @@ -26,7 +26,7 @@ use crate::services::tier::{ tier_config::TierS3, warm_backend::{ TransitionCandidateIdentity, TransitionCandidateProbe, TransitionCandidateReconciler, WarmBackend, WarmBackendGetOpts, - build_transition_put_options, endpoint_authority, + build_transition_put_options, endpoint_authority, transition_client_timeouts_from_env, }, }; use http::HeaderMap; @@ -139,6 +139,7 @@ impl WarmBackendS3 { } else { return Err(std::io::Error::other("insufficient parameters for S3 backend authentication")); } + let timeouts = transition_client_timeouts_from_env(); let opts = Options { creds, secure: u.scheme() == "https", @@ -147,7 +148,7 @@ impl WarmBackendS3 { ..Default::default() }; let endpoint = endpoint_authority(&u)?; - let client = TransitionClient::new(&endpoint, opts, tier_type).await?; + let client = TransitionClient::new_with_timeouts(&endpoint, opts, tier_type, timeouts).await?; let client = Arc::new(client); let core = TransitionCore(Arc::clone(&client)); diff --git a/crates/s3-client/src/api_get_object.rs b/crates/s3-client/src/api_get_object.rs index 872c1a90d..6cbe2fc02 100644 --- a/crates/s3-client/src/api_get_object.rs +++ b/crates/s3-client/src/api_get_object.rs @@ -120,18 +120,10 @@ impl TransitionClient { let h = resp.headers().clone(); - let mut body = resp.into_body(); let body_vec = if let Some(limit) = max_response_bytes { - collect_response_body(body, limit).await? + self.collect_response_body(resp.into_body(), limit).await? } else { - let mut body_vec = Vec::new(); - while let Some(frame) = body.frame().await { - let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; - if let Some(data) = frame.data_ref() { - body_vec.extend_from_slice(data); - } - } - body_vec + self.collect_response_body_unbounded(resp.into_body()).await? }; Ok((object_stat, h, BufReader::new(Cursor::new(body_vec)))) } @@ -143,7 +135,7 @@ mod bounded_response_tests { use crate::{ api_get_options::GetObjectOptions, credentials::{Credentials, SignatureType, Static, Value}, - transition_api::{BucketLookupType, Options, TransitionClient, collect_response_body}, + transition_api::{BucketLookupType, Options, TransitionClient, TransitionClientTimeouts, collect_response_body}, }; use http_body_util::Full; use hyper::body::Bytes; @@ -175,7 +167,31 @@ mod bounded_response_tests { assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); } - async fn bounded_get_fixture(body: &'static [u8]) -> Option<(TransitionClient, tokio::task::JoinHandle)> { + fn test_options() -> Options { + Options { + creds: Credentials::new(Static(Value { + access_key_id: "access-key".to_string(), + secret_access_key: "secret-key".to_string(), + signer_type: SignatureType::SignatureV4, + ..Default::default() + })), + region: "us-east-1".to_string(), + bucket_lookup: BucketLookupType::BucketLookupPath, + max_retries: 1, + ..Default::default() + } + } + + async fn client_for_endpoint(endpoint: &str, timeouts: TransitionClientTimeouts) -> TransitionClient { + TransitionClient::new_with_timeouts(endpoint, test_options(), "", timeouts) + .await + .expect("fixture client should build") + } + + async fn bounded_get_fixture_with_timeouts( + body: &'static [u8], + timeouts: TransitionClientTimeouts, + ) -> Option<(TransitionClient, 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, @@ -209,27 +225,14 @@ mod bounded_response_tests { stream.write_all(body).await.expect("fixture should write response body"); request }); - let client = TransitionClient::new( - &endpoint, - Options { - creds: Credentials::new(Static(Value { - access_key_id: "access-key".to_string(), - secret_access_key: "secret-key".to_string(), - signer_type: SignatureType::SignatureV4, - ..Default::default() - })), - region: "us-east-1".to_string(), - bucket_lookup: BucketLookupType::BucketLookupPath, - max_retries: 1, - ..Default::default() - }, - "", - ) - .await - .expect("fixture client should build"); + let client = client_for_endpoint(&endpoint, timeouts).await; Some((client, request)) } + async fn bounded_get_fixture(body: &'static [u8]) -> Option<(TransitionClient, tokio::task::JoinHandle)> { + bounded_get_fixture_with_timeouts(body, TransitionClientTimeouts::default()).await + } + #[tokio::test] async fn real_transport_accepts_the_exact_closed_range_length() { let Some((client, request)) = bounded_get_fixture(b"RustFS!").await else { @@ -292,24 +295,7 @@ mod bounded_response_tests { .local_addr() .expect("listener local address should be available") .to_string(); - let client = TransitionClient::new( - &endpoint, - Options { - creds: Credentials::new(Static(Value { - access_key_id: "access-key".to_string(), - secret_access_key: "secret-key".to_string(), - signer_type: SignatureType::SignatureV4, - ..Default::default() - })), - region: "us-east-1".to_string(), - bucket_lookup: BucketLookupType::BucketLookupPath, - max_retries: 1, - ..Default::default() - }, - "", - ) - .await - .expect("fixture client should build"); + let client = client_for_endpoint(&endpoint, TransitionClientTimeouts::default()).await; let mut opts = GetObjectOptions::default(); opts.headers .insert("range".to_string(), "bytes=0-18446744073709551615".to_string()); @@ -326,6 +312,176 @@ mod bounded_response_tests { .is_err() ); } + + #[tokio::test] + async fn connection_refused_returns_without_waiting_for_the_request_timeout() { + let listener = match TcpListener::bind("127.0.0.1:0").await { + Ok(listener) => listener, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("test listener should bind: {err}"), + }; + let endpoint = listener + .local_addr() + .expect("listener local address should be available") + .to_string(); + drop(listener); + + let client = client_for_endpoint( + &endpoint, + TransitionClientTimeouts::new(Duration::from_secs(1), Duration::from_secs(5), Duration::from_secs(1)), + ) + .await; + let mut opts = GetObjectOptions::default(); + opts.set_range(0, 6).expect("the probe range should be valid"); + + let result = tokio::time::timeout(Duration::from_secs(2), client.get_object_inner("bucket", "probe", &opts)) + .await + .expect("connection refused should return before the broader request timeout"); + + assert!(result.is_err(), "connection refused must fail instead of hanging"); + } + + #[tokio::test] + async fn response_header_stall_returns_timed_out() { + let listener = match TcpListener::bind("127.0.0.1:0").await { + Ok(listener) => listener, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("test listener should bind: {err}"), + }; + let endpoint = listener + .local_addr() + .expect("listener local address should be available") + .to_string(); + let fixture = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("fixture should accept one GET"); + let mut request = Vec::new(); + let mut buffer = [0; 1024]; + loop { + let read = stream.read(&mut buffer).await.expect("fixture should read request headers"); + assert_ne!(read, 0, "connection closed before request headers were received"); + request.extend_from_slice(&buffer[..read]); + if request.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + tokio::time::sleep(Duration::from_millis(200)).await; + }); + let client = client_for_endpoint( + &endpoint, + TransitionClientTimeouts::new(Duration::from_secs(1), Duration::from_millis(50), Duration::from_secs(1)), + ) + .await; + let mut opts = GetObjectOptions::default(); + opts.set_range(0, 6).expect("the probe range should be valid"); + + let err = client + .get_object_inner("bucket", "probe", &opts) + .await + .expect_err("response header stalls must be bounded"); + + assert_eq!(err.kind(), std::io::ErrorKind::TimedOut); + fixture.await.expect("fixture should join"); + } + + #[tokio::test] + async fn response_body_idle_stall_returns_timed_out() { + let listener = match TcpListener::bind("127.0.0.1:0").await { + Ok(listener) => listener, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("test listener should bind: {err}"), + }; + let endpoint = listener + .local_addr() + .expect("listener local address should be available") + .to_string(); + let fixture = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("fixture should accept one GET"); + let mut request = Vec::new(); + let mut buffer = [0; 1024]; + loop { + let read = stream.read(&mut buffer).await.expect("fixture should read request headers"); + assert_ne!(read, 0, "connection closed before request headers were received"); + request.extend_from_slice(&buffer[..read]); + if request.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + stream + .write_all(b"HTTP/1.1 206 Partial Content\r\nContent-Length: 7\r\nConnection: close\r\n\r\nRu") + .await + .expect("fixture should write the first body chunk"); + tokio::time::sleep(Duration::from_millis(200)).await; + }); + let client = client_for_endpoint( + &endpoint, + TransitionClientTimeouts::new(Duration::from_secs(1), Duration::from_secs(1), Duration::from_millis(50)), + ) + .await; + let mut opts = GetObjectOptions::default(); + opts.set_range(0, 6).expect("the probe range should be valid"); + + let err = client + .get_object_inner("bucket", "probe", &opts) + .await + .expect_err("body stalls after partial progress must be bounded"); + + assert_eq!(err.kind(), std::io::ErrorKind::TimedOut); + fixture.await.expect("fixture should join"); + } + + #[tokio::test] + async fn response_body_idle_timer_resets_on_progress() { + let listener = match TcpListener::bind("127.0.0.1:0").await { + Ok(listener) => listener, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("test listener should bind: {err}"), + }; + let endpoint = listener + .local_addr() + .expect("listener local address should be available") + .to_string(); + let fixture = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("fixture should accept one GET"); + let mut request = Vec::new(); + let mut buffer = [0; 1024]; + loop { + let read = stream.read(&mut buffer).await.expect("fixture should read request headers"); + assert_ne!(read, 0, "connection closed before request headers were received"); + request.extend_from_slice(&buffer[..read]); + if request.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + stream + .write_all(b"HTTP/1.1 206 Partial Content\r\nContent-Length: 7\r\nConnection: close\r\n\r\n") + .await + .expect("fixture should write response headers"); + for byte in b"RustFS!" { + stream.write_all(&[*byte]).await.expect("fixture should write body progress"); + tokio::time::sleep(Duration::from_millis(20)).await; + } + }); + let client = client_for_endpoint( + &endpoint, + TransitionClientTimeouts::new(Duration::from_millis(10), Duration::from_secs(1), Duration::from_millis(100)), + ) + .await; + let mut opts = GetObjectOptions::default(); + opts.set_range(0, 6).expect("the probe range should be valid"); + + let (_, _, mut reader) = client + .get_object_inner("bucket", "probe", &opts) + .await + .expect("continuous body progress must not be killed by the idle timer"); + let mut body = Vec::new(); + reader + .read_to_end(&mut body) + .await + .expect("bounded response should be readable"); + + assert_eq!(body, b"RustFS!"); + fixture.await.expect("fixture should join"); + } } #[derive(Default)] diff --git a/crates/s3-client/src/api_list.rs b/crates/s3-client/src/api_list.rs index dab74cf84..0bdebbf5c 100644 --- a/crates/s3-client/src/api_list.rs +++ b/crates/s3-client/src/api_list.rs @@ -27,7 +27,6 @@ use crate::{ transition_api::{ReaderImpl, RequestMetadata, TransitionClient, collect_response_body}, }; use http::{HeaderMap, StatusCode}; -use http_body_util::BodyExt; use hyper::body::Body; use hyper::body::Bytes; use rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE; @@ -124,14 +123,9 @@ impl TransitionClient { } //let mut list_bucket_result = ListBucketV2Result::default(); - let mut body_vec = Vec::new(); - let mut body = resp.into_body(); - while let Some(frame) = body.frame().await { - let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; - if let Some(data) = frame.data_ref() { - body_vec.extend_from_slice(data); - } - } + let body_vec = self + .collect_response_body(resp.into_body(), MAX_S3_CLIENT_RESPONSE_SIZE) + .await?; let mut list_bucket_result = match quick_xml::de::from_str::(&String::from_utf8_lossy(&body_vec)) { Ok(result) => result, Err(err) => { @@ -214,7 +208,9 @@ impl TransitionClient { let resp_status = resp.status(); let headers = resp.headers().clone(); - let body = collect_response_body(resp.into_body(), MAX_S3_CLIENT_RESPONSE_SIZE).await?; + let body = self + .collect_response_body(resp.into_body(), MAX_S3_CLIENT_RESPONSE_SIZE) + .await?; if resp_status != StatusCode::OK { return Err(std::io::Error::other(http_resp_to_error_response( resp_status, @@ -428,6 +424,30 @@ fn decode_s3_name(name: &str, encoding_type: &str) -> Result Options { + Options { + creds: Credentials::new(Static(Value { + access_key_id: "access-key".to_string(), + secret_access_key: "secret-key".to_string(), + signer_type: SignatureType::SignatureV4, + ..Default::default() + })), + region: "us-east-1".to_string(), + bucket_lookup: BucketLookupType::BucketLookupPath, + max_retries: 1, + ..Default::default() + } + } #[test] fn list_versions_xml_preserves_versions_and_delete_markers() { @@ -525,4 +545,56 @@ mod tests { assert_eq!(parsed.common_prefixes.len(), 1); assert_eq!(parsed.common_prefixes[0].prefix, "subdir/"); } + + #[tokio::test] + async fn list_objects_v2_body_stall_returns_timed_out() { + let listener = match TcpListener::bind("127.0.0.1:0").await { + Ok(listener) => listener, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("test listener should bind: {err}"), + }; + let endpoint = listener + .local_addr() + .expect("listener local address should be available") + .to_string(); + let fixture = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("fixture should accept one list request"); + let mut request = Vec::new(); + let mut buffer = [0; 1024]; + loop { + let read = stream.read(&mut buffer).await.expect("fixture should read request headers"); + assert_ne!(read, 0, "connection closed before request headers were received"); + request.extend_from_slice(&buffer[..read]); + if request.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 512\r\nConnection: close\r\n\r\nwarm") + .await + .expect("fixture should write a partial list response"); + tokio::time::sleep(Duration::from_millis(200)).await; + }); + let client = TransitionClient::new_with_timeouts( + &endpoint, + timeout_test_options(), + "", + TransitionClientTimeouts::new(Duration::from_secs(1), Duration::from_secs(1), Duration::from_millis(50)), + ) + .await + .expect("fixture client should build"); + client + .bucket_loc_cache + .lock() + .expect("location cache should lock") + .set("bucket", "us-east-1"); + + let err = client + .list_objects_v2_query("bucket", "", "", false, false, "", "", 1, HeaderMap::new()) + .await + .expect_err("a stalled ListObjectsV2 body must be bounded"); + + assert_eq!(err.kind(), std::io::ErrorKind::TimedOut); + fixture.await.expect("fixture should join"); + } } diff --git a/crates/s3-client/src/api_put_object_multipart.rs b/crates/s3-client/src/api_put_object_multipart.rs index d7d1ba7a1..8262a6598 100644 --- a/crates/s3-client/src/api_put_object_multipart.rs +++ b/crates/s3-client/src/api_put_object_multipart.rs @@ -18,7 +18,6 @@ #![allow(clippy::all)] use http::{HeaderMap, HeaderName, StatusCode}; -use http_body_util::BodyExt; use hyper::body::Bytes; use s3s::S3ErrorCode; use std::collections::HashMap; @@ -247,14 +246,9 @@ impl TransitionClient { // Parse the CreateMultipartUpload response for the UploadId. Returning a // default (empty) result here made every multipart transition fail at the // first UploadPart with "UploadID cannot be empty" (rustfs/rustfs#4811). - let mut body_vec = Vec::new(); - let mut body = resp.into_body(); - while let Some(frame) = body.frame().await { - let frame = frame.map_err(|e| std::io::Error::other(e.to_string()))?; - if let Some(data) = frame.data_ref() { - body_vec.extend_from_slice(data); - } - } + let body_vec = self + .collect_response_body(resp.into_body(), rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE) + .await?; let initiate_multipart_upload_result = quick_xml::de::from_str::(&String::from_utf8_lossy(&body_vec)) .map_err(|e| std::io::Error::other(format!("failed to parse CreateMultipartUpload response: {e}")))?; diff --git a/crates/s3-client/src/api_remove.rs b/crates/s3-client/src/api_remove.rs index 5e063e14f..4a552cfcd 100644 --- a/crates/s3-client/src/api_remove.rs +++ b/crates/s3-client/src/api_remove.rs @@ -19,7 +19,6 @@ #![allow(clippy::all)] use http::{HeaderMap, HeaderValue, Method, StatusCode}; -use http_body_util::BodyExt; use hyper::body::Body; use hyper::body::Bytes; use rustfs_utils::HashAlgorithm; @@ -351,14 +350,9 @@ impl TransitionClient { ) .await?; - let mut body_vec = Vec::new(); - let mut body = resp.into_body(); - while let Some(frame) = body.frame().await { - let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; - if let Some(data) = frame.data_ref() { - body_vec.extend_from_slice(data); - } - } + let body_vec = self + .collect_response_body(resp.into_body(), rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE) + .await?; process_remove_multi_objects_response( ReaderImpl::Body(Bytes::from(body_vec)), bucket_name, diff --git a/crates/s3-client/src/api_stat.rs b/crates/s3-client/src/api_stat.rs index bac59c8ed..9386a209e 100644 --- a/crates/s3-client/src/api_stat.rs +++ b/crates/s3-client/src/api_stat.rs @@ -19,7 +19,6 @@ #![allow(clippy::all)] use http::{HeaderMap, HeaderValue, StatusCode}; -use http_body_util::BodyExt; use hyper::body::Body; use hyper::body::Bytes; use rustfs_utils::EMPTY_STRING_SHA256_HASH; @@ -119,14 +118,9 @@ impl TransitionClient { let resp_status = resp.status(); let h = resp.headers().clone(); - let mut body_vec = Vec::new(); - let mut body = resp.into_body(); - while let Some(frame) = body.frame().await { - let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; - if let Some(data) = frame.data_ref() { - body_vec.extend_from_slice(data); - } - } + let body_vec = self + .collect_response_body(resp.into_body(), rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE) + .await?; let resperr = http_resp_to_error_response(resp_status, &h, body_vec, bucket_name, ""); warn!("bucket exists, resperr: {:?}", resperr); @@ -170,11 +164,13 @@ impl TransitionClient { let resp_status = resp.status(); let h = resp.headers().clone(); - let body_vec = collect_response_body(resp.into_body(), rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE).await?; + let body_vec = self + .collect_response_body(resp.into_body(), rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE) + .await?; parse_bucket_versioning_response(resp_status, &h, body_vec, bucket_name) } - Err(err) => Err(std::io::Error::other(err)), + Err(err) => Err(err), } } @@ -274,8 +270,14 @@ impl TransitionClient { #[cfg(test)] mod tests { use super::parse_bucket_versioning_response; + use crate::{ + credentials::{Credentials, SignatureType, Static, Value}, + transition_api::{BucketLookupType, Options, TransitionClient, TransitionClientTimeouts}, + }; use http::{HeaderMap, StatusCode}; use s3s::dto::BucketVersioningStatus; + use std::time::Duration; + use tokio::{io::AsyncReadExt, net::TcpListener}; #[test] fn parses_bucket_versioning_statuses_mfa_delete_and_unversioned_state() { @@ -338,4 +340,63 @@ mod tests { assert_eq!(strict_err.kind(), std::io::ErrorKind::InvalidData); } } + + #[tokio::test] + async fn get_bucket_versioning_preserves_request_timeout_kind() { + let listener = match TcpListener::bind("127.0.0.1:0").await { + Ok(listener) => listener, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("test listener should bind: {err}"), + }; + let endpoint = listener + .local_addr() + .expect("listener local address should be available") + .to_string(); + let fixture = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("fixture should accept one versioning request"); + let mut request = Vec::new(); + let mut buffer = [0; 1024]; + loop { + let read = stream.read(&mut buffer).await.expect("fixture should read request headers"); + assert_ne!(read, 0, "connection closed before request headers were received"); + request.extend_from_slice(&buffer[..read]); + if request.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + tokio::time::sleep(Duration::from_millis(200)).await; + }); + let client = TransitionClient::new_with_timeouts( + &endpoint, + Options { + creds: Credentials::new(Static(Value { + access_key_id: "access-key".to_string(), + secret_access_key: "secret-key".to_string(), + signer_type: SignatureType::SignatureV4, + ..Default::default() + })), + region: "us-east-1".to_string(), + bucket_lookup: BucketLookupType::BucketLookupPath, + max_retries: 1, + ..Default::default() + }, + "", + TransitionClientTimeouts::new(Duration::from_secs(1), Duration::from_millis(50), Duration::from_secs(1)), + ) + .await + .expect("fixture client should build"); + client + .bucket_loc_cache + .lock() + .expect("location cache should lock") + .set("bucket", "us-east-1"); + + let err = client + .get_bucket_versioning("bucket") + .await + .expect_err("a stalled versioning request must time out"); + + assert_eq!(err.kind(), std::io::ErrorKind::TimedOut); + fixture.await.expect("fixture should join"); + } } diff --git a/crates/s3-client/src/bucket_cache.rs b/crates/s3-client/src/bucket_cache.rs index c891d65a0..c3429da19 100644 --- a/crates/s3-client/src/bucket_cache.rs +++ b/crates/s3-client/src/bucket_cache.rs @@ -26,7 +26,6 @@ use crate::{ transition_api::{CreateBucketConfiguration, LocationConstraint, TransitionClient}, }; use http::Request; -use http_body_util::BodyExt; use hyper::StatusCode; use hyper::body::Body; use hyper::body::Bytes; @@ -86,7 +85,7 @@ impl TransitionClient { let req = self.get_bucket_location_request(bucket_name)?; let mut resp = self.doit(req).await?; - location = process_bucket_location_response(resp, bucket_name, &self.tier_type).await?; + location = process_bucket_location_response(self, resp, bucket_name, &self.tier_type).await?; { if let Ok(mut bucket_loc_cache) = self.bucket_loc_cache.lock() { bucket_loc_cache.set(bucket_name, &location); @@ -198,6 +197,7 @@ impl TransitionClient { } async fn process_bucket_location_response( + client: &TransitionClient, mut resp: http::Response, bucket_name: &str, tier_type: &str, @@ -237,14 +237,9 @@ async fn process_bucket_location_response( } //} - let mut body_vec = Vec::new(); - let mut body = resp.into_body(); - while let Some(frame) = body.frame().await { - let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; - if let Some(data) = frame.data_ref() { - body_vec.extend_from_slice(data); - } - } + let body_vec = client + .collect_response_body(resp.into_body(), MAX_S3_CLIENT_RESPONSE_SIZE) + .await?; let mut location = "".to_string(); if tier_type == "huaweicloud" { if let Ok(body_str) = String::from_utf8(body_vec) { diff --git a/crates/s3-client/src/transition_api.rs b/crates/s3-client/src/transition_api.rs index 6f45c76e2..bd3c62fb5 100644 --- a/crates/s3-client/src/transition_api.rs +++ b/crates/s3-client/src/transition_api.rs @@ -41,7 +41,7 @@ use http::{ request::{Builder, Request}, }; use http_body::Body; -use http_body_util::{BodyExt, LengthLimitError, Limited}; +use http_body_util::BodyExt; use hyper::body::Bytes; use hyper::body::Incoming; use hyper_rustls::{ConfigBuilderExt, HttpsConnector}; @@ -67,10 +67,12 @@ use s3s::dto::Owner; use s3s::dto::ReplicationStatus; use serde::{Deserialize, Serialize}; use sha2::Sha256; +use std::error::Error as StdError; use std::io::Cursor; use std::pin::Pin; use std::sync::atomic::{AtomicI32, Ordering}; use std::task::{Context, Poll}; +use std::time::Duration as StdDuration; use std::{ collections::HashMap, sync::{Arc, Mutex}, @@ -79,28 +81,108 @@ use time::Duration; use time::OffsetDateTime; use tokio::io::BufReader; use tokio::io::{AsyncRead, AsyncReadExt}; -use tracing::{debug, error, warn}; +use tracing::{debug, error, trace, warn}; use url::{Url, form_urlencoded}; use uuid::Uuid; const C_USER_AGENT: &str = "RustFS (linux; x86)"; pub const MAX_S3_ERROR_RESPONSE_SIZE: usize = 64 * 1024; +const EVENT_TIER_REMOTE_TRANSPORT: &str = "tier_remote_transport"; +const LOG_COMPONENT_S3_CLIENT: &str = "s3_client"; +const LOG_SUBSYSTEM_TIER: &str = "tier"; const SUCCESS_STATUS: [StatusCode; 3] = [StatusCode::OK, StatusCode::NO_CONTENT, StatusCode::PARTIAL_CONTENT]; +fn response_body_exceeds_limit_error() -> std::io::Error { + std::io::Error::new(std::io::ErrorKind::InvalidData, "remote tier response body exceeds limit") +} + +fn remote_tier_timeout_error(message: &'static str) -> std::io::Error { + std::io::Error::new(std::io::ErrorKind::TimedOut, message) +} + +fn source_chain_has_io_kind(error: &(dyn StdError + 'static), kind: std::io::ErrorKind) -> bool { + let mut current = Some(error); + while let Some(error) = current { + if error + .downcast_ref::() + .is_some_and(|io_error| io_error.kind() == kind) + { + return true; + } + current = error.source(); + } + false +} + +fn transition_transport_error(err: hyper_util::client::legacy::Error) -> std::io::Error { + if source_chain_has_io_kind(&err, std::io::ErrorKind::TimedOut) { + return remote_tier_timeout_error("remote tier connection timed out"); + } + std::io::Error::other(err) +} + +async fn next_response_body_data( + mut body: Pin<&mut B>, + idle_timeout: Option, +) -> Result, std::io::Error> +where + B: Body, + B::Error: Into>, +{ + let next_nonempty_data = async { + loop { + let Some(frame) = std::future::poll_fn(|cx| body.as_mut().poll_frame(cx)).await else { + return Ok(None); + }; + let frame = frame.map_err(std::io::Error::other)?; + let Ok(data) = frame.into_data() else { + continue; + }; + if !data.is_empty() { + return Ok(Some(data)); + } + } + }; + + if let Some(idle_timeout) = idle_timeout { + tokio::time::timeout(idle_timeout, next_nonempty_data) + .await + .map_err(|_| remote_tier_timeout_error("remote tier response body stalled"))? + } else { + next_nonempty_data.await + } +} + +async fn collect_response_body_inner( + body: B, + limit: Option, + idle_timeout: Option, +) -> Result, std::io::Error> +where + B: Body, + B::Error: Into>, +{ + let mut body_vec = Vec::new(); + let mut body = std::pin::pin!(body); + while let Some(data) = next_response_body_data(body.as_mut(), idle_timeout).await? { + let Some(new_len) = body_vec.len().checked_add(data.len()) else { + return Err(response_body_exceeds_limit_error()); + }; + if limit.is_some_and(|limit| new_len > limit) { + return Err(response_body_exceeds_limit_error()); + } + body_vec.extend_from_slice(&data); + } + Ok(body_vec) +} + pub async fn collect_response_body(body: B, limit: usize) -> Result, std::io::Error> where B: Body, - B::Error: Into>, + B::Error: Into>, { - let body = Limited::new(body, limit).collect().await.map_err(|err| { - if err.is::() { - std::io::Error::new(std::io::ErrorKind::InvalidData, "remote tier response body exceeds limit") - } else { - std::io::Error::other(err) - } - })?; - Ok(body.to_bytes().to_vec()) + collect_response_body_inner(body, Some(limit), None).await } const C_UNKNOWN: i32 = -1; @@ -196,6 +278,62 @@ pub struct TransitionClient { pub trailing_header_support: bool, pub max_retries: i64, pub tier_type: String, + pub timeouts: TransitionClientTimeouts, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct TransitionClientTimeouts { + pub connect_timeout: StdDuration, + pub request_timeout: StdDuration, + pub response_body_idle_timeout: StdDuration, +} + +impl TransitionClientTimeouts { + pub const fn new( + connect_timeout: StdDuration, + request_timeout: StdDuration, + response_body_idle_timeout: StdDuration, + ) -> Self { + Self { + connect_timeout, + request_timeout, + response_body_idle_timeout, + } + } + + fn validate(self) -> Result { + if self.connect_timeout.is_zero() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "remote tier connect timeout must be greater than zero", + )); + } + if self.request_timeout.is_zero() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "remote tier request timeout must be greater than zero", + )); + } + if self.response_body_idle_timeout.is_zero() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "remote tier response body idle timeout must be greater than zero", + )); + } + Ok(self) + } +} + +impl Default for TransitionClientTimeouts { + fn default() -> Self { + Self { + connect_timeout: StdDuration::from_secs(rustfs_config::DEFAULT_TIER_REMOTE_CONNECT_TIMEOUT_SECS), + request_timeout: StdDuration::from_secs(rustfs_config::DEFAULT_TIER_REMOTE_REQUEST_TIMEOUT_SECS), + response_body_idle_timeout: StdDuration::from_secs( + rustfs_config::DEFAULT_TIER_REMOTE_RESPONSE_BODY_IDLE_TIMEOUT_SECS, + ), + } + } } #[derive(Debug, Default)] @@ -288,12 +426,28 @@ async fn build_tls_config() -> Result { impl TransitionClient { pub async fn new(endpoint: &str, opts: Options, tier_type: &str) -> Result { - let client = Self::private_new(endpoint, opts, tier_type).await?; - - Ok(client) + Self::private_new(endpoint, opts, tier_type, TransitionClientTimeouts::default()).await } - async fn private_new(endpoint: &str, opts: Options, tier_type: &str) -> Result { + /// Builds a transition client with explicit transport timeout budgets. + /// + /// [`Self::new`] keeps the historical constructor surface and uses the + /// production defaults from [`TransitionClientTimeouts::default`]. + pub async fn new_with_timeouts( + endpoint: &str, + opts: Options, + tier_type: &str, + timeouts: TransitionClientTimeouts, + ) -> Result { + Self::private_new(endpoint, opts, tier_type, timeouts).await + } + + async fn private_new( + endpoint: &str, + opts: Options, + tier_type: &str, + timeouts: TransitionClientTimeouts, + ) -> Result { if rustls::crypto::CryptoProvider::get_default().is_none() { // No default provider is set yet; try to install aws-lc-rs. // `install_default` can only fail if another thread races us and installs a provider @@ -306,15 +460,19 @@ impl TransitionClient { } let endpoint_url = get_endpoint_url(endpoint, opts.secure)?; + let timeouts = timeouts.validate()?; let tls = build_tls_config().await?; + let mut http = HttpConnector::new(); + http.enforce_http(false); + http.set_connect_timeout(Some(timeouts.connect_timeout)); let https = hyper_rustls::HttpsConnectorBuilder::new() .with_tls_config(tls) .https_or_http() .enable_http1() .enable_http2() - .build(); + .wrap_connector(http); let http_client = Client::builder(TokioExecutor::new()).build(https); let mut client = TransitionClient { @@ -337,6 +495,7 @@ impl TransitionClient { trailing_header_support: opts.trailing_headers, max_retries: opts.max_retries, tier_type: tier_type.to_string(), + timeouts, }; { @@ -501,29 +660,43 @@ impl TransitionClient { } pub async fn doit(&self, req: Request) -> Result, std::io::Error> { - let req_method; - let req_uri; - let resp; let http_client = self.http_client.clone(); - { - req_method = req.method().clone(); - req_uri = req.uri().clone(); - - debug!("endpoint_url: {}", self.endpoint_url.as_str().to_string()); - resp = http_client.request(req); - } - let resp = resp.await; - debug!("http_client url: {} {}", req_method, req_uri); - if let Err(err) = resp { - error!("http_client call error: {:?}", err); - return Err(std::io::Error::other(err)); - } - + let req_method = req.method().clone(); + let resp = tokio::time::timeout(self.timeouts.request_timeout, http_client.request(req)).await; let resp = match resp { - Ok(r) => r, - Err(_) => return Err(std::io::Error::other("Unexpected error in response")), + Ok(Ok(resp)) => resp, + Ok(Err(err)) => { + let err = transition_transport_error(err); + error!( + event = EVENT_TIER_REMOTE_TRANSPORT, + component = LOG_COMPONENT_S3_CLIENT, + subsystem = LOG_SUBSYSTEM_TIER, + method = %req_method, + error_kind = ?err.kind(), + "remote tier request failed" + ); + return Err(err); + } + Err(_) => { + warn!( + event = EVENT_TIER_REMOTE_TRANSPORT, + component = LOG_COMPONENT_S3_CLIENT, + subsystem = LOG_SUBSYSTEM_TIER, + method = %req_method, + timeout_ms = self.timeouts.request_timeout.as_millis(), + "remote tier request timed out before response headers" + ); + return Err(remote_tier_timeout_error("remote tier request timed out before response headers")); + } }; - debug!(status = %resp.status(), "remote tier response received"); + trace!( + event = EVENT_TIER_REMOTE_TRANSPORT, + component = LOG_COMPONENT_S3_CLIENT, + subsystem = LOG_SUBSYSTEM_TIER, + method = %req_method, + status = %resp.status(), + "remote tier response received" + ); //let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec(); //debug!("http_resp_body: {}", String::from_utf8(b).unwrap()); @@ -537,7 +710,15 @@ impl TransitionClient { .and_then(|value| value.to_str().ok()) .unwrap_or_default() .to_string(); - warn!(status = %status, request_id, "remote tier request rejected"); + warn!( + event = EVENT_TIER_REMOTE_TRANSPORT, + component = LOG_COMPONENT_S3_CLIENT, + subsystem = LOG_SUBSYSTEM_TIER, + method = %req_method, + status = %status, + request_id, + "remote tier request rejected" + ); } Ok(resp) } @@ -581,7 +762,9 @@ impl TransitionClient { let resp_status = resp.status(); let h = resp.headers().clone(); - let body_vec = collect_response_body(resp.into_body(), MAX_S3_ERROR_RESPONSE_SIZE).await?; + let body_vec = self + .collect_response_body(resp.into_body(), MAX_S3_ERROR_RESPONSE_SIZE) + .await?; let parsed_error = http_resp_to_error_response(resp_status, &h, body_vec, &metadata.bucket_name, &metadata.object_name); let routing_region = parsed_error.region; @@ -635,6 +818,22 @@ impl TransitionClient { Err(std::io::Error::other("remote tier request did not produce a response")) } + pub async fn collect_response_body(&self, body: B, limit: usize) -> Result, std::io::Error> + where + B: Body, + B::Error: Into>, + { + collect_response_body_inner(body, Some(limit), Some(self.timeouts.response_body_idle_timeout)).await + } + + pub async fn collect_response_body_unbounded(&self, body: B) -> Result, std::io::Error> + where + B: Body, + B::Error: Into>, + { + collect_response_body_inner(body, None, Some(self.timeouts.response_body_idle_timeout)).await + } + async fn new_request( &self, method: &http::Method, @@ -1504,12 +1703,17 @@ pub struct CreateBucketConfiguration { mod tests { use super::{ MAX_S3_CLIENT_RESPONSE_SIZE, MAX_S3_ERROR_RESPONSE_SIZE, SignatureType, build_tls_config, collect_response_body, - signer_error_to_io_error, to_object_info_for_provider, validate_header_values, with_rustls_init_guard, + collect_response_body_inner, signer_error_to_io_error, to_object_info_for_provider, validate_header_values, + with_rustls_init_guard, }; use crate::provider_versions::{BucketVersioningState, ProviderVersionCapabilities, RemoteVersion}; - use http::{HeaderMap, HeaderValue}; - use http_body_util::Full; + use futures::stream; + use http::{HeaderMap, HeaderValue, Request}; + use http_body::Frame; + use http_body_util::{Full, StreamBody}; use hyper::body::Bytes; + use std::time::Duration as StdDuration; + use tokio::net::TcpListener; use uuid::Uuid; #[tokio::test] @@ -1540,6 +1744,77 @@ mod tests { assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); } + #[tokio::test] + async fn empty_data_frames_do_not_reset_the_body_idle_timeout() { + let frames = stream::unfold((), |_| async { + tokio::time::sleep(StdDuration::from_millis(10)).await; + Some((Ok::<_, std::io::Error>(Frame::data(Bytes::new())), ())) + }); + let body = StreamBody::new(Box::pin(frames)); + + let err = tokio::time::timeout( + StdDuration::from_millis(200), + collect_response_body_inner(body, Some(1), Some(StdDuration::from_millis(50))), + ) + .await + .expect("the collector should enforce its own body idle timeout") + .expect_err("empty frames must not count as body progress"); + + assert_eq!(err.kind(), std::io::ErrorKind::TimedOut); + } + + #[tokio::test] + async fn public_body_collector_accepts_non_unpin_bodies() { + let body = StreamBody::new(stream::once(async { Ok::<_, std::io::Error>(Frame::data(Bytes::from_static(b"ok"))) })); + + let collected = collect_response_body(body, 2) + .await + .expect("the public collector should pin non-Unpin bodies internally"); + + assert_eq!(collected, b"ok"); + } + + #[tokio::test] + async fn https_endpoints_reach_the_transport_connector() { + let listener = match TcpListener::bind("127.0.0.1:0").await { + Ok(listener) => listener, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("test listener should bind: {err}"), + }; + let endpoint = listener + .local_addr() + .expect("listener local address should be available") + .to_string(); + let accepted = tokio::spawn(async move { + let (stream, _) = tokio::time::timeout(StdDuration::from_secs(1), listener.accept()) + .await + .expect("HTTPS connector should reach the TCP listener") + .expect("fixture should accept the HTTPS connection"); + drop(stream); + }); + let client = super::TransitionClient::new_with_timeouts( + &endpoint, + super::Options { + secure: true, + ..Default::default() + }, + "", + super::TransitionClientTimeouts::new(StdDuration::from_secs(1), StdDuration::from_secs(1), StdDuration::from_secs(1)), + ) + .await + .expect("fixture client should build"); + let request = Request::builder() + .uri(format!("https://{endpoint}/")) + .body(s3s::Body::empty()) + .expect("fixture request should build"); + + client + .doit(request) + .await + .expect_err("the fixture closes before completing the TLS handshake"); + accepted.await.expect("fixture should join"); + } + #[test] fn rustls_guard_converts_panics_to_io_errors() { let err = with_rustls_init_guard(|| -> Result<(), std::io::Error> { panic!("missing provider") }) @@ -1573,6 +1848,18 @@ mod tests { assert!(outcome.is_ok(), "provider install guard must not panic when a provider is already set"); } + #[test] + fn transition_timeouts_reject_zero_budgets() { + for timeouts in [ + super::TransitionClientTimeouts::new(StdDuration::ZERO, StdDuration::from_secs(1), StdDuration::from_secs(1)), + super::TransitionClientTimeouts::new(StdDuration::from_secs(1), StdDuration::ZERO, StdDuration::from_secs(1)), + super::TransitionClientTimeouts::new(StdDuration::from_secs(1), StdDuration::from_secs(1), StdDuration::ZERO), + ] { + let err = timeouts.validate().expect_err("zero timeout budgets must fail closed"); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); + } + } + #[test] fn validate_header_values_returns_header_name_for_non_utf8_values() { let mut headers = HeaderMap::new(); diff --git a/docs/operations/replication-outbound-transport.md b/docs/operations/replication-outbound-transport.md index 8ff7f285a..f3149e21c 100644 --- a/docs/operations/replication-outbound-transport.md +++ b/docs/operations/replication-outbound-transport.md @@ -29,6 +29,18 @@ Both knobs are read by the RustFS process that owns the replication target, at client build time; restart the server after changing them. +### Remote tier transport timeouts + +Remote tier S3-compatible clients use separate transport budgets. These settings do not change bucket or site replication clients. + +| Variable | Default | Meaning | +| --- | --- | --- | +| `RUSTFS_TIER_REMOTE_CONNECT_TIMEOUT_SECS` | `10` | Maximum time to establish the remote tier TCP connection. | +| `RUSTFS_TIER_REMOTE_REQUEST_TIMEOUT_SECS` | `86400` | Maximum time for a remote tier request to reach response headers. The long default preserves large transition-upload headroom. | +| `RUSTFS_TIER_REMOTE_RESPONSE_BODY_IDLE_TIMEOUT_SECS` | `60` | Maximum time without a non-empty response-body chunk. Empty HTTP/2 frames do not count as progress. | + +All three values must be positive integers. Zero fails tier client initialization instead of silently disabling the boundary. An invalid integer is logged and falls back to the default; very large values are accepted and provide a correspondingly long effective budget. The values are read when the tier client is built; recreate or reload the tier configuration after changing them. + ## Before changing any of this Follow the SOP in `docs/postmortems/2026-09-03-replication-checksum-default-regression.md`: inventory the target-side rules the current default satisfies, run the outbound target matrix, and document any new knob here in the same PR.