feat(tier): fence legacy state reconciliation

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/api/mod.rs

@@ -479,9 +479,11 @@ pub mod notification {
     #[cfg(any(test, feature = "test-util"))]
     pub use crate::services::notification_sys::rotate_cross_pool_fence_fleet_proof_for_test;
     pub use crate::services::notification_sys::{
-        ClusterTierDailyStats, CrossPoolFenceFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant,
-        acquire_cross_pool_fence_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
-        new_global_notification_sys, scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
+        ClusterTierDailyStats, CrossPoolFenceFleetProofToken, LegacyTransitionStateReconcileFleetProofToken, NotificationPeerErr,
+        NotificationSys, ScannerPublicationLeaseGrant, acquire_cross_pool_fence_fleet_proof,
+        acquire_legacy_transition_state_reconcile_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
+        legacy_transition_state_reconcile_fleet_proof_matches, new_global_notification_sys,
+        scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
     };
 }
 

crates/ecstore/src/bucket/on_demand_migration/source_client.rs

@@ -14,9 +14,9 @@
 
 //! Outbound client for an on-demand migration source bucket.
 //!
-//! `SourceClient` maps local keys onto a read-only `SourceBackend`. The
-//! S3 backend uses the shared remote builder and exposes the surface the
-//! migration path needs (HEAD, ranged streaming GET, ListObjectsV2, GetObjectTagging, a
+//! `SourceClient` wraps an `aws_sdk_s3::Client` built through the shared
+//! remote builder and exposes the read-only surface the migration path
+//! needs (HEAD, ranged streaming GET, ListObjectsV2, GetObjectTagging, a
 //! probe for admin validation). Every request carries the
 //! `source-proxy-request` anti-loop marker in both the `x-rustfs-` and
 //! `x-minio-` prefixes so a RustFS/MinIO source answers locally instead of
