fix(quota): close reservation bypasses

crates/config/src/constants/app.rs

@@ -353,6 +353,11 @@ pub const DEFAULT_OBS_TRACES_EXPORT_ENABLED: bool = true;
 /// Environment variable: RUSTFS_OBS_METRICS_EXPORT_ENABLED
 pub const DEFAULT_OBS_METRICS_EXPORT_ENABLED: bool = true;
 
+/// Default detailed PUT stage metrics enabled
+/// Default value: false
+/// Environment variable: RUSTFS_OBS_PUT_STAGE_METRICS_ENABLED
+pub const DEFAULT_OBS_PUT_STAGE_METRICS_ENABLED: bool = false;
+
 /// Default logs export enabled
 /// It is used to enable or disable exporting logs
 /// Default value: true

crates/config/src/observability/mod.rs

@@ -44,6 +44,10 @@ pub const ENV_OBS_METRICS_EXPORT_ENABLED: &str = "RUSTFS_OBS_METRICS_EXPORT_ENAB
 pub const ENV_OBS_LOGS_EXPORT_ENABLED: &str = "RUSTFS_OBS_LOGS_EXPORT_ENABLED";
 pub const ENV_OBS_PROFILING_EXPORT_ENABLED: &str = "RUSTFS_OBS_PROFILING_EXPORT_ENABLED";
 
+/// Enables detailed per-stage PUT metrics. Disabled by default because each
+/// PUT records multiple timers and histograms when attribution is active.
+pub const ENV_OBS_PUT_STAGE_METRICS_ENABLED: &str = "RUSTFS_OBS_PUT_STAGE_METRICS_ENABLED";
+
 pub const ENV_OBS_LOGGER_LEVEL: &str = "RUSTFS_OBS_LOGGER_LEVEL";
 pub const ENV_OBS_LOG_STDOUT_ENABLED: &str = "RUSTFS_OBS_LOG_STDOUT_ENABLED";
 pub const ENV_OBS_LOG_DIRECTORY: &str = "RUSTFS_OBS_LOG_DIRECTORY";
@@ -141,6 +145,7 @@ mod tests {
         assert_eq!(ENV_OBS_METRICS_EXPORT_ENABLED, "RUSTFS_OBS_METRICS_EXPORT_ENABLED");
         assert_eq!(ENV_OBS_LOGS_EXPORT_ENABLED, "RUSTFS_OBS_LOGS_EXPORT_ENABLED");
         assert_eq!(ENV_OBS_PROFILING_EXPORT_ENABLED, "RUSTFS_OBS_PROFILING_EXPORT_ENABLED");
+        assert_eq!(ENV_OBS_PUT_STAGE_METRICS_ENABLED, "RUSTFS_OBS_PUT_STAGE_METRICS_ENABLED");
         // Test log cleanup related env keys
         assert_eq!(ENV_OBS_LOG_MAX_TOTAL_SIZE_BYTES, "RUSTFS_OBS_LOG_MAX_TOTAL_SIZE_BYTES");
         assert_eq!(ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, "RUSTFS_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES");

crates/ecstore/src/api/mod.rs

@@ -405,6 +405,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,
@@ -468,7 +470,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::{PutObjectCommitBarrier, PutObjectCommitPause};
+        pub use crate::set_disk::{MultipartCommitBarrier, MultipartCommitPause, PutObjectCommitBarrier, PutObjectCommitPause};
     }
 }
 

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

@@ -52,6 +52,7 @@ 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 {
@@ -67,6 +68,7 @@ impl QuotaChecker {
                     quota_limit: None,
                     operation_size,
                     remaining: None,
+                    uses_durable_reservations,
                 });
             }
             Some(q) => q,
@@ -74,14 +76,17 @@ 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 + operation_size,
+            QuotaOperation::PutObject | QuotaOperation::PostObject | QuotaOperation::CopyObject => {
+                current_usage.saturating_add(admission_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, operation_size)
+                quota_config.check_operation_allowed(current_usage, admission_size)
             }
             QuotaOperation::DeleteObject => true,
         };
@@ -105,6 +110,7 @@ impl QuotaChecker {
             quota_limit: Some(quota_limit),
             operation_size,
             remaining,
+            uses_durable_reservations,
         };
 
         let duration = start_time.elapsed();
@@ -375,6 +381,7 @@ mod tests {
             quota_limit: None,
             operation_size: 1024,
             remaining: None,
+            uses_durable_reservations: false,
         };
 
         assert!(result.allowed);
@@ -398,4 +405,13 @@ 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

@@ -20,7 +20,7 @@ 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};
+use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _};
 use thiserror::Error;
 use time::OffsetDateTime;
 
