Merge branch 'main' into docs/scanner-heal-v2-plan

Signed-off-by: houseme <housemecn@gmail.com>

Cargo.lock

@@ -315,7 +315,7 @@ dependencies = [
  "strum",
  "thiserror 2.0.20",
  "uuid",
- "zstd",
+ "zstd 0.13.3",
 ]
 
 [[package]]
@@ -508,7 +508,7 @@ dependencies = [
  "arrow-select",
  "flatbuffers",
  "lz4_flex",
- "zstd",
+ "zstd 0.13.3",
 ]
 
 [[package]]
@@ -2249,8 +2249,8 @@ dependencies = [
  "liblzma",
  "lz4",
  "memchr",
- "zstd",
- "zstd-safe",
+ "zstd 0.13.3",
+ "zstd-safe 7.3.0",
 ]
 
 [[package]]
@@ -4067,7 +4067,7 @@ dependencies = [
  "uuid",
  "walkdir",
  "zip",
- "zstd",
+ "zstd 0.14.0",
 ]
 
 [[package]]
@@ -5971,7 +5971,7 @@ dependencies = [
  "lz4",
  "snap",
  "uuid",
- "zstd",
+ "zstd 0.13.3",
 ]
 
 [[package]]
@@ -7658,7 +7658,7 @@ dependencies = [
  "snap",
  "tokio",
  "twox-hash",
- "zstd",
+ "zstd 0.13.3",
 ]
 
 [[package]]
@@ -9618,7 +9618,7 @@ dependencies = [
  "x509-parser",
  "zeroize",
  "zip",
- "zstd",
+ "zstd 0.14.0",
 ]
 
 [[package]]
@@ -10228,7 +10228,7 @@ dependencies = [
  "thiserror 2.0.20",
  "walkdir",
  "zip",
- "zstd",
+ "zstd 0.14.0",
 ]
 
 [[package]]
@@ -10395,7 +10395,7 @@ dependencies = [
  "tracing-opentelemetry",
  "tracing-subscriber",
  "url",
- "zstd",
+ "zstd 0.14.0",
 ]
 
 [[package]]
@@ -10985,7 +10985,7 @@ dependencies = [
  "transform-stream",
  "url",
  "windows",
- "zstd",
+ "zstd 0.14.0",
 ]
 
 [[package]]
@@ -14095,7 +14095,7 @@ dependencies = [
  "typed-path",
  "zeroize",
  "zopfli",
- "zstd",
+ "zstd 0.13.3",
 ]
 
 [[package]]
@@ -14128,7 +14128,16 @@ version = "0.13.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a"
 dependencies = [
- "zstd-safe",
+ "zstd-safe 7.3.0",
+]
+
+[[package]]
+name = "zstd"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bf06bd8162af0734b344780deb55b42a2429ae430870d13fcc12f238e880fe6e"
+dependencies = [
+ "zstd-safe 8.0.0",
 ]
 
 [[package]]
@@ -14140,6 +14149,15 @@ dependencies = [
  "zstd-sys",
 ]
 
+[[package]]
+name = "zstd-safe"
+version = "8.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ae42c0555055784c70058d19ba8e275528e8a99a706684868ace5da4e716a4ab"
+dependencies = [
+ "zstd-sys",
+]
+
 [[package]]
 name = "zstd-sys"
 version = "2.1.0+zstd.1.5.7"

Cargo.toml

@@ -199,10 +199,10 @@ serde_urlencoded = "0.7.1"
 # matching stable releases are not available yet, while previous stable lines
 # have incompatible APIs. Keep them exact-pinned and monitor upstream for stable
 # releases.
-aes-gcm = { version = "=0.11.1" }
-argon2 = { version = "=0.6.0" }
-blake2 = "=0.11.0"
-chacha20poly1305 = { version = "=0.11.0" }
+aes-gcm = { version = "0.11.1" }
+argon2 = { version = "0.6.0" }
+blake2 = "0.11.0"
+chacha20poly1305 = { version = "0.11.0" }
 crc-fast = "1.10.0"
 hmac = { version = "0.13.0" }
 jsonwebtoken = { version = "11.0.0" }