@@ -511,7 +511,7 @@ pub struct SourceObject {
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct SourcePage {
     pub objects: Vec<SourceObject>,
-    /// Rolled-up prefixes, in the same namespace as `objects`; always empty when the
+    /// Rolled-up prefixes, in the local namespace; always empty when the
     /// request carried no delimiter.
     pub common_prefixes: Vec<String>,
     pub is_truncated: bool,
@@ -519,8 +519,7 @@ pub struct SourcePage {
 }
 
 /// One `ListObjectsV2` page request against the source. Keys are given in the
-/// local namespace at `SourceClient`, and in the source namespace at
-/// `SourceBackend`; `SourceClient` maps them through `source_prefix`.
+/// local namespace; `SourceClient` maps them through `source_prefix`.
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct SourceListRequest<'a> {
     pub prefix: Option<&'a str>,
@@ -576,30 +575,8 @@ impl Intercept for SourceProxyMarkerInterceptor {
     }
 }
 
-/// Read-only provider operations in the source bucket namespace.
-///
-/// Implementations must preserve streaming, honor the requested range and
-/// pagination cursor, and classify failures without including credentials.
-/// `SourceClient` owns prefix mapping so every provider shares the same local
-/// namespace. Continuation tokens are opaque and must never be prefix-mapped.
-#[async_trait::async_trait]
-pub trait SourceBackend: Send + Sync {
-    async fn head(&self, key: &str) -> Result<SourceHead, SourceError>;
-    async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError>;
-    async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError>;
-    async fn tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError>;
-    /// Verify bucket access; `SourceClient` separately probes a filtered listing.
-    async fn probe(&self) -> Result<(), SourceError>;
-}
-
-/// S3-compatible implementation, including request signing and anti-loop headers.
-pub struct S3SourceBackend {
-    client: S3Client,
-    bucket: String,
-}
-
 pub struct SourceClient {
-    backend: Box<dyn SourceBackend>,
+    client: S3Client,
     endpoint: String,
     bucket: String,
     source_prefix: Option<String>,
@@ -632,10 +609,7 @@ impl SourceClient {
     fn from_config_builder(config: aws_sdk_s3::config::Builder, endpoint: String, spec: &SourceClientSpec) -> Self {
         let client = S3Client::from_conf(config.interceptor(SourceProxyMarkerInterceptor::new()).build());
         Self {
-            backend: Box::new(S3SourceBackend {
-                client,
-                bucket: spec.bucket.clone(),
-            }),
+            client,
             endpoint,
             bucket: spec.bucket.clone(),
             source_prefix: spec.source_prefix.clone().filter(|prefix| !prefix.is_empty()),
@@ -678,15 +652,35 @@ impl SourceClient {
     }
 
     pub async fn head_object(&self, key: &str) -> Result<SourceHead, SourceError> {
-        self.backend.head(&self.source_key(key)).await
+        let output = self
+            .client
+            .head_object()
+            .bucket(&self.bucket)
+            .key(self.source_key(key))
+            .send()
+            .await
+            .map_err(classify_sdk_error)?;
+        source_head_from_head_output(output)
     }
 
-    /// Streams the object, preserving an optional HTTP byte range.
+    /// Streams the object; `range` is passed through as an HTTP `Range`
+    /// header and omitted entirely when `None`.
     pub async fn get_object(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
-        self.backend.get(&self.source_key(key), range).await
+        let range = range.map(range_header_value).transpose()?;
+        let output = self
+            .client
+            .get_object()
+            .bucket(&self.bucket)
+            .key(self.source_key(key))
+            .set_range(range)
+            .send()
+            .await
+            .map_err(classify_sdk_error)?;
+        source_get_from_output(output)
     }
 
-    /// Lists one page under the local prefix.
+    /// Lists one page under the local `prefix`. Keys are returned in the
+    /// local namespace; entries outside `source_prefix` are skipped.
     pub async fn list_objects_v2(
         &self,
         prefix: Option<&str>,
@@ -702,81 +696,9 @@ impl SourceClient {
         .await
     }
 
-    /// Maps keys and common prefixes while leaving opaque cursors untouched.
+    /// [`Self::list_objects_v2`] with the delimiter and start-after the
+    /// list-through merge needs (rustfs/backlog#2164).
     pub async fn list_page(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
-        let prefix = self.source_key(request.prefix.unwrap_or_default());
-        let start_after = request.start_after.map(|key| self.source_key(key));
-        let mut page = self
-            .backend
-            .list(&SourceListRequest {
-                prefix: Some(&prefix),
-                start_after: start_after.as_deref(),
-                ..*request
-            })
-            .await?;
-        page.objects = page
-            .objects
-            .into_iter()
-            .filter_map(|object| self.local_object(object))
-            .collect();
-        page.common_prefixes = page
-            .common_prefixes
-            .into_iter()
-            .filter_map(|prefix| self.local_key(&prefix).map(str::to_string))
-            .collect();
-        Ok(page)
-    }
-
-    fn local_object(&self, mut object: SourceObject) -> Option<SourceObject> {
-        object.key = self.local_key(&object.key)?.to_string();
-        Some(object)
-    }
-
-    pub async fn get_object_tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
-        self.backend.tagging(&self.source_key(key)).await
-    }
-
-    pub async fn probe(&self) -> Result<SourceProbe, SourceError> {
-        self.backend.probe().await?;
-        let page = self.list_objects_v2(None, None, 1).await?;
-        Ok(SourceProbe {
-            sample_object: page.objects.into_iter().next(),
-            has_more_objects: page.is_truncated,
-        })
-    }
-}
-
-#[async_trait::async_trait]
-impl SourceBackend for S3SourceBackend {
-    async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
-        let output = self
-            .client
-            .head_object()
-            .bucket(&self.bucket)
-            .key(key)
-            .send()
-            .await
-            .map_err(classify_sdk_error)?;
-        source_head_from_head_output(output)
-    }
-
-    /// Streams the object; `range` is passed through as an HTTP `Range`
-    /// header and omitted entirely when `None`.
-    async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
-        let range = range.map(range_header_value).transpose()?;
-        let output = self
-            .client
-            .get_object()
-            .bucket(&self.bucket)
-            .key(key)
-            .set_range(range)
-            .send()
-            .await
-            .map_err(classify_sdk_error)?;
-        source_get_from_output(output)
-    }
-
-    async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
         // `start_after` is silently ignored by S3 once a continuation token is
         // present; refuse the ambiguous pair rather than list from the wrong
         // position.
@@ -789,9 +711,9 @@ impl SourceBackend for S3SourceBackend {
             .client
             .list_objects_v2()
             .bucket(&self.bucket)
-            .prefix(request.prefix.unwrap_or_default())
+            .prefix(self.source_key(request.prefix.unwrap_or_default()))
             .set_delimiter(request.delimiter.map(str::to_string))
-            .set_start_after(request.start_after.map(str::to_string))
+            .set_start_after(request.start_after.map(|after| self.source_key(after)))
             .set_continuation_token(request.continuation_token.map(str::to_string))
             .max_keys(request.max_keys)
             .send()
@@ -809,13 +731,13 @@ impl SourceBackend for S3SourceBackend {
             .contents
             .unwrap_or_default()
             .into_iter()
-            .filter_map(s3_source_object)
+            .filter_map(|object| self.source_object(object))
             .collect();
         let common_prefixes = output
             .common_prefixes
             .unwrap_or_default()
             .into_iter()
-            .filter_map(|prefix| prefix.prefix)
+            .filter_map(|prefix| Some(self.local_key(prefix.prefix.as_deref()?)?.to_string()))
             .collect();
 
         Ok(SourcePage {
@@ -826,43 +748,48 @@ impl SourceBackend for S3SourceBackend {
         })
     }
 
-    async fn tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
+    fn source_object(&self, object: SdkObject) -> Option<SourceObject> {
+        let key = self.local_key(object.key.as_deref()?)?.to_string();
+        let etag = normalize_etag(object.e_tag);
+        let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag);
+        Some(SourceObject {
+            key,
+            etag,
+            size: object.size.and_then(|size| u64::try_from(size).ok()).unwrap_or(0),
+            last_modified: system_time(object.last_modified),
+            storage_class: object.storage_class.map(|class| class.as_str().to_string()),
+            is_multipart_etag,
+        })
+    }
+
+    pub async fn get_object_tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
         let output = self
             .client
             .get_object_tagging()
             .bucket(&self.bucket)
-            .key(key)
+            .key(self.source_key(key))
             .send()
             .await
             .map_err(classify_sdk_error)?;
         Ok(output.tag_set.into_iter().map(|tag| (tag.key, tag.value)).collect())
     }
 
-    async fn probe(&self) -> Result<(), SourceError> {
+    /// Admin validation: HeadBucket plus a one-key listing under the prefix.
+    pub async fn probe(&self) -> Result<SourceProbe, SourceError> {
         self.client
             .head_bucket()
             .bucket(&self.bucket)
             .send()
             .await
             .map_err(classify_sdk_error)?;
-        Ok(())
+        let page = self.list_objects_v2(None, None, 1).await?;
+        Ok(SourceProbe {
+            sample_object: page.objects.into_iter().next(),
+            has_more_objects: page.is_truncated,
+        })
     }
 }
 
-fn s3_source_object(object: SdkObject) -> Option<SourceObject> {
-    let key = object.key?;
-    let etag = normalize_etag(object.e_tag);
-    let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag);
-    Some(SourceObject {
-        key,
-        etag,
-        size: object.size.and_then(|size| u64::try_from(size).ok()).unwrap_or(0),
-        last_modified: system_time(object.last_modified),
-        storage_class: object.storage_class.map(|class| class.as_str().to_string()),
-        is_multipart_etag,
-    })
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1267,68 +1194,6 @@ mod tests {
         assert!(requests[1].uri.contains("continuation-token=token-1"), "{}", requests[1].uri);
     }
 
-    #[tokio::test]
-    async fn list_page_maps_delimiter_prefixes_and_start_after_but_not_cursors() {
-        let body = r#"<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
-<IsTruncated>true</IsTruncated><NextContinuationToken>data/opaque</NextContinuationToken>
-<CommonPrefixes><Prefix>data/photos/</Prefix></CommonPrefixes>
-<CommonPrefixes><Prefix>outside/</Prefix></CommonPrefixes>
-</ListBucketResult>"#;
-        let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), body)]).await;
-        let first = client
-            .list_page(&SourceListRequest {
-                prefix: Some("photos/"),
-                delimiter: Some("/"),
-                start_after: Some("photos/a"),
-                max_keys: 2,
-                ..Default::default()
-            })
-            .await
-            .expect("delimiter listing should succeed");
-        assert_eq!(first.common_prefixes, vec!["photos/"]);
-        assert_eq!(first.next_continuation_token.as_deref(), Some("data/opaque"));
-        let second = client
-            .list_page(&SourceListRequest {
-                continuation_token: first.next_continuation_token.as_deref(),
-                max_keys: 2,
-                ..Default::default()
-            })
-            .await
-            .expect("opaque continuation should succeed");
-        assert_eq!(second.common_prefixes, first.common_prefixes);
-        let requests = recorded(&requests);
-        let query = |request: &RecordedRequest| {
-            Url::parse(&request.uri)
-                .expect("request URI")
-                .query_pairs()
-                .into_owned()
-                .collect::<HashMap<_, _>>()
-        };
-        let first_query = query(&requests[0]);
-        assert_eq!(first_query.get("prefix").map(String::as_str), Some("data/photos/"));
-        assert_eq!(first_query.get("start-after").map(String::as_str), Some("data/photos/a"));
-        assert_eq!(first_query.get("delimiter").map(String::as_str), Some("/"));
-        let second_query = query(&requests[1]);
-        assert_eq!(second_query.get("continuation-token").map(String::as_str), Some("data/opaque"));
-        assert!(!second_query.contains_key("start-after"));
-    }
-
-    #[tokio::test]
-    async fn list_page_rejects_ambiguous_cursor_before_sending() {
-        let (client, requests) = scripted_client(&spec(Some("data/")), vec![]).await;
-        let err = client
-            .list_page(&SourceListRequest {
-                start_after: Some("a"),
-                continuation_token: Some("opaque"),
-                max_keys: 1,
-                ..Default::default()
-            })
-            .await
-            .expect_err("ambiguous list position must fail");
-        assert!(matches!(err, SourceError::Other(_)));
-        assert!(recorded(&requests).is_empty(), "invalid request must never reach the source");
-    }
-
     #[tokio::test]
     async fn list_objects_v2_rejects_truncated_page_without_token() {
         let (client, _) = scripted_client(&spec(None), vec![ok(Vec::new(), LIST_TRUNCATED_WITHOUT_TOKEN)]).await;
@@ -1492,14 +1357,11 @@ mod tests {
 
     fn prefix_client(prefix: Option<String>) -> SourceClient {
         SourceClient {
-            backend: Box::new(S3SourceBackend {
-                client: S3Client::from_conf(
-                    aws_sdk_s3::Config::builder()
-                        .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
-                        .build(),
-                ),
-                bucket: "bucket".to_string(),
-            }),
+            client: S3Client::from_conf(
+                aws_sdk_s3::Config::builder()
+                    .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+                    .build(),
+            ),
             endpoint: "https://source.example.com".to_string(),
             bucket: "bucket".to_string(),
             source_prefix: prefix.filter(|prefix| !prefix.is_empty()),

crates/ecstore/src/services/notification_sys.rs

@@ -62,12 +62,27 @@ const REMOTE_VERSION_STATE_PROOF_TTL: Duration = Duration::from_secs(30);
 const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 2;
 const TIER_DELETE_JOURNAL_POLICY_SUPPORTED_VERSION: u32 = 3;
 const DECOMMISSION_TARGET_FENCE_POLICY_SUPPORTED_VERSION: u32 = 4;
+// Keep this synchronized with the version served by node_service. Including
+// the local member in the minimum prevents an older coordinator from
+// self-authorizing a policy implemented only by newer remote peers.
+const LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION: u32 = 4;
+/// Version 5 is reserved for a fleet whose every metadata writer preserves
+/// explicit transition version state and destination identity, and implements
+/// conditional per-generation `xl.meta` writes with strong readback. The node
+/// service must not advertise this version until the conditional writer from
+/// rustfs/backlog#684 is available.
+const LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION: u32 = 5;
 type CrossPoolFencePolicyResult = Result<BTreeMap<String, Uuid>>;
 
 fn cross_pool_fence_policy_results(
     peer_epochs: BTreeMap<String, Uuid>,
     minimum_version: u32,
-) -> (CrossPoolFencePolicyResult, CrossPoolFencePolicyResult, CrossPoolFencePolicyResult) {
+) -> (
+    CrossPoolFencePolicyResult,
+    CrossPoolFencePolicyResult,
+    CrossPoolFencePolicyResult,
+    CrossPoolFencePolicyResult,
+) {
     let journal_result = if minimum_version >= TIER_DELETE_JOURNAL_POLICY_SUPPORTED_VERSION {
         Ok(peer_epochs.clone())
     } else {
@@ -78,7 +93,18 @@ fn cross_pool_fence_policy_results(
     } else {
         Err(Error::other("decommission target fence policy capability version is unsupported"))
     };
-    (Ok(peer_epochs), journal_result, decommission_target_fence_result)
+    let legacy_transition_state_reconcile_result =
+        if minimum_version >= LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION {
+            Ok(peer_epochs.clone())
+        } else {
+            Err(Error::other("legacy transition state reconcile policy capability version is unsupported"))
+        };
+    (
+        Ok(peer_epochs),
+        journal_result,
+        decommission_target_fence_result,
+        legacy_transition_state_reconcile_result,
+    )
 }
 
 #[derive(Clone, Debug)]
@@ -252,10 +278,21 @@ pub(crate) struct TierDeleteJournalFleetProofToken {
     _permit: FleetCapabilityProofPermit,
 }
 
+/// Effect-window authority for one legacy transition-state reconciliation.
+///
+/// The token intentionally cannot be cloned. Its permit keeps the admitted
+/// fleet generation alive until the caller finishes the final strong
+/// readback, while revocation makes every later validation fail immediately.
+pub struct LegacyTransitionStateReconcileFleetProofToken {
+    token: FleetCapabilityProofToken,
+    _permit: FleetCapabilityProofPermit,
+}
+
 static REMOTE_VERSION_STATE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static CROSS_POOL_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static TIER_DELETE_JOURNAL_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static DECOMMISSION_TARGET_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
+static LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock<String> = OnceLock::new();
 
 fn cross_pool_fence_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
@@ -274,6 +311,10 @@ fn decommission_target_fence_fleet_proof_slot() -> &'static std::sync::RwLock<Fl
     DECOMMISSION_TARGET_FENCE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
 }
 
+fn legacy_transition_state_reconcile_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
+    LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
+}
+
 fn revoke_fleet_capability_proof_state(state: &mut FleetCapabilityProofState) {
     if let Some(proof) = state.proof.take() {
         proof.generation.revoke();
@@ -444,6 +485,125 @@ pub(crate) fn tier_delete_journal_topology_generation(proof: &TierDeleteJournalF
     stable_tier_delete_journal_topology_generation(&proof.token.topology_fingerprint)
 }
 
+/// Acquire one non-cloneable authority that must span the complete reconcile
+/// effect window, including its final strong readback.
+pub async fn acquire_legacy_transition_state_reconcile_fleet_proof() -> Option<LegacyTransitionStateReconcileFleetProofToken> {
+    let expected_topology = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get()?;
+    let proof = {
+        let state = legacy_transition_state_reconcile_fleet_proof_slot()
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, expected_topology, Instant::now())?
+    };
+    let observed_peer_epochs = observe_legacy_transition_state_reconcile_fleet(expected_topology).await?;
+    let state = legacy_transition_state_reconcile_fleet_proof_slot()
+        .read()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
+        &state,
+        &proof,
+        expected_topology,
+        &observed_peer_epochs,
+        Instant::now(),
+    )
+    .then_some(proof)
+}
+
+fn acquire_legacy_transition_state_reconcile_fleet_proof_from(
+    state: &FleetCapabilityProofState,
+    expected_topology: &str,
+    now: Instant,
+) -> Option<LegacyTransitionStateReconcileFleetProofToken> {
+    let token = acquire_fleet_capability_proof_from(state, expected_topology, now)?;
+    let permit = state.proof.as_ref()?.generation.try_acquire()?;
+    Some(LegacyTransitionStateReconcileFleetProofToken { token, _permit: permit })
+}
+
+async fn observe_legacy_transition_state_reconcile_fleet(expected_topology: &str) -> Option<BTreeMap<String, Uuid>> {
+    let notification_sys = get_global_notification_sys()?;
+    let (peer_epochs, minimum_version) = timeout(
+        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
+        notification_sys.probe_cross_pool_fence_fleet(expected_topology),
+    )
+    .await
+    .ok()?
+    .ok()?;
+    let (_, _, _, reconcile_result) = cross_pool_fence_policy_results(peer_epochs, minimum_version);
+    reconcile_result.ok()
+}
+
+/// Revalidate the exact fleet generation captured by a reconcile token with a
+/// fresh synchronous observation. Callers must await this before each
+/// conditional metadata write and after the final strong readback.
+pub async fn legacy_transition_state_reconcile_fleet_proof_matches(
+    proof: &LegacyTransitionStateReconcileFleetProofToken,
+) -> bool {
+    let Some(expected_topology) = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get() else {
+        return false;
+    };
+    legacy_transition_state_reconcile_fleet_proof_matches_with_observer(
+        legacy_transition_state_reconcile_fleet_proof_slot(),
+        proof,
+        expected_topology,
+        || observe_legacy_transition_state_reconcile_fleet(expected_topology),
+    )
+    .await
+}
+
+async fn legacy_transition_state_reconcile_fleet_proof_matches_with_observer<F, Fut>(
+    slot: &std::sync::RwLock<FleetCapabilityProofState>,
+    proof: &LegacyTransitionStateReconcileFleetProofToken,
+    expected_topology: &str,
+    observe: F,
+) -> bool
+where
+    F: FnOnce() -> Fut,
+    Fut: Future<Output = Option<BTreeMap<String, Uuid>>>,
+{
+    {
+        let state = slot.read().unwrap_or_else(std::sync::PoisonError::into_inner);
+        if !legacy_transition_state_reconcile_fleet_proof_matches_at(&state, proof, expected_topology, Instant::now()) {
+            return false;
+        }
+    }
+    let Some(observed_peer_epochs) = observe().await else {
+        return false;
+    };
+    let state = slot.read().unwrap_or_else(std::sync::PoisonError::into_inner);
+    legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
+        &state,
+        proof,
+        expected_topology,
+        &observed_peer_epochs,
+        Instant::now(),
+    )
+}
+
+fn legacy_transition_state_reconcile_fleet_proof_matches_at(
+    state: &FleetCapabilityProofState,
+    proof: &LegacyTransitionStateReconcileFleetProofToken,
+    expected_topology: &str,
+    now: Instant,
+) -> bool {
+    proof._permit.generation.is_accepting()
+        && fleet_capability_proof_matches_at(state, &proof.token, expected_topology, now)
+        && state
+            .proof
+            .as_ref()
+            .is_some_and(|current| Arc::ptr_eq(&current.generation, &proof._permit.generation))
+}
+
+fn legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
+    state: &FleetCapabilityProofState,
+    proof: &LegacyTransitionStateReconcileFleetProofToken,
+    expected_topology: &str,
+    observed_peer_epochs: &BTreeMap<String, Uuid>,
+    now: Instant,
+) -> bool {
+    legacy_transition_state_reconcile_fleet_proof_matches_at(state, proof, expected_topology, now)
+        && proof.token.peer_epochs.as_ref() == observed_peer_epochs
+}
+
 #[cfg(all(test, feature = "test-util"))]
 pub(crate) fn tier_delete_journal_fleet_proof_has_inflight_for_test() -> bool {
     let state = tier_delete_journal_fleet_proof_slot()
@@ -766,6 +926,7 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 cross_pool_fence_fleet_proof_slot(),
                 tier_delete_journal_fleet_proof_slot(),
                 decommission_target_fence_fleet_proof_slot(),
+                legacy_transition_state_reconcile_fleet_proof_slot(),
             ] {
                 mark_fleet_capability_topology_conflict(slot);
             }
@@ -798,11 +959,12 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 .unwrap_or_else(|_| Err(Error::other("cross-pool fence fleet capability probe timed out"))),
                 None => Err(Error::other("cross-pool fence fleet capability notification system is unavailable")),
             };