@@ -90,7 +90,10 @@ impl<'de> Deserialize<'de> for BucketQuota {
     {
         let wire = BucketQuotaWire::deserialize(deserializer)?;
         let quota = if wire.reservation_protocol == Some(QUOTA_RESERVATION_PROTOCOL_V1) {
-            wire.reservation_quota
+            Some(
+                wire.reservation_quota
+                    .ok_or_else(|| D::Error::custom("reservation_quota is required for reservation protocol v1"))?,
+            )
         } else {
             wire.quota
         };
@@ -164,6 +167,7 @@ pub struct QuotaCheckResult {
     pub quota_limit: Option<u64>,
     pub operation_size: u64,
     pub remaining: Option<u64>,
+    pub uses_durable_reservations: bool,
 }
 
 #[derive(Debug)]
@@ -327,6 +331,14 @@ mod tests {
         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"));
+    }
+
     /// unmarshal accepts format without quota_type
     #[test]
     fn unmarshal_format_without_quota_type() {

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

@@ -209,6 +209,7 @@ pub(crate) struct QuotaContext {
     quota_limit: Option<u64>,
     capability_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
     snapshot_admission: Option<QuotaAdmission>,
+    legacy_data_movement: bool,
     metadata_guard: Option<NamespaceLockGuard>,
     pool_index: Option<usize>,
     set_index: Option<usize>,
@@ -233,6 +234,12 @@ impl QuotaContext {
             }
             return Ok(QuotaReservation::unlimited(self.metadata_guard));
         }
+        if self.legacy_data_movement {
+            if new_size > old_size {
+                return Err(StorageError::PartMissingOrCorrupt);
+            }
+            return Ok(QuotaReservation::unlimited(self.metadata_guard));
+        }
         let store = self.store.ok_or(StorageError::PartMissingOrCorrupt)?;
         let bucket_incarnation = self.bucket_incarnation.ok_or(StorageError::PartMissingOrCorrupt)?;
         let quota_revision = self.quota_revision.ok_or(StorageError::PartMissingOrCorrupt)?;
@@ -431,7 +438,8 @@ pub(crate) async fn begin(
     ctx: &crate::runtime::instance::InstanceContext,
     bucket: &str,
     object: &str,
-    _snapshot_admission: Option<QuotaAdmission>,
+    snapshot_admission: Option<QuotaAdmission>,
+    data_movement: bool,
     pool_index: usize,
     set_index: usize,
 ) -> Result<QuotaContext> {
@@ -446,13 +454,14 @@ pub(crate) async fn begin(
             quota_limit: None,
             capability_proof: None,
             snapshot_admission: None,
+            legacy_data_movement: false,
             metadata_guard: None,
             pool_index: None,
             set_index: None,
         });
     }
     #[cfg(test)]
-    if let Some(snapshot_admission) = _snapshot_admission {
+    if let Some(snapshot_admission) = snapshot_admission {
         return Ok(QuotaContext {
             store: None,
             bucket: bucket.to_string(),
@@ -463,6 +472,7 @@ pub(crate) async fn begin(
             quota_limit: Some(snapshot_admission.quota_limit()),
             capability_proof: None,
             snapshot_admission: Some(snapshot_admission),
+            legacy_data_movement: false,
             metadata_guard: None,
             pool_index: Some(pool_index),
             set_index: Some(set_index),
@@ -480,6 +490,7 @@ pub(crate) async fn begin(
             quota_limit: None,
             capability_proof: None,
             snapshot_admission: None,
+            legacy_data_movement: false,
             metadata_guard: None,
             pool_index: None,
             set_index: None,
@@ -504,7 +515,8 @@ pub(crate) async fn begin(
     {
         return Err(StorageError::PartMissingOrCorrupt);
     }
-    let capability_proof = if quota.as_ref().is_some_and(|quota| quota.uses_durable_reservations()) {
+    let durable_quota = quota.as_ref().filter(|quota| quota.uses_durable_reservations());
+    let capability_proof = if durable_quota.is_some() {
         Some(
             crate::services::notification_sys::acquire_cross_pool_fence_fleet_proof()
                 .ok_or_else(|| quota_capability_error(bucket, &ledger_object(bucket)))?,
@@ -512,10 +524,28 @@ pub(crate) async fn begin(
     } else {
         None
     };
-    let quota_limit = quota
-        .filter(crate::bucket::quota::BucketQuota::uses_durable_reservations)
-        .and_then(|quota| quota.quota);
-    let store = if quota_limit.is_some() {
+    let durable_quota_limit = durable_quota.and_then(|quota| quota.quota);
+    let snapshot_admission = match quota.as_ref().filter(|quota| !quota.uses_durable_reservations()) {
+        Some(quota) => match (quota.quota, snapshot_admission) {
+            (Some(limit), Some(admission)) if admission.quota_limit() == limit => Some(admission),
+            (Some(_), None) if data_movement => None,
+            (Some(_), _) => return Err(StorageError::PartMissingOrCorrupt),
+            (None, _) => None,
+        },
+        None => None,
+    };
+    let legacy_data_movement = durable_quota_limit.is_none()
+        && quota.as_ref().and_then(|quota| quota.quota).is_some()
+        && snapshot_admission.is_none()
+        && data_movement;
+    let quota_limit = durable_quota_limit
+        .or_else(|| snapshot_admission.map(QuotaAdmission::quota_limit))
+        .or_else(|| {
+            legacy_data_movement
+                .then(|| quota.as_ref().and_then(|quota| quota.quota))
+                .flatten()
+        });
+    let store = if durable_quota_limit.is_some() {
         Some(metadata_sys::object_store_in(ctx).await?)
     } else {
         None
@@ -529,7 +559,8 @@ pub(crate) async fn begin(
         quota_revision: Some(quota_revision),
         quota_limit,
         capability_proof,
-        snapshot_admission: None,
+        snapshot_admission,
+        legacy_data_movement,
         metadata_guard: Some(metadata_guard),
         pool_index: Some(pool_index),
         set_index: Some(set_index),

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

@@ -846,31 +846,49 @@ impl RemoteDisk {
     /// default to 1 (see [`internode_idempotent_read_retries`]). MUST NOT be used for write/lock
     /// RPCs — those must never auto-retry (quorum/idempotency safety). The `operation` closure is
     /// re-invoked per attempt, so it must be `Fn` (rebuild the request from borrowed inputs, do not
-    /// move captured state out).
+    /// move captured state out). Attempts and backoff share one total timeout budget.
     async fn execute_read_with_retry<T, F, Fut>(&self, op: &'static str, operation: F, timeout_duration: Duration) -> Result<T>
     where
         F: Fn() -> Fut,
         Fut: std::future::Future<Output = Result<T>>,
     {
+        let deadline = (!timeout_duration.is_zero()).then(|| {
+            time::Instant::now()
+                .checked_add(timeout_duration)
+                .unwrap_or_else(|| time::sleep(timeout_duration).deadline())
+        });
         let max_retries = internode_idempotent_read_retries();
         let mut attempt = 0usize;
         loop {
-            // Only the final attempt marks the disk faulty / evicts the channel. Earlier retries
-            // ignore the failure, so a transient error cannot flip the disk into a faulty
-            // short-circuit (which would defeat the retry) or over-count failures.
+            let attempt_timeout = deadline
+                .map(|deadline| deadline.saturating_duration_since(time::Instant::now()))
+                .unwrap_or(Duration::ZERO);
+            if deadline.is_some() && attempt_timeout.is_zero() {
+                self.record_timeout(op, timeout_duration);
+                return Err(DiskError::Timeout);
+            }
+
             let health_action = if attempt >= max_retries {
                 FailureHealthAction::MarkFailure
             } else {
                 FailureHealthAction::IgnoreFailure
             };
             match self
-                .execute_with_timeout_for_op_and_health_action(op, &operation, timeout_duration, health_action)
+                .execute_with_timeout_for_op_and_health_action(op, &operation, attempt_timeout, health_action)
                 .await
             {
                 Err(err) if attempt < max_retries && is_network_like_disk_error(&err) => {
+                    if matches!(err, DiskError::Timeout) && deadline.is_some_and(|deadline| time::Instant::now() >= deadline) {
+                        self.mark_faulty("read_operation_deadline");
+                        return Err(err);
+                    }
                     attempt += 1;
                     let backoff = REMOTE_DISK_READ_RETRY_BASE_BACKOFF
                         .saturating_mul(1u32 << u32::try_from(attempt - 1).unwrap_or(4).min(4));
+                    if deadline.is_some_and(|deadline| deadline.saturating_duration_since(time::Instant::now()) <= backoff) {
+                        attempt = max_retries;
+                        continue;
+                    }
                     debug!(
                         endpoint = %self.endpoint,
                         addr = %self.addr,
@@ -878,7 +896,17 @@ impl RemoteDisk {
                         attempt,
                         "retrying idempotent read-only RPC after transient network error"
                     );
-                    tokio::time::sleep(backoff).await;
+                    if let Some(deadline) = deadline {
+                        if time::timeout_at(deadline, time::sleep(backoff)).await.is_err() {
+                            self.record_timeout(op, timeout_duration);
+                            return Err(DiskError::Timeout);
+                        }
+                    } else {
+                        time::sleep(backoff).await;
+                    }
+                    if self.health.is_faulty() {
+                        return Err(DiskError::FaultyDisk);
+                    }
                 }
                 other => return other,
             }
@@ -957,32 +985,35 @@ impl RemoteDisk {
                 operation_result
             }
             Err(_) => {
-                // Timeout occurred, mark disk as potentially faulty
-                counter!(
-                    "rustfs_drive_op_timeout_total",
-                    "endpoint" => self.endpoint.to_string(),
-                    "op" => op.to_string()
-                )
-                .increment(1);
+                self.record_timeout(op, timeout_duration);
                 if failure_health_action == FailureHealthAction::MarkFailure {
                     self.mark_faulty_and_evict("operation_timeout").await;
                 }
-                warn!(
-                    event = EVENT_REMOTE_DISK_RPC,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
-                    endpoint = %self.endpoint,
-                    addr = %self.addr,
-                    op,
-                    timeout_ms = timeout_duration.as_millis(),
-                    state = "timeout",
-                    "Remote disk operation timed out"
-                );
                 Err(DiskError::Timeout)
             }
         }
     }
 
+    fn record_timeout(&self, op: &'static str, timeout_duration: Duration) {
+        counter!(
+            "rustfs_drive_op_timeout_total",
+            "endpoint" => self.endpoint.to_string(),
+            "op" => op.to_string()
+        )
+        .increment(1);
+        warn!(
+            event = EVENT_REMOTE_DISK_RPC,
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
+            endpoint = %self.endpoint,
+            addr = %self.addr,
+            op,
+            timeout_ms = timeout_duration.as_millis(),
+            state = "timeout",
+            "Remote disk operation timed out"
+        );
+    }
+
     async fn handle_network_like_error<T>(
         &self,
         op: &'static str,
@@ -1016,7 +1047,7 @@ impl RemoteDisk {
         }
     }
 
-    async fn mark_faulty_and_evict(&self, reason: &'static str) {
+    fn mark_faulty(&self, reason: &'static str) -> bool {
         let previous_state = self.runtime_state();
         let transitioned_to_offline = self.mark_suspect_or_offline(reason);
         let state = self.runtime_state();
@@ -1053,6 +1084,12 @@ impl RemoteDisk {
                     "Remote disk marked suspect"
                 );
             }
+        }
+        state != previous_state
+    }
+
+    async fn mark_faulty_and_evict(&self, reason: &'static str) {
+        if self.mark_faulty(reason) {
             counter!(
                 "rustfs_drive_connection_evict_total",
                 "endpoint" => self.endpoint.to_string(),
@@ -2068,7 +2105,7 @@ impl DiskAPI for RemoteDisk {
 
                 Ok(file_info)
             },
-            get_max_timeout_duration(),
+            get_drive_metadata_timeout(),
         )
         .await
     }
@@ -5094,6 +5131,452 @@ mod tests {
         );
     }
 
+    #[tokio::test(start_paused = true)]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_reset_during_backoff_preserves_recovery() {
+        let remote_disk = Arc::new(new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await);
+        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let first_attempt = Arc::new(tokio::sync::Notify::new());
+        let started = time::Instant::now();
+
+        let task_disk = Arc::clone(&remote_disk);
+        let task_attempts = Arc::clone(&attempts);
+        let task_first_attempt = Arc::clone(&first_attempt);
+        let task = tokio::spawn(async move {
+            task_disk
+                .execute_read_with_retry(
+                    "read_version",
+                    move || {
+                        let attempt = task_attempts.fetch_add(1, Ordering::SeqCst);
+                        let first_attempt = Arc::clone(&task_first_attempt);
+                        async move {
+                            if attempt == 0 {
+                                time::sleep(Duration::from_millis(20)).await;
+                                first_attempt.notify_one();
+                                return Err::<(), Error>(DiskError::Io(std_io::Error::new(
+                                    std_io::ErrorKind::ConnectionRefused,
+                                    "connection refused",
+                                )));
+                            }
+                            Ok(())
+                        }
+                    },
+                    Duration::from_millis(100),
+                )
+                .await
+        });
+
+        first_attempt.notified().await;
+        tokio::task::yield_now().await;
+        remote_disk.health.reset_for_store_init_retry(&remote_disk.endpoint);
+        let channel = TonicEndpoint::from_shared(remote_disk.addr.clone())
+            .expect("remote disk address should parse")
+            .connect_lazy();
+        runtime_sources::cache_test_node_channel(remote_disk.addr.clone(), channel).await;
+        task.await
+            .expect("retry task should finish")
+            .expect("the retry should succeed after the health reset");
+
+        assert_eq!(attempts.load(Ordering::SeqCst), 2);
+        assert_eq!(started.elapsed(), Duration::from_millis(70));
+        assert_eq!(
+            remote_disk.health.waiting_count(),
+            0,
+            "health reset must not underflow the waiting counter"
+        );
+        assert_eq!(remote_disk.runtime_state(), RuntimeDriveHealthState::Online);
+        assert!(
+            runtime_sources::test_node_channel_is_cached(&remote_disk.addr).await,
+            "a recovered channel must survive the retry backoff"
+        );
+        remote_disk.cancel_token.cancel();
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_still_retries_within_shared_deadline() {
+        let remote_disk = new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await;
+        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let channel = TonicEndpoint::from_shared(remote_disk.addr.clone())
+            .expect("remote disk address should parse")
+            .connect_lazy();
+        runtime_sources::cache_test_node_channel(remote_disk.addr.clone(), channel).await;
+
+        remote_disk
+            .execute_read_with_retry(
+                "read_version",
+                || {
+                    let attempt = attempts.fetch_add(1, Ordering::SeqCst);
+                    async move {
+                        if attempt == 0 {
+                            return Err::<(), Error>(DiskError::Io(std_io::Error::new(
+                                std_io::ErrorKind::ConnectionReset,
+                                "connection reset",
+                            )));
+                        }
+                        Ok(())
+                    }
+                },
+                Duration::from_millis(100),
+            )
+            .await
+            .expect("a retry that fits the shared deadline should succeed");
+
+        assert_eq!(attempts.load(Ordering::SeqCst), 2);
+        assert_eq!(remote_disk.runtime_state(), RuntimeDriveHealthState::Online);
+        assert!(runtime_sources::test_node_channel_is_cached(&remote_disk.addr).await);
+        remote_disk.cancel_token.cancel();
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_uses_remaining_budget_for_final_attempt() {
+        let remote_disk = new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await;
+        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let started = time::Instant::now();
+        let channel = TonicEndpoint::from_shared(remote_disk.addr.clone())
+            .expect("remote disk address should parse")
+            .connect_lazy();
+        runtime_sources::cache_test_node_channel(remote_disk.addr.clone(), channel).await;
+
+        let err = remote_disk
+            .execute_read_with_retry(
+                "read_version",
+                || {
+                    let attempt = attempts.fetch_add(1, Ordering::SeqCst);
+                    async move {
+                        if attempt == 0 {
+                            time::sleep(Duration::from_millis(20)).await;
+                            return Err::<(), Error>(DiskError::Io(std_io::Error::new(
+                                std_io::ErrorKind::ConnectionRefused,
+                                "connection refused",
+                            )));
+                        }
+                        std::future::pending::<Result<()>>().await
+                    }
+                },
+                Duration::from_millis(100),
+            )
+            .await
+            .expect_err("the final retry should consume only the remaining total budget");
+
+        assert_eq!(err, DiskError::Timeout);
+        assert_eq!(attempts.load(Ordering::SeqCst), 2);
+        assert_eq!(started.elapsed(), Duration::from_millis(100));
+        assert_eq!(remote_disk.runtime_state(), RuntimeDriveHealthState::Suspect);
+        assert!(!runtime_sources::test_node_channel_is_cached(&remote_disk.addr).await);
+        remote_disk.cancel_token.cancel();
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_uses_final_attempt_at_exact_backoff_boundary() {
+        let remote_disk = new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await;
+        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let started = time::Instant::now();
+
+        let err = remote_disk
+            .execute_read_with_retry(
+                "read_version",
+                || {
+                    let attempt = attempts.fetch_add(1, Ordering::SeqCst);
+                    async move {
+                        if attempt == 0 {
+                            time::sleep(Duration::from_millis(50)).await;
+                            return Err::<(), Error>(DiskError::Io(std_io::Error::new(
+                                std_io::ErrorKind::ConnectionRefused,
+                                "connection refused",
+                            )));
+                        }
+                        std::future::pending::<Result<()>>().await
+                    }
+                },
+                Duration::from_millis(100),
+            )
+            .await
+            .expect_err("the exact backoff boundary should be reserved for a final attempt");
+
+        assert_eq!(err, DiskError::Timeout);
+        assert_eq!(attempts.load(Ordering::SeqCst), 2);
+        assert_eq!(started.elapsed(), Duration::from_millis(100));
+        assert_eq!(remote_disk.runtime_state(), RuntimeDriveHealthState::Suspect);
+        remote_disk.cancel_token.cancel();
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_uses_final_attempt_below_backoff_budget() {
+        let remote_disk = new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await;
+        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let started = time::Instant::now();
+
+        let err = remote_disk
+            .execute_read_with_retry(
+                "read_version",
+                || {
+                    let attempt = attempts.fetch_add(1, Ordering::SeqCst);
+                    async move {
+                        if attempt == 0 {
+                            time::sleep(Duration::from_millis(80)).await;
+                            return Err::<(), Error>(DiskError::Io(std_io::Error::new(
+                                std_io::ErrorKind::ConnectionRefused,
+                                "connection refused",
+                            )));
+                        }
+                        std::future::pending::<Result<()>>().await
+                    }
+                },
+                Duration::from_millis(100),
+            )
+            .await
+            .expect_err("remaining budget below backoff should be reserved for a final attempt");
+
+        assert_eq!(err, DiskError::Timeout);
+        assert_eq!(attempts.load(Ordering::SeqCst), 2);
+        assert_eq!(started.elapsed(), Duration::from_millis(100));
+        assert_eq!(remote_disk.runtime_state(), RuntimeDriveHealthState::Suspect);
+        remote_disk.cancel_token.cancel();
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_zero_timeout_disables_the_deadline() {
+        let remote_disk = new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await;
+        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let started = time::Instant::now();
+
+        remote_disk
+            .execute_read_with_retry(
+                "read_version",
+                || {
+                    let attempt = attempts.fetch_add(1, Ordering::SeqCst);
+                    async move {
+                        if attempt == 0 {
+                            return Err::<(), Error>(DiskError::Io(std_io::Error::new(
+                                std_io::ErrorKind::ConnectionReset,
+                                "connection reset",
+                            )));
+                        }
+                        Ok(())
+                    }
+                },
+                Duration::ZERO,
+            )
+            .await
+            .expect("zero timeout should allow a retry without a deadline");
+
+        assert_eq!(attempts.load(Ordering::SeqCst), 2);
+        assert_eq!(started.elapsed(), REMOTE_DISK_READ_RETRY_BASE_BACKOFF);
+        remote_disk.cancel_token.cancel();
+    }
+
+    #[tokio::test]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_accepts_max_metadata_timeout() {
+        temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_METADATA_TIMEOUT_SECS, Some(u64::MAX.to_string()))], async {
+            let remote_disk = new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await;
+
+            remote_disk
+                .execute_read_with_retry("read_version", || async { Ok::<(), Error>(()) }, get_drive_metadata_timeout())
+                .await
+                .expect("the maximum configured metadata timeout must not panic");
+
+            remote_disk.cancel_token.cancel();
+        })
+        .await;
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_zero_retries_runs_once() {
+        temp_env::async_with_vars([(rustfs_config::ENV_INTERNODE_IDEMPOTENT_READ_RETRIES, Some("0"))], async {
+            let remote_disk = new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await;
+            let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+            let started = time::Instant::now();
+            let channel = TonicEndpoint::from_shared(remote_disk.addr.clone())
+                .expect("remote disk address should parse")
+                .connect_lazy();
+            runtime_sources::cache_test_node_channel(remote_disk.addr.clone(), channel).await;
+
+            let err = remote_disk
+                .execute_read_with_retry(
+                    "read_version",
+                    || {
+                        attempts.fetch_add(1, Ordering::SeqCst);
+                        async {
+                            Err::<(), Error>(DiskError::Io(std_io::Error::new(
+                                std_io::ErrorKind::ConnectionReset,
+                                "connection reset",
+                            )))
+                        }
+                    },
+                    Duration::from_secs(1),
+                )
+                .await
+                .expect_err("zero retries should return the first network error");
+
+            assert!(matches!(err, DiskError::Io(ref io_err) if io_err.kind() == std_io::ErrorKind::ConnectionReset));
+            assert_eq!(attempts.load(Ordering::SeqCst), 1);
+            assert_eq!(started.elapsed(), Duration::ZERO);
+            assert_eq!(remote_disk.runtime_state(), RuntimeDriveHealthState::Suspect);
+            assert!(!runtime_sources::test_node_channel_is_cached(&remote_disk.addr).await);
+            remote_disk.cancel_token.cancel();
+        })
+        .await;
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_attempt_timeout_marks_health_without_evicting() {
+        let remote_disk = new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await;
+        let recorder = crate::test_metrics::CapturingRecorder::default();
+        let _recorder_guard = metrics::set_default_local_recorder(&recorder);
+        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let channel = TonicEndpoint::from_shared(remote_disk.addr.clone())
+            .expect("remote disk address should parse")
+            .connect_lazy();
+        runtime_sources::cache_test_node_channel(remote_disk.addr.clone(), channel).await;
+
+        let err = remote_disk
+            .execute_read_with_retry(
+                "read_version",
+                || {
+                    attempts.fetch_add(1, Ordering::SeqCst);
+                    std::future::pending::<Result<()>>()
+                },
+                Duration::from_millis(100),
+            )
+            .await
+            .expect_err("an in-flight attempt that consumes the deadline should time out");
+
+        assert_eq!(err, DiskError::Timeout);
+        assert_eq!(attempts.load(Ordering::SeqCst), 1);
+        assert_eq!(remote_disk.runtime_state(), RuntimeDriveHealthState::Suspect);
+        assert!(runtime_sources::test_node_channel_is_cached(&remote_disk.addr).await);
+        assert_eq!(
+            recorder.counter_value(
+                "rustfs_drive_op_timeout_total",
+                &[
+                    ("endpoint", remote_disk.endpoint.to_string().as_str()),
+                    ("op", "read_version")
+                ]
+            ),
+            1
+        );
+        remote_disk.cancel_token.cancel();
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_does_not_retry_business_errors() {
+        let remote_disk = new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await;
+        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+
+        let err = remote_disk
+            .execute_read_with_retry(
+                "read_version",
+                || {
+                    attempts.fetch_add(1, Ordering::SeqCst);
+                    async { Err::<(), Error>(DiskError::FileNotFound) }
+                },
+                Duration::from_secs(1),
+            )
+            .await
+            .expect_err("business errors should be returned directly");
+
+        assert_eq!(err, DiskError::FileNotFound);
+        assert_eq!(attempts.load(Ordering::SeqCst), 1);
+        remote_disk.cancel_token.cancel();
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_honors_configured_retry_count() {
+        temp_env::async_with_vars([(rustfs_config::ENV_INTERNODE_IDEMPOTENT_READ_RETRIES, Some("2"))], async {
+            let remote_disk = new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await;
+            let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+            let started = time::Instant::now();
+            let channel = TonicEndpoint::from_shared(remote_disk.addr.clone())
+                .expect("remote disk address should parse")
+                .connect_lazy();
+            runtime_sources::cache_test_node_channel(remote_disk.addr.clone(), channel).await;
+
+            let err = remote_disk
+                .execute_read_with_retry(
+                    "read_version",
+                    || {
+                        attempts.fetch_add(1, Ordering::SeqCst);
+                        async {
+                            Err::<(), Error>(DiskError::Io(std_io::Error::new(
+                                std_io::ErrorKind::ConnectionReset,
+                                "connection reset",
+                            )))
+                        }
+                    },
+                    Duration::from_secs(1),
+                )
+                .await
+                .expect_err("exhausted retries should return the last network error");
+
+            assert!(matches!(err, DiskError::Io(ref io_err) if io_err.kind() == std_io::ErrorKind::ConnectionReset));
+            assert_eq!(attempts.load(Ordering::SeqCst), 3);
+            assert_eq!(started.elapsed(), Duration::from_millis(150));
+            assert_eq!(remote_disk.runtime_state(), RuntimeDriveHealthState::Suspect);
+            assert!(!runtime_sources::test_node_channel_is_cached(&remote_disk.addr).await);
+            remote_disk.cancel_token.cancel();
+        })
+        .await;
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial(remote_disk_read_retry)]
+    async fn execute_read_with_retry_stops_when_disk_turns_offline_during_backoff() {
+        let remote_disk = Arc::new(new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await);
+        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let first_attempt = Arc::new(tokio::sync::Notify::new());
+        let task_disk = Arc::clone(&remote_disk);
+        let task_attempts = Arc::clone(&attempts);
+        let task_first_attempt = Arc::clone(&first_attempt);
+
+        let task = tokio::spawn(async move {
+            task_disk
+                .execute_read_with_retry(
+                    "read_version",
+                    move || {
+                        let attempt = task_attempts.fetch_add(1, Ordering::SeqCst);
+                        let first_attempt = Arc::clone(&task_first_attempt);
+                        async move {
+                            if attempt == 0 {
+                                first_attempt.notify_one();
+                                return Err::<(), Error>(DiskError::Io(std_io::Error::new(
+                                    std_io::ErrorKind::ConnectionReset,
+                                    "connection reset",
+                                )));
+                            }
+                            Ok(())
+                        }
+                    },
+                    Duration::from_secs(1),
+                )
+                .await
+        });
+
+        first_attempt.notified().await;
+        tokio::task::yield_now().await;
+        remote_disk
+            .health
+            .force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+        time::advance(REMOTE_DISK_READ_RETRY_BASE_BACKOFF).await;
+        let err = task
+            .await
+            .expect("retry task should finish")
+            .expect_err("an offline disk must stop before the next attempt");
+
+        assert_eq!(err, DiskError::FaultyDisk);
+        assert_eq!(attempts.load(Ordering::SeqCst), 1);
+        remote_disk.cancel_token.cancel();
+    }
+
     #[tokio::test]
     async fn test_execute_with_timeout_evicts_cached_connection() {
         let addr = "http://127.0.0.1:59991".to_string();
@@ -5603,6 +6086,40 @@ mod tests {
         accept_task.abort();
     }
 
+    #[tokio::test]
+    async fn read_version_uses_the_metadata_timeout_on_a_stalled_peer() {
+        runtime_sources::ensure_test_rpc_secret();
+        let Some((base_addr, accept_task)) = spawn_stalled_grpc_peer().await else {
+            return;
+        };
+        let remote_disk = remote_disk_for_addr(&base_addr).await;
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_DRIVE_METADATA_TIMEOUT_SECS, Some("1")),
+                (rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("10")),
+            ],
+            async {
+                let started = time::Instant::now();
+                let err = tokio::time::timeout(
+                    Duration::from_secs(5),
+                    remote_disk.read_version("bucket", "bucket", "object", "", &ReadOptions::default()),
+                )
+                .await
+                .expect("read_version must use the shorter metadata deadline")
+                .expect_err("a stalled peer must fail read_version");
+
+                assert!(matches!(err, DiskError::Timeout), "expected the metadata deadline to fire, got {err:?}");
+                assert!(started.elapsed() >= Duration::from_millis(900));
+                assert!(started.elapsed() < Duration::from_secs(2));
+            },
+        )
+        .await;
+
+        remote_disk.cancel_token.cancel();
+        accept_task.abort();
+    }
+
     #[tokio::test]
     async fn delete_volume_bounds_the_wait_on_a_stalled_peer() {
         runtime_sources::ensure_test_rpc_secret();

crates/ecstore/src/disk/local.rs

@@ -9243,7 +9243,7 @@ impl DiskAPI for LocalDisk {
 
             if let Some(src_file_path_parent) = src_file_path.parent() {
                 if src_volume != super::RUSTFS_META_MULTIPART_BUCKET {
-                    let _ = remove_std(src_file_path_parent);
+                    let _ = std::fs::remove_dir(src_file_path_parent);
                 } else {
                     let _ = self
                         .delete_file(&dst_volume_dir, &src_file_path_parent.to_path_buf(), true, false)
@@ -9568,7 +9568,7 @@ impl DiskAPI for LocalDisk {
             if let Some(ref cleanup) = cleanup_path {
                 let _ = self.delete_file(&dst_volume_dir, cleanup, true, false).await;
             } else if let Some(parent) = src_file_path.parent() {
-                let _ = remove_std(parent);
+                let _ = std::fs::remove_dir(parent);
             }
 
             // Heal reuses a version's `data_dir` and lands the rebuilt shard on
@@ -12565,6 +12565,10 @@ mod test {
             .join(RUSTFS_META_TMP_BUCKET)
             .join(tmp_object)
             .join(new_data_dir.to_string());
+        let tmp_parent = tmp_data_dir
+            .parent()
+            .expect("tmp data dir should have a parent")
+            .to_path_buf();
         fs::create_dir_all(&tmp_data_dir)
             .await
             .expect("new tmp data dir should be created");
@@ -12576,6 +12580,10 @@ mod test {
         disk.rename_data(RUSTFS_META_TMP_BUCKET, tmp_object, new_fi, bucket, object)
             .await
             .expect("rename_data should commit");
+        assert!(
+            !tmp_parent.exists(),
+            "successful non-inline commit should remove the empty staging parent"
+        );
 
         // The tmp xl.meta write point uses SyncMode::FileOnly: its parent dir
         // ({tmp}/{tmp_object}) must not be fsynced.
@@ -12780,6 +12788,9 @@ mod test {
         let tmp_object = "tmp-new-inline";
         ensure_test_volume(&disk, bucket).await;
         ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
+        let tmp_parent = disk
+            .get_object_path(RUSTFS_META_TMP_BUCKET, tmp_object)
+            .expect("tmp parent should resolve");
 
         let _mode = durability_mode_override::set(DurabilityMode::Strict);
         let version_id = Uuid::parse_str("99999999-9999-9999-9999-999999999999").expect("version id should parse");
@@ -12788,6 +12799,7 @@ mod test {
         disk.rename_data(RUSTFS_META_TMP_BUCKET, tmp_object, new_fi, bucket, object)
             .await
             .expect("inline rename_data should commit the new object");
+        assert!(!tmp_parent.exists(), "successful inline commit should remove the empty staging parent");
 
         let bucket_dir = disk.get_bucket_path(bucket).expect("bucket path should resolve");
         let prefix_dir = disk.get_object_path(bucket, "prefix").expect("prefix path should resolve");
@@ -12811,6 +12823,34 @@ mod test {
         );
     }
 
+    #[tokio::test]
+    async fn rename_data_inline_preserves_non_empty_staging_parent() {
+        use tempfile::tempdir;
+
+        let dir = 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 = "inline-staging-sentinel-bucket";
+        let object = "inline-object";
+        let tmp_object = "inline-stage-with-sentinel";
+        ensure_test_volume(&disk, bucket).await;
+        ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
+
+        let tmp_parent = disk
+            .get_object_path(RUSTFS_META_TMP_BUCKET, tmp_object)
+            .expect("tmp parent should resolve");
+        fs::create_dir_all(&tmp_parent).await.expect("tmp parent should be created");
+        let sentinel = tmp_parent.join("sentinel");
+        fs::write(&sentinel, b"keep").await.expect("sentinel should be written");
+
+        let fi = test_file_info(object, Uuid::new_v4(), None, Some(Bytes::from_static(b"inline-payload")));
+        disk.rename_data(RUSTFS_META_TMP_BUCKET, tmp_object, fi, bucket, object)
+            .await
+            .expect("non-empty staging cleanup must not negate the committed object");
+
+        assert_eq!(fs::read(&sentinel).await.expect("sentinel should remain"), b"keep");
+    }
+
     #[cfg(unix)]
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     #[allow(clippy::await_holding_lock)]
@@ -12985,7 +13025,10 @@ mod test {
             .expect("non-inline rename_data should commit");
 
         assert!(!replacement_dir.exists(), "the destination object directory must not be replaced");
-        assert!(staging_parent.exists(), "the guarded staging parent must retain its identity");
+        assert!(
+            !staging_parent.exists(),
+            "successful commit should remove the empty staging parent after releasing its guard"
+        );
         assert!(
             !replacement_staging_parent.exists(),
             "the staging parent must not be replaced between data and metadata publication"

crates/ecstore/src/erasure/coding/encode.rs

@@ -18,10 +18,12 @@ use crate::disk::error_reduce::{
 };
 use crate::erasure::coding::BitrotWriterWrapper;
 use crate::erasure::coding::Erasure;
+use crate::erasure::coding::erasure::EncodedBlock;
 use crate::runtime::sources as runtime_sources;
 use bytes::{Bytes, BytesMut};
 use futures::StreamExt;
 use futures::stream::FuturesUnordered;
+use rustfs_utils::HashAlgorithm;
 use std::sync::Arc;
 use std::time::Instant;
 use std::vec;
@@ -223,8 +225,8 @@ async fn send_queued<T>(
     sender.send(InflightEntry::new(entry, bytes)).await
 }
 
-fn queued_batch_bytes(batch: &[Vec<Bytes>]) -> usize {
-    batch.iter().map(|block| queued_block_bytes(block)).sum()
+fn queued_batch_bytes(batch: &[EncodedBlock]) -> usize {
+    batch.iter().map(EncodedBlock::queued_bytes).sum()
 }
 
 fn dominant_error_summary_label(summary: &WriteQuorumFailureSummary) -> &'static str {
@@ -336,7 +338,7 @@ impl<'a> MultiWriter<'a> {
         }
     }
 
-    async fn write_shard(writer_opt: &mut Option<BitrotWriterWrapper>, err: &mut Option<Error>, shard: &Bytes) {
+    async fn write_shard(writer_opt: &mut Option<BitrotWriterWrapper>, err: &mut Option<Error>, shard: &[u8]) {
         match writer_opt {
             Some(writer) => {
                 match writer.write(shard).await {
@@ -361,12 +363,20 @@ impl<'a> MultiWriter<'a> {
     }
 
     pub async fn write(&mut self, data: Vec<Bytes>) -> std::io::Result<()> {
-        assert_eq!(data.len(), self.writers.len());
+        self.write_shards(data.iter().map(Bytes::as_ref)).await
+    }
+
+    async fn write_block(&mut self, block: &EncodedBlock) -> std::io::Result<()> {
+        self.write_shards(block.shards()).await
+    }
+
+    async fn write_shards<'b>(&mut self, shards: impl ExactSizeIterator<Item = &'b [u8]>) -> std::io::Result<()> {
+        assert_eq!(shards.len(), self.writers.len());
 
         let budget = self.next_progress_budget();
         {
             let mut futures = FuturesUnordered::new();
-            for ((writer_opt, err), shard) in self.writers.iter_mut().zip(self.errs.iter_mut()).zip(data.iter()) {
+            for ((writer_opt, err), shard) in self.writers.iter_mut().zip(self.errs.iter_mut()).zip(shards) {
                 if err.is_some() {
                     continue; // Skip if we already have an error for this writer
                 }
@@ -490,10 +500,10 @@ impl<'a> MultiWriter<'a> {
 }
 
 impl Erasure {
-    async fn encode_block(self: Arc<Self>, encode_buf: Vec<u8>, len: usize) -> std::io::Result<(Vec<Bytes>, Vec<u8>)> {
+    async fn encode_block(self: Arc<Self>, encode_buf: Vec<u8>, len: usize) -> std::io::Result<(EncodedBlock, Vec<u8>)> {
         let encode_stage_start = stage_timer_if_enabled();
         let encode_once = move || {
-            let res = self.encode_data(&encode_buf[..len]);
+            let res = self.encode_data_block(&encode_buf[..len]);
             (res, encode_buf)
         };
 
@@ -518,9 +528,9 @@ impl Erasure {
         Ok((res?, returned_buf))
     }
 
-    async fn encode_block_bytes_mut(self: Arc<Self>, encode_buf: BytesMut, len: usize) -> std::io::Result<Vec<Bytes>> {
+    async fn encode_block_bytes_mut(self: Arc<Self>, encode_buf: BytesMut, len: usize) -> std::io::Result<EncodedBlock> {
         let encode_stage_start = stage_timer_if_enabled();
-        let encode_once = move || self.encode_data_bytes_mut(encode_buf, len);
+        let encode_once = move || self.encode_data_bytes_mut_block(encode_buf, len);
 
         let res = match tokio::runtime::Handle::current().runtime_flavor() {
             // Same rationale as encode_block: inline the short EC burst on the
@@ -583,6 +593,39 @@ impl Erasure {
         Ok((reader, total))
     }
 
+    /// Encode a small inline object directly into its per-disk bitrot payloads.
+    /// The returned bytes are the same `[hash][shard]` representation produced
+    /// by `BitrotWriter`, ready to be embedded in each disk's staged `xl.meta`.
+    #[hotpath::measure(impl_type = "Erasure")]
+    pub(crate) async fn encode_inline_shards_with_size_hint<R>(
+        self: Arc<Self>,
+        mut reader: R,
+        size_hint: usize,
+    ) -> std::io::Result<(R, usize, Vec<Bytes>)>
+    where
+        R: AsyncRead + Send + Sync + Unpin,
+    {
+        use tokio::io::AsyncReadExt;
+
+        let mut buf = Vec::with_capacity(small_ingest_capacity(&self, size_hint));
+        let total = reader.read_to_end(&mut buf).await?;
+        if total == 0 {
+            return Ok((reader, 0, Vec::new()));
+        }
+
+        let shards = self.encode_data_owned(buf)?;
+        let mut inline_shards = Vec::with_capacity(shards.len());
+        for shard in shards {
+            let hash = HashAlgorithm::HighwayHash256S.hash_encode(&shard);
+            let mut encoded = BytesMut::with_capacity(hash.as_ref().len() + shard.len());
+            encoded.extend_from_slice(hash.as_ref());
+            encoded.extend_from_slice(&shard);
+            inline_shards.push(encoded.freeze());
+        }
+
+        Ok((reader, total, inline_shards))
+    }
+
     #[hotpath::measure(impl_type = "Erasure")]
     pub async fn encode<R>(
         self: Arc<Self>,
@@ -624,7 +667,7 @@ impl Erasure {
         let expanded_block_bytes = self.shard_size().saturating_mul(self.total_shard_count());
         let max_inflight_bytes = erasure_encode_max_inflight_bytes();
         let inflight_blocks = encode_channel_capacity(expanded_block_bytes, max_inflight_bytes);
-        let (tx, mut rx) = mpsc::channel::<InflightEntry<Vec<Bytes>>>(inflight_blocks);
+        let (tx, mut rx) = mpsc::channel::<InflightEntry<EncodedBlock>>(inflight_blocks);
 
         let mut task = AbortOnDropTask::new(tokio::spawn(async move {
             let block_size = self.block_size;
@@ -646,7 +689,7 @@ impl Erasure {
                             let encode_buf = buf;
                             let res = self.clone().encode_block_bytes_mut(encode_buf, n).await?;
                             buf = BytesMut::with_capacity(ingest_capacity);
-                            let queued_bytes = queued_block_bytes(&res);
+                            let queued_bytes = res.queued_bytes();
                             let _producer_stage = rustfs_io_metrics::track_ec_encode_producer_bytes(queued_bytes);
                             let send_wait_stage_start = stage_timer_if_enabled();
                             if let Err(err) = send_queued(&tx, res, queued_bytes).await {
@@ -676,7 +719,7 @@ impl Erasure {
                             let encode_buf = std::mem::take(&mut buf);
                             let (res, returned_buf) = self.clone().encode_block(encode_buf, n).await?;
                             buf = returned_buf;
-                            let queued_bytes = queued_block_bytes(&res);
+                            let queued_bytes = res.queued_bytes();
                             let _producer_stage = rustfs_io_metrics::track_ec_encode_producer_bytes(queued_bytes);
                             let send_wait_stage_start = stage_timer_if_enabled();
                             if let Err(err) = send_queued(&tx, res, queued_bytes).await {
@@ -720,9 +763,9 @@ impl Erasure {
             if block.is_empty() {
                 break;
             }
-            let _writer_stage = rustfs_io_metrics::track_ec_encode_writer_bytes(queued_block_bytes(&block));
+            let _writer_stage = rustfs_io_metrics::track_ec_encode_writer_bytes(block.queued_bytes());
             let write_stage_start = stage_timer_if_enabled();
-            if let Err(err) = writers.write(block).await {
+            if let Err(err) = writers.write_block(&block).await {
                 write_err = Some(err);
                 break;
             }
@@ -769,7 +812,7 @@ impl Erasure {
         let inflight_blocks = encode_channel_capacity(expanded_block_bytes, max_inflight_bytes);
         let batch_blocks = encode_batch_block_count().min(inflight_blocks);
         let channel_capacity = inflight_blocks.div_ceil(batch_blocks).max(1);
-        let (tx, mut rx) = mpsc::channel::<InflightEntry<Vec<Vec<Bytes>>>>(channel_capacity);
+        let (tx, mut rx) = mpsc::channel::<InflightEntry<Vec<EncodedBlock>>>(channel_capacity);
 
         let mut task = AbortOnDropTask::new(tokio::spawn(async move {
             let block_size = self.block_size;
@@ -786,7 +829,7 @@ impl Erasure {
                         let encode_buf = std::mem::take(&mut buf);
                         let (res, returned_buf) = self.clone().encode_block(encode_buf, n).await?;
                         buf = returned_buf;
-                        let queued_bytes = queued_block_bytes(&res);
+                        let queued_bytes = res.queued_bytes();
                         pending_batch_bytes = pending_batch_bytes.saturating_add(queued_bytes);
                         pending_batch.push(res);
                         drop(pending_batch_stage.take());
@@ -845,7 +888,7 @@ impl Erasure {
             let _writer_stage = rustfs_io_metrics::track_ec_encode_writer_bytes(queued_batch_bytes(&batch));
             let write_stage_start = stage_timer_if_enabled();
             for block in batch {
-                if let Err(err) = writers.write(block).await {
+                if let Err(err) = writers.write_block(&block).await {
                     write_err = Some(err);
                     break;
                 }
@@ -1895,7 +1938,11 @@ mod tests {
         let baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes();
         let (tx, rx) = mpsc::channel(2);
         let mut rx = rx;
-        let batch = vec![vec![Bytes::from_static(b"queued")], vec![Bytes::from_static(b"batch")]];
+        let erasure = Erasure::new(1, 0, 16);
+        let batch = vec![
+            erasure.encode_data_block(b"queued").expect("first block should encode"),
+            erasure.encode_data_block(b"batch").expect("second block should encode"),
+        ];
         let batch_bytes = queued_batch_bytes(&batch);
 
         send_queued(&tx, batch, batch_bytes).await.expect("batch should be queued");
@@ -2236,11 +2283,11 @@ mod tests {
             .expect("bytesmut encode should succeed on current-thread runtime");
 
         let expected_shard_size = payload.len().div_ceil(erasure.data_shards);
-        assert_eq!(shards.len(), erasure.total_shard_count());
-        assert!(shards.iter().all(|shard| shard.len() == expected_shard_size));
+        assert_eq!(shards.shards().len(), erasure.total_shard_count());
+        assert!(shards.shards().all(|shard| shard.len() == expected_shard_size));
 
         let mut restored = Vec::new();
-        for shard in shards.iter().take(erasure.data_shards) {
+        for shard in shards.shards().take(erasure.data_shards) {
             restored.extend_from_slice(shard);
         }
         restored.truncate(payload.len());
@@ -2343,6 +2390,35 @@ mod tests {
         assert!(committed.lock().unwrap().is_empty());
     }
 
+    #[tokio::test]
+    async fn encode_inline_shards_matches_writer_bitrot_layout() {
+        const DATA_SHARDS: usize = 2;
+        const PARITY_SHARDS: usize = 2;
+        const BLOCK_SIZE: usize = 64;
+        let payload = b"inline commit payload".to_vec();
+        let checksum_algo = HashAlgorithm::HighwayHash256S;
+        let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE));
+        let reader = tokio::io::BufReader::new(Cursor::new(payload.clone()));
+
+        let (_reader, total, inline_shards) = erasure
+            .clone()
+            .encode_inline_shards_with_size_hint(reader, payload.len())
+            .await
+            .expect("inline shards should encode");
+        let raw_shards = erasure
+            .encode_data_owned(payload.clone())
+            .expect("reference shards should encode");
+
+        assert_eq!(total, payload.len());
+        assert_eq!(inline_shards.len(), DATA_SHARDS + PARITY_SHARDS);
+        for (inline, raw) in inline_shards.iter().zip(raw_shards) {
+            let mut writer = BitrotWriterWrapper::new(CustomWriter::new_inline_buffer(), raw.len(), checksum_algo.clone());
+            writer.write(&raw).await.expect("reference writer should accept shard");
+            writer.shutdown().await.expect("reference writer should shutdown");
+            assert_eq!(inline.as_ref(), writer.into_inline_data().expect("reference writer should retain bytes"));
+        }
+    }
+
     /// encode_inline_small: small payload is encoded into the correct number of shards
     /// and each writer receives data after shutdown.
     #[tokio::test]
@@ -2506,7 +2582,7 @@ mod tests {
         assert_eq!(&next[..], &data[16..]);
     }
 
-    async fn committed_shards_for_ingest_mode(use_bytesmut_ingest: bool, uses_legacy: bool, payload: &[u8]) -> Vec<Vec<u8>> {
+    async fn committed_shards_for_pipeline(pipeline: EncodePipeline, uses_legacy: bool, payload: &[u8]) -> Vec<Vec<u8>> {
         const DATA_SHARDS: usize = 2;
         const PARITY_SHARDS: usize = 2;
         const TOTAL_SHARDS: usize = DATA_SHARDS + PARITY_SHARDS;
@@ -2520,10 +2596,16 @@ mod tests {
 
         let erasure = Arc::new(Erasure::new_with_options(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE, uses_legacy));
         let reader = tokio::io::BufReader::new(Cursor::new(payload.to_vec()));
-        let (_reader, total) = erasure
-            .encode_with_ingest_mode(reader, &mut writers, DATA_SHARDS, use_bytesmut_ingest)
-            .await
-            .expect("encode should succeed");
+        let (_reader, total) = match pipeline {
+            EncodePipeline::Vec => {
+                erasure
+                    .encode_with_ingest_mode(reader, &mut writers, DATA_SHARDS, false)
+                    .await
+            }
+            EncodePipeline::BytesMut => erasure.encode_with_ingest_mode(reader, &mut writers, DATA_SHARDS, true).await,
+            EncodePipeline::Batched => erasure.encode_batched(reader, &mut writers, DATA_SHARDS).await,
+        }
+        .expect("encode should succeed");
         assert_eq!(total, payload.len());
 
         committed
@@ -2532,31 +2614,64 @@ mod tests {
             .collect()
     }
 
-    /// HP-10 (rustfs/backlog#931) merge gate: the BytesMut ingest path must produce
-    /// byte-for-byte identical shard streams to the default Vec ingest path, for both
-    /// legacy-aware shard-size formulas, across empty, sub-block, exactly-full-block,
-    /// and multi-block-with-partial-tail payloads.
+    async fn expected_committed_shards(uses_legacy: bool, payload: &[u8]) -> Vec<Vec<u8>> {
+        const DATA_SHARDS: usize = 2;
+        const PARITY_SHARDS: usize = 2;
+        const TOTAL_SHARDS: usize = DATA_SHARDS + PARITY_SHARDS;
+        const BLOCK_SIZE: usize = 64;
+
+        let committed: Vec<Arc<Mutex<Vec<u8>>>> = (0..TOTAL_SHARDS).map(|_| Arc::new(Mutex::new(Vec::new()))).collect();
+        let mut writers: Vec<BitrotWriterWrapper> = committed
+            .iter()
+            .map(|c| bitrot_writer(DeferredCommitWriter::new(c.clone()), BLOCK_SIZE / DATA_SHARDS))
+            .collect();
+        let erasure = Erasure::new_with_options(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE, uses_legacy);
+
+        for block in payload.chunks(BLOCK_SIZE) {
+            let shards = erasure.encode_data(block).expect("reference block should encode");
+            for (writer, shard) in writers.iter_mut().zip(shards) {
+                let written = writer.write(&shard).await.expect("reference shard should write");
+                assert_eq!(written, shard.len());
+            }
+        }
+        for writer in &mut writers {
+            writer.shutdown().await.expect("reference writer should commit");
+        }
+
+        committed
+            .iter()
+            .map(|c| c.lock().expect("committed buffer should be lockable").clone())
+            .collect()
+    }
+
+    /// The streaming and batched paths must produce the same bitrot-wrapped shard
+    /// bytes as the public block encoder for both shard-size formulas and all block
+    /// boundary shapes.
     #[tokio::test]
     async fn bytesmut_ingest_matches_vec_ingest_byte_for_byte() {
         const BLOCK_SIZE: usize = 64;
         let payloads: Vec<Vec<u8>> = vec![
             Vec::new(),
-            b"tiny".to_vec(),
+            vec![1],
+            vec![2; BLOCK_SIZE - 1],
             (0..BLOCK_SIZE as u32).map(|i| i as u8).collect(), // exactly one full block
-            vec![3u8; BLOCK_SIZE * 4],                         // whole number of blocks
+            vec![4; BLOCK_SIZE + 1],
+            vec![3u8; BLOCK_SIZE * 4],                                           // whole number of blocks
             (0..(BLOCK_SIZE * 3 + 7) as u32).map(|i| (i % 251) as u8).collect(), // partial tail
         ];
 
         for uses_legacy in [false, true] {
             for payload in &payloads {
-                let vec_path = committed_shards_for_ingest_mode(false, uses_legacy, payload).await;
-                let bytesmut_path = committed_shards_for_ingest_mode(true, uses_legacy, payload).await;
-                assert_eq!(
-                    vec_path,
-                    bytesmut_path,
-                    "ingest paths must be byte-identical (legacy={uses_legacy}, payload_len={})",
-                    payload.len()
-                );
+                let expected = expected_committed_shards(uses_legacy, payload).await;
+                for pipeline in [EncodePipeline::Vec, EncodePipeline::BytesMut, EncodePipeline::Batched] {
+                    let actual = committed_shards_for_pipeline(pipeline, uses_legacy, payload).await;
+                    assert_eq!(
+                        actual,
+                        expected,
+                        "streaming shards must match the public block encoder (legacy={uses_legacy}, payload_len={})",
+                        payload.len()
+                    );
+                }
             }
         }
     }

crates/ecstore/src/erasure/coding/erasure.rs

@@ -29,6 +29,46 @@ use tokio::io::AsyncRead;
 use tracing::warn;
 use uuid::Uuid;
 
+pub(crate) struct EncodedBlock {
+    data: Bytes,
+    shard_size: usize,
+}
+
+impl EncodedBlock {
+    fn empty() -> Self {
+        Self {
+            data: Bytes::new(),
+            shard_size: 0,
+        }
+    }
+
+    pub(crate) fn is_empty(&self) -> bool {
+        self.data.is_empty()
+    }
+
+    pub(crate) fn queued_bytes(&self) -> usize {
+        self.data.len()
+    }
+
+    pub(crate) fn shards(&self) -> impl ExactSizeIterator<Item = &[u8]> {
+        debug_assert!(self.shard_size > 0, "only non-empty encoded blocks reach shard writers");
+        debug_assert_eq!(self.data.len() % self.shard_size, 0);
+        self.data.chunks_exact(self.shard_size)
+    }
+
+    fn into_shards(mut self, shard_count: usize) -> Vec<Bytes> {
+        if self.shard_size == 0 {
+            return vec![Bytes::new(); shard_count];
+        }
+
+        let mut shards = Vec::with_capacity(shard_count);
+        for _ in 0..shard_count {
+            shards.push(self.data.split_to(self.shard_size));
+        }
+        shards
+    }
+}
+
 const MODERN_MAX_TOTAL_SHARDS: usize = <reed_solomon_erasure::galois_8::Field as reed_solomon_erasure::Field>::ORDER;
 const MODERN_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 64;
 
@@ -675,6 +715,17 @@ impl Erasure {
     #[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))]
     #[hotpath::measure(impl_type = "Erasure")]
     pub fn encode_data(&self, data: &[u8]) -> io::Result<Vec<Bytes>> {
+        self.encode_data_block_inner(data)
+            .map(|block| block.into_shards(self.total_shard_count()))
+    }
+
+    #[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))]
+    #[hotpath::measure(label = "Erasure::encode_data", impl_type = "Erasure")]
+    pub(crate) fn encode_data_block(&self, data: &[u8]) -> io::Result<EncodedBlock> {
+        self.encode_data_block_inner(data)
+    }
+
+    fn encode_data_block_inner(&self, data: &[u8]) -> io::Result<EncodedBlock> {
         let shard_size_fn = if self.uses_legacy {
             calc_shard_size_legacy
         } else {
@@ -682,7 +733,7 @@ impl Erasure {
         };
         let per_shard_size = shard_size_fn(data.len(), self.data_shards);
         if per_shard_size == 0 {
-            return Ok(vec![Bytes::new(); self.total_shard_count()]);
+            return Ok(EncodedBlock::empty());
         }
         let need_total_size = per_shard_size * self.total_shard_count();
 
@@ -708,15 +759,10 @@ impl Erasure {
             }
         }
 
-        // Zero-copy split, all shards reference data_buffer
-        let mut data_buffer = data_buffer.freeze();
-        let mut shards = Vec::with_capacity(self.total_shard_count());
-        for _ in 0..self.total_shard_count() {
-            let shard = data_buffer.split_to(per_shard_size);
-            shards.push(shard);
-        }
-
-        Ok(shards)
+        Ok(EncodedBlock {
+            data: data_buffer.freeze(),
+            shard_size: per_shard_size,
+        })
     }
 
     /// Encode owned data, avoiding a copy when the caller already has a heap buffer.
@@ -786,7 +832,17 @@ impl Erasure {
     /// `data_len <= block_size` — both shard-size formulas are monotone in
     /// `data_len` — so this function never reallocates the buffer.
     #[hotpath::measure(impl_type = "Erasure")]
-    pub fn encode_data_bytes_mut(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result<Vec<Bytes>> {
+    pub fn encode_data_bytes_mut(&self, data_buffer: BytesMut, data_len: usize) -> io::Result<Vec<Bytes>> {
+        self.encode_data_bytes_mut_block_inner(data_buffer, data_len)
+            .map(|block| block.into_shards(self.total_shard_count()))
+    }
+
+    #[hotpath::measure(label = "Erasure::encode_data_bytes_mut", impl_type = "Erasure")]
+    pub(crate) fn encode_data_bytes_mut_block(&self, data_buffer: BytesMut, data_len: usize) -> io::Result<EncodedBlock> {
+        self.encode_data_bytes_mut_block_inner(data_buffer, data_len)
+    }
+
+    fn encode_data_bytes_mut_block_inner(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result<EncodedBlock> {
         let shard_size_fn = if self.uses_legacy {
             calc_shard_size_legacy
         } else {
@@ -794,7 +850,7 @@ impl Erasure {
         };
         let per_shard_size = shard_size_fn(data_len, self.data_shards);
         if per_shard_size == 0 {
-            return Ok(vec![Bytes::new(); self.total_shard_count()]);
+            return Ok(EncodedBlock::empty());
         }
         let need_total_size = per_shard_size * self.total_shard_count();
 
@@ -821,14 +877,10 @@ impl Erasure {
             }
         }
 
-        let mut data_buffer = data_buffer.freeze();
-        let mut shards = Vec::with_capacity(self.total_shard_count());
-        for _ in 0..self.total_shard_count() {
-            let shard = data_buffer.split_to(per_shard_size);
-            shards.push(shard);
-        }
-
-        Ok(shards)
+        Ok(EncodedBlock {
+            data: data_buffer.freeze(),
+            shard_size: per_shard_size,
+        })
     }
 
     /// Decode and reconstruct missing data shards in-place.
@@ -1547,6 +1599,37 @@ mod tests {
         }
     }
 
+    #[test]
+    fn streaming_encoded_block_uses_one_contiguous_backing_buffer() {
+        let erasure = Erasure::new(8, 8, 64);
+
+        for data_len in [1, 63, 64] {
+            let data = (0..data_len).map(|i| i as u8).collect::<Vec<_>>();
+            let expected = erasure.encode_data(&data).expect("public encode should succeed");
+            let borrowed = erasure
+                .encode_data_block(&data)
+                .expect("borrowed streaming encode should succeed");
+            let owned = erasure
+                .encode_data_bytes_mut_block(BytesMut::from(&data[..]), data.len())
+                .expect("BytesMut streaming encode should succeed");
+
+            assert!(borrowed.shards().eq(expected.iter().map(Bytes::as_ref)));
+            assert!(owned.shards().eq(expected.iter().map(Bytes::as_ref)));
+            assert_eq!(borrowed.shards().len(), 16);
+            assert_eq!(borrowed.queued_bytes(), owned.queued_bytes());
+
+            let first = borrowed.shards().next().expect("encoded block should have shards").as_ptr();
+            for (index, shard) in borrowed.shards().enumerate() {
+                assert_eq!(shard.as_ptr(), first.wrapping_add(index * shard.len()));
+            }
+        }
+        assert_eq!(
+            std::mem::size_of::<EncodedBlock>(),
+            std::mem::size_of::<Bytes>() + std::mem::size_of::<usize>(),
+            "queue entries must contain one backing buffer handle, not per-shard handles"
+        );
+    }
+
     /// HP-10 capacity invariant: both shard-size formulas are monotone in `data_len`,
     /// so pre-reserving `shard_size(block_size) * total_shard_count` covers the
     /// `need_total_size` of every block-or-smaller payload and the ingest buffer

crates/ecstore/src/services/notification_sys.rs

@@ -210,6 +210,22 @@ pub fn cross_pool_fence_fleet_proof_matches(proof: &CrossPoolFenceFleetProofToke
     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,

crates/ecstore/src/set_disk/metadata.rs

@@ -1079,6 +1079,25 @@ impl SetDisks {
         shuffled_disks
     }
 
+    pub(super) fn shuffle_disks_owned(mut disks: Vec<Option<DiskStore>>, distribution: &[usize]) -> Vec<Option<DiskStore>> {
+        if distribution.is_empty() {
+            return disks;
+        }
+
+        let mut shuffled_disks = vec![None; disks.len()];
+        for (index, disk) in disks.iter_mut().enumerate() {
+            let Some(slot) = distribution
+                .get(index)
+                .and_then(|block_index| block_index.checked_sub(1))
+                .filter(|slot| *slot < shuffled_disks.len())
+            else {
+                continue;
+            };
+            shuffled_disks[slot] = disk.take();
+        }
+        shuffled_disks
+    }
+
     pub(super) fn shuffle_check_parts(parts_errs: &[usize], distribution: &[usize]) -> Vec<usize> {
         if distribution.is_empty() {
             return parts_errs.to_vec();
@@ -1390,6 +1409,23 @@ mod tests {
         assert_eq!(owned_slots, expected_slots, "fallback disk slots must match the borrowing variant");
     }
 
+    #[tokio::test]
+    async fn owned_shuffle_preserves_fresh_put_metadata() {
+        let tempdir = tempfile::tempdir().expect("tempdir should be created");
+        let fi = FileInfo::new("bucket/object", 2, 1);
+        let parts = vec![fi.clone(); fi.erasure.distribution.len()];
+        let disks = shuffle_test_disks(&tempdir, parts.len()).await;
+
+        let (owned_disks, owned_parts) = SetDisks::shuffle_disks_and_parts_metadata_by_index_owned(disks, parts, &fi);
+
+        assert!(owned_disks.iter().all(Option::is_some), "fresh PUT must retain every online disk");
+        assert_eq!(
+            owned_parts,
+            vec![fi; owned_disks.len()],
+            "fresh PUT metadata with pending shard indexes must survive init fallback"
+        );
+    }
+
     // backlog#949: corrupt/adversarial distribution values (0 or > N) must not
     // trigger a `usize` underflow / out-of-bounds panic in the shuffle helpers.
     #[test]
@@ -1419,6 +1455,22 @@ mod tests {
         assert_eq!(result.len(), disks.len(), "output length must be preserved");
     }
 
+    #[tokio::test]
+    async fn owned_disk_shuffle_matches_borrowing_variant() {
+        let tempdir = tempfile::tempdir().expect("tempdir should be created");
+        let mut disks = shuffle_test_disks(&tempdir, 4).await;
+        disks[1] = None;
+        disks[3] = None;
+        let distribution = [3, 1, 4, 2];
+
+        let expected = SetDisks::shuffle_disks(&disks, &distribution);
+        let actual = SetDisks::shuffle_disks_owned(disks, &distribution);
+
+        let expected_slots = expected.iter().map(Option::is_some).collect::<Vec<_>>();
+        let actual_slots = actual.iter().map(Option::is_some).collect::<Vec<_>>();
+        assert_eq!(actual_slots, expected_slots, "owned shuffle must preserve disk placement");
+    }
+
     #[tokio::test]
     async fn shuffle_disks_and_parts_metadata_survives_corrupt_distribution() {
         let tempdir = tempfile::tempdir().expect("tempdir should be created");

crates/ecstore/src/set_disk/mod.rs

@@ -718,8 +718,8 @@ pub(crate) use core::io_primitives::disk_call_counters;
 mod ctx;
 mod metadata;
 mod ops;
-#[cfg(test)]
-pub(crate) use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause};
+#[cfg(any(test, feature = "test-util"))]
+pub 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;

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

@@ -27,24 +27,25 @@ use crate::crash_inject::{self, CrashPoint};
 use crate::multipart_listing::paginate_multipart_listing;
 use futures::{StreamExt, stream};
 use std::future::Future;
-#[cfg(test)]
+#[cfg(any(test, feature = "test-util"))]
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::Duration;
 use tokio::task::JoinSet;
 
 const MULTIPART_LIST_IO_CONCURRENCY: usize = 16;
 
-#[cfg(test)]
+#[cfg(any(test, feature = "test-util"))]
 #[derive(Clone, Copy, PartialEq, Eq)]
-pub(crate) enum MultipartCommitPause {
+pub enum MultipartCommitPause {
     PutPartBeforeLockAcquire,
     PutPartBeforeLockLost,
     PutPartAfterRename,
     BeforeLockLost,
+    BeforeQuotaRename,
     AfterRename,
 }
 
-#[cfg(test)]
+#[cfg(any(test, feature = "test-util"))]
 struct MultipartCommitBarrierState {
     bucket: String,
     object: String,
@@ -55,27 +56,22 @@ struct MultipartCommitBarrierState {
     release: tokio::sync::Semaphore,
 }
 
-#[cfg(test)]
-pub(crate) struct MultipartCommitBarrier {
+#[cfg(any(test, feature = "test-util"))]
+pub struct MultipartCommitBarrier {
     state: Arc<MultipartCommitBarrierState>,
 }
 
-#[cfg(test)]
+#[cfg(any(test, feature = "test-util"))]
 static MULTIPART_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Arc<MultipartCommitBarrierState>>>> =
     std::sync::OnceLock::new();
 
-#[cfg(test)]
+#[cfg(any(test, feature = "test-util"))]
 impl MultipartCommitBarrier {
-    pub(crate) fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
+    pub fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
         Self::install_for_arrivals(bucket, object, pause, 1)
     }
 
-    pub(crate) fn install_for_arrivals(
-        bucket: &str,
-        object: &str,
-        pause: MultipartCommitPause,
-        expected_arrivals: usize,
-    ) -> Self {
+    pub 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(),
@@ -96,7 +92,7 @@ impl MultipartCommitBarrier {
         Self { state }
     }
 
-    pub(crate) async fn wait_until_paused(&self) {
+    pub async fn wait_until_paused(&self) {
         tokio::time::timeout(Duration::from_secs(30), async {
             loop {
                 let arrived = self.state.arrived.notified();
@@ -110,12 +106,12 @@ impl MultipartCommitBarrier {
         .expect("multipart completion should reach the deterministic commit barrier");
     }
 
-    pub(crate) fn release(&self) {
+    pub fn release(&self) {
         self.state.release.add_permits(self.state.expected_arrivals);
     }
 }
 
-#[cfg(test)]
+#[cfg(any(test, feature = "test-util"))]
 impl Drop for MultipartCommitBarrier {
     fn drop(&mut self) {
         self.release();
@@ -129,7 +125,7 @@ impl Drop for MultipartCommitBarrier {
     }
 }
 
-#[cfg(test)]
+#[cfg(any(test, feature = "test-util"))]
 async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartCommitPause) {
     let barrier = MULTIPART_COMMIT_BARRIER
         .get_or_init(|| std::sync::Mutex::new(None))
@@ -1761,8 +1757,16 @@ 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, self.pool_index, self.set_index).await?;
+        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);
@@ -2125,7 +2129,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         }
 
         let quota_old_size = if quota_context.is_enforced() {
-            reservation::replaced_logical_size(&self, bucket, object, opts).await?
+            if opts.data_movement {
+                quota_new_size
+            } else {
+                reservation::replaced_logical_size(&self, bucket, object, opts).await?
+            }
         } else {
             0
         };
@@ -2228,7 +2236,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             .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

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

@@ -62,6 +62,10 @@ fn duration_millis_f64(duration: std::time::Duration) -> f64 {
     duration.as_secs_f64() * 1000.0
 }
 
+fn committed_response_metadata_slot<D>(committed_disks: &[Option<D>], fallback_slot: usize) -> usize {
+    committed_disks.iter().position(Option::is_some).unwrap_or(fallback_slot)
+}
+
 #[cfg(test)]
 mod duration_metrics_tests {
     use super::duration_millis_f64;
@@ -73,6 +77,38 @@ mod duration_metrics_tests {
     }
 }
 
+#[cfg(test)]
+mod put_metadata_tests {
+    use super::*;
+
+    #[test]
+    fn committed_file_info_follows_exact_quorum_success_slot() {
+        let mut first_success = FileInfo::new("bucket/object", 2, 2);
+        first_success.name = "first-success".to_string();
+        let mut second_success = first_success.clone();
+        second_success.name = "second-success".to_string();
+        let mut parts_metadata = [FileInfo::default(), first_success, second_success, FileInfo::default()];
+        let committed_disks = [None, Some(()), Some(()), None];
+
+        assert_eq!(
+            committed_disks.iter().filter(|disk| disk.is_some()).count(),
+            2,
+            "fixture must meet exact quorum"
+        );
+        let selected_slot = committed_response_metadata_slot(&committed_disks, 3);
+        let selected = std::mem::take(&mut parts_metadata[selected_slot]);
+
+        assert_eq!(selected.name, "first-success");
+        assert_eq!(parts_metadata[1], FileInfo::default(), "selected metadata should move without cloning");
+        assert_eq!(parts_metadata[2].name, "second-success", "other committed metadata must remain available");
+        assert_eq!(
+            committed_response_metadata_slot::<()>(&[None, None, None, None], 3),
+            3,
+            "a violated post-commit success-mask invariant must not turn a durable PUT into an error"
+        );
+    }
+}
+
 fn is_restore_control_metadata(key: &str) -> bool {
     key.eq_ignore_ascii_case(X_AMZ_RESTORE.as_str())
         || key.eq_ignore_ascii_case(rustfs_utils::http::headers::AMZ_RESTORE_EXPIRY_DAYS)
@@ -1060,10 +1096,7 @@ impl SetDisks {
         }
 
         fi.data_dir = Some(Uuid::new_v4());
-
-        let parts_metadata = vec![fi.clone(); disks.len()];
-
-        let (mut shuffle_disks, mut parts_metadatas) = Self::shuffle_disks_and_parts_metadata(&disks, &parts_metadata, &fi);
+        let mut shuffle_disks = Self::shuffle_disks_owned(disks, &fi.erasure.distribution);
 
         let tmp_dir = Uuid::new_v4().to_string();
 
@@ -1075,61 +1108,91 @@ impl SetDisks {
             let put_object_size = known_put_object_storage_size(data.size());
             let is_inline_buffer = storage_class_config.should_inline(erasure.shard_file_size(put_object_size), opts.versioned);
 
+            let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
             let shard_file_size = erasure.shard_file_size(put_object_size);
             let shard_size = erasure.shard_size();
-            let writer_setup_stage_start = Instant::now();
-            let writer_futs: Vec<_> = shuffle_disks
-                .iter()
-                .map(|disk_op| {
-                    let tmp_obj = tmp_object.clone();
-                    async move {
-                        if let Some(disk) = disk_op
-                            && disk.is_online().await
-                        {
-                            match create_bitrot_writer(
-                                is_inline_buffer,
-                                Some(disk),
-                                RUSTFS_META_TMP_BUCKET,
-                                &tmp_obj,
-                                shard_file_size,
-                                shard_size,
-                                HashAlgorithm::HighwayHash256S,
-                            )
-                            .await
-                            {
-                                Ok(writer) => (Some(writer), None),
-                                Err(err) => {
-                                    warn!(
-                                        event = EVENT_SET_DISK_WRITE,
-                                        component = LOG_COMPONENT_ECSTORE,
-                                        subsystem = LOG_SUBSYSTEM_SET_DISK,
-                                        disk = ?disk,
-                                        state = "bitrot_writer_skipped",
-                                        error = ?err,
-                                        "Set disk bitrot writer skipped"
-                                    );
-                                    (None, Some(err))
-                                }
-                            }
-                        } else {
-                            (None, Some(DiskError::DiskNotFound))
-                        }
+            let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size);
+            let direct_inline_commit = matches!(write_path, SmallWritePath::Inline);
+            rustfs_io_metrics::record_put_object_path(write_path.metric_label());
+            let writer_setup_stage_start = collect_stage_timing.then(Instant::now);
+            let (mut writers, errors) = if direct_inline_commit {
+                let online = join_all(shuffle_disks.iter().map(|disk| async move {
+                    if let Some(disk) = disk {
+                        disk.is_online().await
+                    } else {
+                        false
                     }
-                })
-                .collect();
-            let writer_results = join_all(writer_futs).await;
-            let mut writers = Vec::with_capacity(writer_results.len());
-            let mut errors = Vec::with_capacity(writer_results.len());
-            for (w, e) in writer_results {
-                writers.push(w);
-                errors.push(e);
+                }))
+                .await;
+                let mut errors = Vec::with_capacity(online.len());
+                for (disk, is_online) in shuffle_disks.iter_mut().zip(online) {
+                    if is_online {
+                        errors.push(None);
+                    } else {
+                        *disk = None;
+                        errors.push(Some(DiskError::DiskNotFound));
+                    }
+                }
+                (std::iter::repeat_with(|| None).take(shuffle_disks.len()).collect(), errors)
+            } else {
+                let writer_futs: Vec<_> = shuffle_disks
+                    .iter()
+                    .map(|disk_op| {
+                        let tmp_obj = tmp_object.clone();
+                        async move {
+                            if let Some(disk) = disk_op
+                                && disk.is_online().await
+                            {
+                                match create_bitrot_writer(
+                                    is_inline_buffer,
+                                    Some(disk),
+                                    RUSTFS_META_TMP_BUCKET,
+                                    &tmp_obj,
+                                    shard_file_size,
+                                    shard_size,
+                                    HashAlgorithm::HighwayHash256S,
+                                )
+                                .await
+                                {
+                                    Ok(writer) => (Some(writer), None),
+                                    Err(err) => {
+                                        warn!(
+                                            event = EVENT_SET_DISK_WRITE,
+                                            component = LOG_COMPONENT_ECSTORE,
+                                            subsystem = LOG_SUBSYSTEM_SET_DISK,
+                                            disk = ?disk,
+                                            state = "bitrot_writer_skipped",
+                                            error = ?err,
+                                            "Set disk bitrot writer skipped"
+                                        );
+                                        (None, Some(err))
+                                    }
+                                }
+                            } else {
+                                (None, Some(DiskError::DiskNotFound))
+                            }
+                        }
+                    })
+                    .collect();
+                let writer_results = join_all(writer_futs).await;
+                let mut writers = Vec::with_capacity(writer_results.len());
+                let mut errors = Vec::with_capacity(writer_results.len());
+                for (writer, error) in writer_results {
+                    writers.push(writer);
+                    errors.push(error);
+                }
+                (writers, errors)
+            };
+            let writer_setup_elapsed = writer_setup_stage_start.map(|stage_start| stage_start.elapsed());
+            let writer_setup_ms = writer_setup_elapsed
+                .map(|elapsed| elapsed.as_millis() as u64)
+                .unwrap_or_default();
+            if let Some(writer_setup_elapsed) = writer_setup_elapsed {
+                rustfs_io_metrics::record_put_object_stage_duration(
+                    "set_disk_writer_setup",
+                    duration_millis_f64(writer_setup_elapsed),
+                );
             }
-            let writer_setup_elapsed = writer_setup_stage_start.elapsed();
-            let writer_setup_ms = writer_setup_elapsed.as_millis() as u64;
-            rustfs_io_metrics::record_put_object_stage_duration(
-                "set_disk_writer_setup",
-                duration_millis_f64(writer_setup_elapsed),
-            );
 
             let nil_count = errors.iter().filter(|&e| e.is_none()).count();
             if nil_count < write_quorum {
@@ -1157,21 +1220,23 @@ impl SetDisks {
                 HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?,
             );
 
-            let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size);
-            rustfs_io_metrics::record_put_object_path(write_path.metric_label());
             let small_size_hint = if matches!(write_path, SmallWritePath::Inline | SmallWritePath::SingleBlockNonInline) {
                 usize::try_from(put_object_size).map_err(Error::other)?
             } else {
                 0
             };
 
-            let encode_stage_start = Instant::now();
+            let encode_stage_start = collect_stage_timing.then(Instant::now);
+            let mut inline_shards = None;
             let (reader, w_size) = match write_path {
                 SmallWritePath::Inline => match Arc::clone(&erasure)
-                    .encode_inline_small_with_size_hint(stream, &mut writers, write_quorum, small_size_hint)
+                    .encode_inline_shards_with_size_hint(stream, small_size_hint)
                     .await
                 {
-                    Ok((r, w)) => (r, w),
+                    Ok((r, w, shards)) => {
+                        inline_shards = Some(shards);
+                        (r, w)
+                    }
                     Err(e) => {
                         error!("encode_inline_small err {:?}", e);
                         return Err(e.into());
@@ -1204,9 +1269,11 @@ impl SetDisks {
                     }
                 },
             };
-            let encode_elapsed = encode_stage_start.elapsed();
-            let encode_ms = encode_elapsed.as_millis() as u64;
-            rustfs_io_metrics::record_put_object_stage_duration("set_disk_encode", duration_millis_f64(encode_elapsed));
+            let encode_elapsed = encode_stage_start.map(|stage_start| stage_start.elapsed());
+            let encode_ms = encode_elapsed.map(|elapsed| elapsed.as_millis() as u64).unwrap_or_default();
+            if let Some(encode_elapsed) = encode_elapsed {
+                rustfs_io_metrics::record_put_object_stage_duration("set_disk_encode", duration_millis_f64(encode_elapsed));
+            }
 
             let _ = mem::replace(&mut data.stream, reader);
             // if let Err(err) = close_bitrot_writers(&mut writers).await {
@@ -1292,37 +1359,67 @@ impl SetDisks {
             // drop it below reconstructable quorum (backlog#852 / #799 B3).
             // `rename_data` re-checks write quorum over the surviving disks and
             // rolls back if too few remain.
-            let committed_shards = drop_failed_writer_disks(&mut shuffle_disks, &writers);
+            let committed_shards = if matches!(write_path, SmallWritePath::Inline) {
+                shuffle_disks.iter().filter(|disk| disk.is_some()).count()
+            } else {
+                drop_failed_writer_disks(&mut shuffle_disks, &writers)
+            };
             if committed_shards < write_quorum {
                 return Err(Error::other(format!(
                     "put_object write quorum unavailable after encode: {committed_shards} shard(s) committed, need {write_quorum}"
                 )));
             }
 
-            for (i, pfi) in parts_metadatas.iter_mut().enumerate() {
-                pfi.metadata = user_defined.clone();
+            fi.metadata = user_defined;
+            fi.mod_time = mod_time;
+            fi.size = w_size as i64;
+            fi.versioned = opts.versioned || opts.version_suspended;
+            fi.add_object_part(1, etag, w_size, mod_time, actual_size, index_op, None);
+            if opts.data_movement {
+                fi.set_data_moved();
+            }
+            let parity_blocks = fi.erasure.parity_blocks;
+
+            let response_metadata_slot = shuffle_disks
+                .iter()
+                .rposition(Option::is_some)
+                .ok_or_else(|| Error::other("put_object write quorum unavailable after encode"))?;
+            let mut base_file_info = fi;
+            let mut parts_metadatas = Vec::with_capacity(shuffle_disks.len());
+            for (i, disk) in shuffle_disks.iter().enumerate() {
+                if disk.is_none() {
+                    parts_metadatas.push(FileInfo::default());
+                    continue;
+                }
+
+                let mut pfi = if i == response_metadata_slot {
+                    std::mem::take(&mut base_file_info)
+                } else {
+                    base_file_info.clone()
+                };
                 if is_inline_buffer {
-                    if let Some(writer) = writers[i].take() {
+                    if let Some(shards) = inline_shards.as_ref() {
+                        pfi.data = Some(
+                            shards
+                                .get(i)
+                                .cloned()
+                                .ok_or_else(|| Error::other(format!("inline encoder omitted disk shard {i}")))?,
+                        );
+                    } else if let Some(writer) = writers[i].take() {
                         pfi.data = Some(writer.into_inline_data().map(Bytes::from).unwrap_or_default());
                     }
 
                     pfi.set_inline_data();
                 }
-
-                pfi.mod_time = mod_time;
-                pfi.size = w_size as i64;
-                pfi.versioned = opts.versioned || opts.version_suspended;
-                pfi.add_object_part(1, etag.clone(), w_size, mod_time, actual_size, index_op.clone(), None);
-                pfi.checksum = fi.checksum.clone();
-
-                if opts.data_movement {
-                    pfi.set_data_moved();
-                }
+                parts_metadatas.push(pfi);
             }
+            let committed_version_id = parts_metadatas[response_metadata_slot].version_id;
+            let committed_data_dir = parts_metadatas[response_metadata_slot].data_dir;
+            let is_compressed = parts_metadatas[response_metadata_slot].is_compressed();
 
             drop(writers); // drop writers to close all files, this is to prevent FileAccessDenied errors when renaming data
 
-            if fi.erasure.parity_blocks == 0 {
+            if parity_blocks == 0 {
                 let written_size = i64::try_from(w_size).map_err(|_| Error::other("put_object written size overflows i64"))?;
                 let logical_shard_size = usize::try_from(erasure.shard_file_size(written_size))
                     .map_err(|_| Error::other("put_object shard size overflows usize"))?;
@@ -1496,8 +1593,16 @@ impl SetDisks {
                 });
             }
 
-            let quota_context =
-                reservation::begin(&self.ctx, bucket, object, opts.quota_admission, self.pool_index, self.set_index).await?;
+            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 mut replication_quota_size = None;
 
@@ -1506,17 +1611,16 @@ impl SetDisks {
                     return Err(Error::PartMissingOrCorrupt);
                 }
                 if quota_context.is_enforced() {
+                    let persisted_metadata = &parts_metadatas[response_metadata_slot].metadata;
                     let observed_size = u64::try_from(actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
                     let physical_size = u64::try_from(w_size).map_err(|_| Error::PartMissingOrCorrupt)?;
-                    let transformed = contains_key_str(&user_defined, SUFFIX_COMPRESSION)
-                        || parts_metadatas
-                            .first()
-                            .is_some_and(|metadata| should_persist_encryption_original_size(&metadata.metadata));
-                    let declared_size = get_str(&user_defined, SUFFIX_ACTUAL_SIZE)
+                    let transformed = contains_key_str(persisted_metadata, SUFFIX_COMPRESSION)
+                        || should_persist_encryption_original_size(persisted_metadata);
+                    let declared_size = get_str(persisted_metadata, SUFFIX_ACTUAL_SIZE)
                         .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(&user_defined)
+                    let declared_encryption_size = rustfs_utils::http::get_object_encryption_original_size(persisted_metadata)
                         .map_err(Error::other)?
                         .map(u64::try_from)
                         .transpose()
@@ -1544,10 +1648,9 @@ impl SetDisks {
                 }
             } else if actual_size >= 0 {
                 let observed_size = u64::try_from(actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
-                let transformed = contains_key_str(&user_defined, SUFFIX_COMPRESSION)
-                    || parts_metadatas
-                        .first()
-                        .is_some_and(|metadata| should_persist_encryption_original_size(&metadata.metadata));
+                let persisted_metadata = &parts_metadatas[response_metadata_slot].metadata;
+                let transformed = contains_key_str(persisted_metadata, SUFFIX_COMPRESSION)
+                    || should_persist_encryption_original_size(persisted_metadata);
                 let server_observed_size = if transformed {
                     observed_size
                 } else {
@@ -1574,7 +1677,12 @@ impl SetDisks {
                         .map_err(|_| Error::PartMissingOrCorrupt)?
                         .max(u64::try_from(w_size).map_err(|_| Error::PartMissingOrCorrupt)?),
                 };
-                (reservation::replaced_logical_size(self, bucket, object, opts).await?, new_size)
+                let old_size = if opts.data_movement {
+                    new_size
+                } else {
+                    reservation::replaced_logical_size(self, bucket, object, opts).await?
+                };
+                (old_size, new_size)
             } else {
                 (0, 0)
             };
@@ -1674,6 +1782,7 @@ impl SetDisks {
             #[cfg(any(test, feature = "test-util"))]
             pause_put_object_commit(bucket, object, PutObjectCommitPause::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())
                 || opts
                     .namespace_lock_fence
@@ -1736,7 +1845,7 @@ impl SetDisks {
                     Some(self.pool_index),
                     Some(self.set_index),
                 );
-                request.object_version_id = fi.version_id.map(|version_id| version_id.to_string());
+                request.object_version_id = committed_version_id.map(|version_id| version_id.to_string());
                 tokio::spawn(async move {
                     let _ = rustfs_common::heal_channel::send_heal_request(request).await;
                 });
@@ -1771,7 +1880,7 @@ impl SetDisks {
 
             let mut cleanup_stage_ms: Option<u64> = None;
             if let Some(old_dir) = op_old_dir {
-                let committed_dir = fi.data_dir.unwrap_or_default().to_string();
+                let committed_dir = committed_data_dir.unwrap_or_default().to_string();
                 let cleanup_stage_start = Instant::now();
                 // backlog#898: reclaiming the dereferenced old data dir is
                 // best-effort and returns a receipt (never `Err`). A failed GC
@@ -1808,16 +1917,10 @@ impl SetDisks {
                 }
             }
 
-            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;
-                }
-            }
+            let committed_metadata_slot = committed_response_metadata_slot(&online_disks, response_metadata_slot);
+            let mut fi = std::mem::take(&mut parts_metadatas[committed_metadata_slot]);
 
-            if fi.is_compressed() {
+            if is_compressed {
                 record_compression_total_memory(actual_size as u64, w_size as u64).await;
             }
             self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks);
@@ -5906,6 +6009,264 @@ mod replication_quota_safety_tests {
     }
 }
 
+#[cfg(test)]
+mod inline_put_commit_path_tests {
+    use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;
+    use super::*;
+    use crate::disk::{DiskAPI as _, ReadOptions};
+    use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
+    use tokio::io::AsyncReadExt;
+
+    async fn make_bucket(disks: &[DiskStore], bucket: &str) {
+        for disk in disks {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+    }
+
+    #[tokio::test]
+    async fn inline_put_direct_commit_round_trips_verified_bitrot_shards() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "inline-direct-commit";
+        let object = "object.bin";
+        let payload: Vec<u8> = (0..16 * 1024).map(|index| (index % 251) as u8).collect();
+        make_bucket(&disk_stores, bucket).await;
+
+        let mut reader = PutObjReader::from_vec(payload.clone());
+        set_disks
+            .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("inline PUT should commit");
+
+        let read_data = ReadOptions {
+            read_data: true,
+            ..Default::default()
+        };
+        for (disk_index, disk) in disk_stores.iter().enumerate() {
+            let file_info = disk
+                .read_version("", bucket, object, "", &read_data)
+                .await
+                .unwrap_or_else(|err| panic!("disk {disk_index} should persist inline metadata: {err}"));
+            assert!(file_info.inline_data(), "disk {disk_index} should mark the shard inline");
+            let inline_data = file_info
+                .data
+                .as_ref()
+                .unwrap_or_else(|| panic!("disk {disk_index} should persist inline bitrot bytes"));
+            let erasure = erasure_from_file_info(&file_info, false).expect("persisted erasure layout should be valid");
+            let logical_shard_size =
+                usize::try_from(erasure.shard_file_size(payload.len() as i64)).expect("logical shard size should fit usize");
+            coding::bitrot_verify(
+                Cursor::new(inline_data.clone()),
+                inline_data.len(),
+                logical_shard_size,
+                HashAlgorithm::HighwayHash256S,
+                erasure.shard_size(),
+            )
+            .await
+            .unwrap_or_else(|err| panic!("disk {disk_index} inline shard should pass bitrot verification: {err}"));
+        }
+
+        let mut object_reader = set_disks
+            .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("committed inline object should be readable");
+        let mut restored = Vec::new();
+        object_reader
+            .stream
+            .read_to_end(&mut restored)
+            .await
+            .expect("inline object should stream");
+        assert_eq!(restored, payload);
+    }
+
+    #[tokio::test]
+    async fn inline_put_direct_commit_accepts_exact_quorum_and_rejects_quorum_minus_one() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "inline-direct-quorum";
+        let exact_quorum_object = "exact-quorum.bin";
+        let below_quorum_object = "below-quorum.bin";
+        make_bucket(&disk_stores, bucket).await;
+        {
+            let mut disks = set_disks.disks.write().await;
+            disks[3] = None;
+        }
+
+        let mut reader = PutObjReader::from_vec(vec![0x5a; 4 * 1024]);
+        set_disks
+            .put_object(bucket, exact_quorum_object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("three online disks should satisfy the four-disk write quorum");
+        for (disk_index, disk) in disk_stores.iter().enumerate() {
+            let persisted = disk
+                .read_version("", bucket, exact_quorum_object, "", &ReadOptions::default())
+                .await;
+            assert_eq!(
+                persisted.is_ok(),
+                disk_index < 3,
+                "exact-quorum commit should publish only on the three online disks"
+            );
+        }
+
+        set_disks.disks.write().await[2] = None;
+        let mut reader = PutObjReader::from_vec(vec![0xa5; 4 * 1024]);
+        set_disks
+            .put_object(bucket, below_quorum_object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect_err("two online disks are one below the four-disk write quorum");
+
+        for (disk_index, disk) in disk_stores.iter().enumerate() {
+            assert!(
+                disk.read_version("", bucket, below_quorum_object, "", &ReadOptions::default())
+                    .await
+                    .is_err(),
+                "disk {disk_index} must not expose an object after pre-commit quorum failure"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn inline_put_direct_commit_handles_post_encode_rename_failures() {
+        use crate::disk::health_state::RuntimeDriveHealthState;
+
+        let payload = vec![0x5a; 4 * 1024];
+
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "inline-direct-post-encode-quorum";
+        let object = "exact-quorum.bin";
+        make_bucket(&disk_stores, bucket).await;
+        let barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::AfterNamespace);
+        let put = {
+            let set_disks = Arc::clone(&set_disks);
+            let payload = payload.clone();
+            tokio::spawn(async move {
+                let mut reader = PutObjReader::from_vec(payload);
+                set_disks
+                    .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+                    .await
+            })
+        };
+        barrier.wait_until_paused().await;
+        disk_stores[3].force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+        barrier.release();
+        put.await
+            .expect("exact-quorum PUT task should complete")
+            .expect("one post-encode rename failure should preserve write quorum");
+        disk_stores[3].force_runtime_state_for_test(RuntimeDriveHealthState::Online);
+        for (disk_index, disk) in disk_stores.iter().enumerate() {
+            let persisted = disk.read_version("", bucket, object, "", &ReadOptions::default()).await;
+            assert_eq!(
+                persisted.is_ok(),
+                disk_index < 3,
+                "only disks that completed rename_data may publish the exact-quorum object"
+            );
+        }
+
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "inline-direct-post-encode-rollback";
+        let object = "rollback.bin";
+        let old_payload = vec![0x31; 4 * 1024];
+        make_bucket(&disk_stores, bucket).await;
+        let mut old_reader = PutObjReader::from_vec(old_payload.clone());
+        set_disks
+            .put_object(bucket, object, &mut old_reader, &ObjectOptions::default())
+            .await
+            .expect("old inline object should commit");
+        let read_data = ReadOptions {
+            read_data: true,
+            ..Default::default()
+        };
+        let mut old_disk_data = Vec::with_capacity(disk_stores.len());
+        for disk in &disk_stores {
+            old_disk_data.push(
+                disk.read_version("", bucket, object, "", &read_data)
+                    .await
+                    .expect("old inline shard should be readable before overwrite")
+                    .data,
+            );
+        }
+
+        let barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::AfterNamespace);
+        let put = {
+            let set_disks = Arc::clone(&set_disks);
+            let payload = payload.clone();
+            tokio::spawn(async move {
+                let mut reader = PutObjReader::from_vec(payload);
+                set_disks
+                    .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+                    .await
+            })
+        };
+        barrier.wait_until_paused().await;
+        for disk in &disk_stores[2..] {
+            disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+        }
+        barrier.release();
+        put.await
+            .expect("quorum-minus-one PUT task should complete")
+            .expect_err("two post-encode rename failures must fail write quorum");
+        for disk in &disk_stores[2..] {
+            disk.force_runtime_state_for_test(RuntimeDriveHealthState::Online);
+        }
+
+        for (disk_index, disk) in disk_stores.iter().enumerate() {
+            let restored = disk
+                .read_version("", bucket, object, "", &read_data)
+                .await
+                .unwrap_or_else(|err| panic!("disk {disk_index} should retain the old inline object: {err}"));
+            assert_eq!(restored.data, old_disk_data[disk_index]);
+        }
+        let mut object_reader = set_disks
+            .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("old object should remain readable after quorum rollback");
+        let mut restored = Vec::new();
+        object_reader
+            .stream
+            .read_to_end(&mut restored)
+            .await
+            .expect("old object should stream after quorum rollback");
+        assert_eq!(restored, old_payload);
+    }
+
+    #[tokio::test]
+    async fn zero_length_put_keeps_existing_pipeline_layout_and_round_trips() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "zero-length-put";
+        let object = "empty.bin";
+        make_bucket(&disk_stores, bucket).await;
+
+        let mut reader = PutObjReader::from_vec(Vec::new());
+        set_disks
+            .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("zero-length PUT should commit through the existing pipeline");
+
+        let read_data = ReadOptions {
+            read_data: true,
+            ..Default::default()
+        };
+        for (disk_index, disk) in disk_stores.iter().enumerate() {
+            let file_info = disk
+                .read_version("", bucket, object, "", &read_data)
+                .await
+                .unwrap_or_else(|err| panic!("disk {disk_index} should persist empty-object metadata: {err}"));
+            assert_eq!(file_info.size, 0);
+            assert_eq!(file_info.data.as_deref(), Some(&[][..]));
+        }
+
+        let mut object_reader = set_disks
+            .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("empty object should be readable");
+        let mut restored = Vec::new();
+        object_reader
+            .stream
+            .read_to_end(&mut restored)
+            .await
+            .expect("empty object should stream");
+        assert!(restored.is_empty());
+    }
+}
+
 #[cfg(test)]
 mod get_object_downstream_close_accounting_tests {
     use super::hermetic_set_disks_support::hermetic_set_disks;

crates/io-metrics/src/lib.rs

@@ -58,7 +58,7 @@ use std::sync::{
 /// When `false`, `record_put_object_path` and `record_put_object_stage_duration`
 /// become no-ops, and callers can skip the `Instant::now()` syscalls entirely.
 ///
-/// Set to `true` during startup when OTEL metric export is enabled.
+/// Enabled only through an explicit runtime opt-in.
 static PUT_STAGE_METRICS_ENABLED: AtomicBool = AtomicBool::new(false);
 static GET_STAGE_METRICS_ENABLED: AtomicBool = AtomicBool::new(false);
 
@@ -78,7 +78,7 @@ static METRICS_ENABLED: AtomicBool = AtomicBool::new(false);
 
 /// Enable or disable detailed per-stage PUT metrics.
 ///
-/// Called once during startup, typically gated by `rustfs_obs::observability_metric_enabled()`.
+/// Called once during startup after applying the detailed PUT attribution opt-in.
 pub fn set_put_stage_metrics_enabled(enabled: bool) {
     PUT_STAGE_METRICS_ENABLED.store(enabled, Ordering::Relaxed);
 }
@@ -103,6 +103,12 @@ pub fn put_stage_metrics_enabled() -> bool {
     PUT_STAGE_METRICS_ENABLED.load(Ordering::Relaxed)
 }
 
+/// Start a PUT-stage timer only when detailed PUT attribution is enabled.
+#[inline(always)]
+pub fn put_stage_timer() -> Option<std::time::Instant> {
+    put_stage_metrics_enabled().then(std::time::Instant::now)
+}
+
 #[inline(always)]
 pub fn get_stage_metrics_enabled() -> bool {
     GET_STAGE_METRICS_ENABLED.load(Ordering::Relaxed)
@@ -434,7 +440,7 @@ pub fn record_get_object_request_result(status: &str, duration_secs: f64) {
 /// Record PutObject request start.
 #[inline(always)]
 pub fn record_put_object_request_start(concurrent_requests: usize) {
-    if !put_stage_metrics_enabled() {
+    if !metrics_enabled() {
         return;
     }
     counter!("rustfs_io_put_object_requests_total").increment(1);
@@ -444,7 +450,7 @@ pub fn record_put_object_request_start(concurrent_requests: usize) {
 /// Record PutObject request result.
 #[inline(always)]
 pub fn record_put_object_request_result(status: &str, duration_secs: f64) {
-    if !put_stage_metrics_enabled() {
+    if !metrics_enabled() {
         return;
     }
     counter!("rustfs_io_put_object_request_results_total", "status" => status.to_string()).increment(1);
@@ -1905,7 +1911,7 @@ pub fn record_get_object(duration_ms: f64, size_bytes: i64) {
 /// * `zero_copy_eligible` - Whether the request was eligible for a zero-copy path
 #[inline(always)]
 pub fn record_put_object(duration_ms: f64, size_bytes: i64, zero_copy_eligible: bool) {
-    if !put_stage_metrics_enabled() {
+    if !metrics_enabled() {
         return;
     }
     counter!("rustfs_s3_put_object_total").increment(1);
@@ -2004,6 +2010,13 @@ pub fn record_put_object_stage_duration(stage: &'static str, duration_ms: f64) {
     histogram!("rustfs_s3_put_object_stage_duration_ms", "stage" => stage).record(duration_ms);
 }
 
+#[inline(always)]
+pub fn record_put_object_stage_duration_from(stage: &'static str, started_at: Option<std::time::Instant>) {
+    if let Some(started_at) = started_at {
+        record_put_object_stage_duration(stage, started_at.elapsed().as_secs_f64() * 1000.0);
+    }
+}
+
 /// Record generic internal operation stage duration (non-PUT paths).
 /// Use this for metacache walks, listing, lifecycle, and other background
 /// operations that are NOT part of the PUT object hot path.
@@ -2819,6 +2832,56 @@ mod tests {
         assert!(!put_stage_metrics_enabled());
     }
 
+    #[test]
+    fn put_stage_gate_does_not_disable_basic_put_metrics() {
+        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_metrics_enabled(true);
+            set_put_stage_metrics_enabled(false);
+            record_put_object_request_start(1);
+            record_put_object_request_result("ok", 0.001);
+            record_put_object(1.0, 1024, false);
+            record_put_object_stage_duration("disabled_stage", 0.5);
+
+            set_put_stage_metrics_enabled(true);
+            record_put_object_stage_duration("enabled_stage", 0.5);
+
+            set_put_stage_metrics_enabled(false);
+            set_metrics_enabled(false);
+        });
+
+        let metrics = snapshotter.snapshot().into_vec();
+        assert!(metrics.iter().any(|(composite, _, _, _)| {
+            composite.kind() == MetricKind::Counter && composite.key().name() == "rustfs_s3_put_object_total"
+        }));
+        assert!(metrics.iter().any(|(composite, _, _, _)| {
+            composite.kind() == MetricKind::Counter && composite.key().name() == "rustfs_io_put_object_requests_total"
+        }));
+
+        let stages = metrics
+            .iter()
+            .filter(|(composite, _, _, _)| {
+                composite.kind() == MetricKind::Histogram && composite.key().name() == "rustfs_s3_put_object_stage_duration_ms"
+            })
+            .flat_map(|(composite, _, _, _)| composite.key().labels().map(|label| label.value().to_string()))
+            .collect::<Vec<_>>();
+        assert_eq!(stages, ["enabled_stage"]);
+    }
+
+    #[test]
+    fn test_put_stage_timer_follows_metrics_switch() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        set_put_stage_metrics_enabled(false);
+        assert!(put_stage_timer().is_none());
+
+        set_put_stage_metrics_enabled(true);
+        assert!(put_stage_timer().is_some());
+        set_put_stage_metrics_enabled(false);
+    }
+
     #[test]
     fn test_record_get_object_path_and_stage() {
         let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());

crates/kms/src/backends/vault.rs

@@ -208,6 +208,7 @@ impl<'de> Deserialize<'de> for VaultKeyData {
             RotatedAt,
             EncryptedKeyMaterial,
             BaselineVersion,
+            WrapBudgetReserved,
             Unknown(BoundedUnknownFieldName),
         }
 
@@ -242,6 +243,7 @@ impl<'de> Deserialize<'de> for VaultKeyData {
                             "rotated_at" => Field::RotatedAt,
                             "encrypted_key_material" => Field::EncryptedKeyMaterial,
                             "baseline_version" => Field::BaselineVersion,
+                            "wrap_budget_reserved" => Field::WrapBudgetReserved,
                             _ => Field::Unknown(BoundedUnknownFieldName::new(value)),
                         })
                     }
@@ -285,6 +287,7 @@ impl<'de> Deserialize<'de> for VaultKeyData {
                 let mut rotated_at = None;
                 let mut encrypted_key_material = None;
                 let mut baseline_version = None;
+                let mut wrap_budget_reserved = None;
                 let mut unknown_fields = UnknownFieldSummary::default();
 
                 while let Some(field) = map.next_key()? {
@@ -301,6 +304,7 @@ impl<'de> Deserialize<'de> for VaultKeyData {
                         Field::RotatedAt => read_field!(rotated_at, "rotated_at"),
                         Field::EncryptedKeyMaterial => read_field!(encrypted_key_material, "encrypted_key_material"),
                         Field::BaselineVersion => read_field!(baseline_version, "baseline_version"),
+                        Field::WrapBudgetReserved => read_field!(wrap_budget_reserved, "wrap_budget_reserved"),
                         Field::Unknown(field) => {
                             let _: IgnoredAny = map.next_value()?;
                             unknown_fields.observe(field);
@@ -322,6 +326,10 @@ impl<'de> Deserialize<'de> for VaultKeyData {
                     encrypted_key_material: encrypted_key_material
                         .ok_or_else(|| de::Error::missing_field("encrypted_key_material"))?,
                     baseline_version: baseline_version.unwrap_or(None),
+                    // Absent on records written before wrap accounting existed, and
+                    // on records an older build rewrote; zero restarts the
+                    // reservation rather than blocking a wrap.
+                    wrap_budget_reserved: wrap_budget_reserved.unwrap_or(0),
                 };
                 unknown_fields.record_for_vault_kv2_key();
                 Ok(key_data)
@@ -341,6 +349,7 @@ impl<'de> Deserialize<'de> for VaultKeyData {
             "rotated_at",
             "encrypted_key_material",
             "baseline_version",
+            "wrap_budget_reserved",
         ];
         deserializer.deserialize_struct("VaultKeyData", FIELDS, VaultKeyDataVisitor)
     }
@@ -3014,6 +3023,54 @@ mod tests {
         assert_eq!(legacy.version, 1);
     }
 
+    /// Every declared `VaultKeyData` field must survive a serialize/deserialize
+    /// round trip through the hand-written `Deserialize`.
+    ///
+    /// The hand-written impl lists its fields three times (the `Field` enum, the
+    /// match arms, the struct literal), so a field added to the struct alone
+    /// compiles on its own branch and only breaks once both branches merge —
+    /// which is exactly how `wrap_budget_reserved` briefly broke the build.
+    /// Asserting against the serialized key set makes the deserializer's
+    /// coverage a test failure rather than a merge-order accident.
+    #[test]
+    fn vault_key_data_deserializer_covers_every_serialized_field() {
+        let mut key_data = healthy_key_data();
+        key_data.wrap_budget_reserved = 7_000_000;
+        key_data.baseline_version = Some(2);
+        key_data.rotated_at = Some(Zoned::now());
+        key_data.deletion_date = Some(Zoned::now());
+        key_data.description = Some("described".to_string());
+
+        let value = serde_json::to_value(&key_data).expect("serialize key data");
+        let serialized_fields: Vec<String> = value
+            .as_object()
+            .expect("key data serializes to an object")
+            .keys()
+            .cloned()
+            .collect();
+
+        // Every serialized field must be a known field: an unknown one would be
+        // counted by the unknown-field observer instead of being read back.
+        let recorder = metrics_util::debugging::DebuggingRecorder::new();
+        let restored: VaultKeyData =
+            metrics::with_local_recorder(&recorder, || serde_json::from_value(value).expect("round trip"));
+        assert_eq!(
+            crate::test_support::unknown_field_metric(&recorder, "vault-kv2-key"),
+            0,
+            "a serialized field was not recognized by the deserializer; fields: {serialized_fields:?}"
+        );
+
+        // And every value must survive, not just parse.
+        assert_eq!(restored.wrap_budget_reserved, key_data.wrap_budget_reserved);
+        assert_eq!(restored.baseline_version, key_data.baseline_version);
+        assert_eq!(restored.version, key_data.version);
+        assert_eq!(restored.status, key_data.status);
+        assert_eq!(restored.description, key_data.description);
+        assert_eq!(restored.encrypted_key_material, key_data.encrypted_key_material);
+        assert!(restored.rotated_at.is_some());
+        assert!(restored.deletion_date.is_some());
+    }
+
     #[test]
     fn vault_key_data_unknown_fields_remain_readable_and_are_observed() {
         // A record written by a newer build carries fields this build does not

crates/protocols/src/swift/handler.rs

@@ -748,9 +748,10 @@ async fn handle_authenticated_request(
                                 }
 
                                 for (key, value) in info.user_defined.iter() {
-                                    if key != "content-type" {
-                                        let header_name = format!("x-object-meta-{}", key);
-                                        response = response.header(header_name, value.as_str());
+                                    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());
                                     }
                                 }
 
@@ -817,9 +818,10 @@ 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 header_name = format!("x-object-meta-{}", key);
-                            response = response.header(header_name, value.as_str());
+                        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());
                         }
                     }
 
@@ -856,9 +858,10 @@ 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 header_name = format!("x-object-meta-{}", key);
-                            response = response.header(header_name, value.as_str());
+                        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());
                         }
                     }
 
@@ -1168,9 +1171,10 @@ async fn handle_object_get(
                 }
 
                 for (key, value) in info.user_defined.iter() {
-                    if key != "content-type" {
-                        let header_name = format!("x-object-meta-{}", key);
-                        response = response.header(header_name, value.as_str());
+                    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());
                     }
                 }
 
@@ -1237,9 +1241,10 @@ 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 header_name = format!("x-object-meta-{}", key);
-            response = response.header(header_name, 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());
         }
     }
 
@@ -1293,9 +1298,10 @@ 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 header_name = format!("x-object-meta-{}", key);
-            response = response.header(header_name, 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());
         }
     }
 

crates/protocols/src/swift/object.rs

@@ -84,6 +84,24 @@ fn stored_swift_user_metadata_key(key: &str) -> 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;
@@ -1142,6 +1160,19 @@ mod tests {
         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("");