@@ -343,7 +343,7 @@ windows = { version = "0.62.2" }
 windows-sys = "0.61.2"
 xxhash-rust = { version = "0.8.18" }
 zip = "8.6.0"
-zstd = "0.13.3"
+zstd = "0.14.0"
 
 # Observability and Metrics
 metrics = "0.24.6"

crates/config/README.md

@@ -130,21 +130,6 @@ 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`

crates/config/src/constants/object.rs

@@ -137,28 +137,6 @@ 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.
 ///
@@ -834,16 +812,6 @@ 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");

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, TransitionClientTimeouts, TransitionCore};
+use rustfs_s3_client::transition_api::{BucketLookupType, Options, TransitionClient, TransitionCore};
 use rustfs_s3_client::{
     admin_handler_utils::AdminError,
     api_error_response::to_error_response,
@@ -320,27 +320,6 @@ pub(crate) fn endpoint_authority(url: &url::Url) -> Result<String, std::io::Erro
     }
 }
 
-fn transition_timeout_from_env(env_key: &str, default_secs: u64) -> 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
@@ -371,7 +350,6 @@ 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",
@@ -384,7 +362,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_with_timeouts(&endpoint, opts, params.provider_tag, timeouts).await?;
+    let client = TransitionClient::new(&endpoint, opts, params.provider_tag).await?;
 
     let client = Arc::new(client);
     let core = TransitionCore(Arc::clone(&client));

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, transition_client_timeouts_from_env,
+        build_transition_put_options, endpoint_authority,
     },
 };
 use http::HeaderMap;
@@ -139,7 +139,6 @@ 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",
@@ -148,7 +147,7 @@ impl WarmBackendS3 {
             ..Default::default()
         };
         let endpoint = endpoint_authority(&u)?;
-        let client = TransitionClient::new_with_timeouts(&endpoint, opts, tier_type, timeouts).await?;
+        let client = TransitionClient::new(&endpoint, opts, tier_type).await?;
 
         let client = Arc::new(client);
         let core = TransitionCore(Arc::clone(&client));

crates/s3-client/src/api_get_object.rs

@@ -120,10 +120,18 @@ impl TransitionClient {
 
         let h = resp.headers().clone();
 
+        let mut body = resp.into_body();
         let body_vec = if let Some(limit) = max_response_bytes {
-            self.collect_response_body(resp.into_body(), limit).await?
+            collect_response_body(body, limit).await?
         } else {
-            self.collect_response_body_unbounded(resp.into_body()).await?
+            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
         };
         Ok((object_stat, h, BufReader::new(Cursor::new(body_vec))))
     }
@@ -135,7 +143,7 @@ mod bounded_response_tests {
     use crate::{
         api_get_options::GetObjectOptions,
         credentials::{Credentials, SignatureType, Static, Value},
-        transition_api::{BucketLookupType, Options, TransitionClient, TransitionClientTimeouts, collect_response_body},
+        transition_api::{BucketLookupType, Options, TransitionClient, collect_response_body},
     };
     use http_body_util::Full;
     use hyper::body::Bytes;
@@ -167,31 +175,7 @@ mod bounded_response_tests {
         assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
     }
 
-    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<String>)> {
+    async fn bounded_get_fixture(body: &'static [u8]) -> Option<(TransitionClient, tokio::task::JoinHandle<String>)> {
         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,
@@ -225,14 +209,27 @@ mod bounded_response_tests {
             stream.write_all(body).await.expect("fixture should write response body");
             request
         });
-        let client = client_for_endpoint(&endpoint, timeouts).await;
+        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");
         Some((client, request))
     }
 
-    async fn bounded_get_fixture(body: &'static [u8]) -> Option<(TransitionClient, tokio::task::JoinHandle<String>)> {
-        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 {
@@ -295,7 +292,24 @@ mod bounded_response_tests {
             .local_addr()
             .expect("listener local address should be available")
             .to_string();
-        let client = client_for_endpoint(&endpoint, TransitionClientTimeouts::default()).await;
+        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 mut opts = GetObjectOptions::default();
         opts.headers
             .insert("range".to_string(), "bytes=0-18446744073709551615".to_string());
@@ -312,176 +326,6 @@ 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)]

crates/s3-client/src/api_list.rs

@@ -27,6 +27,7 @@ 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;
@@ -123,9 +124,14 @@ impl TransitionClient {
         }
 
         //let mut list_bucket_result = ListBucketV2Result::default();
-        let body_vec = self
-            .collect_response_body(resp.into_body(), MAX_S3_CLIENT_RESPONSE_SIZE)
-            .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 mut list_bucket_result = match quick_xml::de::from_str::<ListBucketV2Result>(&String::from_utf8_lossy(&body_vec)) {
             Ok(result) => result,
             Err(err) => {
@@ -208,9 +214,7 @@ impl TransitionClient {
 
         let resp_status = resp.status();
         let headers = resp.headers().clone();
-        let body = self
-            .collect_response_body(resp.into_body(), MAX_S3_CLIENT_RESPONSE_SIZE)
-            .await?;
+        let body = 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,
@@ -424,30 +428,6 @@ fn decode_s3_name(name: &str, encoding_type: &str) -> Result<String, std::io::Er
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{
-        credentials::{Credentials, SignatureType, Static, Value},
-        transition_api::{BucketLookupType, Options, TransitionClientTimeouts},
-    };
-    use std::time::Duration;
-    use tokio::{
-        io::{AsyncReadExt, AsyncWriteExt},
-        net::TcpListener,
-    };
-
-    fn timeout_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()
-        }
-    }
 
     #[test]
     fn list_versions_xml_preserves_versions_and_delete_markers() {
@@ -545,56 +525,4 @@ 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\n<ListBucketResult><Name>warm")
-                .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");
-    }
 }

crates/s3-client/src/api_put_object_multipart.rs

@@ -18,6 +18,7 @@
 #![allow(clippy::all)]
 
 use http::{HeaderMap, HeaderName, StatusCode};
+use http_body_util::BodyExt;
 use hyper::body::Bytes;
 use s3s::S3ErrorCode;
 use std::collections::HashMap;
@@ -246,9 +247,14 @@ 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 body_vec = self
-            .collect_response_body(resp.into_body(), rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE)
-            .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::other(e.to_string()))?;
+            if let Some(data) = frame.data_ref() {
+                body_vec.extend_from_slice(data);
+            }
+        }
         let initiate_multipart_upload_result =
             quick_xml::de::from_str::<InitiateMultipartUploadResult>(&String::from_utf8_lossy(&body_vec))
                 .map_err(|e| std::io::Error::other(format!("failed to parse CreateMultipartUpload response: {e}")))?;

crates/s3-client/src/api_remove.rs

@@ -19,6 +19,7 @@
 #![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;
@@ -350,9 +351,14 @@ impl TransitionClient {
                 )
                 .await?;
 
-            let body_vec = self
-                .collect_response_body(resp.into_body(), rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE)
-                .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);
+                }
+            }
             process_remove_multi_objects_response(
                 ReaderImpl::Body(Bytes::from(body_vec)),
                 bucket_name,

crates/s3-client/src/api_stat.rs

@@ -19,6 +19,7 @@
 #![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;
@@ -118,9 +119,14 @@ impl TransitionClient {
             let resp_status = resp.status();
             let h = resp.headers().clone();
 
-            let body_vec = self
-                .collect_response_body(resp.into_body(), rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE)
-                .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 resperr = http_resp_to_error_response(resp_status, &h, body_vec, bucket_name, "");
 
             warn!("bucket exists, resperr: {:?}", resperr);
@@ -164,13 +170,11 @@ impl TransitionClient {
                 let resp_status = resp.status();
                 let h = resp.headers().clone();
 
-                let body_vec = self
-                    .collect_response_body(resp.into_body(), rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE)
-                    .await?;
+                let body_vec = 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(err),
+            Err(err) => Err(std::io::Error::other(err)),
         }
     }
 
@@ -270,14 +274,8 @@ 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() {
@@ -340,63 +338,4 @@ 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");
-    }
 }

crates/s3-client/src/bucket_cache.rs

@@ -26,6 +26,7 @@ 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;
@@ -85,7 +86,7 @@ impl TransitionClient {
         let req = self.get_bucket_location_request(bucket_name)?;
 
         let mut resp = self.doit(req).await?;
-        location = process_bucket_location_response(self, resp, bucket_name, &self.tier_type).await?;
+        location = process_bucket_location_response(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);
@@ -197,7 +198,6 @@ impl TransitionClient {
 }
 
 async fn process_bucket_location_response(
-    client: &TransitionClient,
     mut resp: http::Response<Incoming>,
     bucket_name: &str,
     tier_type: &str,
@@ -237,9 +237,14 @@ async fn process_bucket_location_response(
     }
     //}
 
-    let body_vec = client
-        .collect_response_body(resp.into_body(), MAX_S3_CLIENT_RESPONSE_SIZE)
-        .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 mut location = "".to_string();
     if tier_type == "huaweicloud" {
         if let Ok(body_str) = String::from_utf8(body_vec) {

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;
+use http_body_util::{BodyExt, LengthLimitError, Limited};
 use hyper::body::Bytes;
 use hyper::body::Incoming;
 use hyper_rustls::{ConfigBuilderExt, HttpsConnector};
@@ -67,12 +67,10 @@ 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},
@@ -81,108 +79,28 @@ use time::Duration;
 use time::OffsetDateTime;
 use tokio::io::BufReader;
 use tokio::io::{AsyncRead, AsyncReadExt};
-use tracing::{debug, error, trace, warn};
+use tracing::{debug, error, 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::<std::io::Error>()
-            .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<B>(
-    mut body: Pin<&mut B>,
-    idle_timeout: Option<StdDuration>,
-) -> Result<Option<Bytes>, std::io::Error>
-where
-    B: Body<Data = Bytes>,
-    B::Error: Into<Box<dyn StdError + Send + Sync>>,
-{
-    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<B>(
-    body: B,
-    limit: Option<usize>,
-    idle_timeout: Option<StdDuration>,
-) -> Result<Vec<u8>, std::io::Error>
-where
-    B: Body<Data = Bytes>,
-    B::Error: Into<Box<dyn StdError + Send + Sync>>,
-{
-    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<B>(body: B, limit: usize) -> Result<Vec<u8>, std::io::Error>
 where
     B: Body<Data = Bytes>,
-    B::Error: Into<Box<dyn StdError + Send + Sync>>,
+    B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
 {
-    collect_response_body_inner(body, Some(limit), None).await
+    let body = Limited::new(body, limit).collect().await.map_err(|err| {
+        if err.is::<LengthLimitError>() {
+            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())
 }
 
 const C_UNKNOWN: i32 = -1;
@@ -278,62 +196,6 @@ 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<Self, std::io::Error> {
-        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)]
@@ -426,28 +288,12 @@ async fn build_tls_config() -> Result<rustls::ClientConfig, std::io::Error> {
 
 impl TransitionClient {
     pub async fn new(endpoint: &str, opts: Options, tier_type: &str) -> Result<TransitionClient, std::io::Error> {
-        Self::private_new(endpoint, opts, tier_type, TransitionClientTimeouts::default()).await
+        let client = Self::private_new(endpoint, opts, tier_type).await?;
+
+        Ok(client)
     }
 
-    /// 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<TransitionClient, std::io::Error> {
-        Self::private_new(endpoint, opts, tier_type, timeouts).await
-    }
-
-    async fn private_new(
-        endpoint: &str,
-        opts: Options,
-        tier_type: &str,
-        timeouts: TransitionClientTimeouts,
-    ) -> Result<TransitionClient, std::io::Error> {
+    async fn private_new(endpoint: &str, opts: Options, tier_type: &str) -> Result<TransitionClient, std::io::Error> {
         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
@@ -460,19 +306,15 @@ 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()
-            .wrap_connector(http);
+            .build();
         let http_client = Client::builder(TokioExecutor::new()).build(https);
 
         let mut client = TransitionClient {
@@ -495,7 +337,6 @@ impl TransitionClient {
             trailing_header_support: opts.trailing_headers,
             max_retries: opts.max_retries,
             tier_type: tier_type.to_string(),
-            timeouts,
         };
 
         {
@@ -660,43 +501,29 @@ impl TransitionClient {
     }
 
     pub async fn doit(&self, req: Request<s3s::Body>) -> Result<Response<Incoming>, std::io::Error> {
+        let req_method;
+        let req_uri;
+        let resp;
         let http_client = self.http_client.clone();
-        let req_method = req.method().clone();
-        let resp = tokio::time::timeout(self.timeouts.request_timeout, http_client.request(req)).await;
+        {
+            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 resp = match resp {
-            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"));
-            }
+            Ok(r) => r,
+            Err(_) => return Err(std::io::Error::other("Unexpected error in response")),
         };
-        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"
-        );
+        debug!(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());
@@ -710,15 +537,7 @@ impl TransitionClient {
                 .and_then(|value| value.to_str().ok())
                 .unwrap_or_default()
                 .to_string();
-            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"
-            );
+            warn!(status = %status, request_id, "remote tier request rejected");
         }
         Ok(resp)
     }
@@ -762,9 +581,7 @@ impl TransitionClient {
             let resp_status = resp.status();
             let h = resp.headers().clone();
 
-            let body_vec = self
-                .collect_response_body(resp.into_body(), MAX_S3_ERROR_RESPONSE_SIZE)
-                .await?;
+            let body_vec = 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;
@@ -818,22 +635,6 @@ impl TransitionClient {
         Err(std::io::Error::other("remote tier request did not produce a response"))
     }
 
-    pub async fn collect_response_body<B>(&self, body: B, limit: usize) -> Result<Vec<u8>, std::io::Error>
-    where
-        B: Body<Data = Bytes>,
-        B::Error: Into<Box<dyn StdError + Send + Sync>>,
-    {
-        collect_response_body_inner(body, Some(limit), Some(self.timeouts.response_body_idle_timeout)).await
-    }
-
-    pub async fn collect_response_body_unbounded<B>(&self, body: B) -> Result<Vec<u8>, std::io::Error>
-    where
-        B: Body<Data = Bytes>,
-        B::Error: Into<Box<dyn StdError + Send + Sync>>,
-    {
-        collect_response_body_inner(body, None, Some(self.timeouts.response_body_idle_timeout)).await
-    }
-
     async fn new_request(
         &self,
         method: &http::Method,
@@ -1703,17 +1504,12 @@ pub struct CreateBucketConfiguration {
 mod tests {
     use super::{
         MAX_S3_CLIENT_RESPONSE_SIZE, MAX_S3_ERROR_RESPONSE_SIZE, SignatureType, build_tls_config, collect_response_body,
-        collect_response_body_inner, signer_error_to_io_error, to_object_info_for_provider, validate_header_values,
-        with_rustls_init_guard,
+        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 futures::stream;
-    use http::{HeaderMap, HeaderValue, Request};
-    use http_body::Frame;
-    use http_body_util::{Full, StreamBody};
+    use http::{HeaderMap, HeaderValue};
+    use http_body_util::Full;
     use hyper::body::Bytes;
-    use std::time::Duration as StdDuration;
-    use tokio::net::TcpListener;
     use uuid::Uuid;
 
     #[tokio::test]
@@ -1744,77 +1540,6 @@ 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") })
@@ -1848,18 +1573,6 @@ 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();

docs/operations/replication-outbound-transport.md

@@ -29,18 +29,6 @@
 
 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.