-            let (fence_result, journal_result, decommission_target_fence_result) = match fence_probe {
+            let (fence_result, journal_result, decommission_target_fence_result, reconcile_result) = match fence_probe {
                 Ok((peer_epochs, minimum_version)) => cross_pool_fence_policy_results(peer_epochs, minimum_version),
                 Err(err) => {
                     let message = err.to_string();
                     (
+                        Err(Error::other(message.clone())),
                         Err(Error::other(message.clone())),
                         Err(Error::other(message.clone())),
                         Err(Error::other(message)),
@@ -818,6 +980,7 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 revoke_fleet_capability_proof(cross_pool_fence_fleet_proof_slot());
                 revoke_fleet_capability_proof(tier_delete_journal_fleet_proof_slot());
                 revoke_fleet_capability_proof(decommission_target_fence_fleet_proof_slot());
+                revoke_fleet_capability_proof(legacy_transition_state_reconcile_fleet_proof_slot());
             } else if let Some(err) = publish_fleet_capability_probe_result(
                 remote_version_state_fleet_proof_slot(),
                 &topology_fingerprint,
@@ -880,6 +1043,24 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                     "notification capability probe"
                 );
             }
+            if !topology_conflict
+                && let Some(err) = publish_fleet_capability_probe_result(
+                    legacy_transition_state_reconcile_fleet_proof_slot(),
+                    &topology_fingerprint,
+                    reconcile_result,
+                    Instant::now(),
+                )
+            {
+                debug!(
+                    event = EVENT_NOTIFICATION_CAPABILITY_PROBE,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_NOTIFICATION,
+                    capability = "legacy_transition_state_reconcile_v1",
+                    state = "failed_closed",
+                    error = %err,
+                    "notification capability probe"
+                );
+            }
             sleep(REMOTE_VERSION_STATE_PROBE_INTERVAL).await;
         }
     });
