feat(ecstore): expose read version stage metrics (#6073)

Record local read_version path resolution, path length check, xl.meta read, and metadata decode durations through the existing GET stage metrics channel. The new samples are gated by GET stage metrics so metrics-off reads avoid timer and recorder work.

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/api/mod.rs

@@ -131,8 +131,6 @@ pub mod bucket {
     }
 
     pub mod metadata_sys {
-        #[cfg(feature = "test-util")]
-        pub use crate::bucket::metadata_sys::ConfigWriteLockProbe;
         pub use crate::bucket::metadata_sys::{
             BucketMetadataMutationGuard, BucketMetadataSys, ObjectLockConfigState, acquire_bucket_metadata_transaction_lock,
             capture_bucket_metadata_incarnation, delete, delete_if_incarnation, get, get_accelerate_config, get_bucket_policy,
@@ -142,7 +140,7 @@ pub mod bucket {
             get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config,
             get_website_config, init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata,
             set_bucket_metadata, update, update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation,
-            update_quota_if_incarnation, update_under_transaction_lock,
+            update_under_transaction_lock,
         };
     }
 
@@ -318,7 +316,7 @@ pub mod data_usage {
         DATA_USAGE_CACHE_NAME, apply_bucket_usage_memory_overlay, compute_bucket_usage,
         init_compression_total_memory_from_backend, invalidate_admin_data_usage_snapshot_cache,
         invalidate_data_usage_snapshot_cache, live_bucket_usage_computations, load_admin_data_usage_from_backend_cached,
-        load_compression_total_from_memory, load_data_usage_from_backend, load_data_usage_from_backend_cached, quota_object_size,
+        load_compression_total_from_memory, load_data_usage_from_backend, load_data_usage_from_backend_cached,
         record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory,
         record_bucket_object_write_memory, record_bucket_object_write_unknown_previous_memory, record_compression_total_memory,
         refresh_bucket_usage_from_object_layer, refresh_versioned_bucket_usage_from_object_layer,
@@ -405,11 +403,8 @@ pub mod metrics {
 }
 
 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::{
-        CrossPoolFenceFleetProofToken, NotificationPeerErr, NotificationSys, acquire_cross_pool_fence_fleet_proof,
-        cross_pool_fence_fleet_proof_matches, get_global_notification_sys, new_global_notification_sys,
+        NotificationPeerErr, NotificationSys, get_global_notification_sys, new_global_notification_sys,
         start_remote_version_state_fleet_probe,
     };
 }
@@ -469,8 +464,7 @@ pub mod set_disk {
 
     #[cfg(feature = "test-util")]
     pub mod test_util {
-        pub use crate::bucket::quota::reservation::fail_next_quota_ledger_save_for_test;
-        pub use crate::set_disk::{MultipartCommitBarrier, MultipartCommitPause, PutObjectCommitBarrier, PutObjectCommitPause};
+        pub use crate::set_disk::{PutObjectCommitBarrier, PutObjectCommitPause};
     }
 }
 

crates/ecstore/src/bucket/metadata_sys.rs

@@ -50,72 +50,6 @@ use uuid::Uuid;
 
 const BUCKET_METADATA_REFRESH_INTERVAL: Duration = Duration::from_secs(15 * 60);
 
-#[cfg(any(test, feature = "test-util"))]
-struct ConfigWriteLockProbeState {
-    bucket: String,
-    arrived: tokio::sync::Notify,
-}
-
-#[cfg(any(test, feature = "test-util"))]
-static CONFIG_WRITE_LOCK_PROBES: std::sync::OnceLock<StdMutex<Vec<Arc<ConfigWriteLockProbeState>>>> = std::sync::OnceLock::new();
-
-#[cfg(any(test, feature = "test-util"))]
-pub struct ConfigWriteLockProbe {
-    state: Arc<ConfigWriteLockProbeState>,
-}
-
-#[cfg(any(test, feature = "test-util"))]
-impl ConfigWriteLockProbe {
-    pub fn install(bucket: &str) -> Self {
-        let state = Arc::new(ConfigWriteLockProbeState {
-            bucket: bucket.to_string(),
-            arrived: tokio::sync::Notify::new(),
-        });
-        let mut probes = CONFIG_WRITE_LOCK_PROBES
-            .get_or_init(|| StdMutex::new(Vec::new()))
-            .lock()
-            .expect("config write lock probe mutex should not poison");
-        assert!(
-            !probes.iter().any(|current| current.bucket == state.bucket),
-            "config write lock probe must be unique for a bucket"
-        );
-        probes.push(Arc::clone(&state));
-        drop(probes);
-        Self { state }
-    }
-
-    pub async fn wait_until_attempted(&self) {
-        tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
-            .await
-            .expect("bucket config update should attempt the transaction lock");
-    }
-}
-
-#[cfg(any(test, feature = "test-util"))]
-impl Drop for ConfigWriteLockProbe {
-    fn drop(&mut self) {
-        let mut probes = CONFIG_WRITE_LOCK_PROBES
-            .get_or_init(|| StdMutex::new(Vec::new()))
-            .lock()
-            .expect("config write lock probe mutex should not poison");
-        probes.retain(|state| !Arc::ptr_eq(state, &self.state));
-    }
-}
-
-#[cfg(any(test, feature = "test-util"))]
-fn notify_config_write_lock_attempt(bucket: &str) {
-    let probe = CONFIG_WRITE_LOCK_PROBES
-        .get_or_init(|| StdMutex::new(Vec::new()))
-        .lock()
-        .expect("config write lock probe mutex should not poison")
-        .iter()
-        .find(|probe| probe.bucket == bucket)
-        .cloned();
-    if let Some(probe) = probe {
-        probe.arrived.notify_one();
-    }
-}
-
 #[derive(Clone, Copy)]
 enum MetadataLoadMode {
     Initial,
@@ -656,31 +590,6 @@ pub async fn update_under_transaction_lock(
     update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data).await
 }
 
-pub async fn update_quota_if_incarnation(
-    bucket: &str,
-    data: Vec<u8>,
-    expected_incarnation_id: Uuid,
-    proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
-) -> Result<OffsetDateTime> {
-    let sys = get_bucket_metadata_sys()?;
-    let guard = Box::pin(acquire_config_write_guard_for_incarnation(
-        sys.clone(),
-        bucket,
-        Some(expected_incarnation_id),
-    ))
-    .await?;
-    if !crate::services::notification_sys::cross_pool_fence_fleet_proof_matches(proof) {
-        return Err(Error::NamespaceLockQuorumUnavailable {
-            mode: "quota_capability",
-            bucket: bucket.to_string(),
-            object: rustfs_config::QUOTA_CONFIG_FILE.to_string(),
-            required: 1,
-            achieved: 0,
-        });
-    }
-    update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data).await
-}
-
 pub async fn update_bucket_targets_under_transaction_lock(
     guard: &BucketMetadataMutationGuard,
     bucket: &str,
@@ -825,26 +734,7 @@ async fn acquire_transaction_lock_with_sys(
     let lock = api
         .new_ns_lock(RUSTFS_META_BUCKET, &bucket_metadata_transaction_lock_key(bucket))
         .await?;
-    let acquire = lock.get_write_lock(crate::set_disk::get_lock_acquire_timeout());
-    #[cfg(any(test, feature = "test-util"))]
-    {
-        tokio::pin!(acquire);
-        let mut notified = false;
-        let guard = futures::future::poll_fn(|cx| match std::future::Future::poll(acquire.as_mut(), cx) {
-            std::task::Poll::Pending => {
-                if !notified {
-                    notify_config_write_lock_attempt(bucket);
-                    notified = true;
-                }
-                std::task::Poll::Pending
-            }
-            std::task::Poll::Ready(result) => std::task::Poll::Ready(result),
-        })
-        .await?;
-        Ok(guard)
-    }
-    #[cfg(not(any(test, feature = "test-util")))]
-    Ok(acquire.await?)
+    Ok(lock.get_write_lock(crate::set_disk::get_lock_acquire_timeout()).await?)
 }
 
 /// The lock resource name is deliberately still the `bucket-targets` one it
@@ -999,37 +889,6 @@ pub(crate) async fn get_object_lock_config_and_incarnation_from_disk_in(
     }
 }
 
-/// Re-read the quota configuration and bucket incarnation from the same
-/// authoritative metadata blob while the caller holds the bucket metadata
-/// transaction read lock.
-pub(crate) async fn get_quota_config_and_incarnation_from_disk_in(
-    ctx: &crate::runtime::instance::InstanceContext,
-    bucket: &str,
-) -> Result<(Option<BucketQuota>, Uuid, OffsetDateTime)> {
-    let bucket_meta_sys_lock = bucket_metadata_sys_of(ctx)?;
-    let bucket_meta_sys = bucket_meta_sys_lock.read().await.clone();
-
-    match bucket_meta_sys
-        .read_authoritative_metadata_from_disk_under_transaction_lock(bucket)
-        .await?
-    {
-        BucketMetadataAuthority::Authoritative(metadata)
-            if metadata.bucket_incarnation_sidecar && !metadata.bucket_incarnation_id.is_nil() =>
-        {
-            Ok((
-                metadata.quota_config.clone(),
-                metadata.bucket_incarnation_id,
-                metadata.quota_config_updated_at,
-            ))
-        }
-        BucketMetadataAuthority::Authoritative(_) => {
-            Err(Error::other(format!("bucket incarnation metadata is not authoritative: {bucket}")))
-        }
-        BucketMetadataAuthority::MissingBucket => Err(Error::BucketNotFound(bucket.to_string())),
-        BucketMetadataAuthority::Fabricated => Err(Error::other(format!("bucket quota metadata is not authoritative: {bucket}"))),
-    }
-}
-
 pub async fn get_replication_config(bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;

crates/ecstore/src/bucket/quota/checker.rs

@@ -52,7 +52,6 @@ impl QuotaChecker {
     ) -> Result<QuotaCheckResult, QuotaError> {
         let start_time = Instant::now();
         let quota_config = self.get_quota_config(bucket).await?;
-        let uses_durable_reservations = quota_config.uses_durable_reservations();
 
         // If no quota limit is set, allow operation
         let quota_limit = match quota_config.quota {
@@ -68,7 +67,6 @@ impl QuotaChecker {
                     quota_limit: None,
                     operation_size,
                     remaining: None,
-                    uses_durable_reservations,
                 });
             }
             Some(q) => q,
@@ -76,17 +74,14 @@ impl QuotaChecker {
 
         let current_usage = self.get_real_time_usage(bucket).await?;
 
-        let admission_size = if uses_durable_reservations { 0 } else { operation_size };
         let expected_usage = match operation {
-            QuotaOperation::PutObject | QuotaOperation::PostObject | QuotaOperation::CopyObject => {
-                current_usage.saturating_add(admission_size)
-            }
+            QuotaOperation::PutObject | QuotaOperation::PostObject | QuotaOperation::CopyObject => current_usage + operation_size,
             QuotaOperation::DeleteObject => current_usage.saturating_sub(operation_size),
         };
 
         let allowed = match operation {
             QuotaOperation::PutObject | QuotaOperation::PostObject | QuotaOperation::CopyObject => {
-                quota_config.check_operation_allowed(current_usage, admission_size)
+                quota_config.check_operation_allowed(current_usage, operation_size)
             }
             QuotaOperation::DeleteObject => true,
         };
@@ -110,7 +105,6 @@ impl QuotaChecker {
             quota_limit: Some(quota_limit),
             operation_size,
             remaining,
-            uses_durable_reservations,
         };
 
         let duration = start_time.elapsed();
@@ -164,26 +158,6 @@ impl QuotaChecker {
             .await
     }
 
-    pub async fn set_durable_quota_config_if_incarnation(
-        &mut self,
-        bucket: &str,
-        quota: BucketQuota,
-        expected_incarnation_id: uuid::Uuid,
-        proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
-    ) -> Result<OffsetDateTime, QuotaError> {
-        let json_data = serde_json::to_vec(&quota).map_err(|e| QuotaError::InvalidConfig {
-            reason: format!("Failed to serialize quota config: {}", e),
-        })?;
-        let start_time = Instant::now();
-        let updated_at =
-            crate::bucket::metadata_sys::update_quota_if_incarnation(bucket, json_data, expected_incarnation_id, proof)
-                .await
-                .map_err(QuotaError::StorageError)?;
-
-        rustfs_common::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed());
-        Ok(updated_at)
-    }
-
     async fn set_quota_config_for_incarnation(
         &mut self,
         bucket: &str,
@@ -381,7 +355,6 @@ mod tests {
             quota_limit: None,
             operation_size: 1024,
             remaining: None,
-            uses_durable_reservations: false,
         };
 
         assert!(result.allowed);
@@ -405,13 +378,4 @@ mod tests {
         let allowed = quota.check_operation_allowed(512, 1024);
         assert!(!allowed);
     }
-
-    #[test]
-    fn legacy_quota_rejects_full_operation_while_v1_defers_net_growth() {
-        let legacy: BucketQuota = serde_json::from_str(r#"{"quota":5}"#).expect("legacy quota should parse");
-        let durable = BucketQuota::new(Some(5));
-
-        assert!(!legacy.check_operation_allowed(4, 2));
-        assert!(durable.uses_durable_reservations());
-    }
 }

crates/ecstore/src/bucket/quota/mod.rs

@@ -13,98 +13,38 @@
 // limitations under the License.
 
 pub mod checker;
-pub(crate) mod reservation;
 
 use crate::error::Result;
 use rustfs_config::{
     QUOTA_API_PATH, QUOTA_EXCEEDED_ERROR_CODE, QUOTA_INTERNAL_ERROR_CODE, QUOTA_INVALID_CONFIG_ERROR_CODE,
     QUOTA_NOT_FOUND_ERROR_CODE,
 };
-use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _};
+use serde::{Deserialize, Serialize};
 use thiserror::Error;
 use time::OffsetDateTime;
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
 pub enum QuotaType {
-    /// Hard quota accounting.
+    /// Hard quota: reject immediately when exceeded
     #[default]
     #[serde(alias = "HARD", alias = "hard")]
     Hard,
 }
 
-pub(crate) const QUOTA_RESERVATION_PROTOCOL_V1: u32 = 1;
-
 /// Bucket quota configuration. quota_type defaults to Hard when omitted.
-#[derive(Debug, Default, Clone, PartialEq)]
+#[derive(Debug, Deserialize, Serialize, Default, Clone, PartialEq)]
 pub struct BucketQuota {
+    #[serde(default)]
     pub quota: Option<u64>,
     /// Defaults to Hard when missing.
+    #[serde(default)]
     pub quota_type: QuotaType,
-    /// Optional durable reservation protocol. The wire format gives older
-    /// nodes a zero hard quota so a mixed-version fleet fails closed.
-    pub reservation_protocol: Option<u32>,
     /// Timestamp when this quota configuration was set (for audit purposes)
+    #[serde(default, with = "time::serde::rfc3339::option")]
     pub created_at: Option<OffsetDateTime>,
     /// Accept updated_at for compatibility; not used.
-    pub updated_at: Option<OffsetDateTime>,
-}
-
-#[derive(Deserialize, Serialize)]
-struct BucketQuotaWire {
-    #[serde(default)]
-    quota: Option<u64>,
-    #[serde(default)]
-    quota_type: QuotaType,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    reservation_protocol: Option<u32>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    reservation_quota: Option<u64>,
-    #[serde(default, with = "time::serde::rfc3339::option")]
-    created_at: Option<OffsetDateTime>,
     #[serde(default, with = "time::serde::rfc3339::option", skip_serializing_if = "Option::is_none")]
-    updated_at: Option<OffsetDateTime>,
-}
-
-impl Serialize for BucketQuota {
-    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
-    where
-        S: Serializer,
-    {
-        let durable = self.uses_durable_reservations();
-        BucketQuotaWire {
-            quota: if durable { Some(0) } else { self.quota },
-            quota_type: self.quota_type.clone(),
-            reservation_protocol: self.reservation_protocol,
-            reservation_quota: if durable { self.quota } else { None },
-            created_at: self.created_at,
-            updated_at: self.updated_at,
-        }
-        .serialize(serializer)
-    }
-}
-
-impl<'de> Deserialize<'de> for BucketQuota {
-    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        let wire = BucketQuotaWire::deserialize(deserializer)?;
-        let quota = if wire.reservation_protocol == Some(QUOTA_RESERVATION_PROTOCOL_V1) {
-            Some(
-                wire.reservation_quota
-                    .ok_or_else(|| D::Error::custom("reservation_quota is required for reservation protocol v1"))?,
-            )
-        } else {
-            wire.quota
-        };
-        Ok(Self {
-            quota,
-            quota_type: wire.quota_type,
-            reservation_protocol: wire.reservation_protocol,
-            created_at: wire.created_at,
-            updated_at: wire.updated_at,
-        })
-    }
+    pub updated_at: Option<OffsetDateTime>,
 }
 
 impl BucketQuota {
@@ -123,7 +63,6 @@ impl BucketQuota {
         Self {
             quota,
             quota_type: QuotaType::Hard,
-            reservation_protocol: quota.map(|_| QUOTA_RESERVATION_PROTOCOL_V1),
             created_at: Some(now),
             updated_at: None,
         }
@@ -133,19 +72,7 @@ impl BucketQuota {
         self.quota
     }
 
-    pub fn uses_durable_reservations(&self) -> bool {
-        self.reservation_protocol == Some(QUOTA_RESERVATION_PROTOCOL_V1)
-    }
-
-    pub fn has_unsupported_reservation_protocol(&self) -> bool {
-        self.reservation_protocol
-            .is_some_and(|version| version != QUOTA_RESERVATION_PROTOCOL_V1)
-    }
-
     pub fn check_operation_allowed(&self, current_usage: u64, operation_size: u64) -> bool {
-        if operation_size == 0 {
-            return true;
-        }
         if let Some(quota_limit) = self.quota {
             current_usage.saturating_add(operation_size) <= quota_limit
         } else {
@@ -167,7 +94,6 @@ pub struct QuotaCheckResult {
     pub quota_limit: Option<u64>,
     pub operation_size: u64,
     pub remaining: Option<u64>,
-    pub uses_durable_reservations: bool,
 }
 
 #[derive(Debug)]
@@ -284,59 +210,7 @@ mod tests {
         let buf = q.marshal_msg().expect("marshal");
         let restored = BucketQuota::unmarshal(&buf).expect("unmarshal");
         assert_eq!(q.quota, restored.quota);
-        assert_eq!(restored.quota_type, QuotaType::Hard);
-        assert_eq!(restored.reservation_protocol, Some(QUOTA_RESERVATION_PROTOCOL_V1));
-    }
-
-    #[test]
-    fn clearing_quota_keeps_the_legacy_compatible_type() {
-        let quota = BucketQuota::new(None);
-
-        assert_eq!(quota.quota_type, QuotaType::Hard);
-        assert_eq!(quota.reservation_protocol, None);
-        assert!(!quota.uses_durable_reservations());
-    }
-
-    #[test]
-    fn durable_quota_makes_legacy_nodes_fail_closed() {
-        let json = serde_json::to_vec(&BucketQuota::new(Some(2048))).expect("durable quota should serialize");
-        let quota: BucketQuota = serde_json::from_slice(&json).expect("current quota version should parse");
-        assert!(quota.uses_durable_reservations());
-        assert_eq!(quota.quota, Some(2048));
-
-        #[derive(Deserialize)]
-        enum LegacyQuotaType {
-            Hard,
-        }
-        #[derive(Deserialize)]
-        struct LegacyBucketQuota {
-            #[allow(dead_code)]
-            quota: Option<u64>,
-            #[allow(dead_code)]
-            quota_type: LegacyQuotaType,
-        }
-        let legacy = serde_json::from_slice::<LegacyBucketQuota>(&json)
-            .expect("legacy readers should ignore the reservation protocol field");
-        assert_eq!(legacy.quota, Some(0));
-        assert!(matches!(legacy.quota_type, LegacyQuotaType::Hard));
-    }
-
-    #[test]
-    fn unknown_reservation_protocol_does_not_activate_v1() {
-        let quota: BucketQuota =
-            serde_json::from_str(r#"{"quota":0,"quota_type":"Hard","reservation_protocol":2,"reservation_quota":2048}"#)
-                .expect("future protocol should remain parseable");
-
-        assert!(!quota.uses_durable_reservations());
-        assert!(quota.has_unsupported_reservation_protocol());
-    }
-
-    #[test]
-    fn reservation_protocol_v1_requires_reservation_quota() {
-        let err = serde_json::from_str::<BucketQuota>(r#"{"quota":0,"quota_type":"Hard","reservation_protocol":1}"#)
-            .expect_err("v1 without its authoritative quota must fail closed");
-
-        assert!(err.to_string().contains("reservation_quota is required"));
+        assert_eq!(q.quota_type, restored.quota_type);
     }
 
     /// unmarshal accepts format without quota_type

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -248,16 +248,6 @@ fn decode_remote_version_state_capability(expected_member: &str, result: &[u8])
     Ok(server_epoch)
 }
 
-fn decode_cross_pool_fence_capability(expected_member: &str, result: &[u8]) -> Result<(u32, Uuid)> {
-    let version = result
-        .get(..4)
-        .and_then(|value| value.try_into().ok())
-        .map(u32::from_be_bytes)
-        .ok_or_else(|| Error::other("peer returned an invalid cross-pool fence capability version"))?;
-    let epoch = decode_remote_version_state_capability(expected_member, &result[4..])?;
-    Ok((version, epoch))
-}
-
 #[derive(Clone, Debug)]
 pub struct PeerLiveEventsBatch {
     pub events: Vec<u8>,
@@ -1298,16 +1288,6 @@ impl PeerRestClient {
         Ok((self.topology_member.clone(), epoch))
     }
 
-    pub async fn probe_cross_pool_fence(&self, topology_fingerprint: String) -> Result<(String, u32, Uuid)> {
-        let mut probe = rustfs_protos::CROSS_POOL_FENCE_CAPABILITY_PROBE_PREFIX.to_vec();
-        probe.extend_from_slice(Uuid::new_v4().as_bytes());
-        let result = self
-            .heal_control(rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, topology_fingerprint, probe)
-            .await?;
-        let (supported_version, epoch) = decode_cross_pool_fence_capability(&self.topology_member, &result)?;
-        Ok((self.topology_member.clone(), supported_version, epoch))
-    }
-
     pub async fn load_bucket_metadata(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
         self.finalize_result(
             async {
@@ -2758,24 +2738,6 @@ mod tests {
         assert!(decode_remote_version_state_capability("node-a:9000", &nil).is_err());
     }
 
-    #[test]
-    fn cross_pool_fence_capability_decoder_fails_closed() {
-        let epoch = Uuid::new_v4();
-        let result = rustfs_protos::encode_cross_pool_fence_capability(1, "node-a:9000", epoch.as_bytes())
-            .expect("small capability response should encode");
-        assert_eq!(
-            decode_cross_pool_fence_capability("node-a:9000", &result).expect("valid capability should decode"),
-            (1, epoch)
-        );
-        for malformed in [&[][..], &[0, 0, 0][..], &result[..result.len() - 1]] {
-            assert!(decode_cross_pool_fence_capability("node-a:9000", malformed).is_err());
-        }
-        assert!(decode_cross_pool_fence_capability("node-b:9000", &result).is_err());
-        let nil = rustfs_protos::encode_cross_pool_fence_capability(1, "node-a:9000", Uuid::nil().as_bytes())
-            .expect("small capability response should encode");
-        assert!(decode_cross_pool_fence_capability("node-a:9000", &nil).is_err());
-    }
-
     struct TierMutationResponseFixture<'a> {
         version: u32,
         phase: TierMutationRpcPhase,

crates/ecstore/src/data_usage/mod.rs

@@ -1355,7 +1355,7 @@ impl BucketUsageAccumulator {
             return Ok(());
         }
 
-        let object_size = quota_object_size(object)?;
+        let object_size = object.size.max(0) as u64;
         self.current_live_versions = self.current_live_versions.saturating_add(1);
         self.size_histogram.add(object_size);
         self.total_size = self.total_size.saturating_add(object_size);
@@ -1385,20 +1385,6 @@ impl BucketUsageAccumulator {
     }
 }
 
-pub fn quota_object_size(object: &ObjectInfo) -> Result<u64, Error> {
-    let logical_size = u64::try_from(object.get_actual_size().map_err(Error::other)?).map_err(|_| Error::PartMissingOrCorrupt)?;
-    let persisted_part_size = if object.parts.is_empty() {
-        u64::try_from(object.size).map_err(|_| Error::PartMissingOrCorrupt)?
-    } else {
-        object.parts.iter().try_fold(0_u64, |total, part| {
-            let actual_size = u64::try_from(part.actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
-            let part_size = actual_size.max(u64::try_from(part.size).map_err(|_| Error::PartMissingOrCorrupt)?);
-            total.checked_add(part_size).ok_or(Error::PartMissingOrCorrupt)
-        })?
-    };
-    Ok(logical_size.max(persisted_part_size))
-}
-
 type UsageVersionPage = StorageListObjectVersionsInfo<ObjectInfo>;
 
 pub async fn compute_bucket_usage(store: Arc<ECStore>, bucket_name: &str) -> Result<BucketUsageInfo, Error> {
@@ -3138,80 +3124,6 @@ mod tests {
         assert_eq!(usage.object_versions_histogram.get("BETWEEN_1000_AND_10000"), Some(&1));
     }
 
-    #[test]
-    fn bucket_usage_uses_the_larger_of_logical_and_physical_size() {
-        let mut metadata = HashMap::new();
-        rustfs_utils::http::insert_str(
-            &mut metadata,
-            rustfs_utils::http::SUFFIX_COMPRESSION,
-            "klauspost/compress/s2".to_string(),
-        );
-        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "4096".to_string());
-        let object = ObjectInfo {
-            name: "compressed".to_string(),
-            size: 128,
-            user_defined: Arc::new(metadata),
-            ..Default::default()
-        };
-        let mut usage = BucketUsageAccumulator::default();
-        usage
-            .record("bucket", &object)
-            .expect("valid compressed metadata should be counted");
-        assert_eq!(usage.finish().size, 4096);
-
-        let mut framed_metadata = HashMap::new();
-        rustfs_utils::http::insert_str(
-            &mut framed_metadata,
-            rustfs_utils::http::SUFFIX_COMPRESSION,
-            "klauspost/compress/s2".to_string(),
-        );
-        rustfs_utils::http::insert_str(&mut framed_metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "1".to_string());
-        let framed = ObjectInfo {
-            name: "framed".to_string(),
-            size: 17,
-            user_defined: Arc::new(framed_metadata),
-            ..Default::default()
-        };
-        assert_eq!(quota_object_size(&framed).expect("physical framing must remain quota-accounted"), 17);
-
-        let mut corrupt_metadata = (*object.user_defined).clone();
-        rustfs_utils::http::insert_str(&mut corrupt_metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "-1".to_string());
-        let corrupt = ObjectInfo {
-            user_defined: Arc::new(corrupt_metadata),
-            ..object
-        };
-        assert!(
-            matches!(
-                BucketUsageAccumulator::default().record("bucket", &corrupt),
-                Err(Error::PartMissingOrCorrupt)
-            ),
-            "negative logical metadata must not become a smaller quota baseline"
-        );
-
-        let mut poisoned_metadata = HashMap::new();
-        rustfs_utils::http::insert_str(
-            &mut poisoned_metadata,
-            rustfs_utils::http::SUFFIX_COMPRESSION,
-            "klauspost/compress/s2".to_string(),
-        );
-        rustfs_utils::http::insert_str(&mut poisoned_metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "1".to_string());
-        let poisoned = ObjectInfo {
-            name: "legacy-swift-metadata".to_string(),
-            size: 4096,
-            user_defined: Arc::new(poisoned_metadata),
-            parts: Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
-                size: 4096,
-                actual_size: 4096,
-                ..Default::default()
-            }]),
-            ..Default::default()
-        };
-        assert_eq!(
-            quota_object_size(&poisoned).expect("persisted part accounting must bound legacy user metadata"),
-            4096
-        );
-    }
-
     #[tokio::test]
     #[serial]
     async fn live_bucket_usage_refreshes_are_coalesced_only_while_in_flight() {

crates/ecstore/src/diagnostics/get.rs

@@ -84,6 +84,10 @@ pub(crate) const GET_STAGE_READER_STREAM_FIRST_READ: &str = "reader_stream_first
 pub(crate) const GET_STAGE_READER_TASK_BITROT_READER_INIT: &str = "reader_task_bitrot_reader_init";
 pub(crate) const GET_STAGE_READER_TASK_FILE_OPEN: &str = "reader_task_file_open";
 pub(crate) const GET_STAGE_READER_TASK_READER_CONSTRUCTION: &str = "reader_task_reader_construction";
+pub(crate) const GET_STAGE_READ_VERSION_DECODE: &str = "read_version_decode";
+pub(crate) const GET_STAGE_READ_VERSION_PATH_CHECK: &str = "read_version_path_check";
+pub(crate) const GET_STAGE_READ_VERSION_PATH_RESOLVE: &str = "read_version_path_resolve";
+pub(crate) const GET_STAGE_READ_VERSION_XLMETA_READ: &str = "read_version_xlmeta_read";
 pub(crate) const GET_STAGE_RECONSTRUCT: &str = "reconstruct";
 pub(crate) const GET_STAGE_RESPONSE_HANDOFF: &str = "response_handoff";
 pub(crate) const GET_STAGE_SLOWEST_METADATA_RESPONSE: &str = "slowest_metadata_response";
@@ -442,6 +446,10 @@ mod tests {
         assert_eq!(GET_STAGE_QUORUM_REACHED, "quorum_reached");
         assert_eq!(GET_STAGE_RANGE, "range");
         assert_eq!(GET_STAGE_READER_SETUP, "reader_setup");
+        assert_eq!(GET_STAGE_READ_VERSION_DECODE, "read_version_decode");
+        assert_eq!(GET_STAGE_READ_VERSION_PATH_CHECK, "read_version_path_check");
+        assert_eq!(GET_STAGE_READ_VERSION_PATH_RESOLVE, "read_version_path_resolve");
+        assert_eq!(GET_STAGE_READ_VERSION_XLMETA_READ, "read_version_xlmeta_read");
         assert_eq!(GET_STAGE_RECONSTRUCT, "reconstruct");
         assert_eq!(GET_STAGE_RESPONSE_HANDOFF, "response_handoff");
         assert_eq!(GET_STAGE_SLOWEST_METADATA_RESPONSE, "slowest_metadata_response");

crates/ecstore/src/disk/local.rs

@@ -15,6 +15,11 @@
 use crate::config::storageclass::DEFAULT_INLINE_BLOCK;
 use crate::crash_inject::{self, CrashPoint};
 use crate::data_usage::local_snapshot::ensure_data_usage_layout;
+use crate::diagnostics::get::{
+    GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_READ_VERSION_DECODE,
+    GET_STAGE_READ_VERSION_PATH_CHECK, GET_STAGE_READ_VERSION_PATH_RESOLVE, GET_STAGE_READ_VERSION_XLMETA_READ,
+    get_stage_timer_if_enabled, record_get_stage_duration_if_enabled,
+};
 #[cfg(test)]
 use crate::disk::HEALING_MARKER_PATH;
 use crate::disk::disk_store::{get_drive_walkdir_stall_timeout, get_object_disk_read_timeout};
@@ -22,18 +27,17 @@ use crate::disk::{
     BUCKET_META_PREFIX, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN,
     CHECK_PART_VOLUME_NOT_FOUND, CheckPartsResp, ConditionalFileUpdate, DataDirDeleteStatus, DeleteOptions, DiskAPI, DiskInfo,
     DiskInfoOptions, DiskLocation, DiskMetrics, FileInfoVersions, FileReader, FileWriter, MmapCopyStageMetrics, OldCurrentSize,
-    PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction,
-    QUOTA_MUTATION_FENCE_METADATA_SUFFIX, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET, RUSTFS_META_TMP_DELETED_BUCKET,
-    ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, STORAGE_FORMAT_FILE_BACKUP,
-    SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, conv_part_err_to_int,
+    PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction, RUSTFS_META_BUCKET,
+    RUSTFS_META_TMP_BUCKET, RUSTFS_META_TMP_DELETED_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
+    STORAGE_FORMAT_FILE, STORAGE_FORMAT_FILE_BACKUP, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
+    conv_part_err_to_int,
     endpoint::Endpoint,
     error::{DiskError, Error, FileAccessDeniedWithContext, Result},
     error_conv::{to_access_error, to_file_error, to_unformatted_disk_error, to_volume_error},
     format::FormatV3,
     fs::{O_APPEND, O_CREATE, O_RDONLY, O_TRUNC, O_WRONLY, access, lstat, lstat_std, remove, remove_all_std, remove_std, rename},
-    is_quota_mutation_fence_path, os,
+    os,
     os::{check_path_length, is_dir_not_empty_error, is_empty_dir, is_root_disk, rename_all, rename_all_ignore_missing_source},
-    quota_mutation_fence_path,
 };
 use crate::erasure::coding::{self, bitrot_verify};
 use crate::runtime::sources as runtime_sources;
@@ -56,7 +60,9 @@ use std::collections::HashMap;
 use std::collections::HashSet;
 use std::fmt::Debug;
 use std::io::{Error as IoError, SeekFrom};
-use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
+#[cfg(target_os = "linux")]
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, OnceLock};
 use std::time::Duration;
 use std::{
@@ -4750,25 +4756,6 @@ struct SnapshotLeaseEntry {
     tokens: HashSet<SnapshotLeaseToken>,
     pending_delete: Option<DeleteOptions>,
     deleting: bool,
-    mutation_fence: Option<Arc<QuotaMutationFenceState>>,
-}
-
-#[derive(Default)]
-struct QuotaMutationFenceState {
-    revoked: AtomicBool,
-    running: AtomicUsize,
-    notify: Notify,
-}
-
-struct QuotaMutationFenceClaim {
-    state: Arc<QuotaMutationFenceState>,
-}
-
-impl Drop for QuotaMutationFenceClaim {
-    fn drop(&mut self) {
-        self.state.running.fetch_sub(1, Ordering::AcqRel);
-        self.state.notify.notify_waiters();
-    }
 }
 
 #[derive(Default)]
@@ -7333,34 +7320,6 @@ fn normalize_path_components(path: impl AsRef<Path>) -> PathBuf {
 }
 
 impl LocalDisk {
-    async fn claim_quota_mutation_fence(
-        &self,
-        volume: &str,
-        path: &str,
-        token: SnapshotLeaseToken,
-    ) -> Result<Arc<QuotaMutationFenceClaim>> {
-        let key = SnapshotLeaseKey {
-            volume: RUSTFS_META_BUCKET.to_string(),
-            path: quota_mutation_fence_path(volume, path),
-        };
-        let state = {
-            let registry = self.snapshot_leases.lock().await;
-            let entry = registry.entries.get(&key).ok_or(DiskError::FileNotFound)?;
-            let state = entry.mutation_fence.as_ref().ok_or(DiskError::FileNotFound)?;
-            if !entry.tokens.contains(&token) || state.revoked.load(Ordering::Acquire) {
-                return Err(DiskError::FileNotFound);
-            }
-            state.running.fetch_add(1, Ordering::AcqRel);
-            Arc::clone(state)
-        };
-        if state.revoked.load(Ordering::Acquire) {
-            state.running.fetch_sub(1, Ordering::AcqRel);
-            state.notify.notify_waiters();
-            return Err(DiskError::FileNotFound);
-        }
-        Ok(Arc::new(QuotaMutationFenceClaim { state }))
-    }
-
     async fn reserve_version_delete(&self, volume: &str, object: &str, data_dir: Uuid, rollback_dir: Uuid) -> Result<bool> {
         let path = format!("{object}/{data_dir}");
         let data_path = self.io_get_object_path(volume, &path)?;
@@ -8694,30 +8653,7 @@ impl DiskAPI for LocalDisk {
         // optimistic; this lease establishes the local commit/delete order and
         // remains owned by any blocking syscall that outlives async cancellation.
         let destination_object_path = self.io_get_object_path(dst_volume, dst_path)?;
-        let quota_fence_token =
-            match rustfs_utils::http::metadata_compat::get_consistent_str(&fi.metadata, QUOTA_MUTATION_FENCE_METADATA_SUFFIX) {
-                Some(value) => {
-                    let token = Uuid::parse_str(value).map_err(|_| DiskError::FileCorrupt)?;
-                    Some(SnapshotLeaseToken::from_slice(token.as_bytes())?)
-                }
-                None if rustfs_utils::http::metadata_compat::contains_key_str(
-                    &fi.metadata,
-                    QUOTA_MUTATION_FENCE_METADATA_SUFFIX,
-                ) =>
-                {
-                    return Err(DiskError::FileCorrupt);
-                }
-                None => None,
-            };
-        rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, QUOTA_MUTATION_FENCE_METADATA_SUFFIX);
-        let quota_fence_claim = match quota_fence_token {
-            Some(token) => Some(self.claim_quota_mutation_fence(dst_volume, dst_path, token).await?),
-            None => None,
-        };
         let mutation_lease = os::acquire_rename_data_mutation_lease(&self.root, dst_volume, &destination_object_path).await;
-        if let Some(claim) = quota_fence_claim {
-            mutation_lease.attach_external_guard(claim);
-        }
         if fi.is_legacy_indexed_delete_marker() {
             fi.erasure.index = 0;
         }
@@ -9707,26 +9643,11 @@ impl DiskAPI for LocalDisk {
     }
 
     async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> Result<SnapshotLeaseToken> {
+        let file_path = self.io_get_object_path(volume, path)?;
         let key = SnapshotLeaseKey {
             volume: volume.to_string(),
             path: path.to_string(),
         };
-        if volume == RUSTFS_META_BUCKET && is_quota_mutation_fence_path(path) {
-            let mut registry = self.snapshot_leases.lock().await;
-            let entry = registry.entries.entry(key).or_default();
-            let state = entry
-                .mutation_fence
-                .get_or_insert_with(|| Arc::new(QuotaMutationFenceState::default()));
-            if state.revoked.load(Ordering::Acquire) {
-                return Err(DiskError::FileNotFound);
-            }
-            let token = SnapshotLeaseToken::new();
-            entry.tokens.insert(token);
-            return Ok(token);
-        }
-
-        let file_path = self.io_get_object_path(volume, path)?;
-        let _mutation_lease = os::acquire_rename_data_mutation_lease(&self.root, volume, &file_path).await;
         let token = {
             let mut registry = self.snapshot_leases.lock().await;
             if registry.entries.get(&key).is_some_and(|entry| entry.deleting) {
@@ -9754,48 +9675,6 @@ impl DiskAPI for LocalDisk {
             volume: volume.to_string(),
             path: path.to_string(),
         };
-        if volume == RUSTFS_META_BUCKET && is_quota_mutation_fence_path(path) {
-            if !token.is_revoke_all() {
-                let mut registry = self.snapshot_leases.lock().await;
-                let Some(entry) = registry.entries.get_mut(&key) else {
-                    return Ok(());
-                };
-                entry.tokens.remove(&token);
-                let removable = entry.tokens.is_empty()
-                    && entry
-                        .mutation_fence
-                        .as_ref()
-                        .is_none_or(|state| state.running.load(Ordering::Acquire) == 0);
-                if removable {
-                    registry.entries.remove(&key);
-                }
-                return Ok(());
-            }
-            let state = {
-                let mut registry = self.snapshot_leases.lock().await;
-                let Some(entry) = registry.entries.get_mut(&key) else {
-                    return Ok(());
-                };
-                let Some(state) = entry.mutation_fence.as_ref().cloned() else {
-                    registry.entries.remove(&key);
-                    return Ok(());
-                };
-                state.revoked.store(true, Ordering::Release);
-                entry.tokens.clear();
-                state
-            };
-            loop {
-                let notified = state.notify.notified();
-                tokio::pin!(notified);
-                notified.as_mut().enable();
-                if state.running.load(Ordering::Acquire) == 0 {
-                    break;
-                }
-                notified.await;
-            }
-            self.snapshot_leases.lock().await.entries.remove(&key);
-            return Ok(());
-        }
         let opts = {
             let mut registry = self.snapshot_leases.lock().await;
             let Some(entry) = registry.entries.get_mut(&key) else {
@@ -9966,6 +9845,12 @@ impl DiskAPI for LocalDisk {
         opts: &ReadOptions,
     ) -> Result<FileInfo> {
         crate::hp_guard!("LocalDisk::read_version");
+        let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
+        let metrics_path = if stage_metrics_enabled && crate::bucket::utils::is_meta_bucketname(volume) {
+            GET_OBJECT_PATH_INTERNAL_META
+        } else {
+            GET_OBJECT_PATH_LEGACY_DUPLEX
+        };
         if !org_volume.is_empty() {
             let org_volume_path = self.io_get_bucket_path(org_volume)?;
             if !skip_access_checks(org_volume) {
@@ -9975,37 +9860,46 @@ impl DiskAPI for LocalDisk {
             }
         }
 
+        let path_resolve_start = get_stage_timer_if_enabled(stage_metrics_enabled);
         let file_path = self.io_get_object_path(volume, path)?;
         let volume_dir = self.io_get_bucket_path(volume)?;
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_PATH_RESOLVE, path_resolve_start);
 
+        let path_check_start = get_stage_timer_if_enabled(stage_metrics_enabled);
         check_path_length(file_path.to_string_lossy().as_ref())?;
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_PATH_CHECK, path_check_start);
 
         let read_data = opts.read_data;
 
-        let (data, _) = self
-            .read_raw(volume, volume_dir.clone(), file_path, read_data)
-            .await
-            .map_err(|e| {
-                if e == DiskError::FileNotFound && !version_id.is_empty() {
-                    DiskError::FileVersionNotFound
-                } else {
-                    e
-                }
-            })?;
+        let xlmeta_read_start = get_stage_timer_if_enabled(stage_metrics_enabled);
+        let raw_read_result = self.read_raw(volume, volume_dir.clone(), file_path, read_data).await;
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_XLMETA_READ, xlmeta_read_start);
+        let (data, _) = raw_read_result.map_err(|e| {
+            if e == DiskError::FileNotFound && !version_id.is_empty() {
+                DiskError::FileVersionNotFound
+            } else {
+                e
+            }
+        })?;
 
-        let mut fi = get_file_info(
-            &data,
-            volume,
-            path,
-            version_id,
-            FileInfoOpts {
-                data: read_data,
-                include_free_versions: opts.incl_free_versions,
-                include_part_checksums: false,
-            },
-        )?;
-
-        fi.validate_for_metadata_read()?;
+        let decode_start = get_stage_timer_if_enabled(stage_metrics_enabled);
+        let file_info_result: Result<FileInfo> = (|| {
+            let fi = get_file_info(
+                &data,
+                volume,
+                path,
+                version_id,
+                FileInfoOpts {
+                    data: read_data,
+                    include_free_versions: opts.incl_free_versions,
+                    include_part_checksums: false,
+                },
+            )?;
+            fi.validate_for_metadata_read()?;
+            Ok(fi)
+        })();
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_DECODE, decode_start);
+        let mut fi = file_info_result?;
         if fi.is_canonical_delete_marker() {
             return Ok(fi);
         }
@@ -10688,6 +10582,108 @@ mod test {
         meta.marshal_msg().expect("test metadata should encode")
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn read_version_records_local_metadata_stage_breakdown() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime should be created");
+        let recorder = crate::test_metrics::CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let dir = tempfile::tempdir().expect("temp dir should be created");
+                let endpoint =
+                    Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
+                let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+                let bucket = "bucket";
+                let object = "stage-breakdown";
+                ensure_test_volume(&disk, bucket).await;
+
+                let object_dir = dir.path().join(bucket).join(object);
+                fs::create_dir_all(&object_dir)
+                    .await
+                    .expect("object directory should be created");
+                fs::write(
+                    object_dir.join(STORAGE_FORMAT_FILE),
+                    test_meta(test_file_info(object, Uuid::new_v4(), None, Some(Bytes::from_static(b"inline")))),
+                )
+                .await
+                .expect("object metadata should be written");
+
+                disk.read_version(
+                    "",
+                    bucket,
+                    object,
+                    "",
+                    &ReadOptions {
+                        read_data: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("read_version should succeed");
+
+                let meta_object = "stage-breakdown-meta";
+                let meta_object_dir = dir.path().join(RUSTFS_META_BUCKET).join(meta_object);
+                fs::create_dir_all(&meta_object_dir)
+                    .await
+                    .expect("internal metadata object directory should be created");
+                fs::write(
+                    meta_object_dir.join(STORAGE_FORMAT_FILE),
+                    test_meta(test_file_info(meta_object, Uuid::new_v4(), None, Some(Bytes::from_static(b"meta")))),
+                )
+                .await
+                .expect("internal metadata should be written");
+
+                disk.read_version(
+                    "",
+                    RUSTFS_META_BUCKET,
+                    meta_object,
+                    "",
+                    &ReadOptions {
+                        read_data: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("internal metadata read_version should succeed");
+            });
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        for stage in [
+            GET_STAGE_READ_VERSION_PATH_RESOLVE,
+            GET_STAGE_READ_VERSION_PATH_CHECK,
+            GET_STAGE_READ_VERSION_XLMETA_READ,
+            GET_STAGE_READ_VERSION_DECODE,
+        ] {
+            assert_eq!(
+                recorder
+                    .histogram_values(
+                        "rustfs_io_get_object_stage_duration_seconds",
+                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX), ("stage", stage)]
+                    )
+                    .len(),
+                1,
+                "{stage} should be recorded once for user-bucket LocalDisk::read_version"
+            );
+            assert_eq!(
+                recorder
+                    .histogram_values(
+                        "rustfs_io_get_object_stage_duration_seconds",
+                        &[("path", GET_OBJECT_PATH_INTERNAL_META), ("stage", stage)]
+                    )
+                    .len(),
+                1,
+                "{stage} should be recorded once for internal-meta LocalDisk::read_version"
+            );
+        }
+    }
+
     #[test]
     fn inline_metadata_rollback_dir_avoids_real_data_dir_collision() {
         let target_version = Uuid::parse_str("11111111-2222-3333-4444-555555555555").expect("version id should parse");
@@ -18893,48 +18889,6 @@ mod test {
         assert!(matches!(disk.read_all(volume, &first_part).await, Err(DiskError::FileNotFound)));
     }
 
-    #[tokio::test]
-    async fn quota_mutation_fence_revoke_waits_for_active_claim_and_rejects_late_claims() {
-        use tempfile::tempdir;
-
-        let root_dir = tempdir().expect("temp dir should be created");
-        let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
-        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
-        let bucket = "quota-fence-volume";
-        let object = "object";
-        let fence_path = quota_mutation_fence_path(bucket, object);
-        let token = disk
-            .acquire_snapshot_lease(RUSTFS_META_BUCKET, &fence_path)
-            .await
-            .expect("quota mutation token should be prepared");
-        let claim = disk
-            .claim_quota_mutation_fence(bucket, object, token)
-            .await
-            .expect("prepared token should be claimable");
-
-        let release_disk = Arc::clone(&disk);
-        let mut release = tokio::spawn(async move {
-            release_disk
-                .release_snapshot_lease(RUSTFS_META_BUCKET, &fence_path, SnapshotLeaseToken::revoke_all())
-                .await
-        });
-        assert!(
-            tokio::time::timeout(Duration::from_millis(50), &mut release).await.is_err(),
-            "revoke must wait until an already claimed mutation has finished"
-        );
-
-        drop(claim);
-        tokio::time::timeout(Duration::from_secs(1), release)
-            .await
-            .expect("revoke should wake after the final claim drops")
-            .expect("revoke task should not panic")
-            .expect("revoke should succeed");
-        assert!(matches!(
-            disk.claim_quota_mutation_fence(bucket, object, token).await,
-            Err(DiskError::FileNotFound)
-        ));
-    }
-
     #[tokio::test]
     async fn delete_version_keeps_later_part_until_snapshot_release() {
         use tempfile::tempdir;

crates/ecstore/src/disk/mod.rs

@@ -72,28 +72,6 @@ use time::OffsetDateTime;
 use tokio::io::{AsyncRead, AsyncWrite};
 use uuid::Uuid;
 
-const QUOTA_MUTATION_FENCE_PREFIX: &str = "tmp/quota-mutation-fences/";
-pub(crate) const QUOTA_MUTATION_FENCE_METADATA_SUFFIX: &str = "quota-mutation-fence-token";
-
-pub(crate) fn quota_mutation_fence_path(bucket: &str, object: &str) -> String {
-    use sha2::{Digest, Sha256};
-
-    let mut input = Vec::with_capacity(bucket.len() + object.len() + 1);
-    input.extend_from_slice(bucket.as_bytes());
-    input.push(0);
-    input.extend_from_slice(object.as_bytes());
-    let digest = Sha256::digest(input);
-    format!(
-        "{QUOTA_MUTATION_FENCE_PREFIX}{}",
-        hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower)
-    )
-}
-
-pub(crate) fn is_quota_mutation_fence_path(path: &str) -> bool {
-    path.strip_prefix(QUOTA_MUTATION_FENCE_PREFIX)
-        .is_some_and(|digest| digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()))
-}
-
 pub type DiskStore = Arc<Disk>;
 
 pub type FileReader = Box<dyn AsyncRead + Send + Sync + Unpin>;
@@ -118,20 +96,6 @@ impl SnapshotLeaseToken {
     pub fn as_bytes(&self) -> &[u8; 16] {
         self.0.as_bytes()
     }
-
-    pub(crate) fn as_uuid(self) -> Uuid {
-        self.0
-    }
-
-    #[doc(hidden)]
-    pub fn revoke_all() -> Self {
-        Self(Uuid::nil())
-    }
-
-    #[doc(hidden)]
-    pub fn is_revoke_all(self) -> bool {
-        self.0.is_nil()
-    }
 }
 
 impl Default for SnapshotLeaseToken {

crates/ecstore/src/disk/os.rs

@@ -306,20 +306,12 @@ fn disk_namespace_mutation_lock(path: &Path) -> Arc<NamespaceMutationLock> {
 pub(crate) struct NamespaceMutationLease {
     _namespace_guard: OwnedMutexGuard<()>,
     _volume_guard: Option<OwnedRwLockReadGuard<()>>,
-    external_guard: Mutex<Option<Arc<dyn Send + Sync>>>,
-}
-
-impl NamespaceMutationLease {
-    pub(crate) fn attach_external_guard(&self, guard: Arc<dyn Send + Sync>) {
-        *self.external_guard.lock() = Some(guard);
-    }
 }
 
 async fn acquire_namespace_mutation_lease(path: &Path) -> Arc<NamespaceMutationLease> {
     Arc::new(NamespaceMutationLease {
         _namespace_guard: disk_namespace_mutation_lock(path).lock_owned().await,
         _volume_guard: None,
-        external_guard: Mutex::new(None),
     })
 }
 
@@ -335,7 +327,6 @@ pub(crate) async fn acquire_rename_data_mutation_lease(
     Arc::new(NamespaceMutationLease {
         _namespace_guard: namespace_guard,
         _volume_guard: Some(volume_guard),
-        external_guard: Mutex::new(None),
     })
 }
 

crates/ecstore/src/services/notification_sys.rs

@@ -44,14 +44,12 @@ const CONSECUTIVE_FAILURE_THRESHOLD: u32 = 3;
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
 const LOG_SUBSYSTEM_NOTIFICATION: &str = "notification";
 const EVENT_NOTIFICATION_PEER_PROPAGATION: &str = "notification_peer_propagation";
-const EVENT_NOTIFICATION_CAPABILITY_PROBE: &str = "notification_capability_probe";
 const SCANNER_ACTIVITY_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
 const TIER_CONFIG_RELOAD_RETRY_BASE: Duration = Duration::from_millis(100);
 const TIER_CONFIG_RELOAD_RETRY_CAP: Duration = Duration::from_secs(5);
 const REMOTE_VERSION_STATE_PROBE_INTERVAL: Duration = Duration::from_secs(10);
 const REMOTE_VERSION_STATE_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
 const REMOTE_VERSION_STATE_PROOF_TTL: Duration = Duration::from_secs(30);
-const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 1;
 
 /// Cached result from the last successful admin call to a peer.
 struct PeerAdminCache {
@@ -97,15 +95,15 @@ lazy_static! {
 }
 
 #[derive(Clone)]
-struct FleetCapabilityProof {
+struct RemoteVersionStateFleetProof {
     topology_fingerprint: String,
     peer_epochs: Arc<BTreeMap<String, Uuid>>,
     expires_at: Instant,
 }
 
-impl FleetCapabilityProof {
-    fn token(&self) -> FleetCapabilityProofToken {
-        FleetCapabilityProofToken {
+impl RemoteVersionStateFleetProof {
+    fn token(&self) -> RemoteVersionStateFleetProofToken {
+        RemoteVersionStateFleetProofToken {
             topology_fingerprint: self.topology_fingerprint.clone(),
             peer_epochs: self.peer_epochs.clone(),
         }
@@ -113,41 +111,37 @@ impl FleetCapabilityProof {
 }
 
 #[derive(Clone, PartialEq, Eq)]
-struct FleetCapabilityProofToken {
+pub(crate) struct RemoteVersionStateFleetProofToken {
     topology_fingerprint: String,
     peer_epochs: Arc<BTreeMap<String, Uuid>>,
 }
 
 #[derive(Default)]
-struct FleetCapabilityProofState {
-    proof: Option<FleetCapabilityProof>,
+struct RemoteVersionStateFleetProofState {
+    proof: Option<RemoteVersionStateFleetProof>,
     topology_conflict: bool,
 }
 
-#[derive(Clone, PartialEq, Eq)]
-pub(crate) struct RemoteVersionStateFleetProofToken(FleetCapabilityProofToken);
-
-#[derive(Clone, PartialEq, Eq)]
-pub struct CrossPoolFenceFleetProofToken(FleetCapabilityProofToken);
-
-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 REMOTE_VERSION_STATE_FLEET_PROOF: OnceLock<std::sync::RwLock<RemoteVersionStateFleetProofState>> = OnceLock::new();
 static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock<String> = OnceLock::new();
 
-fn cross_pool_fence_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
-    CROSS_POOL_FENCE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
+fn remote_version_state_fleet_proof_slot() -> &'static std::sync::RwLock<RemoteVersionStateFleetProofState> {
+    REMOTE_VERSION_STATE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(RemoteVersionStateFleetProofState::default()))
 }
 
-fn remote_version_state_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
-    REMOTE_VERSION_STATE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
+fn replace_remote_version_state_fleet_proof(proof: Option<RemoteVersionStateFleetProof>) {
+    replace_remote_version_state_fleet_proof_in(remote_version_state_fleet_proof_slot(), proof);
 }
 
-fn replace_fleet_capability_proof(slot: &std::sync::RwLock<FleetCapabilityProofState>, proof: Option<FleetCapabilityProof>) {
+fn replace_remote_version_state_fleet_proof_in(
+    slot: &std::sync::RwLock<RemoteVersionStateFleetProofState>,
+    proof: Option<RemoteVersionStateFleetProof>,
+) {
     slot.write().unwrap_or_else(std::sync::PoisonError::into_inner).proof = proof;
 }
 
-fn publish_fleet_capability_probe_result(
-    slot: &std::sync::RwLock<FleetCapabilityProofState>,
+fn publish_remote_version_state_probe_result(
+    slot: &std::sync::RwLock<RemoteVersionStateFleetProofState>,
     topology_fingerprint: &str,
     result: Result<BTreeMap<String, Uuid>>,
     observed_at: Instant,
@@ -161,7 +155,7 @@ fn publish_fleet_capability_probe_result(
                 .filter(|proof| proof.topology_fingerprint == topology_fingerprint && proof.peer_epochs.as_ref() == &peer_epochs)
                 .map(|proof| Arc::clone(&proof.peer_epochs))
                 .unwrap_or_else(|| Arc::new(peer_epochs));
-            state.proof = Some(FleetCapabilityProof {
+            state.proof = Some(RemoteVersionStateFleetProof {
                 topology_fingerprint: topology_fingerprint.to_string(),
                 peer_epochs,
                 expires_at: observed_at + REMOTE_VERSION_STATE_PROOF_TTL,
@@ -169,7 +163,7 @@ fn publish_fleet_capability_probe_result(
             None
         }
         Err(err) => {
-            replace_fleet_capability_proof(slot, None);
+            replace_remote_version_state_fleet_proof_in(slot, None);
             Some(err)
         }
     }
@@ -180,60 +174,27 @@ pub(crate) fn acquire_remote_version_state_fleet_proof() -> Option<RemoteVersion
     let state = remote_version_state_fleet_proof_slot()
         .read()
         .unwrap_or_else(std::sync::PoisonError::into_inner);
-    acquire_fleet_capability_proof_from(&state, expected_topology, Instant::now()).map(RemoteVersionStateFleetProofToken)
+    acquire_remote_version_state_fleet_proof_from(&state, expected_topology, Instant::now())
 }
 
-fn acquire_fleet_capability_proof_from(
-    state: &FleetCapabilityProofState,
+fn acquire_remote_version_state_fleet_proof_from(
+    state: &RemoteVersionStateFleetProofState,
     expected_topology: &str,
     now: Instant,
-) -> Option<FleetCapabilityProofToken> {
-    if state.topology_conflict || !fleet_capability_proof_valid_at(state.proof.as_ref(), expected_topology, now) {
+) -> Option<RemoteVersionStateFleetProofToken> {
+    if state.topology_conflict || !remote_version_state_fleet_proof_valid_at(state.proof.as_ref(), expected_topology, now) {
         return None;
     }
-    state.proof.as_ref().map(FleetCapabilityProof::token)
+    state.proof.as_ref().map(RemoteVersionStateFleetProof::token)
 }
 
 pub(crate) fn remote_version_state_fleet_proof_matches(proof: &RemoteVersionStateFleetProofToken) -> bool {
-    fleet_capability_proof_matches(remote_version_state_fleet_proof_slot(), &proof.0)
-}
-
-pub fn acquire_cross_pool_fence_fleet_proof() -> Option<CrossPoolFenceFleetProofToken> {
-    let expected_topology = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get()?;
-    let state = cross_pool_fence_fleet_proof_slot()
-        .read()
-        .unwrap_or_else(std::sync::PoisonError::into_inner);
-    acquire_fleet_capability_proof_from(&state, expected_topology, Instant::now()).map(CrossPoolFenceFleetProofToken)
-}
-
-pub fn cross_pool_fence_fleet_proof_matches(proof: &CrossPoolFenceFleetProofToken) -> bool {
-    fleet_capability_proof_matches(cross_pool_fence_fleet_proof_slot(), &proof.0)
-}
-
-#[cfg(any(test, feature = "test-util"))]
-pub fn rotate_cross_pool_fence_fleet_proof_for_test() -> bool {
-    let mut state = cross_pool_fence_fleet_proof_slot()
-        .write()
-        .unwrap_or_else(std::sync::PoisonError::into_inner);
-    let Some(current) = state.proof.as_ref() else {
-        return false;
-    };
-    state.proof = Some(FleetCapabilityProof {
-        topology_fingerprint: current.topology_fingerprint.clone(),
-        peer_epochs: Arc::new(current.peer_epochs.as_ref().clone()),
-        expires_at: current.expires_at,
-    });
-    true
-}
-
-fn fleet_capability_proof_matches(
-    slot: &std::sync::RwLock<FleetCapabilityProofState>,
-    proof: &FleetCapabilityProofToken,
-) -> bool {
     let Some(expected_topology) = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get() else {
         return false;
     };
-    let state = slot.read().unwrap_or_else(std::sync::PoisonError::into_inner);
+    let state = remote_version_state_fleet_proof_slot()
+        .read()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
     if state.topology_conflict {
         return false;
     }
@@ -245,7 +206,11 @@ fn fleet_capability_proof_matches(
     })
 }
 
-fn fleet_capability_proof_valid_at(proof: Option<&FleetCapabilityProof>, expected_topology: &str, now: Instant) -> bool {
+fn remote_version_state_fleet_proof_valid_at(
+    proof: Option<&RemoteVersionStateFleetProof>,
+    expected_topology: &str,
+    now: Instant,
+) -> bool {
     proof.is_some_and(|proof| proof.topology_fingerprint == expected_topology && now < proof.expires_at)
 }
 
@@ -259,11 +224,11 @@ fn insert_remote_version_state_peer(peer_epochs: &mut BTreeMap<String, Uuid>, pe
 pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
     if REMOTE_VERSION_STATE_PROBE_TOPOLOGY.set(topology_fingerprint.clone()).is_err() {
         if REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get() != Some(&topology_fingerprint) {
-            for slot in [remote_version_state_fleet_proof_slot(), cross_pool_fence_fleet_proof_slot()] {
-                let mut state = slot.write().unwrap_or_else(std::sync::PoisonError::into_inner);
-                state.topology_conflict = true;
-                state.proof = None;
-            }
+            let mut state = remote_version_state_fleet_proof_slot()
+                .write()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            state.topology_conflict = true;
+            state.proof = None;
         }
         return;
     }
@@ -284,23 +249,13 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 }
                 None => Err(Error::other("remote version state fleet capability notification system is unavailable")),
             };
-            let fence_result = match get_global_notification_sys() {
-                Some(notification_sys) => timeout(
-                    REMOTE_VERSION_STATE_PROBE_TIMEOUT,
-                    notification_sys.probe_cross_pool_fence_fleet(&topology_fingerprint),
-                )
-                .await
-                .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 topology_conflict = remote_version_state_fleet_proof_slot()
                 .read()
                 .unwrap_or_else(std::sync::PoisonError::into_inner)
                 .topology_conflict;
             if topology_conflict {
-                replace_fleet_capability_proof(remote_version_state_fleet_proof_slot(), None);
-                replace_fleet_capability_proof(cross_pool_fence_fleet_proof_slot(), None);
-            } else if let Some(err) = publish_fleet_capability_probe_result(
+                replace_remote_version_state_fleet_proof(None);
+            } else if let Some(err) = publish_remote_version_state_probe_result(
                 remote_version_state_fleet_proof_slot(),
                 &topology_fingerprint,
                 result,
@@ -308,24 +263,6 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
             ) {
                 debug!(error = %err, "remote version state fleet capability probe failed closed");
             }
-            if !topology_conflict
-                && let Some(err) = publish_fleet_capability_probe_result(
-                    cross_pool_fence_fleet_proof_slot(),
-                    &topology_fingerprint,
-                    fence_result,
-                    Instant::now(),
-                )
-            {
-                debug!(
-                    event = EVENT_NOTIFICATION_CAPABILITY_PROBE,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_NOTIFICATION,
-                    capability = "cross_pool_fence_v1",
-                    state = "failed_closed",
-                    error = %err,
-                    "notification capability probe"
-                );
-            }
             sleep(REMOTE_VERSION_STATE_PROBE_INTERVAL).await;
         }
     });
@@ -393,27 +330,6 @@ impl NotificationSys {
         }
         Ok(peer_epochs)
     }
-
-    async fn probe_cross_pool_fence_fleet(&self, topology_fingerprint: &str) -> Result<BTreeMap<String, Uuid>> {
-        if self.peer_clients.len() != self.peer_topology_hosts.len() {
-            return Err(Error::other("cross-pool fence capability fleet membership is incomplete"));
-        }
-        let probes = self.peer_clients.iter().map(|client| async {
-            let client = client
-                .as_ref()
-                .ok_or_else(|| Error::other("cross-pool fence capability peer is unreachable"))?;
-            client.probe_cross_pool_fence(topology_fingerprint.to_string()).await
-        });
-        let mut peer_epochs = BTreeMap::new();
-        for result in join_all(probes).await {
-            let (peer, version, epoch) = result?;
-            if version < CROSS_POOL_FENCE_SUPPORTED_VERSION {
-                return Err(Error::other("cross-pool fence capability version is unsupported"));
-            }
-            insert_remote_version_state_peer(&mut peer_epochs, peer, epoch)?;
-        }
-        Ok(peer_epochs)
-    }
 }
 
 pub struct NotificationPeerErr {
@@ -2229,16 +2145,16 @@ mod tests {
         let now = Instant::now();
         let mut peer_epochs = BTreeMap::new();
         peer_epochs.insert("peer-a".to_string(), Uuid::new_v4());
-        let proof = FleetCapabilityProof {
+        let proof = RemoteVersionStateFleetProof {
             topology_fingerprint: "topology-a".to_string(),
             peer_epochs: Arc::new(peer_epochs),
             expires_at: now + Duration::from_secs(1),
         };
 
-        assert!(fleet_capability_proof_valid_at(Some(&proof), "topology-a", now));
-        assert!(!fleet_capability_proof_valid_at(Some(&proof), "topology-b", now));
-        assert!(!fleet_capability_proof_valid_at(Some(&proof), "topology-a", proof.expires_at));
-        assert!(!fleet_capability_proof_valid_at(None, "topology-a", now));
+        assert!(remote_version_state_fleet_proof_valid_at(Some(&proof), "topology-a", now));
+        assert!(!remote_version_state_fleet_proof_valid_at(Some(&proof), "topology-b", now));
+        assert!(!remote_version_state_fleet_proof_valid_at(Some(&proof), "topology-a", proof.expires_at));
+        assert!(!remote_version_state_fleet_proof_valid_at(None, "topology-a", now));
     }
 
     #[test]
@@ -2252,25 +2168,25 @@ mod tests {
     #[test]
     fn remote_version_state_fleet_proof_accepts_single_node_membership() {
         let now = Instant::now();
-        let proof = FleetCapabilityProof {
+        let proof = RemoteVersionStateFleetProof {
             topology_fingerprint: "topology-a".to_string(),
             peer_epochs: Arc::new(BTreeMap::new()),
             expires_at: now + Duration::from_secs(1),
         };
 
-        assert!(fleet_capability_proof_valid_at(Some(&proof), "topology-a", now));
+        assert!(remote_version_state_fleet_proof_valid_at(Some(&proof), "topology-a", now));
     }
 
     #[test]
     fn remote_version_state_fleet_proof_token_changes_with_process_epoch() {
         let now = Instant::now();
-        let proof = FleetCapabilityProof {
+        let proof = RemoteVersionStateFleetProof {
             topology_fingerprint: "topology-a".to_string(),
             peer_epochs: Arc::new(BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())])),
             expires_at: now + Duration::from_secs(1),
         };
         let captured = proof.token();
-        let restarted = FleetCapabilityProof {
+        let restarted = RemoteVersionStateFleetProof {
             topology_fingerprint: proof.topology_fingerprint.clone(),
             peer_epochs: Arc::new(BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())])),
             expires_at: proof.expires_at,
@@ -2281,11 +2197,11 @@ mod tests {
 
     #[test]
     fn remote_version_state_fleet_proof_renewal_preserves_only_same_epoch_token() {
-        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let slot = std::sync::RwLock::new(RemoteVersionStateFleetProofState::default());
         let now = Instant::now();
         let epoch = Uuid::new_v4();
         let peers = BTreeMap::from([("peer-a".to_string(), epoch)]);
-        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(peers.clone()), now).is_none());
+        assert!(publish_remote_version_state_probe_result(&slot, "topology-a", Ok(peers.clone()), now).is_none());
         let original = slot
             .read()
             .expect("proof slot should not poison")
@@ -2294,7 +2210,9 @@ mod tests {
             .expect("successful probe should publish proof")
             .token();
 
-        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(peers), now + Duration::from_millis(1)).is_none());
+        assert!(
+            publish_remote_version_state_probe_result(&slot, "topology-a", Ok(peers), now + Duration::from_millis(1)).is_none()
+        );
         let renewed = slot
             .read()
             .expect("proof slot should not poison")
@@ -2306,7 +2224,8 @@ mod tests {
 
         let restarted = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
         assert!(
-            publish_fleet_capability_probe_result(&slot, "topology-a", Ok(restarted), now + Duration::from_millis(2)).is_none()
+            publish_remote_version_state_probe_result(&slot, "topology-a", Ok(restarted), now + Duration::from_millis(2))
+                .is_none()
         );
         let replaced = slot
             .read()
@@ -2321,18 +2240,18 @@ mod tests {
     #[test]
     fn remote_version_state_fleet_proof_conflict_revokes_atomic_snapshot() {
         let now = Instant::now();
-        let mut state = FleetCapabilityProofState {
-            proof: Some(FleetCapabilityProof {
+        let mut state = RemoteVersionStateFleetProofState {
+            proof: Some(RemoteVersionStateFleetProof {
                 topology_fingerprint: "topology-a".to_string(),
                 peer_epochs: Arc::new(BTreeMap::new()),
                 expires_at: now + Duration::from_secs(1),
             }),
             topology_conflict: false,
         };
-        assert!(acquire_fleet_capability_proof_from(&state, "topology-a", now).is_some());
+        assert!(acquire_remote_version_state_fleet_proof_from(&state, "topology-a", now).is_some());
 
         state.topology_conflict = true;
-        assert!(acquire_fleet_capability_proof_from(&state, "topology-a", now).is_none());
+        assert!(acquire_remote_version_state_fleet_proof_from(&state, "topology-a", now).is_none());
     }
 
     #[test]
@@ -2348,19 +2267,19 @@ mod tests {
 
     #[test]
     fn remote_version_state_fleet_probe_failure_revokes_previous_proof() {
-        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let slot = std::sync::RwLock::new(RemoteVersionStateFleetProofState::default());
         let now = Instant::now();
         let peer_epochs = BTreeMap::from([("node-a:9000".to_string(), Uuid::new_v4())]);
-        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(peer_epochs), now).is_none());
+        assert!(publish_remote_version_state_probe_result(&slot, "topology-a", Ok(peer_epochs), now).is_none());
         assert!(slot.read().expect("proof slot should not poison").proof.is_some());
 
         assert!(
-            publish_fleet_capability_probe_result(&slot, "topology-a", Err(Error::other("peer unavailable")), now,).is_some()
+            publish_remote_version_state_probe_result(&slot, "topology-a", Err(Error::other("peer unavailable")), now,).is_some()
         );
         assert!(slot.read().expect("proof slot should not poison").proof.is_none());
 
         let peer_epochs = BTreeMap::from([("node-a:9000".to_string(), Uuid::new_v4())]);
-        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(peer_epochs), now).is_none());
+        assert!(publish_remote_version_state_probe_result(&slot, "topology-a", Ok(peer_epochs), now).is_none());
         assert!(slot.read().expect("proof slot should not poison").proof.is_some());
     }
 

crates/ecstore/src/set_disk/mod.rs

@@ -719,8 +719,8 @@ pub(crate) use core::io_primitives::disk_call_counters;
 mod ctx;
 mod metadata;
 mod ops;
-#[cfg(any(test, feature = "test-util"))]
-pub use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause};
+#[cfg(test)]
+pub(crate) use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause};
 #[cfg(feature = "test-util")]
 pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier;
 pub(crate) use ops::object::body_cache_plaintext_len;
@@ -880,12 +880,44 @@ mod prepared_get_object_metadata_tests {
     use super::*;
     use crate::ecstore_validation_blackbox::make_local_set_disks;
     use crate::object_api::{BLOCK_SIZE_V2, PutObjReader};
-    use crate::set_disk::core::io_primitives::disk_call_counters;
+    use crate::set_disk::core::io_primitives::{bounded_metadata_fanout_order, disk_call_counters, rename_fanout_barrier};
     use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
+    use crate::test_metrics::CapturingRecorder;
     use http::HeaderMap;
     use tokio::io::AsyncReadExt;
 
+    const READ_VERSION_BARRIER_GUARD: std::time::Duration = std::time::Duration::from_secs(10);
+
+    fn object_with_initial_data_shards(bucket: &str, prefix: &str) -> String {
+        (0..1000)
+            .map(|index| format!("{prefix}-{index}.bin"))
+            .find(|name| {
+                let order = bounded_metadata_fanout_order(bucket, name, 4, 2);
+                let distribution = FileInfo::new(&[bucket, name].join("/"), 2, 2).erasure.distribution;
+                let mut seen = [false; 2];
+                for disk_index in order.into_iter().take(3) {
+                    if let Some(block_index @ 1..=2) = distribution.get(disk_index).copied() {
+                        seen[block_index - 1] = true;
+                    }
+                }
+                seen.into_iter().all(|seen| seen)
+            })
+            .expect("test should find an object whose initial fanout covers both data shards")
+    }
+
+    fn bounded_spare_disk_index(bucket: &str, object: &str) -> usize {
+        *bounded_metadata_fanout_order(bucket, object, 4, 2)
+            .get(3)
+            .expect("4-disk test geometry should leave one bounded spare disk")
+    }
+
+    fn bounded_slow_initial_disk_index(bucket: &str, object: &str) -> usize {
+        *bounded_metadata_fanout_order(bucket, object, 4, 2)
+            .get(2)
+            .expect("4-disk test geometry should include a third initial metadata disk")
+    }
+
     #[tokio::test]
     async fn prepared_metadata_is_consumed_exactly_once() {
         let snapshot = GetObjectFileInfo::owned(FileInfo::default(), Vec::new(), Vec::new());
@@ -1002,6 +1034,307 @@ mod prepared_get_object_metadata_tests {
         );
     }
 
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn inline_data_read_early_stop_reader_returns_exact_body() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "inline-data-read-early-stop-reader";
+        let object = object_with_initial_data_shards(bucket, "inline-data-read-early-stop-reader-object");
+        let payload = b"inline early-stop reader payload".repeat(256);
+        let recorder = CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        let (restored, object_size, calls_total) = metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                let opts = ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                };
+
+                set_disks
+                    .make_bucket(bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("bucket should be created");
+                let mut put_reader = PutObjReader::from_vec(payload.clone());
+                set_disks
+                    .put_object(bucket, &object, &mut put_reader, &opts)
+                    .await
+                    .expect("inline object should be written");
+
+                temp_env::async_with_vars(
+                    [
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                    ],
+                    async {
+                        let slow_initial_disk = bounded_slow_initial_disk_index(bucket, &object);
+                        let barrier =
+                            rename_fanout_barrier::arm(&object, slow_initial_disk, rename_fanout_barrier::PHASE_READ_VERSION);
+                        let calls = disk_call_counters::observe(&object);
+                        let set_disks_for_read = Arc::clone(&set_disks);
+                        let opts_for_read = opts.clone();
+                        let object_for_read = object.clone();
+                        let mut open_reader = tokio::spawn(async move {
+                            set_disks_for_read
+                                .get_object_reader(bucket, &object_for_read, None, HeaderMap::new(), &opts_for_read)
+                                .await
+                        });
+
+                        tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused())
+                            .await
+                            .expect("bounded inline GET should pause a slow initial metadata read");
+                        let mut reader = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut open_reader)
+                            .await
+                            .expect("production inline GET should return before the paused metadata response")
+                            .expect("inline GET reader task should not panic")
+                            .expect("inline GET reader should open");
+                        let object_size = reader.object_info.size;
+                        let mut restored = Vec::new();
+                        reader
+                            .stream
+                            .read_to_end(&mut restored)
+                            .await
+                            .expect("inline GET body should stream");
+
+                        (restored, object_size, calls.total(disk_call_counters::KIND_READ_VERSION))
+                    },
+                )
+                .await
+            })
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        assert_eq!(object_size, payload.len() as i64);
+        assert_eq!(restored, payload);
+        assert_eq!(calls_total, 4, "bounded production GET should schedule the initial quorum plus one spare");
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_scheduled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "bounded production GET should record all scheduled metadata tasks"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_completed",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![3.0],
+            "bounded production GET should record only observed metadata responses as completed"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_cancelled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![1.0],
+            "bounded production GET should record the aborted slow metadata task"
+        );
+    }
+
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn prepared_metadata_uses_full_fanout_even_when_data_read_early_stop_is_enabled() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "prepared-metadata-early-stop-enabled";
+        let object = object_with_initial_data_shards(bucket, "prepared-metadata-early-stop-enabled-object");
+        let payload = b"prepared metadata early-stop enabled payload".repeat(16);
+        let recorder = CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        let (restored, calls_total) = metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                let opts = ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                };
+
+                set_disks
+                    .make_bucket(bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("bucket should be created");
+                let mut put_reader = PutObjReader::from_vec(payload.clone());
+                set_disks
+                    .put_object(bucket, &object, &mut put_reader, &opts)
+                    .await
+                    .expect("object should be written");
+
+                temp_env::async_with_vars(
+                    [
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                    ],
+                    async {
+                        let calls = disk_call_counters::observe(&object);
+                        let metadata = set_disks
+                            .prepare_get_object_metadata(bucket, &object, &opts)
+                            .await
+                            .expect("prepared metadata should resolve");
+                        let calls_total = calls.total(disk_call_counters::KIND_READ_VERSION);
+
+                        let mut reader = set_disks
+                            .get_object_reader_with_prepared_metadata(bucket, &object, None, HeaderMap::new(), &opts, metadata)
+                            .await
+                            .expect("prepared body reader should open");
+                        let mut restored = Vec::new();
+                        reader
+                            .stream
+                            .read_to_end(&mut restored)
+                            .await
+                            .expect("prepared body should stream");
+                        (restored, calls_total)
+                    },
+                )
+                .await
+            })
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        assert_eq!(restored, payload);
+        assert_eq!(
+            calls_total, 4,
+            "prepared metadata must opt out of data-read early-stop until the read shape is known"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_scheduled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "prepared metadata should schedule the full metadata fanout"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_completed",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "prepared metadata must wait for every scheduled metadata response"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_cancelled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![0.0],
+            "prepared metadata must not cancel metadata responses"
+        );
+    }
+
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn data_read_early_stop_request_shapes_full_wait_in_production_reader() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "data-read-early-stop-shape-reader";
+        let payload = b"shape-gated inline reader payload".repeat(256);
+
+        for (object_prefix, range, configure_opts, expected_body) in [
+            (
+                "data-read-early-stop-range-reader-object",
+                Some(HTTPRangeSpec {
+                    start: 0,
+                    end: 3,
+                    is_suffix_length: false,
+                }),
+                None,
+                payload[..4].to_vec(),
+            ),
+            ("data-read-early-stop-part-reader-object", None, Some(1), payload.clone()),
+        ] {
+            let recorder = CapturingRecorder::default();
+            let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+            rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+            let (restored, calls_total) = metrics::with_local_recorder(&recorder, || {
+                runtime.block_on(async {
+                    let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                    let object = object_with_initial_data_shards(bucket, object_prefix);
+                    let mut opts = ObjectOptions {
+                        no_lock: true,
+                        ..Default::default()
+                    };
+                    opts.part_number = configure_opts;
+
+                    set_disks
+                        .make_bucket(bucket, &MakeBucketOptions::default())
+                        .await
+                        .expect("bucket should be created");
+                    let mut put_reader = PutObjReader::from_vec(payload.clone());
+                    set_disks
+                        .put_object(bucket, &object, &mut put_reader, &opts)
+                        .await
+                        .expect("inline object should be written");
+
+                    temp_env::async_with_vars(
+                        [
+                            ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                            ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                            ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                        ],
+                        async {
+                            let calls = disk_call_counters::observe(&object);
+                            let mut reader = set_disks
+                                .get_object_reader(bucket, &object, range, HeaderMap::new(), &opts)
+                                .await
+                                .expect("shape-gated GET reader should open");
+                            let mut restored = Vec::new();
+                            reader
+                                .stream
+                                .read_to_end(&mut restored)
+                                .await
+                                .expect("shape-gated GET body should stream");
+                            (restored, calls.total(disk_call_counters::KIND_READ_VERSION))
+                        },
+                    )
+                    .await
+                })
+            });
+            rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+            assert_eq!(restored, expected_body);
+            assert_eq!(calls_total, 4, "shape-gated production GET should keep full metadata fanout");
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_scheduled",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![4.0],
+                "shape-gated production GET should schedule the full metadata fanout"
+            );
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_completed",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![4.0],
+                "shape-gated production GET must wait for every scheduled metadata response"
+            );
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_cancelled",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![0.0],
+                "shape-gated production GET must not cancel metadata responses"
+            );
+        }
+    }
+
     #[tokio::test]
     #[serial_test::serial(body_cache_hook)]
     async fn prepared_reader_rebuilds_object_info_when_precomputed_value_is_absent() {
@@ -1105,7 +1438,7 @@ impl SetDisks {
         object: &str,
         opts: &ObjectOptions,
     ) -> Result<PreparedGetObjectMetadata> {
-        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, true).await?;
+        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, false).await?;
         let object_info = build_get_object_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended);
         Ok(PreparedGetObjectMetadata {
             snapshot,
@@ -3411,9 +3744,15 @@ fn collect_inline_data_shard_fileinfos_by_index<'a>(
         if block_index == 0 || block_index > data_shards {
             continue;
         }
+        if file_info.erasure.index != block_index {
+            continue;
+        }
         if !file_info.has_valid_erasure_geometry() {
             continue;
         }
+        if !core::io_primitives::metadata_early_stop_candidate_matches(file_info, fi) {
+            continue;
+        }
         if file_info.data.as_ref().is_none_or(|data| data.is_empty()) {
             continue;
         }
@@ -9221,6 +9560,9 @@ mod tests {
             HashAlgorithm::HighwayHash256S
         };
         let shards = erasure.encode_data(payload).expect("payload should encode");
+        let version_id = Some(Uuid::new_v4());
+        let data_dir = Some(Uuid::new_v4());
+        let mod_time = Some(OffsetDateTime::now_utc());
         let mut files = Vec::with_capacity(shards.len());
 
         for shard in shards {
@@ -9233,6 +9575,16 @@ mod tests {
             writer.shutdown().await.expect("inline writer should shutdown");
             let data = writer.into_inline_data().expect("inline data should be retained");
             let mut file = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
+            file.volume = "bucket".to_string();
+            file.name = "object".to_string();
+            file.size = i64::try_from(payload.len()).expect("test payload should fit i64");
+            file.is_latest = true;
+            file.version_id = version_id;
+            file.data_dir = data_dir;
+            file.mod_time = mod_time;
+            file.metadata.insert("etag".to_string(), "etag-inline".to_string());
+            file.add_object_part(1, "part-etag-inline".to_string(), payload.len(), file.mod_time, file.size, None, None);
+            file.set_inline_data();
             file.erasure.index = files.len() + 1;
             file.data = Some(Bytes::from(data));
             files.push(file);
@@ -9245,16 +9597,40 @@ mod tests {
         inline_bitrot_files_for_payload_with_mode(payload, false).await
     }
 
+    fn disk_ordered_fileinfos(files: &[FileInfo]) -> Vec<FileInfo> {
+        let distribution = &files
+            .first()
+            .expect("inline data shard fixture should include metadata")
+            .erasure
+            .distribution;
+        distribution
+            .iter()
+            .map(|block_index| {
+                files
+                    .get(block_index.checked_sub(1).expect("erasure block indexes are one-based"))
+                    .expect("inline data shard fixture should include every distributed shard")
+                    .clone()
+            })
+            .collect()
+    }
+
     fn inline_data_shard_fileinfo(
-        name: &str,
         data_blocks: usize,
         parity_blocks: usize,
         erasure_index: usize,
         distribution: &[usize],
         data: Option<&'static [u8]>,
     ) -> FileInfo {
-        let mut fi = FileInfo::new(name, data_blocks, parity_blocks);
-        fi.name = name.to_string();
+        let mut fi = FileInfo::new("object", data_blocks, parity_blocks);
+        fi.name = "object".to_string();
+        fi.volume = "bucket".to_string();
+        fi.size = 4;
+        fi.is_latest = true;
+        fi.data_dir = Some(Uuid::nil());
+        fi.mod_time = Some(OffsetDateTime::UNIX_EPOCH);
+        fi.metadata.insert("etag".to_string(), "etag-inline".to_string());
+        fi.add_object_part(1, "part-etag-inline".to_string(), 4, fi.mod_time, 4, None, None);
+        fi.set_inline_data();
         fi.erasure.index = erasure_index;
         fi.erasure.distribution = distribution.to_vec();
         fi.data = data.map(Bytes::from_static);
@@ -9264,36 +9640,41 @@ mod tests {
     #[test]
     fn collect_inline_data_shards_by_index_uses_distribution_order() {
         let distribution = vec![3, 1, 5, 2, 4, 6];
-        let mut fi = FileInfo::new("object", 4, 2);
+        let mut fi = inline_data_shard_fileinfo(4, 2, 1, &distribution, Some(b"x"));
+        fi.erasure.index = 1;
         fi.erasure.distribution = distribution.clone();
         let files = vec![
-            inline_data_shard_fileinfo("block-3", 4, 2, 3, &distribution, Some(b"c")),
-            inline_data_shard_fileinfo("block-1", 4, 2, 1, &distribution, Some(b"a")),
-            inline_data_shard_fileinfo("parity-5", 4, 2, 5, &distribution, Some(b"p")),
-            inline_data_shard_fileinfo("block-2", 4, 2, 2, &distribution, Some(b"b")),
-            inline_data_shard_fileinfo("block-4", 4, 2, 4, &distribution, Some(b"d")),
-            inline_data_shard_fileinfo("parity-6", 4, 2, 6, &distribution, Some(b"q")),
+            inline_data_shard_fileinfo(4, 2, 3, &distribution, Some(b"c")),
+            inline_data_shard_fileinfo(4, 2, 1, &distribution, Some(b"a")),
+            inline_data_shard_fileinfo(4, 2, 5, &distribution, Some(b"p")),
+            inline_data_shard_fileinfo(4, 2, 2, &distribution, Some(b"b")),
+            inline_data_shard_fileinfo(4, 2, 4, &distribution, Some(b"d")),
+            inline_data_shard_fileinfo(4, 2, 6, &distribution, Some(b"q")),
         ];
 
         let data_files =
             collect_inline_data_shard_fileinfos_by_index(&files, &fi, 4, |_| true).expect("all data shards should be collected");
 
         assert_eq!(
-            data_files.iter().map(|file| file.name.as_str()).collect::<Vec<_>>(),
-            ["block-1", "block-2", "block-3", "block-4"]
+            data_files
+                .iter()
+                .map(|file| file.data.as_deref().expect("fixture carries inline bytes"))
+                .collect::<Vec<_>>(),
+            [b"a".as_slice(), b"b".as_slice(), b"c".as_slice(), b"d".as_slice()]
         );
     }
 
     #[test]
     fn collect_inline_data_shards_by_index_rejects_missing_data_shard() {
         let distribution = vec![1, 2, 3, 4];
-        let mut fi = FileInfo::new("object", 2, 2);
+        let mut fi = inline_data_shard_fileinfo(2, 2, 1, &distribution, Some(b"x"));
+        fi.erasure.index = 1;
         fi.erasure.distribution = distribution.clone();
         let files = vec![
-            inline_data_shard_fileinfo("block-1", 2, 2, 1, &distribution, Some(b"a")),
-            inline_data_shard_fileinfo("block-2", 2, 2, 2, &distribution, None),
-            inline_data_shard_fileinfo("parity-3", 2, 2, 3, &distribution, Some(b"p")),
-            inline_data_shard_fileinfo("parity-4", 2, 2, 4, &distribution, Some(b"q")),
+            inline_data_shard_fileinfo(2, 2, 1, &distribution, Some(b"a")),
+            inline_data_shard_fileinfo(2, 2, 2, &distribution, None),
+            inline_data_shard_fileinfo(2, 2, 3, &distribution, Some(b"p")),
+            inline_data_shard_fileinfo(2, 2, 4, &distribution, Some(b"q")),
         ];
 
         assert!(collect_inline_data_shard_fileinfos_by_index(&files, &fi, 2, |_| true).is_none());
@@ -9426,10 +9807,8 @@ mod tests {
 
         let payload = vec![b'i'; 192 * 1024];
         let (erasure, files, _read_length, _checksum_algo) = inline_bitrot_files_for_payload(&payload).await;
-        let mut fi = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
-        fi.size = payload.len() as i64;
-        fi.data = files[0].data.clone();
-        fi.add_object_part(1, String::new(), payload.len(), None, payload.len() as i64, None, None);
+        let fi = files[0].clone();
+        let disk_files = disk_ordered_fileinfos(&files);
 
         let disks = vec![Some(disk); erasure.total_shard_count()];
         let metrics_size_bucket = rustfs_io_metrics::get_object_size_bucket(fi.size);
@@ -9438,7 +9817,7 @@ mod tests {
             "bucket",
             "object",
             &fi,
-            &files,
+            &disk_files,
             &disks,
             true,
             GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,
@@ -9469,10 +9848,8 @@ mod tests {
         let payload = vec![b'v'; 64 * 1024];
         let payload_size = i64::try_from(payload.len()).expect("test payload size should fit i64");
         let (erasure, files, _read_length, _checksum_algo) = inline_bitrot_files_for_payload(&payload).await;
-        let mut fi = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
-        fi.size = payload_size;
-        fi.data = files[0].data.clone();
-        fi.add_object_part(1, String::new(), payload.len(), None, payload_size, None, None);
+        let fi = files[0].clone();
+        let disk_files = disk_ordered_fileinfos(&files);
 
         let mut object_info = ObjectInfo {
             size: payload_size,
@@ -9503,7 +9880,7 @@ mod tests {
             "bucket",
             "object",
             &fi,
-            &files,
+            &disk_files,
             &vec![Some(disk); erasure.total_shard_count()],
             true,
             GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -22,12 +22,11 @@
 
 use super::super::*;
 use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards};
-use crate::bucket::quota::reservation;
 use crate::crash_inject::{self, CrashPoint};
 use crate::multipart_listing::paginate_multipart_listing;
 use futures::{StreamExt, stream};
 use std::future::Future;
-#[cfg(any(test, feature = "test-util"))]
+#[cfg(test)]
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::Duration;
 use tokio::task::JoinSet;
@@ -57,18 +56,17 @@ impl StaleMultipartCleanupGuard {
     }
 }
 
-#[cfg(any(test, feature = "test-util"))]
+#[cfg(test)]
 #[derive(Clone, Copy, PartialEq, Eq)]
-pub enum MultipartCommitPause {
+pub(crate) enum MultipartCommitPause {
     PutPartBeforeLockAcquire,
     PutPartBeforeLockLost,
     PutPartAfterRename,
     BeforeLockLost,
-    BeforeQuotaRename,
     AfterRename,
 }
 
-#[cfg(any(test, feature = "test-util"))]
+#[cfg(test)]
 struct MultipartCommitBarrierState {
     bucket: String,
     object: String,
@@ -79,22 +77,27 @@ struct MultipartCommitBarrierState {
     release: tokio::sync::Semaphore,
 }
 
-#[cfg(any(test, feature = "test-util"))]
-pub struct MultipartCommitBarrier {
+#[cfg(test)]
+pub(crate) struct MultipartCommitBarrier {
     state: Arc<MultipartCommitBarrierState>,
 }
 
-#[cfg(any(test, feature = "test-util"))]
+#[cfg(test)]
 static MULTIPART_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Arc<MultipartCommitBarrierState>>>> =
     std::sync::OnceLock::new();
 
-#[cfg(any(test, feature = "test-util"))]
+#[cfg(test)]
 impl MultipartCommitBarrier {
-    pub fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
+    pub(crate) fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
         Self::install_for_arrivals(bucket, object, pause, 1)
     }
 
-    pub fn install_for_arrivals(bucket: &str, object: &str, pause: MultipartCommitPause, expected_arrivals: usize) -> Self {
+    pub(crate) fn install_for_arrivals(
+        bucket: &str,
+        object: &str,
+        pause: MultipartCommitPause,
+        expected_arrivals: usize,
+    ) -> Self {
         assert!(expected_arrivals > 0, "multipart commit barrier must wait for at least one arrival");
         let state = Arc::new(MultipartCommitBarrierState {
             bucket: bucket.to_string(),
@@ -115,7 +118,7 @@ impl MultipartCommitBarrier {
         Self { state }
     }
 
-    pub async fn wait_until_paused(&self) {
+    pub(crate) async fn wait_until_paused(&self) {
         tokio::time::timeout(Duration::from_secs(30), async {
             loop {
                 let arrived = self.state.arrived.notified();
@@ -129,12 +132,12 @@ impl MultipartCommitBarrier {
         .expect("multipart completion should reach the deterministic commit barrier");
     }
 
-    pub fn release(&self) {
+    pub(crate) fn release(&self) {
         self.state.release.add_permits(self.state.expected_arrivals);
     }
 }
 
-#[cfg(any(test, feature = "test-util"))]
+#[cfg(test)]
 impl Drop for MultipartCommitBarrier {
     fn drop(&mut self) {
         self.release();
@@ -148,7 +151,7 @@ impl Drop for MultipartCommitBarrier {
     }
 }
 
-#[cfg(any(test, feature = "test-util"))]
+#[cfg(test)]
 async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartCommitPause) {
     let barrier = MULTIPART_COMMIT_BARRIER
         .get_or_init(|| std::sync::Mutex::new(None))
@@ -1870,24 +1873,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         fi.parts = Vec::with_capacity(uploaded_parts.len());
 
-        let quota_context = reservation::begin(
-            &self.ctx,
-            bucket,
-            object,
-            opts.quota_admission,
-            opts.data_movement,
-            self.pool_index,
-            self.set_index,
-        )
-        .await?;
-        let quota_mutation_fence = quota_context.is_enforced() || opts.quota_admission.is_some();
-        let preserve_replication_ciphertext = opts.replication_request
-            && contains_key_str(&fi.metadata, rustfs_utils::http::SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT);
-        if quota_context.is_enforced() && preserve_replication_ciphertext {
-            return Err(Error::PartMissingOrCorrupt);
-        }
-        let transformed_object = fi.is_compressed() || should_persist_encryption_original_size(&fi.metadata);
-
         let mut object_size: usize = 0;
         let mut object_actual_size: i64 = 0;
 
@@ -2014,23 +1999,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 checksum_combined.extend_from_slice(cs.raw.as_slice());
             }
 
-            object_size = object_size.checked_add(ext_part.size).ok_or(Error::PartMissingOrCorrupt)?;
-            if ext_part.actual_size < 0 && (!opts.replication_request || quota_context.is_enforced()) {
+            object_size += ext_part.size;
+            if opts.quota_admission.is_some() && ext_part.actual_size < 0 {
                 return Err(Error::PartMissingOrCorrupt);
             }
-            let normalized_actual_size = if ext_part.actual_size >= 0 && !transformed_object {
-                ext_part
-                    .actual_size
-                    .max(i64::try_from(ext_part.size).map_err(|_| Error::PartMissingOrCorrupt)?)
-            } else {
-                ext_part.actual_size
-            };
             object_actual_size = object_actual_size
-                .checked_add(normalized_actual_size)
+                .checked_add(ext_part.actual_size)
                 .ok_or(Error::PartMissingOrCorrupt)?;
-            let mut completed_part = completed_multipart_object_part(p.part_num, ext_part);
-            completed_part.actual_size = normalized_actual_size;
-            fi.parts.push(completed_part);
+
+            fi.parts.push(completed_multipart_object_part(p.part_num, ext_part));
         }
 
         if let Some(wtcs) = opts.want_checksum.as_ref() {
@@ -2057,34 +2034,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             }
         }
 
-        let declared_replication_actual_size = opts
-            .replication_request
-            .then(|| get_str(&opts.user_defined, SUFFIX_ACTUAL_OBJECT_SIZE_CAP))
-            .flatten();
-        let replication_actual_size = if opts.replication_request && quota_context.is_enforced() {
-            let observed_size = u64::try_from(object_actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
-            let declared_cap = declared_replication_actual_size
-                .as_deref()
-                .map(|value| value.parse::<u64>().map_err(|_| Error::PartMissingOrCorrupt))
-                .transpose()?
-                .unwrap_or(0);
-            let declared_encryption_size = rustfs_utils::http::get_object_encryption_original_size(&fi.metadata)
-                .map_err(Error::other)?
-                .map(u64::try_from)
-                .transpose()
-                .map_err(|_| Error::PartMissingOrCorrupt)?
-                .unwrap_or(0);
-            Some(observed_size.max(declared_cap).max(declared_encryption_size))
-        } else {
-            None
-        };
-        let quota_new_size = match replication_actual_size {
-            Some(size) => size.max(u64::try_from(object_size).map_err(|_| Error::PartMissingOrCorrupt)?),
-            None if quota_context.is_enforced() => u64::try_from(object_actual_size)
-                .map_err(|_| Error::PartMissingOrCorrupt)?
-                .max(u64::try_from(object_size).map_err(|_| Error::PartMissingOrCorrupt)?),
-            None => 0,
-        };
+        if let Some(admission) = opts.quota_admission {
+            let quota_operation_size = u64::try_from(object_actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
+            if quota_operation_size > admission.remaining() {
+                return Err(Error::QuotaExceeded {
+                    current: admission.current_usage(),
+                    limit: admission.quota_limit(),
+                });
+            }
+        }
         if let Some(rc_crc) = get_header_map(&opts.user_defined, SUFFIX_REPLICATION_SSEC_CRC) {
             if let Ok(rc_crc_bytes) = base64_simd::STANDARD.decode_to_vec(&rc_crc) {
                 fi.checksum = Some(Bytes::from(rc_crc_bytes));
@@ -2157,13 +2115,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                     .insert("x-rustfs-encryption-original-size".to_string(), actual_size.to_string());
             }
         } else if opts.replication_request {
-            if let Some(actual_size) = replication_actual_size {
-                insert_str(&mut fi.metadata, SUFFIX_ACTUAL_SIZE, actual_size.to_string());
-                if persist_encryption_original_size {
-                    fi.metadata
-                        .insert("x-rustfs-encryption-original-size".to_string(), actual_size.to_string());
-                }
-            } else if let Some(actual_size) = declared_replication_actual_size {
+            if let Some(actual_size) = get_str(&opts.user_defined, SUFFIX_ACTUAL_OBJECT_SIZE_CAP) {
                 insert_str(&mut fi.metadata, SUFFIX_ACTUAL_SIZE, actual_size.clone());
                 if persist_encryption_original_size {
                     fi.metadata
@@ -2344,288 +2296,157 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         }
         ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
-        let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
-
-        // Crash-consistency injection: hard power loss after the upload is fully
-        // staged and locked but before the authoritative rename_data commit. No
-        // disk has moved the staged data, so a crash here must leave any prior
-        // committed version byte-for-byte intact (rustfs/backlog#864) and the
-        // upload fully retryable. Compiles to a no-op outside `#[cfg(test)]`.
-        if crash_inject::should_crash_at(CrashPoint::MultipartBeforeCommitRename, object) {
-            return Err(StorageError::Unexpected);
-        }
-
-        let quota_old_size = if quota_context.is_enforced() {
-            if opts.data_movement {
-                quota_new_size
-            } else {
-                reservation::replaced_logical_size(&self, bucket, object, opts).await?
-            }
-        } else {
-            0
-        };
-        let mut quota_reservation = quota_context.reserve(quota_old_size, quota_new_size).await?;
-        let (commit_disks, quota_fence_tokens) = if quota_mutation_fence {
-            match Self::prepare_quota_mutation_fences(&shuffle_disks, bucket, object, write_quorum).await {
-                Ok((disks, tokens)) => {
-                    for (metadata, token) in parts_metadatas.iter_mut().zip(tokens.iter().copied()) {
-                        if let Some(token) = token {
-                            insert_str(
-                                &mut metadata.metadata,
-                                crate::disk::QUOTA_MUTATION_FENCE_METADATA_SUFFIX,
-                                token.as_uuid().to_string(),
-                            );
-                        }
-                    }
-                    (disks, tokens)
-                }
-                Err(err) => {
-                    quota_reservation.abort().await;
-                    return Err(err);
-                }
-            }
-        } else {
-            (shuffle_disks.clone(), vec![None; shuffle_disks.len()])
-        };
-        if quota_reservation.is_lock_lost()
-            || !quota_reservation.capability_proof_matches()
-            || object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
-            || upload_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
-            || opts
-                .namespace_lock_fence
-                .as_ref()
-                .is_some_and(NamespaceLockFence::is_lock_lost)
-        {
-            Self::abort_quota_reservation_after_fence(
-                quota_reservation,
-                &commit_disks,
-                &quota_fence_tokens,
-                bucket,
-                object,
-                write_quorum,
-                quota_mutation_fence,
-            )
-            .await;
-            return Err(StorageError::NamespaceLockQuorumUnavailable {
-                mode: "quota_reservation",
-                bucket: bucket.to_string(),
-                object: object.to_string(),
-                required: 1,
-                achieved: 0,
-            });
-        }
-        if let Err(err) = ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts) {
-            Self::abort_quota_reservation_after_fence(
-                quota_reservation,
-                &commit_disks,
-                &quota_fence_tokens,
-                bucket,
-                object,
-                write_quorum,
-                quota_mutation_fence,
-            )
-            .await;
-            return Err(err);
-        }
-        if let Err(err) = self
-            .require_current_restore_operation_id(
-                bucket,
-                object,
-                opts,
-                expected_restore_operation_id,
-                "complete_multipart_upload_quota_reservation",
-            )
-            .await
-        {
-            Self::abort_quota_reservation_after_fence(
-                quota_reservation,
-                &commit_disks,
-                &quota_fence_tokens,
-                bucket,
-                object,
-                write_quorum,
-                quota_mutation_fence,
-            )
-            .await;
-            return Err(err);
-        }
-
-        if let Err(err) = quota_reservation.mark_commit_started().await {
-            Self::abort_quota_reservation_after_fence(
-                quota_reservation,
-                &commit_disks,
-                &quota_fence_tokens,
-                bucket,
-                object,
-                write_quorum,
-                quota_mutation_fence,
-            )
-            .await;
-            return Err(err);
-        }
-        #[cfg(any(test, feature = "test-util"))]
-        pause_multipart_commit(bucket, object, MultipartCommitPause::BeforeQuotaRename).await;
-        if quota_reservation.is_lock_lost()
-            || !quota_reservation.capability_proof_matches()
-            || object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
-            || upload_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
-            || opts
-                .namespace_lock_fence
-                .as_ref()
-                .is_some_and(NamespaceLockFence::is_lock_lost)
-            || opts
-                .bucket_lifecycle_lock_fence
-                .as_ref()
-                .is_some_and(NamespaceLockFence::is_lock_lost)
-        {
-            Self::abort_quota_reservation_after_fence(
-                quota_reservation,
-                &commit_disks,
-                &quota_fence_tokens,
-                bucket,
-                object,
-                write_quorum,
-                quota_mutation_fence,
-            )
-            .await;
-            return Err(StorageError::NamespaceLockQuorumUnavailable {
-                mode: "quota_reservation",
-                bucket: bucket.to_string(),
-                object: object.to_string(),
-                required: 1,
-                achieved: 0,
-            });
-        }
-        let rename_result = Self::rename_data(
-            &commit_disks,
-            RUSTFS_META_MULTIPART_BUCKET,
-            &upload_id_path,
-            &parts_metadatas,
-            bucket,
-            object,
-            write_quorum,
-        )
-        .await;
-        if quota_mutation_fence {
-            let _ = Self::release_quota_mutation_fences(&commit_disks, &quota_fence_tokens, bucket, object, write_quorum).await;
-        }
-        if rename_result.is_ok() {
-            quota_reservation.commit().await;
-        }
-        let (online_disks, convergence, op_old_dir, cleanup_disks, _) = match rename_result {
-            Ok(result) => result,
-            Err(err) => return Err(err.into()),
-        };
-
-        // Detach admission before any post-commit await: client cancellation
-        // must not couple durable convergence repair to cleanup work.
-        if convergence.needs_heal() {
-            let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
-                bucket.to_string(),
-                Some(object.to_string()),
-                false,
-                Some(HealChannelPriority::Normal),
-                Some(self.pool_index),
-                Some(self.set_index),
-            );
-            request.object_version_id = fi
-                .version_id
-                .or_else(|| opts.version_suspended.then(Uuid::nil))
-                .map(|version_id| version_id.to_string());
-            tokio::spawn(async move {
-                let _ = rustfs_common::heal_channel::send_heal_request(request).await;
-            });
-        }
-
-        // Crash-consistency injection: hard power loss after the authoritative
-        // rename_data commit succeeded but before the stale part.N.meta cleanup.
-        // The new version is durably committed and visible, so a crash here must
-        // leave the object readable as the new version; the un-reclaimed staging
-        // parts are swept by a retried completion or upload GC (rustfs/backlog#946).
-        // Compiles to a no-op outside `#[cfg(test)]`.
-        if crash_inject::should_crash_at(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object) {
-            return Err(StorageError::Unexpected);
-        }
-
-        // backlog#946: reclaim the stale per-part metadata (and any superfluous
-        // part.N data files no longer in the completed set) only *after* the
-        // authoritative rename_data commit above has succeeded. If rename_data
-        // fails write quorum and returns via `?`, the upload directory must keep
-        // its part.N.meta so a retried CompleteMultipartUpload can still read the
-        // parts; deleting them before the commit would strand the upload
-        // permanently. This mirrors the "clean up only after commit" pattern
-        // already used for the old data-dir GC and the upload-dir delete_all below.
-        self.cleanup_multipart_path(&parts).await;
-
-        if let Some(old_dir) = op_old_dir {
-            let committed_dir = fi.data_dir.unwrap_or_default().to_string();
-            // backlog#898: best-effort reclaim of the dereferenced old data dir.
-            // Returns a receipt (never `Err`); a failed GC must not turn an
-            // already-committed multipart completion into a 503.
-            let cleanup = self
-                .commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), &committed_dir, write_quorum)
-                .await;
-            self.report_old_data_dir_cleanup(bucket, object, &old_dir.to_string(), &cleanup)
-                .await;
-        }
-
-        if let Some(stage_start) = complete_tail_stage_start {
-            rustfs_io_metrics::record_put_object_stage_duration(
-                "multipart_complete_tail",
-                stage_start.elapsed().as_secs_f64() * 1000.0,
-            );
-        }
-
-        #[cfg(test)]
-        pause_multipart_commit(bucket, object, MultipartCommitPause::AfterRename).await;
-
-        let cleanup_store = self.clone();
-        let cleanup_upload_id_path = upload_id_path.clone();
-        let cleanup_bucket = bucket.to_owned();
-        let cleanup_object = object.to_owned();
-        let cleanup_upload_id = upload_id.to_owned();
-        let cleanup_handle = tokio::spawn(async move {
+        let commit_set = self.clone();
+        let commit_bucket = bucket.to_owned();
+        let commit_object = object.to_owned();
+        let commit_upload_id = upload_id.to_owned();
+        let commit_upload_id_path = upload_id_path.clone();
+        let commit_version_suspended = opts.version_suspended;
+        let commit_is_versioned = opts.versioned || opts.version_suspended;
+        let commit_capacity_scope_token = opts.capacity_scope_token;
+        let commit_object_lock_guard = object_lock_guard.take();
+        let detach_commit_owner = commit_object_lock_guard.is_some() || upload_guard.is_some();
+        let commit = async move {
+            let _object_lock_guard = commit_object_lock_guard;
             let _upload_guard = upload_guard;
-            if let Err(err) = cleanup_store
-                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &cleanup_upload_id_path, write_quorum)
+            let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
+
+            // Crash-consistency injection: hard power loss after the upload is fully
+            // staged and locked but before the authoritative rename_data commit. No
+            // disk has moved the staged data, so a crash here must leave any prior
+            // committed version byte-for-byte intact (rustfs/backlog#864) and the
+            // upload fully retryable. Compiles to a no-op outside `#[cfg(test)]`.
+            if crash_inject::should_crash_at(CrashPoint::MultipartBeforeCommitRename, &commit_object) {
+                return Err(StorageError::Unexpected);
+            }
+
+            // The trailing `_` drops the rename_data old-size backfill
+            // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit
+            // `get_object_info` lookup, so the backfill has no consumer here yet.
+            let (online_disks, convergence, op_old_dir, cleanup_disks, _) = SetDisks::rename_data(
+                &shuffle_disks,
+                RUSTFS_META_MULTIPART_BUCKET,
+                &commit_upload_id_path,
+                &parts_metadatas,
+                &commit_bucket,
+                &commit_object,
+                write_quorum,
+            )
+            .await?;
+
+            // Detach admission before any post-commit await: client cancellation
+            // must not couple durable convergence repair to cleanup work.
+            if convergence.needs_heal() {
+                let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
+                    commit_bucket.clone(),
+                    Some(commit_object.clone()),
+                    false,
+                    Some(HealChannelPriority::Normal),
+                    Some(commit_set.pool_index),
+                    Some(commit_set.set_index),
+                );
+                request.object_version_id = fi
+                    .version_id
+                    .or_else(|| commit_version_suspended.then(Uuid::nil))
+                    .map(|version_id| version_id.to_string());
+                tokio::spawn(async move {
+                    let _ = rustfs_common::heal_channel::send_heal_request(request).await;
+                });
+            }
+
+            // Crash-consistency injection: hard power loss after the authoritative
+            // rename_data commit succeeded but before the stale part.N.meta cleanup.
+            // The new version is durably committed and visible, so a crash here must
+            // leave the object readable as the new version; the un-reclaimed staging
+            // parts are swept by a retried completion or upload GC (rustfs/backlog#946).
+            // Compiles to a no-op outside `#[cfg(test)]`.
+            if crash_inject::should_crash_at(CrashPoint::MultipartAfterCommitBeforePartsCleanup, &commit_object) {
+                return Err(StorageError::Unexpected);
+            }
+
+            // backlog#946: reclaim the stale per-part metadata (and any superfluous
+            // part.N data files no longer in the completed set) only *after* the
+            // authoritative rename_data commit above has succeeded. If rename_data
+            // fails write quorum and returns via `?`, the upload directory must keep
+            // its part.N.meta so a retried CompleteMultipartUpload can still read the
+            // parts; deleting them before the commit would strand the upload
+            // permanently. This mirrors the "clean up only after commit" pattern
+            // already used for the old data-dir GC and the upload-dir delete_all below.
+            commit_set.cleanup_multipart_path(&parts).await;
+
+            if let Some(old_dir) = op_old_dir {
+                let committed_dir = fi.data_dir.unwrap_or_default().to_string();
+                // backlog#898: best-effort reclaim of the dereferenced old data dir.
+                // Returns a receipt (never `Err`); a failed GC must not turn an
+                // already-committed multipart completion into a 503.
+                let cleanup = commit_set
+                    .commit_rename_data_dir(
+                        &cleanup_disks,
+                        &commit_bucket,
+                        &commit_object,
+                        &old_dir.to_string(),
+                        &committed_dir,
+                        write_quorum,
+                    )
+                    .await;
+                commit_set
+                    .report_old_data_dir_cleanup(&commit_bucket, &commit_object, &old_dir.to_string(), &cleanup)
+                    .await;
+            }
+
+            if let Some(stage_start) = complete_tail_stage_start {
+                rustfs_io_metrics::record_put_object_stage_duration(
+                    "multipart_complete_tail",
+                    stage_start.elapsed().as_secs_f64() * 1000.0,
+                );
+            }
+
+            #[cfg(test)]
+            pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::AfterRename).await;
+
+            if let Err(err) = commit_set
+                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path, write_quorum)
                 .await
             {
                 warn!(
-                    bucket = %cleanup_bucket,
-                    object = %cleanup_object,
-                    upload_id = %cleanup_upload_id,
+                    bucket = %commit_bucket,
+                    object = %commit_object,
+                    upload_id = %commit_upload_id,
                     error = ?err,
                     "completed multipart upload staging cleanup did not reach write quorum"
                 );
             }
-        });
-        if let Err(err) = cleanup_handle.await {
-            warn!(
-                bucket = %bucket,
-                object = %object,
-                upload_id = %upload_id,
-                error = ?err,
-                "completed multipart upload staging cleanup task failed"
-            );
-        }
-        drop(object_lock_guard); // drop object lock guard to release the lock
 
-        for (i, op_disk) in online_disks.iter().enumerate() {
-            if let Some(disk) = op_disk
-                && disk.is_online().await
-            {
-                fi = parts_metadatas[i].clone();
-                break;
+            for (i, op_disk) in online_disks.iter().enumerate() {
+                if let Some(disk) = op_disk
+                    && disk.is_online().await
+                {
+                    fi = parts_metadatas[i].clone();
+                    break;
+                }
             }
+
+            commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks);
+
+            fi.is_latest = true;
+
+            commit_set
+                .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
+                .await;
+
+            drop(_object_lock_guard); // drop object lock guard to release the lock
+            drop(_upload_guard);
+
+            Ok(ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned))
+        };
+
+        if detach_commit_owner {
+            tokio::spawn(commit)
+                .await
+                .map_err(|err| Error::other(format!("complete_multipart_upload commit task failed: {err}")))?
+        } else {
+            commit.await
         }
-
-        self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks);
-
-        fi.is_latest = true;
-
-        self.invalidate_get_object_metadata_cache(bucket, object).await;
-
-        Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
     }
 }
 
@@ -3079,7 +2900,7 @@ mod tests {
         let (upload_id, parts) =
             stage_upload_with_create_opts(&set_disks, bucket, object, &payload, &ObjectOptions::default()).await;
         let mut denied_opts = ObjectOptions::default();
-        assert!(denied_opts.set_quota_admission(100, 4180));
+        assert!(denied_opts.set_quota_admission(100, 4195));
 
         let err = set_disks
             .clone()
@@ -3090,7 +2911,7 @@ mod tests {
             err,
             StorageError::QuotaExceeded {
                 current: 100,
-                limit: 4180
+                limit: 4195
             }
         ));
 
@@ -3108,7 +2929,7 @@ mod tests {
         );
 
         let mut allowed_opts = ObjectOptions::default();
-        assert!(allowed_opts.set_quota_admission(100, 4181));
+        assert!(allowed_opts.set_quota_admission(100, 4196));
         let completed = set_disks
             .clone()
             .complete_multipart_upload(bucket, object, &upload_id, parts, &allowed_opts)
@@ -3149,120 +2970,6 @@ mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn replication_quota_uses_server_observed_part_size_as_lower_bound() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "multipart-replication-quota-bucket";
-        let object = "object";
-        make_bucket_on_all(&disk_stores, bucket).await;
-
-        let mut create_opts = ObjectOptions::default();
-        insert_str(&mut create_opts.user_defined, SUFFIX_COMPRESSION, "S2".to_string());
-        let upload = set_disks
-            .new_multipart_upload(bucket, object, &create_opts)
-            .await
-            .expect("replication multipart upload should be created");
-        let part = put_test_part(&set_disks, bucket, object, &upload.upload_id, 1, &[0x72; 4096], 1).await;
-        let mut complete_opts = ObjectOptions {
-            replication_request: true,
-            ..Default::default()
-        };
-        assert!(complete_opts.set_quota_admission(0, 4095));
-
-        let err = set_disks
-            .clone()
-            .complete_multipart_upload(bucket, object, &upload.upload_id, vec![part.clone()], &complete_opts)
-            .await
-            .expect_err("a forged tiny replication logical size must not reduce quota admission");
-        assert!(matches!(err, StorageError::QuotaExceeded { current: 0, limit: 4095 }));
-        assert!(
-            set_disks
-                .check_upload_id_exists(bucket, object, &upload.upload_id, false)
-                .await
-                .is_ok(),
-            "quota rejection must leave replicated multipart parts retryable"
-        );
-
-        assert!(complete_opts.set_quota_admission(0, 4096));
-        let completed = set_disks
-            .clone()
-            .complete_multipart_upload(bucket, object, &upload.upload_id, vec![part], &complete_opts)
-            .await
-            .expect("the physical safety boundary should admit the transformed replica");
-        assert_eq!(completed.get_actual_size().expect("replica logical size should parse"), 1);
-    }
-
-    #[tokio::test]
-    async fn direct_multipart_quota_uses_server_observed_part_size_as_lower_bound() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "multipart-direct-quota-bucket";
-        let object = "object";
-        make_bucket_on_all(&disk_stores, bucket).await;
-
-        let upload = set_disks
-            .new_multipart_upload(bucket, object, &ObjectOptions::default())
-            .await
-            .expect("multipart upload should be created");
-        let part = put_test_part(&set_disks, bucket, object, &upload.upload_id, 1, &[0x73; 4096], 1).await;
-        let mut complete_opts = ObjectOptions::default();
-        assert!(complete_opts.set_quota_admission(0, 4095));
-
-        let err = set_disks
-            .clone()
-            .complete_multipart_upload(bucket, object, &upload.upload_id, vec![part], &complete_opts)
-            .await
-            .expect_err("a forged tiny direct logical size must not reduce quota admission");
-        assert!(matches!(err, StorageError::QuotaExceeded { current: 0, limit: 4095 }));
-        assert!(
-            set_disks
-                .check_upload_id_exists(bucket, object, &upload.upload_id, false)
-                .await
-                .is_ok(),
-            "quota rejection must leave direct multipart parts retryable"
-        );
-    }
-
-    #[tokio::test]
-    async fn quota_rejects_ciphertext_replication_without_a_server_observed_logical_size() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "multipart-compressed-ciphertext-quota-bucket";
-        let object = "object";
-        make_bucket_on_all(&disk_stores, bucket).await;
-
-        let mut create_opts = ObjectOptions::default();
-        insert_str(&mut create_opts.user_defined, SUFFIX_COMPRESSION, "S2".to_string());
-        insert_str(
-            &mut create_opts.user_defined,
-            rustfs_utils::http::SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT,
-            "true".to_string(),
-        );
-        let upload = set_disks
-            .new_multipart_upload(bucket, object, &create_opts)
-            .await
-            .expect("ciphertext multipart upload should be created");
-        let payload = vec![0x74; 4096];
-        let part = put_test_part(&set_disks, bucket, object, &upload.upload_id, 1, &payload, 1).await;
-        let mut complete_opts = ObjectOptions {
-            replication_request: true,
-            ..Default::default()
-        };
-        assert!(complete_opts.set_quota_admission(0, u64::MAX));
-
-        let err = set_disks
-            .clone()
-            .complete_multipart_upload(bucket, object, &upload.upload_id, vec![part], &complete_opts)
-            .await
-            .expect_err("ciphertext replication has no server-observed logical quota size");
-        assert!(matches!(err, StorageError::PartMissingOrCorrupt));
-        assert!(
-            set_disks
-                .check_upload_id_exists(bucket, object, &upload.upload_id, false)
-                .await
-                .is_ok(),
-            "rejection must leave ciphertext multipart parts retryable"
-        );
-    }
-
     #[tokio::test]
     async fn complete_multipart_quota_rejects_invalid_logical_sizes() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
@@ -5193,6 +4900,74 @@ mod tests {
         .await;
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn cancelled_complete_keeps_upload_lock_through_tail_cleanup() {
+        temp_env::async_with_vars(
+            [
+                (crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true")),
+                (rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("60")),
+            ],
+            async {
+                let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+                let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+                let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+                let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+                let bucket = "multipart-cancelled-tail-lock-bucket";
+                let object = "object";
+                make_bucket_on_all(&disk_stores, bucket).await;
+                let (upload_id, parts) =
+                    stage_upload_with_create_opts(&set_disks, bucket, object, &[0x53; 4096], &ObjectOptions::default()).await;
+                let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+                signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path));
+                signaling.clear_observed();
+                let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+                let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::AfterRename);
+
+                let complete_store = set_disks.clone();
+                let complete_upload_id = upload_id.clone();
+                let complete = tokio::spawn(async move {
+                    complete_store
+                        .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default())
+                        .await
+                });
+                barrier.wait_until_paused().await;
+
+                let abort_store = set_disks.clone();
+                let abort_upload_id = upload_id.clone();
+                let abort = tokio::spawn(async move {
+                    abort_store
+                        .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
+                        .await
+                });
+                signaling.wait_for_attempts(2).await;
+                tokio::task::yield_now().await;
+                assert!(!abort.is_finished(), "abort must wait while completion tail owns the upload lock");
+
+                complete.abort();
+                assert!(
+                    complete
+                        .await
+                        .expect_err("the completion request should be cancellable while the tail is paused")
+                        .is_cancelled()
+                );
+                tokio::task::yield_now().await;
+                assert!(!abort.is_finished(), "cancelling the completion waiter must not release the upload lock");
+
+                barrier.release();
+                let abort_err = abort
+                    .await
+                    .expect("abort task should not panic")
+                    .expect_err("the committed upload should no longer exist when abort acquires the lock");
+                assert!(
+                    matches!(abort_err, StorageError::InvalidUploadID(..)),
+                    "abort should return InvalidUploadID after the detached completion tail, got {abort_err:?}"
+                );
+            },
+        )
+        .await;
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn complete_validates_parts_after_an_inflight_upload_part_commit() {

crates/ecstore/src/store/init.rs

@@ -1182,32 +1182,6 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    #[serial_test::serial(storage_class_env)]
-    async fn quota_object_fence_ignores_an_unrelated_offline_pool() {
-        let temp_dir = tempfile::tempdir().expect("create quota fence store dir");
-        let (_ctx, store, shutdown) =
-            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "quota-object-fence", &[4, 4])).await;
-        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
-        let bucket = format!("quota-object-fence-{}", uuid::Uuid::new_v4());
-        let object = "object.bin";
-        store
-            .make_bucket(&bucket, &MakeBucketOptions::default())
-            .await
-            .expect("create quota fence bucket");
-        store.pools[1].disk_set[0].disks.write().await.fill(None);
-
-        crate::bucket::quota::reservation::fence_namespace_mutations_for_test(&store, &bucket, object, Some((0, 0)))
-            .await
-            .expect("the selected pool fence should ignore an unrelated offline pool");
-        let err = crate::bucket::quota::reservation::fence_namespace_mutations_for_test(&store, &bucket, object, None)
-            .await
-            .expect_err("legacy reservations must conservatively fence every pool");
-        assert!(matches!(err, StorageError::ErasureWriteQuorum));
-
-        shutdown.cancel();
-    }
-
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]
     async fn tag_updates_skip_active_rebalance_source_pool() {

crates/ecstore/src/store/list_objects.rs

@@ -641,7 +641,7 @@ pub(crate) fn observe_scanner_namespace_mutations(bucket: &str, delta: u64) {
         .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| Some(current.saturating_add(delta)));
 }
 
-pub(crate) async fn observe_list_objects_mutation(store: &ECStore, bucket: &str) -> u64 {
+pub(super) async fn observe_list_objects_mutation(store: &ECStore, bucket: &str) -> u64 {
     observe_list_objects_mutations(store, bucket, 1).await.unwrap_or_default()
 }
 

crates/ecstore/src/store/object.rs

@@ -4741,10 +4741,9 @@ mod tests {
     #[tokio::test]
     #[serial_test::serial(body_cache_hook)]
     async fn select_snapshot_rejects_latest_versioned_delete_marker_during_prepare() {
-        let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
-        let (_first_dirs, first_set) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await;
-        let (_second_dirs, second_set) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await;
-        let store = new_prepared_reader_test_store_with_ctx(&[Arc::clone(&first_set), Arc::clone(&second_set)], ctx).await;
+        let (_first_dirs, first_set) = make_local_set_disks(4, 2).await;
+        let (_second_dirs, second_set) = make_local_set_disks(4, 2).await;
+        let store = new_prepared_reader_test_store(&[Arc::clone(&first_set), Arc::clone(&second_set)]).await;
         let bucket = "select-snapshot-latest-delete-marker";
         let object = "versioned-object.bin";
         let versioned_opts = ObjectOptions {

crates/heal/tests/endpoint_index_test.rs

@@ -14,6 +14,8 @@
 
 //! test endpoint index settings
 
+#![recursion_limit = "256"]
+
 use std::net::SocketAddr;
 use tempfile::TempDir;
 use tokio_util::sync::CancellationToken;

crates/heal/tests/heal_b5_versioned_regression_test.rs

@@ -22,6 +22,8 @@
 //! bucket-metadata-sys OnceCell) — under `cargo nextest` each test runs
 //! in its own process so the OnceCell never collides.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::{

crates/heal/tests/heal_b920_subquorum_union_test.rs

@@ -21,6 +21,8 @@
 //! These drive the REAL `ECStoreHealStorage` + `ECStore` against real disks.
 //! Every test is `#[serial]`; under `cargo nextest` each runs in its own process.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::storage::{

crates/heal/tests/heal_integration_test.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::{

crates/io-metrics/src/lib.rs

@@ -812,6 +812,17 @@ pub fn record_get_object_metadata_fanout_shape(path: &'static str, total: usize,
         .record(metadata_fanout_count_to_f64(non_valid));
 }
 
+/// Record task lifecycle shape for one GetObject metadata fanout.
+#[inline(always)]
+pub fn record_get_object_metadata_fanout_lifecycle(path: &'static str, scheduled: usize, completed: usize, cancelled: usize) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
+    histogram!("rustfs_io_get_object_metadata_fanout_scheduled", "path" => path).record(metadata_fanout_count_to_f64(scheduled));
+    histogram!("rustfs_io_get_object_metadata_fanout_completed", "path" => path).record(metadata_fanout_count_to_f64(completed));
+    histogram!("rustfs_io_get_object_metadata_fanout_cancelled", "path" => path).record(metadata_fanout_count_to_f64(cancelled));
+}
+
 /// Record a guarded metadata early-stop hit for GetObject.
 #[inline(always)]
 pub fn record_get_object_metadata_early_stop_hit(path: &'static str, reason: &'static str) {
@@ -2692,12 +2703,17 @@ mod tests {
         record_get_object_reader_prefetch_wait("codec_streaming", 0.0002);
         record_get_object_response_handoff("standard", "selected", 8192, 1024, 0.0001);
         record_get_object_metadata_fanout_duration("legacy_duplex", 0.001);
+        record_get_object_stage_duration("legacy_duplex", "read_version_path_resolve", 0.0001);
+        record_get_object_stage_duration("legacy_duplex", "read_version_path_check", 0.0001);
+        record_get_object_stage_duration("legacy_duplex", "read_version_xlmeta_read", 0.0005);
+        record_get_object_stage_duration("legacy_duplex", "read_version_decode", 0.0002);
         record_get_object_first_metadata_response_latency("legacy_duplex", 0.001);
         record_get_object_first_valid_metadata_response_latency("legacy_duplex", 0.001);
         record_get_object_slowest_metadata_response_latency("legacy_duplex", 0.003);
         record_get_object_quorum_reached_latency("legacy_duplex", 0.002);
         record_get_object_metadata_response("legacy_duplex", "valid");
         record_get_object_metadata_fanout_shape("legacy_duplex", 4, 3, 1, 1);
+        record_get_object_metadata_fanout_lifecycle("legacy_duplex", 4, 3, 1);
         record_get_object_metadata_early_stop_hit("legacy_duplex", "valid_quorum");
         record_get_object_metadata_early_stop_miss("legacy_duplex", "insufficient_quorum");
         record_get_object_metadata_early_stop_saved_responses("legacy_duplex", 1);
@@ -2768,6 +2784,38 @@ mod tests {
         assert!(remote_scheduled >= remote_avoid_potential);
     }
 
+    #[test]
+    fn metadata_fanout_lifecycle_records_named_histograms() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+
+        metrics::with_local_recorder(&recorder, || {
+            set_get_stage_metrics_enabled(true);
+            record_get_object_metadata_fanout_lifecycle("legacy_duplex", 4, 3, 1);
+            set_get_stage_metrics_enabled(false);
+        });
+
+        let metrics = snapshotter.snapshot().into_vec();
+        for (name, expected) in [
+            ("rustfs_io_get_object_metadata_fanout_scheduled", 4.0),
+            ("rustfs_io_get_object_metadata_fanout_completed", 3.0),
+            ("rustfs_io_get_object_metadata_fanout_cancelled", 1.0),
+        ] {
+            let value = metrics.iter().find_map(|(composite, _, _, value)| {
+                let has_path = composite
+                    .key()
+                    .labels()
+                    .any(|label| label.key() == "path" && label.value() == "legacy_duplex");
+                (composite.kind() == MetricKind::Histogram && composite.key().name() == name && has_path).then_some(value)
+            });
+            assert!(
+                matches!(value, Some(DebugValue::Histogram(values)) if values.len() == 1 && values[0].0 == expected),
+                "{name} must record the exact fanout lifecycle sample"
+            );
+        }
+    }
+
     #[test]
     fn test_record_get_object_fill_metrics() {
         record_get_object_fill_queued("codec_streaming", "single_inflight", 1);

crates/kms/src/config.rs

@@ -41,6 +41,7 @@ pub const ENV_KMS_AWS_ENDPOINT_URL: &str = "RUSTFS_KMS_AWS_ENDPOINT_URL";
 /// unset leaves rotation readiness unreported. Read once when the manager is
 /// built, by [`crate::manager::KmsManager`].
 pub const ENV_KMS_ROTATION_MAX_AGE_SECS: &str = "RUSTFS_KMS_ROTATION_MAX_AGE_SECS";
+pub const ENV_KMS_ROTATION_MAX_WRAPS: &str = "RUSTFS_KMS_ROTATION_MAX_WRAPS";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "secret";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX: &str = "rustfs/kms/transit-metadata";
 pub const DEFAULT_VAULT_APPROLE_MOUNT: &str = "approle";

crates/kms/src/manager.rs

@@ -17,7 +17,7 @@
 use crate::audit::{KmsAuditOperation, KmsAuditRecord, KmsAuditSink};
 use crate::backends::KmsBackend;
 use crate::cache::{KmsCache, KmsCacheStats};
-use crate::config::{ENV_KMS_ALLOW_IMMEDIATE_DELETION, ENV_KMS_ROTATION_MAX_AGE_SECS, KmsConfig};
+use crate::config::{ENV_KMS_ALLOW_IMMEDIATE_DELETION, ENV_KMS_ROTATION_MAX_AGE_SECS, ENV_KMS_ROTATION_MAX_WRAPS, KmsConfig};
 use crate::deletion_worker::DeletionReferenceChecker;
 use crate::error::{KmsError, Result};
 use crate::types::{
@@ -42,6 +42,13 @@ use tracing::warn;
 /// after it was rotated, which trains operators to ignore the signal.
 const MIN_ROTATION_MAX_AGE: Duration = Duration::from_secs(3600);
 
+/// Smallest wrap budget that can be configured.
+///
+/// Wraps are accounted in reserved blocks, so any threshold below one block
+/// would be crossed by a single reservation and report a key that has barely
+/// wrapped anything as overdue.
+const MIN_ROTATION_MAX_WRAPS: u64 = 1_000_000;
+
 /// Rotation age from the environment, or `None` when the signal is off.
 ///
 /// Unset leaves it off rather than guessing a policy: how often a deployment
@@ -68,6 +75,33 @@ fn parse_rotation_max_age(value: Option<&str>) -> Option<Duration> {
     Some(Duration::from_secs(seconds).max(MIN_ROTATION_MAX_AGE))
 }
 
+/// Wrap budget from the environment, or `None` when the signal is off.
+///
+/// Same discipline as the age threshold: unset means unreported rather than a
+/// guessed policy, and an unparsable value is refused loudly instead of
+/// falling back to a number the operator did not write. Clamped to
+/// [`MIN_ROTATION_MAX_WRAPS`] because the backend accounts for wraps in
+/// reserved blocks, so a threshold below one block would trip on the first
+/// reservation regardless of how many wraps actually happened.
+fn configured_rotation_max_wraps() -> Option<u64> {
+    parse_rotation_max_wraps(std::env::var(ENV_KMS_ROTATION_MAX_WRAPS).ok().as_deref())
+}
+
+fn parse_rotation_max_wraps(value: Option<&str>) -> Option<u64> {
+    let value = value?;
+    let Ok(wraps) = value.trim().parse::<u64>() else {
+        warn!(
+            variable = ENV_KMS_ROTATION_MAX_WRAPS,
+            "ignoring unparsable KMS rotation wrap budget; rotation readiness stays unreported"
+        );
+        return None;
+    };
+    if wraps == 0 {
+        return None;
+    }
+    Some(wraps.max(MIN_ROTATION_MAX_WRAPS))
+}
+
 #[derive(Clone)]
 pub struct KmsManager {
     backend: Arc<dyn KmsBackend>,
@@ -82,6 +116,7 @@ pub struct KmsManager {
     /// the verdict unreported. Read once at construction so a listing cannot
     /// change its answer halfway through.
     rotation_max_age: Option<Duration>,
+    rotation_max_wraps: Option<u64>,
 }
 
 impl KmsManager {
@@ -103,6 +138,7 @@ impl KmsManager {
             allow_immediate_deletion: config.allow_immediate_deletion,
             reference_checker: None,
             rotation_max_age: configured_rotation_max_age(),
+            rotation_max_wraps: configured_rotation_max_wraps(),
         }
     }
 
@@ -314,9 +350,22 @@ impl KmsManager {
             key.rotation_due_reason = Some(RotationDueReason::Unsupported);
             return;
         }
+        key.rotation_due = false;
+        key.rotation_due_reason = None;
+
+        // The wrap budget is checked first: it is the cryptographic bound (the
+        // AES-GCM random-nonce ceiling), whereas the age threshold is a policy
+        // choice, so when both are crossed the reason an operator most needs to
+        // see is the one they cannot negotiate.
+        if let (Some(max_wraps), Some(wraps)) = (self.rotation_max_wraps, key.wrap_budget_reserved)
+            && wraps >= max_wraps
+        {
+            key.rotation_due = true;
+            key.rotation_due_reason = Some(RotationDueReason::Wraps);
+            return;
+        }
+
         let Some(max_age) = self.rotation_max_age else {
-            key.rotation_due = false;
-            key.rotation_due_reason = None;
             return;
         };
 
@@ -333,9 +382,6 @@ impl KmsManager {
         if age >= max_age {
             key.rotation_due = true;
             key.rotation_due_reason = Some(reason);
-        } else {
-            key.rotation_due = false;
-            key.rotation_due_reason = None;
         }
     }
 
@@ -1685,10 +1731,15 @@ mod tests {
     }
 
     fn readiness_manager(rotation_max_age: Option<Duration>) -> KmsManager {
+        readiness_manager_with(rotation_max_age, None)
+    }
+
+    fn readiness_manager_with(rotation_max_age: Option<Duration>, rotation_max_wraps: Option<u64>) -> KmsManager {
         let temp_dir = tempfile::tempdir().expect("temp dir");
         let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
         let mut manager = KmsManager::new(Arc::new(ScriptedBackend::succeeding()), config);
         manager.rotation_max_age = rotation_max_age;
+        manager.rotation_max_wraps = rotation_max_wraps;
         manager
     }
 
@@ -1765,6 +1816,85 @@ mod tests {
         assert!(!key.rotation_due, "clock skew must not manufacture an overdue key");
     }
 
+    /// The wrap-budget half of the verdict: the cryptographic bound, checked
+    /// independently of the age policy and reported under its own reason.
+    #[test]
+    fn rotation_readiness_reports_an_exhausted_wrap_budget() {
+        let now = Zoned::now();
+        let recently = &now - jiff::Span::new().hours(1);
+        let long_ago = &now - jiff::Span::new().days(400);
+        let day = Duration::from_secs(86_400);
+        let budget = 2_000_000;
+
+        let with_wraps = |manager: &KmsManager, wraps: Option<u64>, rotated_at: Option<Zoned>| {
+            let mut key = aged_key(rotated_at, recently.clone());
+            key.wrap_budget_reserved = wraps;
+            manager.apply_rotation_readiness(&mut key, true, &now);
+            (key.rotation_due, key.rotation_due_reason)
+        };
+
+        // Budget configured and exceeded on a freshly rotated key: due, and the
+        // reason names the wrap budget rather than an age nobody crossed.
+        let manager = readiness_manager_with(Some(day), Some(budget));
+        assert_eq!(
+            with_wraps(&manager, Some(budget), Some(recently.clone())),
+            (true, Some(RotationDueReason::Wraps))
+        );
+        // At the threshold exactly, not only past it: the bound is a ceiling.
+        assert_eq!(
+            with_wraps(&manager, Some(budget + 1), Some(recently.clone())),
+            (true, Some(RotationDueReason::Wraps))
+        );
+        // Under the threshold: no verdict from the wrap half.
+        assert_eq!(with_wraps(&manager, Some(budget - 1), Some(recently.clone())), (false, None));
+
+        // The cryptographic bound outranks the policy one when both are crossed.
+        let mut key = aged_key(Some(long_ago.clone()), long_ago);
+        key.wrap_budget_reserved = Some(budget);
+        manager.apply_rotation_readiness(&mut key, true, &now);
+        assert_eq!(key.rotation_due_reason, Some(RotationDueReason::Wraps));
+
+        // No wrap threshold configured: an enormous count reports nothing, the
+        // same way an unset age threshold does.
+        let age_only = readiness_manager_with(Some(day), None);
+        assert_eq!(with_wraps(&age_only, Some(u64::MAX), Some(recently.clone())), (false, None));
+
+        // Backend reports no count (Transit, AWS, or a pre-accounting record):
+        // the wrap half stays silent instead of guessing, and the age half
+        // still decides.
+        let wraps_only = readiness_manager_with(None, Some(budget));
+        assert_eq!(with_wraps(&wraps_only, None, Some(recently.clone())), (false, None));
+        assert_eq!(
+            with_wraps(&wraps_only, Some(budget), Some(recently.clone())),
+            (true, Some(RotationDueReason::Wraps))
+        );
+
+        // A backend that cannot rotate is never told to, whatever it wrapped.
+        let mut key = aged_key(None, recently);
+        key.wrap_budget_reserved = Some(u64::MAX);
+        wraps_only.apply_rotation_readiness(&mut key, false, &now);
+        assert!(!key.rotation_due);
+        assert_eq!(key.rotation_due_reason, Some(RotationDueReason::Unsupported));
+    }
+
+    /// Threshold parsing matches the age threshold's discipline: unset and
+    /// unparsable both disable the signal rather than inventing a policy.
+    #[test]
+    fn rotation_wrap_threshold_parsing_refuses_to_guess() {
+        assert_eq!(parse_rotation_max_wraps(None), None);
+        assert_eq!(parse_rotation_max_wraps(Some("not-a-number")), None);
+        assert_eq!(parse_rotation_max_wraps(Some("")), None);
+        assert_eq!(parse_rotation_max_wraps(Some("-1")), None);
+        assert_eq!(parse_rotation_max_wraps(Some("0")), None);
+        // Clamped: below one reservation block the first reservation would trip it.
+        assert_eq!(parse_rotation_max_wraps(Some("1")), Some(MIN_ROTATION_MAX_WRAPS));
+        assert_eq!(
+            parse_rotation_max_wraps(Some(" 5000000 ")),
+            Some(5_000_000),
+            "a configured budget above the floor is honored verbatim"
+        );
+    }
+
     /// The two fields are additive on the wire: a payload written before they
     /// existed still deserializes, and a key with no verdict serializes exactly
     /// as it did before.

crates/kms/src/types.rs

@@ -217,6 +217,12 @@ pub enum RotationDueReason {
     /// The key has never been rotated and has existed longer than the
     /// configured maximum age.
     NeverRotated,
+    /// The key has wrapped more data keys than the configured maximum.
+    ///
+    /// Counted per key-material version, so a rotation restarts the budget.
+    /// The count is an over-estimate by construction (see the backend's
+    /// reservation accounting), so this verdict errs toward rotating early.
+    Wraps,
     /// The backend cannot rotate keys at all, so no age makes one due.
     Unsupported,
 }

crates/protocols/src/swift/handler.rs

@@ -748,10 +748,9 @@ async fn handle_authenticated_request(
                                 }
 
                                 for (key, value) in info.user_defined.iter() {
-                                    if key != "content-type"
-                                        && let Some(key) = object::swift_response_user_metadata_key(key)
-                                    {
-                                        response = response.header(format!("x-object-meta-{key}"), value.as_str());
+                                    if key != "content-type" {
+                                        let header_name = format!("x-object-meta-{}", key);
+                                        response = response.header(header_name, value.as_str());
                                     }
                                 }
 
@@ -818,10 +817,9 @@ async fn handle_authenticated_request(
 
                     // Add custom metadata headers (X-Object-Meta-*)
                     for (key, value) in info.user_defined.iter() {
-                        if key != "content-type"
-                            && let Some(key) = object::swift_response_user_metadata_key(key)
-                        {
-                            response = response.header(format!("x-object-meta-{key}"), value.as_str());
+                        if key != "content-type" {
+                            let header_name = format!("x-object-meta-{}", key);
+                            response = response.header(header_name, value.as_str());
                         }
                     }
 
@@ -858,10 +856,9 @@ async fn handle_authenticated_request(
 
                     // Add custom metadata headers (X-Object-Meta-*)
                     for (key, value) in info.user_defined.iter() {
-                        if key != "content-type"
-                            && let Some(key) = object::swift_response_user_metadata_key(key)
-                        {
-                            response = response.header(format!("x-object-meta-{key}"), value.as_str());
+                        if key != "content-type" {
+                            let header_name = format!("x-object-meta-{}", key);
+                            response = response.header(header_name, value.as_str());
                         }
                     }
 
@@ -1171,10 +1168,9 @@ async fn handle_object_get(
                 }
 
                 for (key, value) in info.user_defined.iter() {
-                    if key != "content-type"
-                        && let Some(key) = object::swift_response_user_metadata_key(key)
-                    {
-                        response = response.header(format!("x-object-meta-{key}"), value.as_str());
+                    if key != "content-type" {
+                        let header_name = format!("x-object-meta-{}", key);
+                        response = response.header(header_name, value.as_str());
                     }
                 }
 
@@ -1241,10 +1237,9 @@ async fn handle_object_get(
         if key == "x-delete-at" {
             // Add X-Delete-At header directly (not as X-Object-Meta-*)
             response = response.header("x-delete-at", value.as_str());
-        } else if key != "content-type"
-            && let Some(key) = object::swift_response_user_metadata_key(key)
-        {
-            response = response.header(format!("x-object-meta-{key}"), value.as_str());
+        } else if key != "content-type" {
+            let header_name = format!("x-object-meta-{}", key);
+            response = response.header(header_name, value.as_str());
         }
     }
 
@@ -1298,10 +1293,9 @@ async fn handle_object_head(
         if key == "x-delete-at" {
             // Add X-Delete-At header directly (not as X-Object-Meta-*)
             response = response.header("x-delete-at", value.as_str());
-        } else if key != "content-type"
-            && let Some(key) = object::swift_response_user_metadata_key(key)
-        {
-            response = response.header(format!("x-object-meta-{key}"), value.as_str());
+        } else if key != "content-type" {
+            let header_name = format!("x-object-meta-{}", key);
+            response = response.header(header_name, value.as_str());
         }
     }
 

crates/protocols/src/swift/object.rs

@@ -67,57 +67,10 @@ const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
 const LOG_SUBSYSTEM_SWIFT_OBJECT: &str = "swift_object";
 const EVENT_SWIFT_OBJECT_STORAGE_STATE: &str = "swift_object_storage_state";
 const SWIFT_DELETE_AT_METADATA: &str = "x-delete-at";
-const USER_METADATA_PREFIX: &str = "x-amz-meta-";
 
 /// Maximum object size in bytes (5GB - Swift default)
 const MAX_OBJECT_SIZE: i64 = 5 * 1024 * 1024 * 1024;
 
-fn stored_swift_user_metadata_key(key: &str) -> String {
-    if rustfs_utils::http::is_internal_key(key)
-        || rustfs_utils::http::starts_with_ignore_ascii_case(key, "x-amz-")
-        || rustfs_utils::http::starts_with_ignore_ascii_case(key, "x-rustfs-encryption-")
-        || rustfs_utils::http::starts_with_ignore_ascii_case(key, "x-minio-encryption-")
-    {
-        format!("{USER_METADATA_PREFIX}{key}")
-    } else {
-        key.to_string()
-    }
-}
-
-pub(super) fn swift_response_user_metadata_key(key: &str) -> Option<&str> {
-    if rustfs_utils::http::is_internal_key(key)
-        || rustfs_utils::http::starts_with_ignore_ascii_case(key, "x-rustfs-encryption-")
-        || rustfs_utils::http::starts_with_ignore_ascii_case(key, "x-minio-encryption-")
-    {
-        return None;
-    }
-    if let Some(unescaped) = key.strip_prefix(USER_METADATA_PREFIX)
-        && (rustfs_utils::http::is_internal_key(unescaped)
-            || rustfs_utils::http::starts_with_ignore_ascii_case(unescaped, "x-amz-")
-            || rustfs_utils::http::starts_with_ignore_ascii_case(unescaped, "x-rustfs-encryption-")
-            || rustfs_utils::http::starts_with_ignore_ascii_case(unescaped, "x-minio-encryption-"))
-    {
-        return Some(unescaped);
-    }
-    Some(key)
-}
-
-fn swift_user_metadata(headers: &HeaderMap) -> Option<HashMap<String, String>> {
-    let mut metadata = HashMap::new();
-    let mut present = false;
-    for (header_name, header_value) in headers.iter() {
-        let header_name = header_name.as_str().to_lowercase();
-        let Some(key) = header_name.strip_prefix("x-object-meta-") else {
-            continue;
-        };
-        present = true;
-        if let Ok(value) = header_value.to_str() {
-            metadata.insert(stored_swift_user_metadata_key(key), value.to_string());
-        }
-    }
-    present.then_some(metadata)
-}
-
 /// Object key translator for Swift object names
 ///
 /// Handles URL encoding/decoding and path normalization for Swift object keys.
@@ -350,7 +303,15 @@ where
     let bucket = mapper.swift_to_s3_bucket(container, &project_id);
 
     // 5. Extract Swift metadata from X-Object-Meta-* headers
-    let mut user_metadata = swift_user_metadata(headers).unwrap_or_default();
+    let mut user_metadata = HashMap::new();
+    for (header_name, header_value) in headers.iter() {
+        let header_str = header_name.as_str().to_lowercase();
+        if let Some(meta_key) = header_str.strip_prefix("x-object-meta-")
+            && let Ok(value_str) = header_value.to_str()
+        {
+            user_metadata.insert(meta_key.to_string(), value_str.to_string());
+        }
+    }
 
     // 6. Extract Content-Type if provided
     if let Some(content_type) = headers.get("content-type")
@@ -778,7 +739,15 @@ pub async fn update_object_metadata(
     }
 
     // 8. Extract new metadata from X-Object-Meta-* headers
-    let mut new_metadata = swift_user_metadata(headers).unwrap_or_default();
+    let mut new_metadata = HashMap::new();
+    for (header_name, header_value) in headers.iter() {
+        let header_str = header_name.as_str().to_lowercase();
+        if let Some(meta_key) = header_str.strip_prefix("x-object-meta-")
+            && let Ok(value_str) = header_value.to_str()
+        {
+            new_metadata.insert(meta_key.to_string(), value_str.to_string());
+        }
+    }
 
     // 9. Also update Content-Type if provided
     if let Some(content_type) = headers.get("content-type")
@@ -920,8 +889,19 @@ pub async fn copy_object(
     let mut new_metadata = (*src_info.user_defined).clone();
 
     // 11. If custom metadata headers provided, use those instead (Swift behavior)
-    if let Some(custom_metadata) = swift_user_metadata(headers) {
-        new_metadata = custom_metadata;
+    let mut has_custom_meta = false;
+    for (header_name, header_value) in headers.iter() {
+        let header_str = header_name.as_str().to_lowercase();
+        if let Some(meta_key) = header_str.strip_prefix("x-object-meta-") {
+            if !has_custom_meta {
+                // First custom meta header - clear source metadata
+                new_metadata.clear();
+                has_custom_meta = true;
+            }
+            if let Ok(value_str) = header_value.to_str() {
+                new_metadata.insert(meta_key.to_string(), value_str.to_string());
+            }
+        }
     }
 
     // 12. Also check for Content-Type override
@@ -1143,36 +1123,6 @@ mod tests {
         assert!(ObjectKeyMapper::validate_object_name("unicode-文件.txt").is_ok());
     }
 
-    #[test]
-    fn swift_user_metadata_cannot_materialize_internal_storage_keys() {
-        let mut headers = HeaderMap::new();
-        headers.insert("x-object-meta-x-rustfs-internal-actual-size", "1".parse().expect("valid metadata value"));
-        headers.insert("x-object-meta-description", "safe".parse().expect("valid metadata value"));
-        let metadata = swift_user_metadata(&headers).expect("custom metadata should be detected");
-
-        assert_eq!(metadata.get("x-amz-meta-x-rustfs-internal-actual-size").map(String::as_str), Some("1"));
-        assert_eq!(metadata.get("description").map(String::as_str), Some("safe"));
-        assert!(!metadata.contains_key("x-rustfs-internal-actual-size"));
-        assert_eq!(
-            stored_swift_user_metadata_key("x-minio-encryption-original-size"),
-            "x-amz-meta-x-minio-encryption-original-size"
-        );
-        assert_eq!(stored_swift_user_metadata_key("description"), "description");
-    }
-
-    #[test]
-    fn swift_user_metadata_response_mapping_is_reversible_and_filters_internal_keys() {
-        assert_eq!(
-            swift_response_user_metadata_key("x-amz-meta-x-rustfs-internal-actual-size"),
-            Some("x-rustfs-internal-actual-size")
-        );
-        assert_eq!(swift_response_user_metadata_key("x-amz-meta-x-amz-checksum"), Some("x-amz-checksum"));
-        assert_eq!(swift_response_user_metadata_key("x-amz-meta-description"), Some("x-amz-meta-description"));
-        assert_eq!(swift_response_user_metadata_key("description"), Some("description"));
-        assert_eq!(swift_response_user_metadata_key("x-rustfs-internal-actual-size"), None);
-        assert_eq!(swift_response_user_metadata_key("x-minio-internal-actual-size"), None);
-    }
-
     #[test]
     fn test_validate_object_name_empty() {
         let result = ObjectKeyMapper::validate_object_name("");

crates/protocols/tests/swift_metadata_persistence.rs

@@ -23,6 +23,7 @@
 //! two are tested together because a reload is the only way to tell a real
 //! merge from one that happened to look right in the cache.
 
+#![recursion_limit = "256"]
 #![cfg(feature = "swift")]
 
 use std::collections::HashMap;

crates/scanner/src/lib.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
 #![warn(
     // missing_docs,

crates/scanner/tests/lifecycle_integration_test.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
+
 use futures::FutureExt;
 use rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT;
 use rustfs_scanner::scanner_folder::ScannerItem;

crates/utils/src/http/object_encryption_keys.rs

@@ -25,7 +25,8 @@
 // The lowercase stored forms, matching exactly what encryption_material_to_metadata
 // persists. The read-path SSE-C check is case-sensitive, so restoring under any
 // other casing would classify the replica as managed-SSE and reject SSE-C GETs.
-use super::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
+use super::headers::{AMZ_ENCRYPTION_KMS, SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER};
+use std::collections::HashMap;
 
 pub const INTERNAL_ENCRYPTION_KEY_ID_HEADER: &str = "x-rustfs-encryption-key-id";
 pub const INTERNAL_ENCRYPTION_KEY_HEADER: &str = "x-rustfs-encryption-key";
@@ -165,6 +166,143 @@ pub fn is_replication_stripped_encryption_key(key: &str) -> bool {
         || super::starts_with_ignore_ascii_case(key, RUSTFS_INTERNAL_ENCRYPTION_PREFIX)
 }
 
+// ============================================================================
+// Managed-SSE attribution (shared classifier)
+// ============================================================================
+//
+// Single source of truth for classifying stored managed-SSE (SSE-S3 / SSE-KMS)
+// object metadata. These live here — rather than in the `rustfs` binary
+// crate's SSE module — so lower-layer consumers such as the scanner can
+// attribute encrypted objects without growing a second copy of the
+// normalization/classification logic (backlog#1643 PR-B0). The binary crate
+// re-exports them from `rustfs::storage::sse`, and a source-scan test there
+// pins that no second definition reappears.
+//
+// Every metadata lookup below is a case-SENSITIVE exact match on the stored
+// `HashMap<String, String>` keys, mirroring the SSE read path. Do not
+// "harmonize" these with the lowercase-normalizing helpers in
+// `header_compat.rs`: the lowercase `x-amz-*` stored forms and the TitleCase
+// MinIO-internal names are load-bearing exactly as written.
+
+/// Type of encryption used
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SSEType {
+    /// SSE-S3 (AES256)
+    SseS3,
+    /// SSE-KMS (aws:kms)
+    SseKms,
+    /// SSE-C (customer-provided key)
+    SseC,
+}
+
+impl SSEType {
+    /// Stable scheme name for audit consumers.
+    pub fn audit_label(self) -> &'static str {
+        match self {
+            SSEType::SseS3 => "SSE-S3",
+            SSEType::SseKms => "SSE-KMS",
+            SSEType::SseC => "SSE-C",
+        }
+    }
+}
+
+/// Recodes a stored MinIO KMS context value — base64-wrapped JSON under
+/// [`MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER`] — into the plain-JSON form
+/// RustFS stores under [`INTERNAL_ENCRYPTION_CONTEXT_HEADER`].
+///
+/// Injected by callers because this crate deliberately carries no JSON codec.
+/// Returning `None` skips the context mapping, matching the historical
+/// silent-skip on a value that fails to decode.
+pub type KmsContextRecoder = fn(&str) -> Option<String>;
+
+/// True when the stored metadata carries a managed-SSE (SSE-S3 / SSE-KMS)
+/// encryption envelope, under either the RustFS-branded or the MinIO-branded
+/// internal keys.
+pub fn contains_managed_encryption_metadata(metadata: &HashMap<String, String>) -> bool {
+    metadata.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
+        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER)
+        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER)
+        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER)
+        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER)
+}
+
+/// Maps the MinIO-branded internal SSE keys onto the RustFS-branded stored
+/// keys (the dual internal metadata keys invariant). RustFS-branded keys
+/// already present always win; every source lookup is a case-sensitive exact
+/// match on the specific TitleCase MinIO names.
+pub fn normalize_managed_metadata(
+    metadata: &HashMap<String, String>,
+    recode_kms_context: Option<KmsContextRecoder>,
+) -> HashMap<String, String> {
+    let mut normalized = metadata.clone();
+
+    if !normalized.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
+        && let Some(value) = metadata
+            .get(MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER)
+            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER))
+            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER))
+            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER))
+    {
+        normalized.insert(INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), value.clone());
+    }
+
+    if !normalized.contains_key(INTERNAL_ENCRYPTION_IV_HEADER)
+        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_IV_HEADER)
+    {
+        normalized.insert(INTERNAL_ENCRYPTION_IV_HEADER.to_string(), value.clone());
+    }
+
+    if !normalized.contains_key(INTERNAL_ENCRYPTION_ALGORITHM_HEADER)
+        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER)
+    {
+        normalized.insert(INTERNAL_ENCRYPTION_ALGORITHM_HEADER.to_string(), value.clone());
+    }
+
+    if !normalized.contains_key(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
+        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER)
+    {
+        normalized.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), value.clone());
+    }
+
+    if !normalized.contains_key(INTERNAL_ENCRYPTION_CONTEXT_HEADER)
+        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER)
+        && let Some(recode) = recode_kms_context
+        && let Some(encoded) = recode(value)
+    {
+        normalized.insert(INTERNAL_ENCRYPTION_CONTEXT_HEADER.to_string(), encoded);
+    }
+
+    normalized
+}
+
+/// Resolve the scheme and KMS key a stored managed-SSE object was wrapped with.
+///
+/// Mirrors the lookup `apply_managed_decryption_material` performs, so both agree on
+/// which key a read is authorized against.
+///
+/// No [`KmsContextRecoder`] is taken: the context mapping only ever inserts
+/// [`INTERNAL_ENCRYPTION_CONTEXT_HEADER`], which this lookup never reads, so
+/// the result is identical with or without it.
+pub fn stored_managed_encryption_key(metadata: &HashMap<String, String>) -> Option<(SSEType, String)> {
+    if !contains_managed_encryption_metadata(metadata) {
+        return None;
+    }
+
+    // Case-sensitive: the SSE writer stores the scheme under the lowercase
+    // `x-amz-server-side-encryption` key; other casings are not stored forms.
+    let sse_type = match metadata.get("x-amz-server-side-encryption")?.as_str() {
+        AMZ_ENCRYPTION_KMS => SSEType::SseKms,
+        _ => SSEType::SseS3,
+    };
+    let key_id = normalize_managed_metadata(metadata, None)
+        .get(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
+        .or_else(|| metadata.get("x-amz-server-side-encryption-aws-kms-key-id"))
+        .cloned()
+        .unwrap_or_else(|| "default".to_string());
+
+    Some((sse_type, key_id))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -273,6 +411,134 @@ mod tests {
         assert!(!is_replication_stripped_encryption_key("content-type"));
     }
 
+    #[test]
+    fn managed_envelope_predicate_matches_both_key_families() {
+        assert!(!contains_managed_encryption_metadata(&HashMap::new()));
+
+        for key in [
+            INTERNAL_ENCRYPTION_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER,
+            MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER,
+        ] {
+            let single = HashMap::from([(key.to_string(), "value".to_string())]);
+            assert!(contains_managed_encryption_metadata(&single), "{key} must classify as managed SSE");
+        }
+
+        // SSE-C material alone is not a managed envelope.
+        let ssec_only = HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]);
+        assert!(!contains_managed_encryption_metadata(&ssec_only));
+    }
+
+    #[test]
+    fn normalize_maps_minio_keys_onto_missing_rustfs_keys_only() {
+        let metadata = HashMap::from([
+            (MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER.to_string(), "minio-dek".to_string()),
+            (MINIO_INTERNAL_ENCRYPTION_IV_HEADER.to_string(), "minio-iv".to_string()),
+            (MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER.to_string(), "DAREv2-HMAC-SHA256".to_string()),
+            (MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER.to_string(), "minio-key".to_string()),
+        ]);
+
+        let normalized = normalize_managed_metadata(&metadata, None);
+        assert_eq!(normalized.get(INTERNAL_ENCRYPTION_KEY_HEADER).map(String::as_str), Some("minio-dek"));
+        assert_eq!(normalized.get(INTERNAL_ENCRYPTION_IV_HEADER).map(String::as_str), Some("minio-iv"));
+        assert_eq!(
+            normalized.get(INTERNAL_ENCRYPTION_ALGORITHM_HEADER).map(String::as_str),
+            Some("DAREv2-HMAC-SHA256")
+        );
+        assert_eq!(normalized.get(INTERNAL_ENCRYPTION_KEY_ID_HEADER).map(String::as_str), Some("minio-key"));
+
+        // Existing RustFS-branded keys always win over the MinIO twins.
+        let mut both = metadata;
+        both.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "rustfs-key".to_string());
+        assert_eq!(
+            normalize_managed_metadata(&both, None)
+                .get(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
+                .map(String::as_str),
+            Some("rustfs-key")
+        );
+
+        // The mapping is a case-sensitive exact match on the TitleCase MinIO
+        // names; a lowercased twin must not normalize.
+        let lowercased = HashMap::from([(MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER.to_lowercase(), "minio-key".to_string())]);
+        assert!(!normalize_managed_metadata(&lowercased, None).contains_key(INTERNAL_ENCRYPTION_KEY_ID_HEADER));
+    }
+
+    #[test]
+    fn normalize_recodes_kms_context_only_through_the_injected_codec() {
+        let metadata = HashMap::from([(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER.to_string(), "encoded-context".to_string())]);
+
+        // Without a codec the context stays unnormalized.
+        assert!(!normalize_managed_metadata(&metadata, None).contains_key(INTERNAL_ENCRYPTION_CONTEXT_HEADER));
+
+        // A codec that fails to decode also leaves it unnormalized.
+        fn reject(_value: &str) -> Option<String> {
+            None
+        }
+        assert!(!normalize_managed_metadata(&metadata, Some(reject)).contains_key(INTERNAL_ENCRYPTION_CONTEXT_HEADER));
+
+        fn recode(value: &str) -> Option<String> {
+            Some(format!("recoded:{value}"))
+        }
+        assert_eq!(
+            normalize_managed_metadata(&metadata, Some(recode))
+                .get(INTERNAL_ENCRYPTION_CONTEXT_HEADER)
+                .map(String::as_str),
+            Some("recoded:encoded-context")
+        );
+
+        // A stored RustFS context wins without invoking the codec.
+        let mut both = metadata;
+        both.insert(INTERNAL_ENCRYPTION_CONTEXT_HEADER.to_string(), "stored-context".to_string());
+        assert_eq!(
+            normalize_managed_metadata(&both, Some(recode))
+                .get(INTERNAL_ENCRYPTION_CONTEXT_HEADER)
+                .map(String::as_str),
+            Some("stored-context")
+        );
+    }
+
+    #[test]
+    fn stored_managed_encryption_key_attributes_scheme_and_key() {
+        // Plaintext metadata carries no managed envelope.
+        assert!(stored_managed_encryption_key(&HashMap::new()).is_none());
+
+        // A managed envelope without the stored SSE marker cannot be attributed.
+        let envelope_only = HashMap::from([(INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), "dek".to_string())]);
+        assert!(stored_managed_encryption_key(&envelope_only).is_none());
+
+        // The stored SSE marker is the lowercase form; a TitleCase key is not
+        // a stored form and must not be recognized.
+        let mut titlecase = envelope_only.clone();
+        titlecase.insert("X-Amz-Server-Side-Encryption".to_string(), "aws:kms".to_string());
+        assert!(stored_managed_encryption_key(&titlecase).is_none());
+
+        let mut sse_s3 = envelope_only.clone();
+        sse_s3.insert("x-amz-server-side-encryption".to_string(), "AES256".to_string());
+        assert_eq!(stored_managed_encryption_key(&sse_s3), Some((SSEType::SseS3, "default".to_string())));
+
+        let mut sse_kms = envelope_only;
+        sse_kms.insert("x-amz-server-side-encryption".to_string(), "aws:kms".to_string());
+        assert_eq!(stored_managed_encryption_key(&sse_kms), Some((SSEType::SseKms, "default".to_string())));
+
+        // Key-id precedence: RustFS stored key id, then the MinIO twin, then
+        // the lowercase amz key id, then "default".
+        sse_kms.insert("x-amz-server-side-encryption-aws-kms-key-id".to_string(), "amz-key".to_string());
+        assert_eq!(stored_managed_encryption_key(&sse_kms), Some((SSEType::SseKms, "amz-key".to_string())));
+        sse_kms.insert(MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER.to_string(), "minio-key".to_string());
+        assert_eq!(stored_managed_encryption_key(&sse_kms), Some((SSEType::SseKms, "minio-key".to_string())));
+        sse_kms.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "rustfs-key".to_string());
+        assert_eq!(stored_managed_encryption_key(&sse_kms), Some((SSEType::SseKms, "rustfs-key".to_string())));
+    }
+
+    #[test]
+    fn sse_type_audit_labels_are_stable() {
+        assert_eq!(SSEType::SseS3.audit_label(), "SSE-S3");
+        assert_eq!(SSEType::SseKms.audit_label(), "SSE-KMS");
+        assert_eq!(SSEType::SseC.audit_label(), "SSE-C");
+    }
+
     #[test]
     fn transport_prefixes_cover_every_transport_value_key() {
         // Every transport key that carries material must match a redaction

docs/architecture/compat-cleanup-register.md

@@ -16,6 +16,7 @@ for later deletion.
 - `table-catalog-strong-snapshot-v1` durable strong catalog snapshot compatibility: mixed-version deployments continue writing version 1 snapshots until operators confirm that every serving node reads version 2, and version 1 table/view identifier collisions remain available only for cleanup. Remove version 1 writes and collision cleanup after the minimum supported RustFS release reads version 2 and every retained durable strong snapshot is collision-free and has been upgraded to version 2.
 - `table-catalog-migration-fence-v1` durable strong migration fence compatibility: version 1 "PREPARING" fences did not distinguish a known-absent global strong snapshot from an unknown baseline, so retries read them but fail closed if the global snapshot is missing. Version 2 preserves the same JSON shape and records the pre-migration global snapshot ETag in the existing target_snapshot_etag field while the fence is "PREPARING". Remove version 1 reads after every supported direct-upgrade source writes version 2 fences and operators have completed or cancelled every older in-progress backing migration.
 - `table-catalog-backing-manifest-v1-wire-labels` durable strong backing manifest labels: version 1 published "STRONG_KV_WAL" and "CUT_OVER_LINEARIZABLE_READS" before the implementation was narrowed to the ETag-CAS durable snapshot backing. Internal names and operator documentation describe the implemented semantics, while version 1 responses retain those labels for existing clients. Replace the labels only in a new manifest version with an explicit client migration contract.
+- `cross-pool-fence-v1` authenticated unsupported advertisement: predeployment servers recognize the versioned cross-pool fence capability probe but report support version 0, allowing a later all-peer probe to distinguish predeployment nodes without activating a second lock domain. Replace the unsupported advertisement only when composite lock acquisition, a cluster-wide activation fence, complete fleet proof, commit-time proof revalidation, and fail-closed revocation ship together.
 - `table-catalog-dotted-namespace` Iceberg REST namespace path compatibility: existing RustFS clients use dotted namespace paths, while the standard multi-level contract uses the URL-encoded unit separator `%1F`. New servers accept both forms so a rolling upgrade does not invalidate existing catalog configuration. Remove the dotted fallback after the minimum supported RustFS release advertises `%1F` and all supported clients have refreshed their catalog configuration.
 - `rustfs-5509` FileInfo positional MessagePack decoding: beta.11 serialized 28 fields, while beta.12 inserted transition-version fields in the middle and serialized an incompatible 30-field array. New releases write named maps and retain readers for both shipped array layouts so direct and rolling upgrades can read either release. Remove the positional-array readers after every supported direct-upgrade release writes named maps and no retained RPC payload can contain a pre-map FileInfo array.
 - `rustfs-5416` Helm distributed startup wait setting: charts that predate explicit local endpoint identity expose startupWaitTimeoutSeconds for their peer DNS/TCP init gate. The new chart keeps the value accepted but ignores it after moving startup convergence into RustFS. Remove the value and its documentation after the minimum supported direct-upgrade chart includes localEndpointHost.autoInject and no longer renders the peer gate.

docs/operations/kms-backend-security.md

@@ -112,6 +112,8 @@ RustFS does not rotate keys on a schedule. There is no built-in rotation worker,
 
 Set `RUSTFS_KMS_ROTATION_MAX_AGE_SECS` to that period in whole seconds. Unset — the default — leaves the verdict unreported rather than assuming a policy: how often keys must be rotated is a compliance decision, and a built-in default would report keys as overdue against a rule nobody wrote. An unparsable value is treated the same way, with a warning, instead of silently falling back to a number the operator did not choose. Values below one hour are raised to one hour, because a threshold of seconds reports every key as overdue moments after it was rotated and teaches operators to ignore the signal.
 
+A second, independent threshold covers the cryptographic bound rather than the policy one. `RUSTFS_KMS_ROTATION_MAX_WRAPS` is the number of data keys one key's material may wrap before the verdict reports `rotation_due` with reason `wraps`. It follows the same discipline — unset or unparsable leaves the verdict unreported, and values below one million are raised to one million because wraps are accounted in reserved blocks of that size, so a smaller threshold would trip on the first reservation. Only backends where RustFS wraps locally and can rotate report a count (Vault KV2 today); Transit and AWS wrap externally and report none, so the wrap half stays silent there rather than guessing. When both thresholds are crossed the reported reason is `wraps`: the AES-GCM random-nonce ceiling is not negotiable, while the age period is a policy an operator chose.
+
 `GET /rustfs/admin/v3/kms/keys` then carries two additional fields per key:
 
 - `rotation_due` — whether the key has outlived the configured period.

docs/operations/kms-observability-runbook.md

@@ -204,7 +204,7 @@ Meaning: `rustfs_kms_oldest_key_rotation_age_seconds` — seconds since the leas
 
 Investigation:
 
-1. Find which keys are due. The gauge deliberately names no key — a per-key label would carry key identifiers into the metric stream — so read the per-key verdict from the listing: `GET /rustfs/admin/v3/kms/keys` carries `rotation_due` and `rotation_due_reason` (`age`, `never_rotated`, or `unsupported`) per key, computed against `RUSTFS_KMS_ROTATION_MAX_AGE_SECS`. The verdict appears only on the listing, not on single-key describe. If `RUSTFS_KMS_ROTATION_MAX_AGE_SECS` is unset, set it to your policy's rotation period so the per-key verdict and this alert agree on what "overdue" means.
+1. Find which keys are due. The gauge deliberately names no key — a per-key label would carry key identifiers into the metric stream — so read the per-key verdict from the listing: `GET /rustfs/admin/v3/kms/keys` carries `rotation_due` and `rotation_due_reason` (`age`, `never_rotated`, `wraps`, or `unsupported`) per key, computed against `RUSTFS_KMS_ROTATION_MAX_AGE_SECS` and `RUSTFS_KMS_ROTATION_MAX_WRAPS`. A `wraps` reason means the key's material has wrapped more data keys than the configured budget — the AES-GCM random-nonce ceiling rather than an age policy, so it is not satisfied by relaxing the age threshold. The verdict appears only on the listing, not on single-key describe. If `RUSTFS_KMS_ROTATION_MAX_AGE_SECS` is unset, set it to your policy's rotation period so the per-key verdict and this alert agree on what "overdue" means.
 2. If the reason is `unsupported`, the backend cannot rotate at all (Local, Static). There is no key-level response; the decision is a backend migration, and the wrap ceiling above is the reason it cannot be deferred forever. See the [rotation drivers and scheduling matrix](kms-backend-security.md#rotation-drivers-and-scheduling-per-backend).
 3. On a backend that can rotate, act per the driver matrix: on **Vault KV2**, check why your external rotation scheduler did not run (or set one up — RustFS deliberately ships none) and satisfy the [pre-rotation checklist](kms-backend-security.md#rotation-drivers-and-scheduling-per-backend) before rotating, above all the [upgrade-ordering hard constraint](kms-backend-security.md#upgrade-before-first-rotation-hard-constraint) — never respond to this alert by rotating in the middle of a rolling upgrade. On **Vault Transit**, check `auto_rotate_period` on the key in Vault. On **AWS KMS**, check the key's automatic rotation status in AWS — and do not schedule rotation through the RustFS endpoint, which maps to quota-limited `RotateKeyOnDemand`.
 4. Know the gauge's blind spot on Transit and AWS before chasing a rotation that already happened: only KV2 persists a rotation timestamp, so Transit and AWS keys age from creation permanently and this alert will not clear after a rotation there. Confirm the real cadence at the owning system — the Transit key's version history in Vault, or the key's rotation status in AWS — and treat a confirmed-healthy cadence as a known overstatement of this gauge rather than an overdue key.

rustfs/src/admin/handlers/bucket_meta.rs

@@ -465,16 +465,6 @@ impl Operation for ImportBucketMetadata {
             file_contents.push((file_path, content));
         }
 
-        let durable_quota_import = imported_quota_requires_fleet_proof(&file_contents)?;
-        let quota_fleet_proof =
-            if durable_quota_import {
-                Some(crate::admin::storage_api::acquire_cross_pool_fence_fleet_proof().ok_or_else(|| {
-                    s3_error!(ServiceUnavailable, "durable quota capability is not confirmed across the cluster")
-                })?)
-            } else {
-                None
-            };
-
         // Extract bucket names
         let mut bucket_names = Vec::new();
         for (file_path, _) in &file_contents {
@@ -717,6 +707,21 @@ impl Operation for ImportBucketMetadata {
                 }
 
                 BUCKET_QUOTA_CONFIG_FILE => {
+                    if let Err(e) = serde_json::from_slice::<BucketQuota>(&content) {
+                        warn!(
+                            event = EVENT_ADMIN_BUCKET_META_STATE,
+                            component = LOG_COMPONENT_ADMIN,
+                            subsystem = LOG_SUBSYSTEM_BUCKET_META,
+                            action = "import_bucket_metadata",
+                            result = "config_deserialize_failed",
+                            bucket = %bucket_name,
+                            config_name = %conf_name,
+                            error = %e,
+                            "admin bucket meta state"
+                        );
+                        continue;
+                    }
+
                     let metadata = match bucket_metadatas.get_mut(bucket_name) {
                         Some(m) => m,
                         None => continue,
@@ -825,6 +830,10 @@ impl Operation for ImportBucketMetadata {
             }
         }
 
+        // Persist the assembled metadata to disk. Prior to this, the import only mutated the
+        // in-memory `bucket_metadatas` map and returned 200, silently dropping every imported
+        // config. `metadata_sys::update` loads the on-disk metadata, overwrites the given config
+        // field and saves it, preserving any configs not present in the import archive.
         for (bucket_name, metadata) in &bucket_metadatas {
             for (config_file, data) in imported_configs_to_persist(metadata) {
                 let site_replication_item = imported_config_to_site_replication_item(bucket_name, metadata, config_file, &data)?;
@@ -833,21 +842,7 @@ impl Operation for ImportBucketMetadata {
                 } else {
                     None
                 };
-                let persist_result = if config_file == BUCKET_QUOTA_CONFIG_FILE {
-                    let quota: BucketQuota =
-                        serde_json::from_slice(&data).map_err(|e| s3_error!(InvalidRequest, "invalid bucket quota: {e}"))?;
-                    if quota.uses_durable_reservations() {
-                        let proof = quota_fleet_proof.as_ref().ok_or_else(|| {
-                            s3_error!(ServiceUnavailable, "durable quota capability is not confirmed across the cluster")
-                        })?;
-                        metadata_sys::update_quota_if_incarnation(bucket_name, data, metadata.bucket_incarnation_id, proof).await
-                    } else {
-                        metadata_sys::update_if_incarnation(bucket_name, config_file, data, metadata.bucket_incarnation_id).await
-                    }
-                } else {
-                    metadata_sys::update_if_incarnation(bucket_name, config_file, data, metadata.bucket_incarnation_id).await
-                };
-                if let Err(e) = persist_result {
+                if let Err(e) = metadata_sys::update(bucket_name, config_file, data).await {
                     warn!(
                         event = EVENT_ADMIN_BUCKET_META_STATE,
                         component = LOG_COMPONENT_ADMIN,
@@ -890,26 +885,6 @@ impl Operation for ImportBucketMetadata {
     }
 }
 
-fn imported_quota_requires_fleet_proof(file_contents: &[(String, Vec<u8>)]) -> S3Result<bool> {
-    let mut durable = false;
-    for (file_path, content) in file_contents {
-        let mut parts = file_path.split(SLASH_SEPARATOR);
-        let Some(_bucket) = parts.next() else {
-            continue;
-        };
-        if parts.next() != Some(BUCKET_QUOTA_CONFIG_FILE) {
-            continue;
-        }
-        let quota: BucketQuota =
-            serde_json::from_slice(content).map_err(|e| s3_error!(InvalidRequest, "invalid bucket quota: {e}"))?;
-        if quota.has_unsupported_reservation_protocol() {
-            return Err(s3_error!(InvalidRequest, "unsupported bucket quota reservation protocol"));
-        }
-        durable |= quota.uses_durable_reservations();
-    }
-    Ok(durable)
-}
-
 /// The `(config_file, data)` pairs to persist for an imported bucket's metadata: every non-empty
 /// config field keyed by its on-disk config-file name, as owned data ready for
 /// `metadata_sys::update`. Empty fields are skipped so an import never overwrites an existing
@@ -1101,31 +1076,4 @@ mod import_persist_tests {
 
         assert!(!has_site_replication_item);
     }
-
-    #[test]
-    fn quota_import_preflight_rejects_invalid_and_unknown_protocols() {
-        let missing_limit = vec![(
-            format!("bucket/{BUCKET_QUOTA_CONFIG_FILE}"),
-            br#"{"quota":0,"reservation_protocol":1}"#.to_vec(),
-        )];
-        assert!(imported_quota_requires_fleet_proof(&missing_limit).is_err());
-
-        let unknown_protocol = vec![(
-            format!("bucket/{BUCKET_QUOTA_CONFIG_FILE}"),
-            br#"{"quota":0,"reservation_protocol":2,"reservation_quota":1024}"#.to_vec(),
-        )];
-        assert!(imported_quota_requires_fleet_proof(&unknown_protocol).is_err());
-    }
-
-    #[test]
-    fn quota_import_preflight_requires_proof_only_for_durable_quota() {
-        let legacy = vec![(format!("bucket/{BUCKET_QUOTA_CONFIG_FILE}"), br#"{"quota":1024}"#.to_vec())];
-        assert!(!imported_quota_requires_fleet_proof(&legacy).expect("legacy quota should remain compatible"));
-
-        let durable = vec![(
-            format!("bucket/{BUCKET_QUOTA_CONFIG_FILE}"),
-            serde_json::to_vec(&BucketQuota::new(Some(1024))).expect("durable quota should encode"),
-        )];
-        assert!(imported_quota_requires_fleet_proof(&durable).expect("durable quota should pass preflight"));
-    }
 }

rustfs/src/admin/handlers/quota.rs

@@ -22,7 +22,6 @@ use crate::admin::storage_api::bucket::metadata_sys::{self, BucketMetadataSys};
 use crate::admin::storage_api::bucket::quota::checker::QuotaChecker;
 use crate::admin::storage_api::bucket::quota::{BucketQuota, QuotaError, QuotaOperation};
 use crate::auth::{check_key_valid, get_session_token};
-use crate::error::ApiError;
 use crate::server::ADMIN_PREFIX;
 use hyper::{Method, StatusCode};
 use matchit::Params;
@@ -294,36 +293,16 @@ impl Operation for SetBucketQuotaHandler {
             return Err(s3_error!(InvalidArgument, "{}", rustfs_config::QUOTA_INVALID_TYPE_ERROR_MSG));
         }
 
-        let fleet_proof = if request.quota.is_some() {
-            Some(crate::admin::storage_api::acquire_cross_pool_fence_fleet_proof().ok_or_else(|| {
-                S3Error::with_message(
-                    s3s::S3ErrorCode::ServiceUnavailable,
-                    "durable quota capability is not confirmed across the cluster".to_string(),
-                )
-            })?)
-        } else {
-            None
-        };
-
         let quota = BucketQuota::new(request.quota);
 
         let metadata_sys_lock = bucket_metadata_from_context()
             .ok_or_else(|| s3_error!(InternalError, "{}", rustfs_config::QUOTA_METADATA_SYSTEM_ERROR_MSG))?;
         let mut quota_checker = QuotaChecker::new(metadata_sys_lock.clone());
 
-        let updated_at = match fleet_proof.as_ref() {
-            Some(fleet_proof) => {
-                quota_checker
-                    .set_durable_quota_config_if_incarnation(&bucket, quota.clone(), expected_incarnation_id, fleet_proof)
-                    .await
-            }
-            None => {
-                quota_checker
-                    .set_quota_config_if_incarnation(&bucket, quota.clone(), expected_incarnation_id)
-                    .await
-            }
-        }
-        .map_err(ApiError::from)?;
+        let updated_at = quota_checker
+            .set_quota_config_if_incarnation(&bucket, quota.clone(), expected_incarnation_id)
+            .await
+            .map_err(|e| s3_error!(InternalError, "failed to set quota: {}", e))?;
 
         if let Err(err) = site_replication_bucket_meta_hook(SRBucketMeta {
             bucket: bucket.clone(),

rustfs/src/admin/handlers/site_replication.rs

@@ -29,7 +29,6 @@ use crate::admin::storage_api::bucket::metadata::{
     BUCKET_SSECONFIG, BUCKET_TAGGING_CONFIG, BUCKET_TARGETS_FILE, BUCKET_VERSIONING_CONFIG, OBJECT_LOCK_CONFIG,
 };
 use crate::admin::storage_api::bucket::metadata_sys;
-use crate::admin::storage_api::bucket::quota::BucketQuota;
 use crate::admin::storage_api::bucket::replication;
 use crate::admin::storage_api::bucket::target::{ARN, BucketTarget, BucketTargetType, BucketTargets, Credentials};
 use crate::admin::storage_api::bucket::target_sys::BucketTargetSys;
@@ -7896,35 +7895,9 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
 
     if !skip_config_write {
         if let Some(data) = data {
-            if item.r#type == "quota-config" {
-                let quota: BucketQuota = serde_json::from_slice(&data)
-                    .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid bucket quota: {e}")))?;
-                if quota.has_unsupported_reservation_protocol() {
-                    return Err(S3Error::with_message(
-                        S3ErrorCode::InvalidRequest,
-                        "unsupported bucket quota reservation protocol".to_string(),
-                    ));
-                }
-                if quota.uses_durable_reservations() {
-                    let proof = crate::admin::storage_api::acquire_cross_pool_fence_fleet_proof().ok_or_else(|| {
-                        S3Error::with_message(
-                            S3ErrorCode::ServiceUnavailable,
-                            "durable quota capability is not confirmed across the cluster".to_string(),
-                        )
-                    })?;
-                    metadata_sys::update_quota_if_incarnation(&item.bucket, data, expected_incarnation_id, &proof)
-                        .await
-                        .map_err(ApiError::from)?;
-                } else {
-                    metadata_sys::update_if_incarnation(&item.bucket, config_file, data, expected_incarnation_id)
-                        .await
-                        .map_err(ApiError::from)?;
-                }
-            } else {
-                metadata_sys::update_if_incarnation(&item.bucket, config_file, data, expected_incarnation_id)
-                    .await
-                    .map_err(ApiError::from)?;
-            }
+            metadata_sys::update_if_incarnation(&item.bucket, config_file, data, expected_incarnation_id)
+                .await
+                .map_err(ApiError::from)?;
         } else {
             metadata_sys::delete_if_incarnation(&item.bucket, config_file, expected_incarnation_id)
                 .await

rustfs/src/admin/handlers/table_catalog/tests.rs

@@ -11,6 +11,12 @@ use datafusion::{
 };
 use std::sync::Arc;
 
+use crate::table_catalog::test_support::{
+    manifest_avro_bytes as test_manifest_avro_bytes,
+    manifest_avro_bytes_with_nullable_sequences as test_manifest_avro_bytes_with_nullable_sequences,
+    manifest_list_avro_bytes as test_manifest_list_avro_bytes, manifest_list_avro_entries as test_manifest_list_avro_entries,
+    table_metadata_json as test_table_metadata_json,
+};
 use rustfs_iam::store::{Store as _, UserType};
 use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq};
 
@@ -8033,33 +8039,6 @@ fn trusted_table_commit_backend(
     TableCommitObjectBackend::trusted(backend.clone())
 }
 
-fn test_table_metadata_json(table_uuid: &str, location: &str) -> serde_json::Value {
-    serde_json::json!({
-        "format-version": 2,
-        "table-uuid": table_uuid,
-        "location": location,
-        "last-sequence-number": 0,
-        "last-updated-ms": 1,
-        "last-column-id": 1,
-        "schemas": [{
-            "type": "struct",
-            "schema-id": 0,
-            "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
-        }],
-        "current-schema-id": 0,
-        "partition-specs": [{"spec-id": 0, "fields": []}],
-        "default-spec-id": 0,
-        "last-partition-id": 999,
-        "sort-orders": [{"order-id": 0, "fields": []}],
-        "default-sort-order-id": 0,
-        "properties": {},
-        "snapshots": [],
-        "snapshot-log": [],
-        "metadata-log": [],
-        "refs": {}
-    })
-}
-
 fn test_snapshot_object_key(bucket: &str, location: &str) -> String {
     crate::table_catalog::table_catalog_object_key_from_location(bucket, location)
         .expect("test snapshot object location should be valid")
@@ -8078,184 +8057,6 @@ fn test_parquet_i32_bytes(values: &[i32]) -> Vec<u8> {
     bytes
 }
 
-fn test_manifest_list_avro_bytes(manifest_paths: &[&str], sequence_number: i64, snapshot_id: i64) -> Vec<u8> {
-    let manifests = manifest_paths
-        .iter()
-        .map(|manifest_path| (*manifest_path, 0, sequence_number, snapshot_id))
-        .collect::<Vec<_>>();
-    test_manifest_list_avro_entries_with_partition_specs(&manifests)
-}
-
-fn test_manifest_list_avro_entries(manifests: &[(&str, i64, i64)]) -> Vec<u8> {
-    let manifests = manifests
-        .iter()
-        .map(|(manifest_path, sequence_number, snapshot_id)| (*manifest_path, 0, *sequence_number, *snapshot_id))
-        .collect::<Vec<_>>();
-    test_manifest_list_avro_entries_with_partition_specs(&manifests)
-}
-
-fn test_manifest_list_avro_entries_with_partition_specs(manifests: &[(&str, i32, i64, i64)]) -> Vec<u8> {
-    let schema = apache_avro::Schema::parse_str(
-        r#"
-            {
-              "type": "record",
-              "name": "manifest_file",
-              "fields": [
-                {"name": "manifest_path", "type": "string"},
-                {"name": "manifest_length", "type": "long"},
-                {"name": "partition_spec_id", "type": "int"},
-                {"name": "content", "type": "int"},
-                {"name": "sequence_number", "type": "long"},
-                {"name": "min_sequence_number", "type": "long"},
-                {"name": "added_snapshot_id", "type": "long"},
-                {"name": "added_files_count", "type": "int"},
-                {"name": "existing_files_count", "type": "int"},
-                {"name": "deleted_files_count", "type": "int"},
-                {"name": "added_rows_count", "type": "long"},
-                {"name": "existing_rows_count", "type": "long"},
-                {"name": "deleted_rows_count", "type": "long"}
-              ]
-            }
-            "#,
-    )
-    .expect("manifest list avro schema should parse");
-    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest list writer should initialize");
-    for (manifest_path, partition_spec_id, sequence_number, snapshot_id) in manifests {
-        writer
-            .append_value(apache_avro::types::Value::Record(vec![
-                (
-                    "manifest_path".to_string(),
-                    apache_avro::types::Value::String((*manifest_path).to_string()),
-                ),
-                ("manifest_length".to_string(), apache_avro::types::Value::Long(1)),
-                ("partition_spec_id".to_string(), apache_avro::types::Value::Int(*partition_spec_id)),
-                ("content".to_string(), apache_avro::types::Value::Int(0)),
-                ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
-                ("min_sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
-                ("added_snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
-                ("added_files_count".to_string(), apache_avro::types::Value::Int(1)),
-                ("existing_files_count".to_string(), apache_avro::types::Value::Int(0)),
-                ("deleted_files_count".to_string(), apache_avro::types::Value::Int(0)),
-                ("added_rows_count".to_string(), apache_avro::types::Value::Long(1)),
-                ("existing_rows_count".to_string(), apache_avro::types::Value::Long(0)),
-                ("deleted_rows_count".to_string(), apache_avro::types::Value::Long(0)),
-            ]))
-            .expect("manifest list record should append");
-    }
-    writer.into_inner().expect("manifest list avro bytes should flush")
-}
-
-fn test_manifest_avro_bytes(files: &[(&str, i32, i32, i64, i64)]) -> Vec<u8> {
-    let schema = apache_avro::Schema::parse_str(
-        r#"
-            {
-              "type": "record",
-              "name": "manifest_entry",
-              "fields": [
-                {"name": "status", "type": "int"},
-                {"name": "snapshot_id", "type": "long"},
-                {"name": "sequence_number", "type": "long"},
-                {"name": "file_sequence_number", "type": "long"},
-                {
-                  "name": "data_file",
-                  "type": {
-                    "type": "record",
-                    "name": "data_file",
-                    "fields": [
-                      {"name": "content", "type": "int"},
-                      {"name": "file_path", "type": "string"},
-                      {"name": "record_count", "type": "long"},
-                      {"name": "file_size_in_bytes", "type": "long"}
-                    ]
-                  }
-                }
-              ]
-            }
-            "#,
-    )
-    .expect("manifest avro schema should parse");
-    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
-    for (file_path, content, status, snapshot_id, sequence_number) in files {
-        writer
-            .append_value(apache_avro::types::Value::Record(vec![
-                ("status".to_string(), apache_avro::types::Value::Int(*status)),
-                ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
-                ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
-                ("file_sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
-                (
-                    "data_file".to_string(),
-                    apache_avro::types::Value::Record(vec![
-                        ("content".to_string(), apache_avro::types::Value::Int(*content)),
-                        ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
-                        ("record_count".to_string(), apache_avro::types::Value::Long(1)),
-                        ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
-                    ]),
-                ),
-            ]))
-            .expect("manifest record should append");
-    }
-    writer.into_inner().expect("manifest avro bytes should flush")
-}
-
-fn test_nullable_long(value: Option<i64>) -> apache_avro::types::Value {
-    match value {
-        Some(value) => apache_avro::types::Value::Union(1, Box::new(apache_avro::types::Value::Long(value))),
-        None => apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)),
-    }
-}
-
-fn test_manifest_avro_bytes_with_nullable_sequences(files: &[(&str, i32, i32, i64, Option<i64>)]) -> Vec<u8> {
-    let schema = apache_avro::Schema::parse_str(
-        r#"
-            {
-              "type": "record",
-              "name": "manifest_entry",
-              "fields": [
-                {"name": "status", "type": "int"},
-                {"name": "snapshot_id", "type": "long"},
-                {"name": "sequence_number", "type": ["null", "long"], "default": null},
-                {"name": "file_sequence_number", "type": ["null", "long"], "default": null},
-                {
-                  "name": "data_file",
-                  "type": {
-                    "type": "record",
-                    "name": "data_file",
-                    "fields": [
-                      {"name": "content", "type": "int"},
-                      {"name": "file_path", "type": "string"},
-                      {"name": "record_count", "type": "long"},
-                      {"name": "file_size_in_bytes", "type": "long"}
-                    ]
-                  }
-                }
-              ]
-            }
-            "#,
-    )
-    .expect("manifest avro schema should parse");
-    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
-    for (file_path, content, status, snapshot_id, sequence_number) in files {
-        writer
-            .append_value(apache_avro::types::Value::Record(vec![
-                ("status".to_string(), apache_avro::types::Value::Int(*status)),
-                ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
-                ("sequence_number".to_string(), test_nullable_long(*sequence_number)),
-                ("file_sequence_number".to_string(), test_nullable_long(*sequence_number)),
-                (
-                    "data_file".to_string(),
-                    apache_avro::types::Value::Record(vec![
-                        ("content".to_string(), apache_avro::types::Value::Int(*content)),
-                        ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
-                        ("record_count".to_string(), apache_avro::types::Value::Long(1)),
-                        ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
-                    ]),
-                ),
-            ]))
-            .expect("manifest record should append");
-    }
-    writer.into_inner().expect("manifest avro bytes should flush")
-}
-
 async fn seed_test_manifest_list(
     backend: &TestTableCatalogObjectBackend,
     bucket: &str,

rustfs/src/admin/router.rs

@@ -2901,7 +2901,7 @@ async fn handle_misc_extension_request(req: &mut S3Request<Body>, route: &MiscEx
         MiscExtRoute::ObjectLambda { bucket, object } => {
             let get_req = build_object_lambda_get_request(req, bucket, object)?;
             let usecase = default_object_usecase();
-            let get_resp = usecase.execute_get_object(get_req).await?;
+            let get_resp = Box::pin(usecase.execute_get_object(get_req)).await?;
             invoke_object_lambda_target(req, bucket, object, get_resp).await
         }
         MiscExtRoute::ListenNotification { bucket } => {

rustfs/src/admin/storage_api.rs

@@ -64,9 +64,7 @@ mod ecstore_metrics {
 }
 
 mod ecstore_notification {
-    pub(crate) use crate::storage::storage_api::ecstore_notification::{
-        CrossPoolFenceFleetProofToken, NotificationSys, acquire_cross_pool_fence_fleet_proof,
-    };
+    pub(crate) use crate::storage::storage_api::ecstore_notification::NotificationSys;
 }
 
 #[allow(unused_imports)]
@@ -114,10 +112,6 @@ pub(crate) type TierCreds = ecstore_tier::tier_admin::TierCreds;
 pub(crate) type TierType = ecstore_tier::tier_config::TierType;
 pub(crate) type TierConfigUpdateError = crate::storage::storage_api::TierConfigUpdateError;
 
-pub(crate) fn acquire_cross_pool_fence_fleet_proof() -> Option<ecstore_notification::CrossPoolFenceFleetProofToken> {
-    ecstore_notification::acquire_cross_pool_fence_fleet_proof()
-}
-
 pub(crate) mod runtime_sources {
     pub(crate) type DailyAllTierStats = super::DailyAllTierStats;
     pub(crate) type ECStore = super::ECStore;
@@ -302,15 +296,6 @@ pub(crate) mod metadata_sys {
         super::ecstore_bucket::metadata_sys::update_if_incarnation(bucket, config_file, data, expected_incarnation_id).await
     }
 
-    pub(crate) async fn update_quota_if_incarnation(
-        bucket: &str,
-        data: Vec<u8>,
-        expected_incarnation_id: uuid::Uuid,
-        proof: &super::ecstore_notification::CrossPoolFenceFleetProofToken,
-    ) -> Result<OffsetDateTime> {
-        super::ecstore_bucket::metadata_sys::update_quota_if_incarnation(bucket, data, expected_incarnation_id, proof).await
-    }
-
     pub(crate) async fn capture_bucket_metadata_incarnation(bucket: &str) -> Result<uuid::Uuid> {
         super::ecstore_bucket::metadata_sys::capture_bucket_metadata_incarnation(bucket).await
     }

rustfs/src/app/gating_test_env.rs

@@ -23,9 +23,6 @@
 //! `ECStore` and one metadata-sys initialization exist per test binary.
 
 use super::storage_api::test::bucket::metadata_sys;
-use super::storage_api::test::bucket::quota::BucketQuota;
-use super::storage_api::test::bucket::quota::checker::QuotaChecker;
-use super::storage_api::test::contract::bucket::MakeBucketOptions;
 use super::storage_api::test::contract::bucket::{BucketOperations, BucketOptions};
 use super::storage_api::test::{ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints};
 use super::{context::AppContext, object_traffic_health::ObjectTrafficHealth};
@@ -90,17 +87,6 @@ pub(crate) async fn shared_gating_ecstore() -> Arc<ECStore> {
     crate::storage::storage_api::new_global_notification_sys(endpoint_pools.clone())
         .await
         .expect("initialize notification system for gating test env");
-    let topology_fingerprint =
-        crate::storage::storage_api::heal_control_startup_consumer::heal_topology_fingerprint(&endpoint_pools)
-            .expect("single-node gating topology should hash");
-    crate::storage::storage_api::start_remote_version_state_fleet_probe(topology_fingerprint);
-    tokio::time::timeout(std::time::Duration::from_secs(5), async {
-        while crate::storage::storage_api::ecstore_notification::acquire_cross_pool_fence_fleet_proof().is_none() {
-            tokio::task::yield_now().await;
-        }
-    })
-    .await
-    .expect("single-node cross-pool fence capability proof should publish");
 
     let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
     let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new())
@@ -121,24 +107,6 @@ pub(crate) async fn shared_gating_ecstore() -> Arc<ECStore> {
     ecstore
 }
 
-pub(crate) async fn durable_quota_test_bucket(prefix: &str, limit: u64) -> (Arc<ECStore>, String) {
-    let store = shared_gating_ecstore().await;
-    crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
-    let bucket = format!("{prefix:.30}-{}", uuid::Uuid::new_v4().simple());
-    store
-        .make_bucket(&bucket, &MakeBucketOptions::default())
-        .await
-        .expect("create durable quota test bucket");
-    super::storage_api::test::data_usage::seed_bucket_usage_memory_for_test(&bucket, 0).await;
-    let metadata_sys =
-        crate::app::storage_api::test::get_global_bucket_metadata_sys().expect("test app context should expose bucket metadata");
-    QuotaChecker::new(metadata_sys)
-        .set_quota_config(&bucket, BucketQuota::new(Some(limit)))
-        .await
-        .expect("configure durable quota test bucket");
-    (store, bucket)
-}
-
 pub(crate) async fn shared_gating_ambient() -> Arc<AppContext> {
     let store = shared_gating_ecstore().await;
     if let Some(ambient) = crate::runtime_sources::current_app_context() {

rustfs/src/app/multipart_usecase.rs

@@ -33,7 +33,7 @@ use super::storage_api::multipart_usecase::contract::multipart::{CompletePart, M
 use super::storage_api::multipart_usecase::contract::object::{ObjectIO as _, ObjectOperations as _};
 use super::storage_api::multipart_usecase::contract::range::HTTPRangeSpec;
 use super::storage_api::multipart_usecase::data_usage::{
-    quota_object_size, record_bucket_object_version_write_memory, record_bucket_object_write_memory,
+    record_bucket_object_version_write_memory, record_bucket_object_write_memory,
 };
 use super::storage_api::multipart_usecase::error::{StorageError, is_err_object_not_found, is_err_version_not_found};
 use super::storage_api::multipart_usecase::helper::OperationHelper;
@@ -63,10 +63,10 @@ use super::storage_api::multipart_usecase::{
 };
 use crate::app::object_data_cache::{
     ObjectDataCacheAdapter, invalidate_object_data_cache_after_complete_multipart_success,
-    invalidate_object_data_cache_before_mutation,
+    invalidate_object_data_cache_after_delete_success, invalidate_object_data_cache_before_mutation,
 };
 use crate::app::object_usecase::{
-    acquire_copy_bucket_lifecycle_locks, apply_quota_admission, build_put_like_object_lock_metadata, map_quota_check_outcome,
+    acquire_copy_bucket_lifecycle_locks, build_put_like_object_lock_metadata, map_quota_check_outcome,
     validate_existing_object_lock_for_write,
 };
 use crate::app::runtime_sources::{
@@ -82,8 +82,6 @@ use rustfs_io_metrics::record_s3_op;
 use rustfs_s3_ops::S3Operation;
 use rustfs_targets::EventName;
 use rustfs_utils::CompressionAlgorithm;
-#[cfg(test)]
-use rustfs_utils::http::insert_header;
 use rustfs_utils::http::{
     SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP,
     SUFFIX_SOURCE_REPLICATION_REQUEST, contains_key_str, get_header, get_source_scheme,
@@ -227,8 +225,12 @@ fn internal_object_info_lookup_opts(mut opts: ObjectOptions) -> ObjectOptions {
     opts
 }
 
+fn logical_object_size(info: &ObjectInfo) -> Result<u64, StorageError> {
+    u64::try_from(info.get_actual_size()?).map_err(|_| StorageError::PartMissingOrCorrupt)
+}
+
 fn quota_accounting_object_size(info: &ObjectInfo, fail_closed: bool) -> S3Result<u64> {
-    match quota_object_size(info) {
+    match logical_object_size(info) {
         Ok(size) => Ok(size),
         Err(err) if fail_closed => Err(ApiError::from(err).into()),
         Err(_) => Ok(info.size.max(0) as u64),
@@ -504,7 +506,11 @@ impl DefaultMultipartUsecase {
             Ok(existing_obj_info) => {
                 validate_existing_object_lock_for_write(&existing_obj_info, &current_opts)?;
                 let physical_size = existing_obj_info.size.max(0) as u64;
-                let logical_size = quota_object_size(&existing_obj_info);
+                let logical_size = if opts.replication_request {
+                    Ok(physical_size)
+                } else {
+                    logical_object_size(&existing_obj_info)
+                };
                 Some((physical_size, logical_size))
             }
             Err(err) => {
@@ -553,19 +559,32 @@ impl DefaultMultipartUsecase {
         };
 
         let quota_metadata_sys = self.bucket_metadata_sys();
-        let mut quota_enabled = false;
         if let Some(metadata_sys) = quota_metadata_sys.as_ref() {
             let quota_checker = QuotaChecker::new(metadata_sys.clone());
             let check_result =
                 map_quota_check_outcome(&bucket, quota_checker.check_quota(&bucket, QuotaOperation::PutObject, 0).await)?;
-            quota_enabled = check_result.quota_limit.is_some();
-            apply_quota_admission(&mut opts, &check_result)?;
+            // Ciphertext-passthrough replication parts use a different size basis and retain
+            // the existing post-commit accounting path until they carry a trusted logical-size proof.
+            if !opts.replication_request
+                && let Some(quota_limit) = check_result.quota_limit
+            {
+                let installed = check_result
+                    .current_usage
+                    .is_some_and(|current_usage| opts.set_quota_admission(current_usage, quota_limit));
+                if !installed {
+                    return Err(S3Error::with_message(
+                        S3ErrorCode::ServiceUnavailable,
+                        "Bucket quota check temporarily unavailable, please retry".to_string(),
+                    ));
+                }
+            }
         }
 
         let previous_current_size = match previous_current_sizes {
-            Some((_, Ok(logical_size))) if quota_enabled => Some(logical_size),
-            Some((_, Err(err))) if quota_enabled => return Err(ApiError::from(err).into()),
-            Some((physical_size, _)) => Some(physical_size),
+            Some((physical_size, _)) if opts.replication_request => Some(physical_size),
+            Some((_, Ok(logical_size))) => Some(logical_size),
+            Some((_, Err(err))) if opts.quota_admission.is_some() => return Err(ApiError::from(err).into()),
+            Some((physical_size, Err(_))) => Some(physical_size),
             None => None,
         };
 
@@ -577,8 +596,37 @@ impl DefaultMultipartUsecase {
         let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await;
         record_capacity_write(Some(capacity_scope_token)).await;
 
-        if quota_metadata_sys.is_some() {
-            let committed_size = quota_accounting_object_size(&obj_info, quota_enabled)?;
+        if let Some(metadata_sys) = quota_metadata_sys.as_ref() {
+            if opts.replication_request {
+                let quota_checker = QuotaChecker::new(metadata_sys.clone());
+                match quota_checker
+                    .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size.max(0) as u64)
+                    .await
+                {
+                    Ok(check_result) if !check_result.allowed => {
+                        let _ = store.delete_object(&bucket, &key, ObjectOptions::default()).await;
+                        let _ = invalidate_object_data_cache_after_delete_success(&cache_adapter, &bucket, &key).await;
+                        return Err(S3Error::with_message(
+                            S3ErrorCode::InvalidRequest,
+                            format!(
+                                "Bucket quota exceeded. Current usage: {} bytes, limit: {} bytes",
+                                check_result.current_usage.unwrap_or(0),
+                                check_result.quota_limit.unwrap_or(0)
+                            ),
+                        ));
+                    }
+                    Err(err) => {
+                        warn!("Quota check failed for bucket {} after multipart completion: {}", bucket, err);
+                    }
+                    Ok(_) => {}
+                }
+            }
+
+            let committed_size = if opts.replication_request {
+                obj_info.size.max(0) as u64
+            } else {
+                quota_accounting_object_size(&obj_info, opts.quota_admission.is_some())?
+            };
             if versioned {
                 record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await;
             } else {
@@ -769,20 +817,6 @@ impl DefaultMultipartUsecase {
         let ciphertext_passthrough = replication_authorized
             && get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true")
             && rustfs_utils::http::ssec_transport_to_stored_metadata(&req.headers).is_some();
-        if ciphertext_passthrough && let Some(metadata_sys) = self.bucket_metadata_sys() {
-            let check_result = map_quota_check_outcome(
-                &bucket,
-                QuotaChecker::new(metadata_sys)
-                    .check_quota(&bucket, QuotaOperation::PutObject, 0)
-                    .await,
-            )?;
-            if check_result.quota_limit.is_some() {
-                return Err(S3Error::with_message(
-                    S3ErrorCode::InvalidRequest,
-                    "SSE-C ciphertext replication is unavailable for quota-enabled buckets".to_string(),
-                ));
-            }
-        }
         if ciphertext_passthrough {
             insert_str(&mut metadata, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, "true".to_string());
         }
@@ -1616,24 +1650,6 @@ mod tests {
 
         assert_eq!(quota_accounting_object_size(&info, true).expect("logical size should resolve"), 8192);
         assert_eq!(quota_accounting_object_size(&info, false).expect("logical size should resolve"), 8192);
-
-        let mut poisoned_metadata = HashMap::new();
-        insert_str(&mut poisoned_metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "S2".to_string());
-        insert_str(&mut poisoned_metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "1".to_string());
-        let poisoned = ObjectInfo {
-            size: 17,
-            parts: Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
-                size: 4096,
-                actual_size: 4096,
-                ..Default::default()
-            }]),
-            user_defined: Arc::new(poisoned_metadata),
-            ..Default::default()
-        };
-        assert_eq!(
-            quota_accounting_object_size(&poisoned, true).expect("persisted part size must be charged"),
-            4096
-        );
     }
 
     #[test]
@@ -2042,14 +2058,30 @@ mod tests {
     #[tokio::test]
     #[serial_test::serial]
     async fn compressed_complete_records_logical_quota_usage_and_overwrite_delta() {
-        let (store, bucket) = crate::app::gating_test_env::durable_quota_test_bucket("compressed-complete-quota", 16_384).await;
+        use crate::app::storage_api::multipart_usecase::bucket::quota::BucketQuota;
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+        use crate::app::storage_api::test::data_usage::seed_bucket_usage_memory_for_test;
+
+        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+        crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
+
+        let bucket = format!("compressed-complete-quota-{}", Uuid::new_v4());
         let object = "object";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create compressed quota bucket");
+        seed_bucket_usage_memory_for_test(&bucket, 0).await;
 
         let usecase = DefaultMultipartUsecase::from_global();
         let metadata_sys = usecase
             .bucket_metadata_sys()
             .expect("test app context should expose bucket metadata");
-        let quota_checker = QuotaChecker::new(metadata_sys);
+        let mut quota_checker = QuotaChecker::new(metadata_sys);
+        quota_checker
+            .set_quota_config(&bucket, BucketQuota::new(Some(16_384)))
+            .await
+            .expect("configure bucket quota");
 
         for (actual_size, payload_byte) in [(8192_i64, 0x61), (4096_i64, 0x62)] {
             let mut create_opts = ObjectOptions::default();
@@ -2094,173 +2126,6 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn create_multipart_rejects_ciphertext_replication_before_parts_are_staged() {
-        let (_store, bucket) = crate::app::gating_test_env::durable_quota_test_bucket("ciphertext-multipart-quota", 4096).await;
-        let usecase = DefaultMultipartUsecase::from_global();
-        let input = CreateMultipartUploadInput::builder()
-            .bucket(bucket)
-            .key("object".to_string())
-            .build()
-            .expect("create multipart request should build");
-        let mut request = build_request(input, Method::POST);
-        insert_header(&mut request.headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
-        request
-            .headers
-            .insert(rustfs_utils::http::REPLICATION_SSEC_ALGORITHM_HEADER, HeaderValue::from_static("AES256"));
-        request.extensions.insert(crate::storage::access::ReqInfo {
-            replication_request_authorized: true,
-            ..Default::default()
-        });
-
-        let err = usecase
-            .execute_create_multipart_upload(request)
-            .await
-            .expect_err("quota-enabled ciphertext multipart replication should fail before upload creation");
-        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
-    }
-
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn concurrent_completions_share_durable_bucket_quota_reservations() {
-        let (store, bucket) = crate::app::gating_test_env::durable_quota_test_bucket("concurrent-complete-quota", 6000).await;
-
-        let usecase = DefaultMultipartUsecase::from_global();
-
-        let mut inputs = Vec::new();
-        for object in ["first", "second"] {
-            let upload = store
-                .new_multipart_upload(&bucket, object, &ObjectOptions::default())
-                .await
-                .expect("create concurrent multipart upload");
-            let mut reader = PutObjReader::from_vec(vec![0x71; 4096]);
-            let part = store
-                .put_object_part(&bucket, object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default())
-                .await
-                .expect("stage concurrent multipart part");
-            inputs.push(
-                CompleteMultipartUploadInput::builder()
-                    .bucket(bucket.clone())
-                    .key(object.to_string())
-                    .upload_id(upload.upload_id)
-                    .multipart_upload(Some(CompletedMultipartUpload {
-                        parts: Some(vec![CompletedPart {
-                            part_number: Some(1),
-                            e_tag: part.etag.map(|etag| to_s3s_etag(&etag)),
-                            ..Default::default()
-                        }]),
-                    }))
-                    .build()
-                    .expect("build concurrent completion input"),
-            );
-        }
-
-        let first_usecase = usecase.clone();
-        let first = first_usecase.execute_complete_multipart_upload(build_request(inputs.remove(0), Method::POST));
-        let second = usecase.execute_complete_multipart_upload(build_request(inputs.remove(0), Method::POST));
-        let (first, second) = tokio::join!(first, second);
-
-        assert_eq!(usize::from(first.is_ok()) + usize::from(second.is_ok()), 1);
-        let denied = first.err().or_else(|| second.err()).expect("one completion must be denied");
-        assert_eq!(denied.code(), &S3ErrorCode::InvalidRequest);
-    }
-
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn multipart_completion_rejects_rotated_quota_capability_before_rename() {
-        use crate::app::storage_api::test::set_disk::{MultipartCommitBarrier, MultipartCommitPause};
-
-        let (store, bucket) = crate::app::gating_test_env::durable_quota_test_bucket("rotated-proof-mpu-quota", 4096).await;
-        let object = "object";
-        let upload = store
-            .new_multipart_upload(&bucket, object, &ObjectOptions::default())
-            .await
-            .expect("create multipart upload");
-        let mut reader = PutObjReader::from_vec(vec![0x78; 4096]);
-        let part = store
-            .put_object_part(&bucket, object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default())
-            .await
-            .expect("stage multipart part");
-        let barrier = MultipartCommitBarrier::install(&bucket, object, MultipartCommitPause::BeforeQuotaRename);
-        let complete_store = Arc::clone(&store);
-        let complete_bucket = bucket.clone();
-        let upload_id = upload.upload_id.clone();
-        let complete = tokio::spawn(async move {
-            complete_store
-                .complete_multipart_upload(
-                    &complete_bucket,
-                    object,
-                    &upload_id,
-                    vec![CompletePart {
-                        part_num: 1,
-                        etag: part.etag,
-                        ..Default::default()
-                    }],
-                    &ObjectOptions::default(),
-                )
-                .await
-        });
-        barrier.wait_until_paused().await;
-        assert!(
-            crate::storage::storage_api::ecstore_notification::rotate_cross_pool_fence_fleet_proof_for_test(),
-            "the gating environment must have a current fleet proof"
-        );
-        barrier.release();
-
-        let err = complete
-            .await
-            .expect("completion task should not panic")
-            .expect_err("a replaced fleet proof must fence multipart rename");
-        assert!(matches!(
-            err,
-            StorageError::NamespaceLockQuorumUnavailable {
-                mode: "quota_reservation",
-                ..
-            }
-        ));
-        store
-            .get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default())
-            .await
-            .expect("proof rotation must preserve the multipart upload for retry");
-    }
-
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn data_movement_multipart_completion_has_zero_quota_growth() {
-        let (store, bucket) = crate::app::gating_test_env::durable_quota_test_bucket("data-movement-mpu-quota", 0).await;
-        let object = "object";
-        let mut movement_opts = ObjectOptions {
-            data_movement: true,
-            ..Default::default()
-        };
-        let upload = store
-            .new_multipart_upload(&bucket, object, &movement_opts)
-            .await
-            .expect("create data-movement multipart upload");
-        let mut reader = PutObjReader::from_vec(vec![0x7a; 4096]);
-        let part = store
-            .put_object_part(&bucket, object, &upload.upload_id, 1, &mut reader, &movement_opts)
-            .await
-            .expect("stage data-movement multipart part");
-        movement_opts.preserve_etag = Some("movement-etag".to_string());
-        let completed = store
-            .complete_multipart_upload(
-                &bucket,
-                object,
-                &upload.upload_id,
-                vec![CompletePart {
-                    part_num: 1,
-                    etag: part.etag,
-                    ..Default::default()
-                }],
-                &movement_opts,
-            )
-            .await
-            .expect("moving an already-accounted multipart object between pools must have zero quota growth");
-        assert_eq!(completed.size, 4096);
-    }
-
     #[tokio::test]
     #[serial_test::serial]
     async fn rejected_empty_parts_preserve_existing_object_and_staging() {

rustfs/src/app/storage_api.rs

@@ -47,8 +47,6 @@ pub(crate) mod capacity {
 pub(crate) mod data_usage {
     use std::sync::Arc;
 
-    pub(crate) use crate::storage::storage_api::ecstore_data_usage::quota_object_size;
-
     pub(crate) async fn apply_bucket_usage_memory_overlay(data_usage_info: &mut rustfs_data_usage::DataUsageInfo) {
         crate::storage::storage_api::ecstore_data_usage::apply_bucket_usage_memory_overlay(data_usage_info).await;
     }
@@ -1000,7 +998,7 @@ pub(crate) mod options {
 }
 
 pub(crate) mod request_context {
-    pub(crate) use crate::storage::storage_api::request_context_consumer::{RequestContext, spawn_traced};
+    pub(crate) use crate::storage::storage_api::request_context_consumer::{RequestContext, spawn_traced, spawn_traced_join};
 }
 
 pub(crate) mod sse {
@@ -1214,14 +1212,4 @@ pub(crate) mod test {
     pub(crate) use crate::storage::storage_api::{
         ECStore, Endpoint, Endpoints, PoolEndpoints, StorageObjectInfo, StorageObjectOptions, StoragePutObjReader,
     };
-    pub(crate) mod set_disk {
-        pub(crate) use crate::storage::storage_api::ecstore_set_disk::{
-            MultipartCommitBarrier, MultipartCommitPause, PutObjectCommitBarrier, PutObjectCommitPause,
-            fail_next_quota_ledger_save_for_test,
-        };
-    }
-
-    pub(crate) mod metadata_sys {
-        pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::ConfigWriteLockProbe;
-    }
 }

rustfs/src/storage/ecfs.rs

@@ -301,7 +301,7 @@ impl S3 for FS {
     #[instrument(level = "debug", skip(self, req))]
     async fn copy_object(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
         let usecase = s3_api::object_usecase_for(self);
-        usecase.execute_copy_object(req).await
+        Box::pin(usecase.execute_copy_object(req)).await
     }
 
     #[instrument(
@@ -704,7 +704,7 @@ impl S3 for FS {
     async fn get_object(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
         crate::hp_guard!("S3::get_object");
         let usecase = s3_api::object_usecase_for(self);
-        usecase.execute_get_object(req).await
+        Box::pin(usecase.execute_get_object(req)).await
     }
 
     async fn get_object_acl(&self, req: S3Request<GetObjectAclInput>) -> S3Result<S3Response<GetObjectAclOutput>> {

rustfs/src/storage/request_context.rs

@@ -257,6 +257,15 @@ where
     tokio::spawn(tracing::Instrument::instrument(fut, tracing::Span::current()));
 }
 
+/// Spawn a request-internal task and return its join handle to the caller.
+pub fn spawn_traced_join<F>(fut: F) -> tokio::task::JoinHandle<F::Output>
+where
+    F: std::future::Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    tokio::spawn(tracing::Instrument::instrument(fut, tracing::Span::current()))
+}
+
 #[cfg(test)]
 #[allow(unused_imports)]
 mod tests {

rustfs/src/storage/rpc/node_service.rs

@@ -153,7 +153,9 @@ fn remove_heal_control_replay(
 
 static HEAL_CONTROL_REPLAY_CACHE: OnceLock<tokio::sync::Mutex<HashMap<String, Arc<HealControlReplayEntry>>>> = OnceLock::new();
 static NODE_CAPABILITY_SERVER_EPOCH: LazyLock<Uuid> = LazyLock::new(Uuid::new_v4);
-const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 1;
+// RUSTFS_COMPAT_TODO(cross-pool-fence-v1): advertise unsupported during predeployment. Remove after composite acquisition,
+// activation fencing, fleet proof, commit-time proof revalidation, and fail-closed revocation ship together.
+const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 0;
 
 fn admit_heal_control_replay(
     replay_cache: &mut HashMap<String, Arc<HealControlReplayEntry>>,
@@ -3340,7 +3342,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn cross_pool_fence_probe_authenticates_supported_v1_state() {
+    async fn cross_pool_fence_probe_authenticates_unsupported_rollout_state() {
         let _ = rustfs_credentials::set_global_rpc_secret("cross-pool-fence-node-service-test-secret".to_string());
         let endpoints = heal_control_test_endpoints_with_coordinator("node-0", true);
         assert!(
@@ -3405,7 +3407,7 @@ mod tests {
 
         assert!(response.success);
         assert_eq!(response.error_info, None);
-        assert_eq!(&response.result[..4], &1_u32.to_be_bytes());
+        assert_eq!(&response.result[..4], &0_u32.to_be_bytes());
         let (topology_member, process_epoch) = rustfs_protos::decode_remote_version_state_capability(&response.result[4..])
             .expect("capability identity should decode");
         assert_eq!(topology_member, "node-a:9000");

rustfs/src/storage/rpc/node_service/disk.rs

@@ -37,28 +37,19 @@ const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512;
 const FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT: usize = 1024;
 const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
 
-fn snapshot_lease_response(result: Result<SnapshotLeaseToken, DiskError>) -> Response<SnapshotLeaseResponse> {
-    match result {
-        Ok(token) => Response::new(SnapshotLeaseResponse {
-            success: true,
-            token: token.as_bytes().to_vec().into(),
-            protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
-            error: None,
-        }),
-        Err(err) => Response::new(SnapshotLeaseResponse {
-            success: false,
-            token: Bytes::new(),
-            protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
-            error: Some(err.into()),
-        }),
-    }
-}
-
 struct DecodedRpcPayload<T> {
     value: T,
     from_msgpack: bool,
 }
 
+fn snapshot_lease_disabled_response() -> SnapshotLeaseResponse {
+    SnapshotLeaseResponse {
+        success: false,
+        token: Bytes::new(),
+        protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
+        error: Some(DiskError::UnsupportedDisk.into()),
+    }
+}
 fn decode_msgpack_or_json<T: DeserializeOwned>(
     binary: &[u8],
     json: &str,
@@ -250,12 +241,7 @@ impl NodeService {
             rustfs_protos::canonical_snapshot_lease_request_body(request.get_ref()),
             "acquire_snapshot_lease",
         )?;
-        let request = request.into_inner();
-        let result = match self.find_disk(&request.disk).await {
-            Some(disk) => disk.acquire_snapshot_lease(&request.volume, &request.path).await,
-            None => Err(DiskError::other("cannot find disk")),
-        };
-        Ok(snapshot_lease_response(result))
+        Ok(Response::new(snapshot_lease_disabled_response()))
     }
 
     pub(super) async fn handle_renew_snapshot_lease(
@@ -267,14 +253,7 @@ impl NodeService {
             rustfs_protos::canonical_snapshot_lease_renew_request_body(request.get_ref()),
             "renew_snapshot_lease",
         )?;
-        let request = request.into_inner();
-        let token =
-            SnapshotLeaseToken::from_slice(&request.token).map_err(|_| Status::invalid_argument("invalid lease token"))?;
-        let result = match self.find_disk(&request.disk).await {
-            Some(disk) => disk.renew_snapshot_lease(&request.volume, &request.path, token).await,
-            None => Err(DiskError::other("cannot find disk")),
-        };
-        Ok(snapshot_lease_response(result))
+        Ok(Response::new(snapshot_lease_disabled_response()))
     }
 
     pub(super) async fn handle_release_snapshot_lease(
@@ -287,11 +266,8 @@ impl NodeService {
             "release_snapshot_lease",
         )?;
         let request = request.into_inner();
-        let token = if request.token.as_ref() == SnapshotLeaseToken::revoke_all().as_bytes() {
-            SnapshotLeaseToken::revoke_all()
-        } else {
-            SnapshotLeaseToken::from_slice(&request.token).map_err(|_| Status::invalid_argument("invalid lease token"))?
-        };
+        let token =
+            SnapshotLeaseToken::from_slice(&request.token).map_err(|_| Status::invalid_argument("invalid lease token"))?;
         let Some(disk) = self.find_disk(&request.disk).await else {
             return Ok(Response::new(SnapshotLeaseMutationResponse {
                 success: false,
@@ -1518,11 +1494,11 @@ mod tests {
     use super::{
         compat_response_json, decode_msgpack_or_json, decode_rename_data_request_file_info,
         encode_batch_read_version_response_payloads, encode_file_info_msgpack, encode_msgpack, encode_msgpack_named,
-        encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
+        encode_read_multiple_response_payloads, encode_rename_data_response_payloads, snapshot_lease_disabled_response,
     };
     use crate::storage::storage_api::ReadMultipleResp;
-    use crate::storage::storage_api::RenameDataResp;
     use crate::storage::storage_api::rpc_consumer::node_service::BatchReadVersionResp;
+    use crate::storage::storage_api::{DiskError, RenameDataResp};
     use rustfs_filemeta::FileInfo;
     use rustfs_io_metrics::internode_metrics::global_internode_metrics;
     use serde::{Deserialize, Serialize};
@@ -1533,6 +1509,17 @@ mod tests {
         count: u32,
     }
 
+    #[test]
+    fn snapshot_lease_acquire_and_renew_fail_closed() {
+        let response = snapshot_lease_disabled_response();
+        let expected_error = DiskError::UnsupportedDisk.into();
+
+        assert!(!response.success);
+        assert!(response.token.is_empty());
+        assert_eq!(response.protocol_version, 1);
+        assert_eq!(response.error, Some(expected_error));
+    }
+
     #[test]
     fn decode_msgpack_or_json_prefers_binary_payload() {
         let payload = SamplePayload {

rustfs/src/storage/sse.rs

@@ -119,8 +119,14 @@ use rustfs_utils::http::object_encryption_keys::{
     MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER, MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER,
     MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER, MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER,
     MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER, MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER,
-    MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER, SSEC_ORIGINAL_SIZE_HEADER,
+    MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER, SSEC_ORIGINAL_SIZE_HEADER, normalize_managed_metadata,
+    stored_managed_encryption_key,
 };
+// The managed-SSE classifier lives in the shared encryption-keys module so the
+// scanner can reuse it (backlog#1643 PR-B0); these re-exports keep the
+// historical `crate::storage::sse` paths compiling.
+pub use rustfs_utils::http::object_encryption_keys::SSEType;
+pub(crate) use rustfs_utils::http::object_encryption_keys::contains_managed_encryption_metadata;
 #[cfg(feature = "rio-v2")]
 const MINIO_INTERNAL_ENCRYPTION_SEAL_ALGORITHM: &str = "DAREv2-HMAC-SHA256";
 #[cfg(feature = "rio-v2")]
@@ -783,28 +789,6 @@ pub struct DecryptionMaterial {
     pub key_kind: EncryptionKeyKind,
 }
 
-/// Type of encryption used
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum SSEType {
-    /// SSE-S3 (AES256)
-    SseS3,
-    /// SSE-KMS (aws:kms)
-    SseKms,
-    /// SSE-C (customer-provided key)
-    SseC,
-}
-
-impl SSEType {
-    /// Stable scheme name for audit consumers.
-    fn audit_label(self) -> &'static str {
-        match self {
-            SSEType::SseS3 => "SSE-S3",
-            SSEType::SseKms => "SSE-KMS",
-            SSEType::SseC => "SSE-C",
-        }
-    }
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum EncryptionKeyKind {
     Direct,
@@ -1064,28 +1048,6 @@ pub async fn authorize_sse_kms_object_read(
     result
 }
 
-/// Resolve the scheme and KMS key a stored managed-SSE object was wrapped with.
-///
-/// Mirrors the lookup `apply_managed_decryption_material` performs, so both agree on
-/// which key a read is authorized against.
-fn stored_managed_encryption_key(metadata: &HashMap<String, String>) -> Option<(SSEType, String)> {
-    if !contains_managed_encryption_metadata(metadata) {
-        return None;
-    }
-
-    let sse_type = match metadata.get("x-amz-server-side-encryption")?.as_str() {
-        ServerSideEncryption::AWS_KMS => SSEType::SseKms,
-        _ => SSEType::SseS3,
-    };
-    let key_id = normalize_managed_metadata(metadata)
-        .get(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
-        .or_else(|| metadata.get("x-amz-server-side-encryption-aws-kms-key-id"))
-        .cloned()
-        .unwrap_or_else(|| "default".to_string());
-
-    Some((sse_type, key_id))
-}
-
 // ============================================================================
 // Data-plane KMS audit attachment (SSE-S3 / SSE-KMS)
 // ============================================================================
@@ -1339,7 +1301,9 @@ fn envelope_master_key_version(envelope_bytes: &[u8]) -> Option<u32> {
 /// Master-key version of the envelope stored on an object, for the audit
 /// summary of a read against that object.
 fn stored_envelope_master_key_version(metadata: &HashMap<String, String>) -> Option<u32> {
-    let encoded = normalize_managed_metadata(metadata);
+    // No context recoder: the recode only ever inserts the context key, which
+    // this lookup never reads, so the normalized result is identical without it.
+    let encoded = normalize_managed_metadata(metadata, None);
     let encoded = encoded.get(INTERNAL_ENCRYPTION_KEY_HEADER)?;
     let envelope = BASE64_STANDARD.decode(encoded).ok()?;
     envelope_master_key_version(&envelope)
@@ -2487,7 +2451,7 @@ async fn apply_managed_decryption_material_inner(
 
     // Safe: presence is guaranteed by the contains_key check above.
     let server_side_encryption = metadata.get("x-amz-server-side-encryption").cloned().unwrap_or_default();
-    let normalized_metadata = normalize_managed_metadata(metadata);
+    let normalized_metadata = normalize_managed_metadata(metadata, Some(recode_minio_kms_context));
 
     let encryption_type = match server_side_encryption.as_str() {
         ServerSideEncryption::AES256 => SSEType::SseS3,
@@ -3229,14 +3193,6 @@ pub fn mark_encrypted_multipart_metadata(metadata: &mut HashMap<String, String>)
     metadata.insert(MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER.to_string(), String::new());
 }
 
-pub(crate) fn contains_managed_encryption_metadata(metadata: &HashMap<String, String>) -> bool {
-    metadata.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
-        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER)
-        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER)
-        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER)
-        || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER)
-}
-
 #[cfg(feature = "rio-v2")]
 fn is_legacy_rustfs_managed_metadata(metadata: &HashMap<String, String>) -> bool {
     metadata.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
@@ -3282,47 +3238,16 @@ fn parse_minio_managed_sealed_key(
     Ok(Some(ManagedSealedKey { iv, sealed_key }))
 }
 
-fn normalize_managed_metadata(metadata: &HashMap<String, String>) -> HashMap<String, String> {
-    let mut normalized = metadata.clone();
-
-    if !normalized.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER)
-        && let Some(value) = metadata
-            .get(MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER)
-            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER))
-            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER))
-            .or_else(|| metadata.get(MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER))
-    {
-        normalized.insert(INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), value.clone());
-    }
-
-    if !normalized.contains_key(INTERNAL_ENCRYPTION_IV_HEADER)
-        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_IV_HEADER)
-    {
-        normalized.insert(INTERNAL_ENCRYPTION_IV_HEADER.to_string(), value.clone());
-    }
-
-    if !normalized.contains_key(INTERNAL_ENCRYPTION_ALGORITHM_HEADER)
-        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER)
-    {
-        normalized.insert(INTERNAL_ENCRYPTION_ALGORITHM_HEADER.to_string(), value.clone());
-    }
-
-    if !normalized.contains_key(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
-        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER)
-    {
-        normalized.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), value.clone());
-    }
-
-    if !normalized.contains_key(INTERNAL_ENCRYPTION_CONTEXT_HEADER)
-        && let Some(value) = metadata.get(MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER)
-        && let Ok(decoded) = BASE64_STANDARD.decode(value)
-        && let Ok(context) = serde_json::from_slice::<HashMap<String, String>>(&decoded)
-        && let Ok(encoded) = serde_json::to_string(&context)
-    {
-        normalized.insert(INTERNAL_ENCRYPTION_CONTEXT_HEADER.to_string(), encoded);
-    }
-
-    normalized
+/// Recodes a stored MinIO KMS context value (base64-wrapped JSON) into the
+/// plain-JSON form RustFS stores under [`INTERNAL_ENCRYPTION_CONTEXT_HEADER`].
+///
+/// Injected into the shared [`normalize_managed_metadata`] because the shared
+/// crate carries no JSON codec; any decode failure returns `None`, which skips
+/// the context mapping exactly like the historical inline `if let Ok` chain.
+fn recode_minio_kms_context(value: &str) -> Option<String> {
+    let decoded = BASE64_STANDARD.decode(value).ok()?;
+    let context = serde_json::from_slice::<HashMap<String, String>>(&decoded).ok()?;
+    serde_json::to_string(&context).ok()
 }
 
 // ============================================================================
@@ -3473,9 +3398,9 @@ mod tests {
         encryption_material_to_metadata, extract_server_side_encryption_from_headers, extract_ssec_params_from_headers,
         extract_ssekms_context_from_headers, generate_ssec_nonce, is_managed_sse, kms_operation_error,
         map_get_object_reader_error, mark_encrypted_multipart_metadata, md5_base64, normalize_managed_metadata,
-        reset_sse_dek_provider, resolve_effective_kms_key_id, sse_decryption, sse_encryption, sse_prepare_encryption,
-        strip_managed_encryption_metadata, validate_sse_headers_for_read, validate_sse_headers_for_write, validate_ssec_for_read,
-        validate_ssec_params, verify_ssec_key_match,
+        recode_minio_kms_context, reset_sse_dek_provider, resolve_effective_kms_key_id, sse_decryption, sse_encryption,
+        sse_prepare_encryption, strip_managed_encryption_metadata, validate_sse_headers_for_read, validate_sse_headers_for_write,
+        validate_ssec_for_read, validate_ssec_params, verify_ssec_key_match,
     };
     #[cfg(feature = "rio-v2")]
     use super::{
@@ -3484,6 +3409,38 @@ mod tests {
     };
     use rustfs_utils::http::headers::SSEC_ALGORITHM_HEADER;
 
+    /// backlog#1643 PR-B0 acceptance guard: the managed-SSE classifier must
+    /// have exactly one definition — in the shared encryption-keys module —
+    /// so the scanner and the S3 layer can never disagree on attribution.
+    /// This module may only re-export or call it.
+    #[test]
+    fn managed_sse_classifier_has_exactly_one_definition() {
+        let classifier_fns = [
+            "contains_managed_encryption_metadata",
+            "normalize_managed_metadata",
+            "stored_managed_encryption_key",
+        ];
+
+        let sse_src = include_str!("sse.rs");
+        let shared_src =
+            std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/../crates/utils/src/http/object_encryption_keys.rs"))
+                .expect("shared encryption-keys module should be readable");
+
+        for name in classifier_fns {
+            // Built at runtime so this test's own source cannot satisfy the scan.
+            let definition = format!("fn {name}(");
+            assert!(
+                !sse_src.contains(&definition),
+                "{name} must not be redefined in storage/sse.rs; call the shared rustfs_utils::http::object_encryption_keys implementation instead"
+            );
+            assert_eq!(
+                shared_src.matches(&definition).count(),
+                1,
+                "{name} must be defined exactly once, in the shared encryption-keys module"
+            );
+        }
+    }
+
     #[test]
     fn ssec_read_headers_are_sensitive() {
         let headers = super::build_ssec_read_headers(
@@ -4869,7 +4826,7 @@ mod tests {
             (MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER.to_string(), "default".to_string()),
         ]);
 
-        let normalized = normalize_managed_metadata(&metadata);
+        let normalized = normalize_managed_metadata(&metadata, Some(recode_minio_kms_context));
 
         assert_eq!(
             normalized.get(INTERNAL_ENCRYPTION_KEY_HEADER),

rustfs/src/storage/storage_api.rs

@@ -203,7 +203,9 @@ pub(crate) mod options_consumer {
 }
 
 pub(crate) mod request_context_consumer {
-    pub(crate) use super::super::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced};
+    pub(crate) use super::super::request_context::{
+        RequestContext, extract_request_id_from_headers, spawn_traced, spawn_traced_join,
+    };
 }
 
 pub(crate) mod rpc_consumer {
@@ -428,7 +430,7 @@ pub(crate) mod ecstore_config {
 pub(crate) mod ecstore_data_usage {
     pub(crate) use rustfs_ecstore::api::data_usage::{
         apply_bucket_usage_memory_overlay, init_compression_total_memory_from_backend, load_admin_data_usage_from_backend_cached,
-        load_data_usage_from_backend, quota_object_size, record_bucket_delete_marker_memory, record_bucket_object_delete_memory,
+        load_data_usage_from_backend, record_bucket_delete_marker_memory, record_bucket_object_delete_memory,
         record_bucket_object_version_write_memory, record_bucket_object_write_memory,
         record_bucket_object_write_unknown_previous_memory, store_compression_total_in_backend,
     };
@@ -485,12 +487,8 @@ pub(crate) mod ecstore_metrics {
 
 #[allow(unused_imports)]
 pub(crate) mod ecstore_notification {
-    #[cfg(test)]
-    pub(crate) use rustfs_ecstore::api::notification::rotate_cross_pool_fence_fleet_proof_for_test;
     pub(crate) use rustfs_ecstore::api::notification::{
-        CrossPoolFenceFleetProofToken, NotificationSys, acquire_cross_pool_fence_fleet_proof,
-        cross_pool_fence_fleet_proof_matches, get_global_notification_sys, new_global_notification_sys,
-        start_remote_version_state_fleet_probe,
+        NotificationSys, get_global_notification_sys, new_global_notification_sys, start_remote_version_state_fleet_probe,
     };
 }
 
@@ -548,11 +546,6 @@ pub(crate) mod ecstore_test_support {
 }
 
 pub(crate) mod ecstore_set_disk {
-    #[cfg(test)]
-    pub(crate) use rustfs_ecstore::api::set_disk::test_util::{
-        MultipartCommitBarrier, MultipartCommitPause, PutObjectCommitBarrier, PutObjectCommitPause,
-        fail_next_quota_ledger_save_for_test,
-    };
     pub(crate) use rustfs_ecstore::api::set_disk::{
         DEFAULT_READ_BUFFER_SIZE, file_info_quorum_hash, get_lock_acquire_timeout, is_valid_storage_class,
     };
@@ -1137,9 +1130,7 @@ pub(crate) trait StorageDiskRpcExt {
     ) -> DiskResult<()>;
     async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult<bytes::Bytes>;
     async fn delete_paths(&self, volume: &str, paths: &[String]) -> DiskResult<()>;
-    async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> DiskResult<SnapshotLeaseToken>;
     async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<()>;
-    async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<SnapshotLeaseToken>;
     async fn stat_volume(&self, volume: &str) -> DiskResult<VolumeInfo>;
     async fn list_volumes(&self) -> DiskResult<Vec<VolumeInfo>>;
     async fn make_volume(&self, volume: &str) -> DiskResult<()>;
@@ -1267,18 +1258,10 @@ where
         ecstore_disk::DiskAPI::delete_paths(self, volume, paths).await
     }
 
-    async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> DiskResult<SnapshotLeaseToken> {
-        ecstore_disk::DiskAPI::acquire_snapshot_lease(self, volume, path).await
-    }
-
     async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<()> {
         ecstore_disk::DiskAPI::release_snapshot_lease(self, volume, path, token).await
     }
 
-    async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<SnapshotLeaseToken> {
-        ecstore_disk::DiskAPI::renew_snapshot_lease(self, volume, path, token).await
-    }
-
     async fn stat_volume(&self, volume: &str) -> DiskResult<VolumeInfo> {
         ecstore_disk::DiskAPI::stat_volume(self, volume).await
     }

rustfs/src/table_catalog/mod.rs

@@ -361,5 +361,8 @@ fn storage_error_to_catalog(action: &str, err: StorageError) -> TableCatalogStor
     }
 }
 
+#[cfg(test)]
+pub(crate) mod test_support;
+
 #[cfg(test)]
 mod tests;

rustfs/src/table_catalog/test_support.rs

@@ -0,0 +1,228 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Shared table-catalog test fixtures (backlog#1837).
+//!
+//! Pure data constructors for Iceberg metadata JSON and avro
+//! manifest-list/manifest bytes, shared by the store-level tests
+//! (`table_catalog/tests.rs`) and the admin handler tests
+//! (`admin/handlers/table_catalog/tests.rs`). The parameterized admin
+//! variants are canonical; the store tests wrap them with their historical
+//! fixed values (sequence 7 / snapshot 20), which keeps every produced byte
+//! identical to the pre-extraction fixtures.
+
+pub(crate) fn table_metadata_json(table_uuid: &str, location: &str) -> serde_json::Value {
+    serde_json::json!({
+        "format-version": 2,
+        "table-uuid": table_uuid,
+        "location": location,
+        "last-sequence-number": 0,
+        "last-updated-ms": 1,
+        "last-column-id": 1,
+        "schemas": [{
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
+        }],
+        "current-schema-id": 0,
+        "partition-specs": [{"spec-id": 0, "fields": []}],
+        "default-spec-id": 0,
+        "last-partition-id": 999,
+        "sort-orders": [{"order-id": 0, "fields": []}],
+        "default-sort-order-id": 0,
+        "properties": {},
+        "snapshots": [],
+        "snapshot-log": [],
+        "metadata-log": [],
+        "refs": {}
+    })
+}
+
+pub(crate) fn manifest_list_avro_bytes(manifest_paths: &[&str], sequence_number: i64, snapshot_id: i64) -> Vec<u8> {
+    let manifests = manifest_paths
+        .iter()
+        .map(|manifest_path| (*manifest_path, 0, sequence_number, snapshot_id))
+        .collect::<Vec<_>>();
+    manifest_list_avro_entries_with_partition_specs(&manifests)
+}
+
+pub(crate) fn manifest_list_avro_entries(manifests: &[(&str, i64, i64)]) -> Vec<u8> {
+    let manifests = manifests
+        .iter()
+        .map(|(manifest_path, sequence_number, snapshot_id)| (*manifest_path, 0, *sequence_number, *snapshot_id))
+        .collect::<Vec<_>>();
+    manifest_list_avro_entries_with_partition_specs(&manifests)
+}
+
+pub(crate) fn manifest_list_avro_entries_with_partition_specs(manifests: &[(&str, i32, i64, i64)]) -> Vec<u8> {
+    let schema = apache_avro::Schema::parse_str(
+        r#"
+            {
+              "type": "record",
+              "name": "manifest_file",
+              "fields": [
+                {"name": "manifest_path", "type": "string"},
+                {"name": "manifest_length", "type": "long"},
+                {"name": "partition_spec_id", "type": "int"},
+                {"name": "content", "type": "int"},
+                {"name": "sequence_number", "type": "long"},
+                {"name": "min_sequence_number", "type": "long"},
+                {"name": "added_snapshot_id", "type": "long"},
+                {"name": "added_files_count", "type": "int"},
+                {"name": "existing_files_count", "type": "int"},
+                {"name": "deleted_files_count", "type": "int"},
+                {"name": "added_rows_count", "type": "long"},
+                {"name": "existing_rows_count", "type": "long"},
+                {"name": "deleted_rows_count", "type": "long"}
+              ]
+            }
+            "#,
+    )
+    .expect("manifest list avro schema should parse");
+    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest list writer should initialize");
+    for (manifest_path, partition_spec_id, sequence_number, snapshot_id) in manifests {
+        writer
+            .append_value(apache_avro::types::Value::Record(vec![
+                (
+                    "manifest_path".to_string(),
+                    apache_avro::types::Value::String((*manifest_path).to_string()),
+                ),
+                ("manifest_length".to_string(), apache_avro::types::Value::Long(1)),
+                ("partition_spec_id".to_string(), apache_avro::types::Value::Int(*partition_spec_id)),
+                ("content".to_string(), apache_avro::types::Value::Int(0)),
+                ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
+                ("min_sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
+                ("added_snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
+                ("added_files_count".to_string(), apache_avro::types::Value::Int(1)),
+                ("existing_files_count".to_string(), apache_avro::types::Value::Int(0)),
+                ("deleted_files_count".to_string(), apache_avro::types::Value::Int(0)),
+                ("added_rows_count".to_string(), apache_avro::types::Value::Long(1)),
+                ("existing_rows_count".to_string(), apache_avro::types::Value::Long(0)),
+                ("deleted_rows_count".to_string(), apache_avro::types::Value::Long(0)),
+            ]))
+            .expect("manifest list record should append");
+    }
+    writer.into_inner().expect("manifest list avro bytes should flush")
+}
+
+pub(crate) fn manifest_avro_bytes(files: &[(&str, i32, i32, i64, i64)]) -> Vec<u8> {
+    let schema = apache_avro::Schema::parse_str(
+        r#"
+            {
+              "type": "record",
+              "name": "manifest_entry",
+              "fields": [
+                {"name": "status", "type": "int"},
+                {"name": "snapshot_id", "type": "long"},
+                {"name": "sequence_number", "type": "long"},
+                {"name": "file_sequence_number", "type": "long"},
+                {
+                  "name": "data_file",
+                  "type": {
+                    "type": "record",
+                    "name": "data_file",
+                    "fields": [
+                      {"name": "content", "type": "int"},
+                      {"name": "file_path", "type": "string"},
+                      {"name": "record_count", "type": "long"},
+                      {"name": "file_size_in_bytes", "type": "long"}
+                    ]
+                  }
+                }
+              ]
+            }
+            "#,
+    )
+    .expect("manifest avro schema should parse");
+    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
+    for (file_path, content, status, snapshot_id, sequence_number) in files {
+        writer
+            .append_value(apache_avro::types::Value::Record(vec![
+                ("status".to_string(), apache_avro::types::Value::Int(*status)),
+                ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
+                ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
+                ("file_sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
+                (
+                    "data_file".to_string(),
+                    apache_avro::types::Value::Record(vec![
+                        ("content".to_string(), apache_avro::types::Value::Int(*content)),
+                        ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
+                        ("record_count".to_string(), apache_avro::types::Value::Long(1)),
+                        ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
+                    ]),
+                ),
+            ]))
+            .expect("manifest record should append");
+    }
+    writer.into_inner().expect("manifest avro bytes should flush")
+}
+
+pub(crate) fn nullable_long(value: Option<i64>) -> apache_avro::types::Value {
+    match value {
+        Some(value) => apache_avro::types::Value::Union(1, Box::new(apache_avro::types::Value::Long(value))),
+        None => apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)),
+    }
+}
+
+pub(crate) fn manifest_avro_bytes_with_nullable_sequences(files: &[(&str, i32, i32, i64, Option<i64>)]) -> Vec<u8> {
+    let schema = apache_avro::Schema::parse_str(
+        r#"
+            {
+              "type": "record",
+              "name": "manifest_entry",
+              "fields": [
+                {"name": "status", "type": "int"},
+                {"name": "snapshot_id", "type": "long"},
+                {"name": "sequence_number", "type": ["null", "long"], "default": null},
+                {"name": "file_sequence_number", "type": ["null", "long"], "default": null},
+                {
+                  "name": "data_file",
+                  "type": {
+                    "type": "record",
+                    "name": "data_file",
+                    "fields": [
+                      {"name": "content", "type": "int"},
+                      {"name": "file_path", "type": "string"},
+                      {"name": "record_count", "type": "long"},
+                      {"name": "file_size_in_bytes", "type": "long"}
+                    ]
+                  }
+                }
+              ]
+            }
+            "#,
+    )
+    .expect("manifest avro schema should parse");
+    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
+    for (file_path, content, status, snapshot_id, sequence_number) in files {
+        writer
+            .append_value(apache_avro::types::Value::Record(vec![
+                ("status".to_string(), apache_avro::types::Value::Int(*status)),
+                ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
+                ("sequence_number".to_string(), nullable_long(*sequence_number)),
+                ("file_sequence_number".to_string(), nullable_long(*sequence_number)),
+                (
+                    "data_file".to_string(),
+                    apache_avro::types::Value::Record(vec![
+                        ("content".to_string(), apache_avro::types::Value::Int(*content)),
+                        ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
+                        ("record_count".to_string(), apache_avro::types::Value::Long(1)),
+                        ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
+                    ]),
+                ),
+            ]))
+            .expect("manifest record should append");
+    }
+    writer.into_inner().expect("manifest avro bytes should flush")
+}

rustfs/src/table_catalog/tests.rs

@@ -1429,54 +1429,12 @@ fn manifest_list_avro_bytes(manifest_paths: &[&str]) -> Vec<u8> {
 }
 
 fn manifest_list_avro_bytes_with_spec(manifest_paths: &[&str], partition_spec_id: i32) -> Vec<u8> {
-    let schema = apache_avro::Schema::parse_str(
-        r#"
-            {
-              "type": "record",
-              "name": "manifest_file",
-              "fields": [
-                {"name": "manifest_path", "type": "string"},
-                {"name": "manifest_length", "type": "long"},
-                {"name": "partition_spec_id", "type": "int"},
-                {"name": "content", "type": "int"},
-                {"name": "sequence_number", "type": "long"},
-                {"name": "min_sequence_number", "type": "long"},
-                {"name": "added_snapshot_id", "type": "long"},
-                {"name": "added_files_count", "type": "int"},
-                {"name": "existing_files_count", "type": "int"},
-                {"name": "deleted_files_count", "type": "int"},
-                {"name": "added_rows_count", "type": "long"},
-                {"name": "existing_rows_count", "type": "long"},
-                {"name": "deleted_rows_count", "type": "long"}
-              ]
-            }
-            "#,
-    )
-    .expect("manifest list avro schema should parse");
-    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest list writer should initialize");
-    for manifest_path in manifest_paths {
-        writer
-            .append_value(apache_avro::types::Value::Record(vec![
-                (
-                    "manifest_path".to_string(),
-                    apache_avro::types::Value::String((*manifest_path).to_string()),
-                ),
-                ("manifest_length".to_string(), apache_avro::types::Value::Long(1)),
-                ("partition_spec_id".to_string(), apache_avro::types::Value::Int(partition_spec_id)),
-                ("content".to_string(), apache_avro::types::Value::Int(0)),
-                ("sequence_number".to_string(), apache_avro::types::Value::Long(7)),
-                ("min_sequence_number".to_string(), apache_avro::types::Value::Long(7)),
-                ("added_snapshot_id".to_string(), apache_avro::types::Value::Long(20)),
-                ("added_files_count".to_string(), apache_avro::types::Value::Int(1)),
-                ("existing_files_count".to_string(), apache_avro::types::Value::Int(0)),
-                ("deleted_files_count".to_string(), apache_avro::types::Value::Int(0)),
-                ("added_rows_count".to_string(), apache_avro::types::Value::Long(1)),
-                ("existing_rows_count".to_string(), apache_avro::types::Value::Long(0)),
-                ("deleted_rows_count".to_string(), apache_avro::types::Value::Long(0)),
-            ]))
-            .expect("manifest list record should append");
-    }
-    writer.into_inner().expect("manifest list avro bytes should flush")
+    // Historical fixed values of this file's fixtures: sequence 7, snapshot 20.
+    let manifests = manifest_paths
+        .iter()
+        .map(|path| (*path, partition_spec_id, 7_i64, 20_i64))
+        .collect::<Vec<_>>();
+    crate::table_catalog::test_support::manifest_list_avro_entries_with_partition_specs(&manifests)
 }
 
 fn v1_manifest_list_avro_bytes(manifest_path: &str) -> Vec<u8> {
@@ -2336,55 +2294,13 @@ fn manifest_avro_bytes(files: &[(&str, i32)]) -> Vec<u8> {
 }
 
 fn manifest_avro_bytes_with_status(files: &[(&str, i32, i32)]) -> Vec<u8> {
-    let schema = apache_avro::Schema::parse_str(
-        r#"
-            {
-              "type": "record",
-              "name": "manifest_entry",
-              "fields": [
-                {"name": "status", "type": "int"},
-                {"name": "snapshot_id", "type": "long"},
-                {"name": "sequence_number", "type": "long"},
-                {"name": "file_sequence_number", "type": "long"},
-                {
-                  "name": "data_file",
-                  "type": {
-                    "type": "record",
-                    "name": "data_file",
-                    "fields": [
-                      {"name": "content", "type": "int"},
-                      {"name": "file_path", "type": "string"},
-                      {"name": "record_count", "type": "long"},
-                      {"name": "file_size_in_bytes", "type": "long"}
-                    ]
-                  }
-                }
-              ]
-            }
-            "#,
-    )
-    .expect("manifest avro schema should parse");
-    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
-    for (file_path, content, status) in files {
-        writer
-            .append_value(apache_avro::types::Value::Record(vec![
-                ("status".to_string(), apache_avro::types::Value::Int(*status)),
-                ("snapshot_id".to_string(), apache_avro::types::Value::Long(20)),
-                ("sequence_number".to_string(), apache_avro::types::Value::Long(7)),
-                ("file_sequence_number".to_string(), apache_avro::types::Value::Long(7)),
-                (
-                    "data_file".to_string(),
-                    apache_avro::types::Value::Record(vec![
-                        ("content".to_string(), apache_avro::types::Value::Int(*content)),
-                        ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
-                        ("record_count".to_string(), apache_avro::types::Value::Long(1)),
-                        ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
-                    ]),
-                ),
-            ]))
-            .expect("manifest record should append");
-    }
-    writer.into_inner().expect("manifest avro bytes should flush")
+    // Historical fixed values of this file's fixtures: snapshot 20, sequence 7
+    // (the shared constructor takes snapshot_id fourth, sequence fifth).
+    let files = files
+        .iter()
+        .map(|(path, content, status)| (*path, *content, *status, 20_i64, 7_i64))
+        .collect::<Vec<_>>();
+    crate::table_catalog::test_support::manifest_avro_bytes(&files)
 }
 
 fn manifest_avro_bytes_with_dt_partition(files: &[(&str, i32, &str)]) -> Vec<u8> {

scripts/check_s3s_footprint.sh

@@ -26,7 +26,7 @@ cd "$(dirname "$0")/.."
 # Excludes crates/e2e_test/ — test infrastructure legitimately uses s3s
 # to verify S3 behavior and does not widen the production s3s surface.
 S3S_IMPORT_FILES_BASELINE=213
-S3_ERROR_LINES_BASELINE=1620
+S3_ERROR_LINES_BASELINE=1621
 S3S_PATH_PATTERN='(^|[^"[:alnum:]_])s3s::'
 E2E_TEST_GLOB='--glob=!crates/e2e_test/**'