@@ -959,7 +1140,7 @@ impl NotificationSys {
             client.probe_cross_pool_fence(topology_fingerprint.to_string()).await
         });
         let mut peer_epochs = BTreeMap::new();
-        let mut minimum_version = u32::MAX;
+        let mut minimum_version = LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION;
         for result in join_all(probes).await {
             let (peer, version, epoch) = result?;
             if version < CROSS_POOL_FENCE_SUPPORTED_VERSION {
@@ -968,11 +1149,6 @@ impl NotificationSys {
             minimum_version = minimum_version.min(version);
             insert_remote_version_state_peer(&mut peer_epochs, peer, epoch)?;
         }
-        // A single-node deployment has no remote member to lower the local
-        // policy version advertised by this binary.
-        if minimum_version == u32::MAX {
-            minimum_version = DECOMMISSION_TARGET_FENCE_POLICY_SUPPORTED_VERSION;
-        }
         Ok((peer_epochs, minimum_version))
     }
 }
@@ -3190,20 +3366,36 @@ mod tests {
     #[test]
     fn cross_pool_policy_versions_authorize_only_their_supported_protocols() {
         let peers = BTreeMap::from([("node-b:9000".to_string(), Uuid::new_v4())]);
-        let (generic_v2, journal_v2, decommission_v2) = cross_pool_fence_policy_results(peers.clone(), 2);
+        let (generic_v2, journal_v2, decommission_v2, reconcile_v2) = cross_pool_fence_policy_results(peers.clone(), 2);
         assert!(generic_v2.is_ok(), "v2 remains valid for existing cross-pool fencing");
         assert!(journal_v2.is_err(), "a mixed v2/v3 fleet must fail closed for journal-v6 deletion");
         assert!(decommission_v2.is_err(), "v2 cannot authorize the sticky per-target decommission fence");
+        assert!(reconcile_v2.is_err(), "v2 cannot authorize legacy transition-state reconciliation");
 
-        let (generic_v3, journal_v3, decommission_v3) = cross_pool_fence_policy_results(peers.clone(), 3);
+        let (generic_v3, journal_v3, decommission_v3, reconcile_v3) = cross_pool_fence_policy_results(peers.clone(), 3);
         assert!(generic_v3.is_ok());
         assert!(journal_v3.is_ok(), "an all-v3 fleet may authorize journal-v6 deletion");
         assert!(decommission_v3.is_err(), "v3 members do not understand the per-target decommission fence");
+        assert!(reconcile_v3.is_err());
 
-        let (generic_v4, journal_v4, decommission_v4) = cross_pool_fence_policy_results(peers, 4);
+        let (generic_v4, journal_v4, decommission_v4, reconcile_v4) =
+            cross_pool_fence_policy_results(peers.clone(), LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION);
         assert!(generic_v4.is_ok());
         assert!(journal_v4.is_ok());
         assert!(decommission_v4.is_ok(), "an all-v4 fleet may create sticky per-target reservations");
+        assert!(
+            reconcile_v4.is_err(),
+            "the current local policy lacks the conditional xl.meta writer required by reconcile"
+        );
+
+        let (generic_v5, journal_v5, decommission_v5, reconcile_v5) = cross_pool_fence_policy_results(peers, 5);
+        assert!(generic_v5.is_ok());
+        assert!(journal_v5.is_ok());
+        assert!(decommission_v5.is_ok());
+        assert!(
+            reconcile_v5.is_ok(),
+            "only an all-v5 fleet preserves destination identity and conditional reconcile writes"
+        );
     }
 
     #[test]
@@ -3458,6 +3650,234 @@ mod tests {
         );
     }
 
+    #[test]
+    fn legacy_transition_state_reconcile_admits_only_compatible_single_and_multi_node_fleets() {
+        let now = Instant::now();
+        for peers in [
+            BTreeMap::new(),
+            BTreeMap::from([("peer-a".to_string(), Uuid::new_v4()), ("peer-b".to_string(), Uuid::new_v4())]),
+        ] {
+            let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+            let (_, _, _, result) =
+                cross_pool_fence_policy_results(peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+            assert!(publish_fleet_capability_probe_result(&slot, "topology-a", result, now).is_none());
+
+            let admitted = {
+                let state = slot.read().expect("reconcile proof slot should not poison");
+                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                    .expect("an all-compatible fleet should admit reconciliation")
+            };
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            assert!(legacy_transition_state_reconcile_fleet_proof_matches_at(
+                &state,
+                &admitted,
+                "topology-a",
+                now,
+            ));
+        }
+    }
+
+    #[test]
+    fn legacy_transition_state_reconcile_restart_drains_concurrent_effect_windows() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let original_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        let (_, _, _, original_result) =
+            cross_pool_fence_policy_results(original_peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", original_result, now).is_none());
+
+        let (first, second) = {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            (
+                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                    .expect("the first reconcile writer should be admitted"),
+                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                    .expect("the second reconcile writer should be admitted"),
+            )
+        };
+
+        let restarted_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        let (_, _, _, restarted_result) =
+            cross_pool_fence_policy_results(restarted_peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        let blocked =
+            publish_fleet_capability_probe_result(&slot, "topology-a", restarted_result, now + Duration::from_millis(1))
+                .expect("a restarted member must revoke the old generation and wait for both writers");
+        assert!(blocked.to_string().contains("previous generation to drain"));
+        {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            assert!(state.proof.is_none());
+            assert!(state.draining_generation.is_some());
+            assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
+                &state,
+                &first,
+                "topology-a",
+                now + Duration::from_millis(1),
+            ));
+            assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
+                &state,
+                &second,
+                "topology-a",
+                now + Duration::from_millis(1),
+            ));
+        }
+
+        drop(first);
+        let (_, _, _, still_blocked_result) =
+            cross_pool_fence_policy_results(restarted_peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        assert!(
+            publish_fleet_capability_probe_result(&slot, "topology-a", still_blocked_result, now + Duration::from_millis(2),)
+                .is_some(),
+            "one remaining writer must keep the successor generation closed"
+        );
+
+        drop(second);
+        let (_, _, _, admitted_result) =
+            cross_pool_fence_policy_results(restarted_peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        assert!(
+            publish_fleet_capability_probe_result(&slot, "topology-a", admitted_result, now + Duration::from_millis(3),)
+                .is_none(),
+            "the restarted generation may publish only after every old writer drains"
+        );
+    }
+
+    #[test]
+    fn legacy_transition_state_reconcile_fresh_observation_closes_the_polling_window() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let original_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        let (_, _, _, original_result) =
+            cross_pool_fence_policy_results(original_peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", original_result, now).is_none());
+        let admitted = {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                .expect("the original fleet should admit reconciliation")
+        };
+
+        let restarted_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        let state = slot.read().expect("reconcile proof slot should not poison");
+        assert!(
+            legacy_transition_state_reconcile_fleet_proof_matches_at(&state, &admitted, "topology-a", now),
+            "the periodic cache has not observed the restart yet"
+        );
+        assert!(!legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
+            &state,
+            &admitted,
+            "topology-a",
+            &restarted_peers,
+            now,
+        ));
+
+        let (_, _, _, downgraded) = cross_pool_fence_policy_results(original_peers, 4);
+        assert!(
+            downgraded.is_err(),
+            "a synchronous observation of a downgraded peer must fail before any cached proof can authorize a write"
+        );
+    }
+
+    #[tokio::test]
+    async fn legacy_transition_state_reconcile_invalid_token_skips_fleet_observation() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(peers), now).is_none());
+        let admitted = {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                .expect("the original fleet should admit reconciliation")
+        };
+        revoke_fleet_capability_proof(&slot);
+
+        assert!(
+            !legacy_transition_state_reconcile_fleet_proof_matches_with_observer(&slot, &admitted, "topology-a", || async {
+                panic!("an invalid local generation must not trigger a fleet observation");
+            },)
+            .await
+        );
+    }
+
+    #[test]
+    fn legacy_transition_state_reconcile_membership_and_topology_changes_revoke_authority() {
+        let now = Instant::now();
+        for replacement in [
+            BTreeMap::from([("peer-a".to_string(), Uuid::new_v4()), ("peer-b".to_string(), Uuid::new_v4())]),
+            BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]),
+        ] {
+            let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+            let original = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+            assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(original), now).is_none());
+            let admitted = {
+                let state = slot.read().expect("reconcile proof slot should not poison");
+                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                    .expect("the original fleet should admit reconciliation")
+            };
+
+            assert!(
+                publish_fleet_capability_probe_result(&slot, "topology-a", Ok(replacement), now + Duration::from_millis(1),)
+                    .is_some(),
+                "membership or process-epoch replacement must wait for the admitted writer"
+            );
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
+                &state,
+                &admitted,
+                "topology-a",
+                now + Duration::from_millis(1),
+            ));
+        }
+
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(BTreeMap::new()), now).is_none());
+        let admitted = {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                .expect("the original topology should admit reconciliation")
+        };
+        mark_fleet_capability_topology_conflict(&slot);
+        let state = slot.read().expect("reconcile proof slot should not poison");
+        assert!(state.topology_conflict);
+        assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
+            &state,
+            &admitted,
+            "topology-a",
+            now,
+        ));
+    }
+
+    #[test]
+    fn legacy_transition_state_reconcile_capability_downgrade_fails_closed() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        let (_, _, _, compatible_result) =
+            cross_pool_fence_policy_results(peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", compatible_result, now).is_none());
+        let admitted = {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                .expect("v5 should admit reconciliation")
+        };
+
+        let (_, _, _, downgraded_result) =
+            cross_pool_fence_policy_results(peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION - 1);
+        let err = publish_fleet_capability_probe_result(&slot, "topology-a", downgraded_result, now + Duration::from_millis(1))
+            .expect("a v4 member must revoke reconcile authority");
+        assert!(err.to_string().contains("reconcile policy capability version is unsupported"));
+        let state = slot.read().expect("reconcile proof slot should not poison");
+        assert!(state.proof.is_none());
+        assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
+            &state,
+            &admitted,
+            "topology-a",
+            now + Duration::from_millis(1),
+        ));
+        assert!(
+            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now + Duration::from_millis(1),)
+                .is_none(),
+            "a downgraded fleet must remain inspect-only"
+        );
+    }
+
     #[test]
     fn remote_version_state_fleet_proof_conflict_revokes_atomic_snapshot() {
         let now = Instant::now();
@@ -3539,6 +3959,57 @@ mod tests {
         assert!(err.to_string().contains("incomplete"));
     }
 
+    #[tokio::test]
+    async fn legacy_transition_state_reconcile_probe_rejects_missing_or_unreachable_members() {
+        let missing = NotificationSys {
+            peer_clients: Vec::new(),
+            all_peer_clients: vec![None],
+            peer_topology_hosts: vec!["peer-a".to_string()],
+            peer_admin_caches: Vec::new(),
+            tier_config_reload_workers: Default::default(),
+        };
+        let missing_err = missing
+            .probe_cross_pool_fence_fleet("topology-a")
+            .await
+            .expect_err("a missing member slot must prevent reconcile capability proof");
+        assert!(missing_err.to_string().contains("incomplete"));
+
+        let unreachable = NotificationSys {
+            peer_clients: vec![None],
+            all_peer_clients: vec![None, None],
+            peer_topology_hosts: vec!["peer-a".to_string()],
+            peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())],
+            tier_config_reload_workers: Default::default(),
+        };
+        let unreachable_err = unreachable
+            .probe_cross_pool_fence_fleet("topology-a")
+            .await
+            .expect_err("an unreachable member must prevent reconcile capability proof");
+        assert!(unreachable_err.to_string().contains("unreachable"));
+    }
+
+    #[tokio::test]
+    async fn legacy_transition_state_reconcile_single_node_stays_closed_before_local_cas_support() {
+        let notification_sys = NotificationSys {
+            peer_clients: Vec::new(),
+            all_peer_clients: vec![None],
+            peer_topology_hosts: Vec::new(),
+            peer_admin_caches: Vec::new(),
+            tier_config_reload_workers: Default::default(),
+        };
+        let (peers, minimum_version) = notification_sys
+            .probe_cross_pool_fence_fleet("topology-a")
+            .await
+            .expect("a single-node capability probe should complete");
+        assert!(peers.is_empty());
+        assert_eq!(minimum_version, LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION);
+        let (_, _, _, reconcile_result) = cross_pool_fence_policy_results(peers, minimum_version);
+        assert!(
+            reconcile_result.is_err(),
+            "the current node must not self-authorize reconcile before the conditional writer lands"
+        );
+    }
+
     fn build_props(endpoint: &str) -> ServerProperties {
         ServerProperties {
             endpoint: endpoint.to_string(),

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_put_object::{AdvancedPutOptions, PutObjectOptions},
@@ -280,27 +280,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
@@ -331,7 +310,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",
@@ -344,7 +322,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.