fix(ecstore): settle checkpoint capacity retries

crates/config/src/constants/object.rs

@@ -197,6 +197,27 @@ pub const DEFAULT_POOL_META_V3_FLEET_CONFIRMED: bool = false;
 const _: () = assert!(!DEFAULT_POOL_META_V3_WRITE);
 const _: () = assert!(!DEFAULT_POOL_META_V3_FLEET_CONFIRMED);
 
+/// Maximum unpacked size accepted for one Snowball archive member.
+///
+/// The value is expressed in bytes. Invalid values use the default, while
+/// valid values are clamped to [`MAX_SNOWBALL_ENTRY_BYTES`].
+pub const ENV_SNOWBALL_MAX_ENTRY_BYTES: &str = "RUSTFS_SNOWBALL_MAX_ENTRY_BYTES";
+pub const DEFAULT_SNOWBALL_MAX_ENTRY_BYTES: u64 = 1024 * 1024 * 1024;
+pub const MAX_SNOWBALL_ENTRY_BYTES: u64 = 1024 * DEFAULT_SNOWBALL_MAX_ENTRY_BYTES;
+
+/// Maximum cumulative unpacked object bytes accepted from one Snowball
+/// archive request.
+///
+/// This does not include tar headers or bounded PAX metadata. The value is
+/// expressed in bytes and is clamped to
+/// [`MAX_SNOWBALL_UNPACKED_BYTES`].
+pub const ENV_SNOWBALL_MAX_UNPACKED_BYTES: &str = "RUSTFS_SNOWBALL_MAX_UNPACKED_BYTES";
+pub const DEFAULT_SNOWBALL_MAX_UNPACKED_BYTES: u64 = 10 * 1024 * 1024 * 1024;
+pub const MAX_SNOWBALL_UNPACKED_BYTES: u64 = 10 * 1024 * DEFAULT_SNOWBALL_MAX_ENTRY_BYTES;
+
+const _: () = assert!(DEFAULT_SNOWBALL_MAX_ENTRY_BYTES <= MAX_SNOWBALL_ENTRY_BYTES);
+const _: () = assert!(DEFAULT_SNOWBALL_MAX_UNPACKED_BYTES <= MAX_SNOWBALL_UNPACKED_BYTES);
+
 // =============================================================================
 // Concurrent Request Fix - Timeout and Backpressure Configuration
 // =============================================================================
@@ -820,4 +841,10 @@ mod remote_version_state_tests {
         assert_eq!(super::ENV_POOL_META_V3_WRITE, "RUSTFS_POOL_META_V3_WRITE");
         assert_eq!(super::ENV_POOL_META_V3_FLEET_CONFIRMED, "RUSTFS_POOL_META_V3_FLEET_CONFIRMED");
     }
+
+    #[test]
+    fn snowball_limit_environment_names_are_stable() {
+        assert_eq!(super::ENV_SNOWBALL_MAX_ENTRY_BYTES, "RUSTFS_SNOWBALL_MAX_ENTRY_BYTES");
+        assert_eq!(super::ENV_SNOWBALL_MAX_UNPACKED_BYTES, "RUSTFS_SNOWBALL_MAX_UNPACKED_BYTES");
+    }
 }

crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs

@@ -1468,7 +1468,7 @@ async fn save_decommission_manifest_checkpoint_if_match(
         }
         let write_data = next_data.to_vec();
         let write = api
-            .run_decommission_capacity_temporary_mutation_with_capacity_lease(
+            .run_decommission_capacity_non_growing_replacement_with_capacity_lease(
                 target.target_pool_index,
                 Some(target.capacity_owner),
                 Some(next_data.len()),

crates/ecstore/src/data_movement/mod.rs

@@ -234,6 +234,7 @@ async fn pause_data_movement_multipart_before_abort(bucket: &str, object: &str)
 }
 
 fn data_movement_abort_opts(
+    object_info: &ObjectInfo,
     src_pool_idx: usize,
     expected_bucket_incarnation_id: Option<uuid::Uuid>,
     lock_lost_signal: Option<&Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
@@ -242,6 +243,9 @@ fn data_movement_abort_opts(
     let mut opts = ObjectOptions {
         data_movement: true,
         src_pool_idx,
+        versioned: object_info.version_id.is_some(),
+        version_id: object_info.version_id.map(|version_id| version_id.to_string()),
+        mod_time: object_info.mod_time,
         expected_bucket_incarnation_id,
         ..Default::default()
     };
@@ -265,16 +269,21 @@ fn insert_data_movement_checksum(user_defined: &mut HashMap<String, String>, obj
     }
 }
 
-fn data_movement_upload_identity(object_info: &ObjectInfo) -> String {
-    let version_id = object_info
-        .version_id
-        .map_or_else(|| "none".to_string(), |version_id| version_id.to_string());
-    let mod_time = object_info
-        .mod_time
-        .map_or_else(|| "none".to_string(), |mod_time| mod_time.unix_timestamp_nanos().to_string());
+fn data_movement_upload_identity_parts(version_id: Option<&str>, mod_time: Option<time::OffsetDateTime>) -> String {
+    let version_id = version_id.unwrap_or("none");
+    let mod_time = mod_time.map_or_else(|| "none".to_string(), |mod_time| mod_time.unix_timestamp_nanos().to_string());
     format!("v1:{version_id}:{mod_time}")
 }
 
+fn data_movement_upload_identity(object_info: &ObjectInfo) -> String {
+    let version_id = object_info.version_id.map(|version_id| version_id.to_string());
+    data_movement_upload_identity_parts(version_id.as_deref(), object_info.mod_time)
+}
+
+pub(crate) fn data_movement_upload_identity_from_options(opts: &ObjectOptions) -> String {
+    data_movement_upload_identity_parts(opts.version_id.as_deref(), opts.mod_time)
+}
+
 fn data_movement_new_multipart_opts(object_info: &ObjectInfo, src_pool_idx: usize) -> ObjectOptions {
     let mut user_defined = data_movement_user_defined(object_info);
     let upload_identity = data_movement_upload_identity(object_info);
@@ -486,7 +495,7 @@ pub(crate) fn data_movement_target_precondition() -> HTTPPreconditions {
     }
 }
 
-fn is_owned_data_movement_target(target: &ObjectInfo) -> bool {
+pub(crate) fn is_owned_data_movement_target(target: &ObjectInfo) -> bool {
     let rustfs_marker = rustfs_utils::http::internal_key_rustfs(SUFFIX_DATA_MOVED);
     let minio_marker = format!("{}{SUFFIX_DATA_MOVED}", rustfs_utils::http::MINIO_INTERNAL_PREFIX);
     if rustfs_utils::http::get_consistent_str(&target.user_defined, SUFFIX_DATA_MOVED) != Some("true")
@@ -560,9 +569,12 @@ fn resolve_data_movement_abort_result(
     primary_err: Error,
     abort_err: Error,
 ) -> Error {
-    Error::other(format!(
-        "{op_label}: abort_multipart_upload failed for {bucket}/{object} upload {upload_id} after error {primary_err}: {abort_err}"
-    ))
+    data_movement_context_error(
+        format!(
+            "{op_label}: abort_multipart_upload failed for {bucket}/{object} upload {upload_id} after error {primary_err}: {abort_err}"
+        ),
+        abort_err,
+    )
 }
 
 /// A data-movement stage failure that keeps the error it wrapped.
@@ -590,17 +602,23 @@ impl std::error::Error for DataMovementStageError {
     }
 }
 
-fn data_movement_stage_error<E>(op_label: &str, stage: &str, bucket: &str, object: &str, err: E) -> Error
+pub(crate) fn data_movement_context_error<E>(rendered: String, err: E) -> Error
 where
     E: std::error::Error + Send + Sync + 'static,
 {
-    let rendered = format!("{op_label}: {stage} failed for {bucket}/{object}: {err}");
     Error::other(DataMovementStageError {
         rendered,
         source: Box::new(err),
     })
 }
 
+fn data_movement_stage_error<E>(op_label: &str, stage: &str, bucket: &str, object: &str, err: E) -> Error
+where
+    E: std::error::Error + Send + Sync + 'static,
+{
+    data_movement_context_error(format!("{op_label}: {stage} failed for {bucket}/{object}: {err}"), err)
+}
+
 #[cfg(test)]
 pub(crate) fn data_movement_stage_error_for_test(op_label: &str, stage: &str, bucket: &str, object: &str, err: Error) -> Error {
     data_movement_stage_error(op_label, stage, bucket, object, err)
@@ -1662,8 +1680,13 @@ async fn migrate_object_inner(
                 );
                 return Ok(());
             }
-            let mut cleanup_opts =
-                data_movement_abort_opts(pool_idx, source_bucket_incarnation_id, lock_lost_signal.as_ref(), capacity_owner);
+            let mut cleanup_opts = data_movement_abort_opts(
+                &object_info,
+                pool_idx,
+                source_bucket_incarnation_id,
+                lock_lost_signal.as_ref(),
+                capacity_owner,
+            );
             if let Some(anchor) = mutation_fence.as_ref() {
                 anchor.guard().add_namespace_lock_fence(&mut cleanup_opts);
             }
@@ -1865,8 +1888,13 @@ async fn migrate_object_inner(
         .await;
 
         if multipart_result.is_ok() && should_abort_multipart_upload(&abort_multipart_flag) {
-            let mut abort_opts =
-                data_movement_abort_opts(pool_idx, expected_bucket_incarnation_id, lock_lost_signal.as_ref(), capacity_owner);
+            let mut abort_opts = data_movement_abort_opts(
+                &object_info,
+                pool_idx,
+                expected_bucket_incarnation_id,
+                lock_lost_signal.as_ref(),
+                capacity_owner,
+            );
             if let Some(anchor) = mutation_fence.as_ref() {
                 anchor.guard().add_namespace_lock_fence(&mut abort_opts);
             }
@@ -1943,8 +1971,13 @@ async fn migrate_object_inner(
             if should_abort_multipart_upload(&abort_multipart_flag) {
                 #[cfg(all(test, feature = "test-util"))]
                 pause_data_movement_multipart_before_abort(&bucket, &object_info.name).await;
-                let mut abort_opts =
-                    data_movement_abort_opts(pool_idx, expected_bucket_incarnation_id, lock_lost_signal.as_ref(), capacity_owner);
+                let mut abort_opts = data_movement_abort_opts(
+                    &object_info,
+                    pool_idx,
+                    expected_bucket_incarnation_id,
+                    lock_lost_signal.as_ref(),
+                    capacity_owner,
+                );
                 if let Some(anchor) = mutation_fence.as_ref() {
                     anchor.guard().add_namespace_lock_fence(&mut abort_opts);
                 }
@@ -2325,6 +2358,7 @@ mod tests {
         assert!(message.contains("bucket-a/object-a"));
         assert!(message.contains("upload upload-1"));
         assert!(message.contains(Error::SlowDown.to_string().as_str()));
+        assert!(matches!(data_movement_stage_source(&err), Some(Error::OperationCanceled)));
     }
 
     #[test]

crates/ecstore/src/data_usage/mod.rs

@@ -2964,7 +2964,6 @@ mod tests {
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: TokioMutex::new(()),
             pool_meta_save_gate: TokioMutex::default(),
-            decommission_capacity_entry_gate: TokioMutex::default(),
             ctx,
             bucket_fence_registry: Arc::default(),
         })

crates/ecstore/src/services/notification_sys.rs

@@ -57,18 +57,24 @@ const REMOTE_VERSION_STATE_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
 const REMOTE_VERSION_STATE_PROOF_TTL: Duration = Duration::from_secs(30);
 const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 2;
 const TIER_DELETE_JOURNAL_POLICY_SUPPORTED_VERSION: u32 = 3;
+const DECOMMISSION_TARGET_FENCE_POLICY_SUPPORTED_VERSION: u32 = 4;
 type CrossPoolFencePolicyResult = Result<BTreeMap<String, Uuid>>;
 
 fn cross_pool_fence_policy_results(
     peer_epochs: BTreeMap<String, Uuid>,
     minimum_version: u32,
-) -> (CrossPoolFencePolicyResult, CrossPoolFencePolicyResult) {
+) -> (CrossPoolFencePolicyResult, CrossPoolFencePolicyResult, CrossPoolFencePolicyResult) {
     let journal_result = if minimum_version >= TIER_DELETE_JOURNAL_POLICY_SUPPORTED_VERSION {
         Ok(peer_epochs.clone())
     } else {
         Err(Error::other("tier delete journal v6 policy capability version is unsupported"))
     };
-    (Ok(peer_epochs), journal_result)
+    let decommission_target_fence_result = if minimum_version >= DECOMMISSION_TARGET_FENCE_POLICY_SUPPORTED_VERSION {
+        Ok(peer_epochs.clone())
+    } else {
+        Err(Error::other("decommission target fence policy capability version is unsupported"))
+    };
+    (Ok(peer_epochs), journal_result, decommission_target_fence_result)
 }
 
 #[derive(Clone, Debug)]
@@ -231,6 +237,9 @@ pub(crate) struct RemoteVersionStateFleetProofToken(FleetCapabilityProofToken);
 #[derive(Clone, PartialEq, Eq)]
 pub struct CrossPoolFenceFleetProofToken(FleetCapabilityProofToken);
 
+#[derive(Clone, PartialEq, Eq)]
+pub(crate) struct DecommissionTargetFenceFleetProofToken(FleetCapabilityProofToken);
+
 /// A point-in-time proof that every current storage member implements the v6
 /// dispatch-manifest policy. It intentionally has no `Clone` implementation:
 /// one acquisition authorizes one manifest construction attempt.
@@ -242,6 +251,7 @@ pub(crate) struct TierDeleteJournalFleetProofToken {
 static REMOTE_VERSION_STATE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static CROSS_POOL_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static TIER_DELETE_JOURNAL_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
+static DECOMMISSION_TARGET_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock<String> = OnceLock::new();
 
 fn cross_pool_fence_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
@@ -256,6 +266,10 @@ fn tier_delete_journal_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCap
     TIER_DELETE_JOURNAL_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
 }
 
+fn decommission_target_fence_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
+    DECOMMISSION_TARGET_FENCE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
+}
+
 fn revoke_fleet_capability_proof_state(state: &mut FleetCapabilityProofState) {
     if let Some(proof) = state.proof.take() {
         proof.generation.revoke();
@@ -368,6 +382,18 @@ pub fn cross_pool_fence_fleet_proof_matches(proof: &CrossPoolFenceFleetProofToke
     fleet_capability_proof_matches(cross_pool_fence_fleet_proof_slot(), &proof.0)
 }
 
+pub(crate) fn acquire_decommission_target_fence_fleet_proof() -> Option<DecommissionTargetFenceFleetProofToken> {
+    let expected_topology = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get()?;
+    let state = decommission_target_fence_fleet_proof_slot()
+        .read()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    acquire_fleet_capability_proof_from(&state, expected_topology, Instant::now()).map(DecommissionTargetFenceFleetProofToken)
+}
+
+pub(crate) fn decommission_target_fence_fleet_proof_matches(proof: &DecommissionTargetFenceFleetProofToken) -> bool {
+    fleet_capability_proof_matches(decommission_target_fence_fleet_proof_slot(), &proof.0)
+}
+
 pub(crate) fn acquire_tier_delete_journal_fleet_proof() -> Option<TierDeleteJournalFleetProofToken> {
     let expected_topology = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get()?;
     let state = tier_delete_journal_fleet_proof_slot()
@@ -468,6 +494,19 @@ pub(crate) fn install_cross_pool_fence_fleet_proof_for_test() {
     journal_state.topology_conflict = false;
     journal_state.draining_generation = None;
     journal_state.proof = proof.as_ref().map(FleetCapabilityProof::with_fresh_generation);
+    drop(journal_state);
+    let mut decommission_state = decommission_target_fence_fleet_proof_slot()
+        .write()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    debug_assert!(
+        decommission_state
+            .proof
+            .as_ref()
+            .is_none_or(|current| current.generation.is_drained())
+    );
+    decommission_state.topology_conflict = false;
+    decommission_state.draining_generation = None;
+    decommission_state.proof = proof.as_ref().map(FleetCapabilityProof::with_fresh_generation);
 }
 
 #[cfg(test)]
@@ -476,6 +515,8 @@ pub(crate) struct CrossPoolFenceFleetProofGuard {
     previous_topology_conflict: bool,
     previous_journal_proof: Option<FleetCapabilityProof>,
     previous_journal_topology_conflict: bool,
+    previous_decommission_proof: Option<FleetCapabilityProof>,
+    previous_decommission_topology_conflict: bool,
 }
 
 #[cfg(test)]
@@ -502,6 +543,17 @@ impl Drop for CrossPoolFenceFleetProofGuard {
             .map(FleetCapabilityProof::with_fresh_generation);
         journal_state.draining_generation = None;
         journal_state.topology_conflict = self.previous_journal_topology_conflict;
+        drop(journal_state);
+        let mut decommission_state = decommission_target_fence_fleet_proof_slot()
+            .write()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        decommission_state.proof = self
+            .previous_decommission_proof
+            .take()
+            .as_ref()
+            .map(FleetCapabilityProof::with_fresh_generation);
+        decommission_state.draining_generation = None;
+        decommission_state.topology_conflict = self.previous_decommission_topology_conflict;
     }
 }
 
@@ -515,11 +567,16 @@ pub(crate) fn without_cross_pool_fence_fleet_proof_for_test() -> CrossPoolFenceF
     let mut journal_state = tier_delete_journal_fleet_proof_slot()
         .write()
         .unwrap_or_else(std::sync::PoisonError::into_inner);
+    let mut decommission_state = decommission_target_fence_fleet_proof_slot()
+        .write()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
     let guard = CrossPoolFenceFleetProofGuard {
         previous_proof: state.proof.clone(),
         previous_topology_conflict: state.topology_conflict,
         previous_journal_proof: journal_state.proof.clone(),
         previous_journal_topology_conflict: journal_state.topology_conflict,
+        previous_decommission_proof: decommission_state.proof.clone(),
+        previous_decommission_topology_conflict: decommission_state.topology_conflict,
     };
     if let Some(proof) = state.proof.take() {
         proof.generation.revoke();
@@ -535,6 +592,49 @@ pub(crate) fn without_cross_pool_fence_fleet_proof_for_test() -> CrossPoolFenceF
         }
     }
     journal_state.topology_conflict = true;
+    if let Some(proof) = decommission_state.proof.take() {
+        proof.generation.revoke();
+        if !proof.generation.is_drained() {
+            decommission_state.draining_generation = Some(proof.generation);
+        }
+    }
+    decommission_state.topology_conflict = true;
+    guard
+}
+
+#[cfg(test)]
+pub(crate) struct DecommissionTargetFenceFleetProofGuard {
+    previous_proof: Option<FleetCapabilityProof>,
+    previous_topology_conflict: bool,
+}
+
+#[cfg(test)]
+impl Drop for DecommissionTargetFenceFleetProofGuard {
+    fn drop(&mut self) {
+        let mut state = decommission_target_fence_fleet_proof_slot()
+            .write()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        state.proof = self
+            .previous_proof
+            .take()
+            .as_ref()
+            .map(FleetCapabilityProof::with_fresh_generation);
+        state.draining_generation = None;
+        state.topology_conflict = self.previous_topology_conflict;
+    }
+}
+
+#[cfg(test)]
+pub(crate) fn without_decommission_target_fence_fleet_proof_for_test() -> DecommissionTargetFenceFleetProofGuard {
+    let mut state = decommission_target_fence_fleet_proof_slot()
+        .write()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    let guard = DecommissionTargetFenceFleetProofGuard {
+        previous_proof: state.proof.clone(),
+        previous_topology_conflict: state.topology_conflict,
+    };
+    revoke_fleet_capability_proof_state(&mut state);
+    state.topology_conflict = true;
     guard
 }
 
@@ -573,6 +673,15 @@ pub fn rotate_cross_pool_fence_fleet_proof_for_test() -> bool {
     if journal_state.draining_generation.is_none() {
         journal_state.proof = Some(proof.with_fresh_generation());
     }
+    drop(journal_state);
+    let mut decommission_state = decommission_target_fence_fleet_proof_slot()
+        .write()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    decommission_state.topology_conflict = false;
+    revoke_fleet_capability_proof_state(&mut decommission_state);
+    if decommission_state.draining_generation.is_none() {
+        decommission_state.proof = Some(proof.with_fresh_generation());
+    }
     true
 }
 
@@ -652,6 +761,7 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 remote_version_state_fleet_proof_slot(),
                 cross_pool_fence_fleet_proof_slot(),
                 tier_delete_journal_fleet_proof_slot(),
+                decommission_target_fence_fleet_proof_slot(),
             ] {
                 mark_fleet_capability_topology_conflict(slot);
             }
@@ -684,11 +794,15 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 .unwrap_or_else(|_| Err(Error::other("cross-pool fence fleet capability probe timed out"))),
                 None => Err(Error::other("cross-pool fence fleet capability notification system is unavailable")),
             };
-            let (fence_result, journal_result) = match fence_probe {
+            let (fence_result, journal_result, decommission_target_fence_result) = match fence_probe {
                 Ok((peer_epochs, minimum_version)) => cross_pool_fence_policy_results(peer_epochs, minimum_version),
                 Err(err) => {
                     let message = err.to_string();
-                    (Err(Error::other(message.clone())), Err(Error::other(message)))
+                    (
+                        Err(Error::other(message.clone())),
+                        Err(Error::other(message.clone())),
+                        Err(Error::other(message)),
+                    )
                 }
             };
             let topology_conflict = remote_version_state_fleet_proof_slot()
@@ -699,6 +813,7 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 revoke_fleet_capability_proof(remote_version_state_fleet_proof_slot());
                 revoke_fleet_capability_proof(cross_pool_fence_fleet_proof_slot());
                 revoke_fleet_capability_proof(tier_delete_journal_fleet_proof_slot());
+                revoke_fleet_capability_proof(decommission_target_fence_fleet_proof_slot());
             } else if let Some(err) = publish_fleet_capability_probe_result(
                 remote_version_state_fleet_proof_slot(),
                 &topology_fingerprint,
@@ -743,6 +858,24 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                     "notification capability probe"
                 );
             }
+            if !topology_conflict
+                && let Some(err) = publish_fleet_capability_probe_result(
+                    decommission_target_fence_fleet_proof_slot(),
+                    &topology_fingerprint,
+                    decommission_target_fence_result,
+                    Instant::now(),
+                )
+            {
+                debug!(
+                    event = EVENT_NOTIFICATION_CAPABILITY_PROBE,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_NOTIFICATION,
+                    capability = "decommission_target_fence_v2",
+                    state = "failed_closed",
+                    error = %err,
+                    "notification capability probe"
+                );
+            }
             sleep(REMOTE_VERSION_STATE_PROBE_INTERVAL).await;
         }
     });
@@ -834,7 +967,7 @@ impl NotificationSys {
         // A single-node deployment has no remote member to lower the local
         // policy version advertised by this binary.
         if minimum_version == u32::MAX {
-            minimum_version = TIER_DELETE_JOURNAL_POLICY_SUPPORTED_VERSION;
+            minimum_version = DECOMMISSION_TARGET_FENCE_POLICY_SUPPORTED_VERSION;
         }
         Ok((peer_epochs, minimum_version))
     }
@@ -2804,15 +2937,22 @@ mod tests {
     use super::*;
 
     #[test]
-    fn cross_pool_v2_remains_generic_but_cannot_authorize_v6_journal() {
+    fn cross_pool_policy_versions_authorize_only_their_supported_protocols() {
         let peers = BTreeMap::from([("node-b:9000".to_string(), Uuid::new_v4())]);
-        let (generic_v2, journal_v2) = cross_pool_fence_policy_results(peers.clone(), 2);
+        let (generic_v2, journal_v2, decommission_v2) = cross_pool_fence_policy_results(peers.clone(), 2);
         assert!(generic_v2.is_ok(), "v2 remains valid for existing cross-pool fencing");
         assert!(journal_v2.is_err(), "a mixed v2/v3 fleet must fail closed for journal-v6 deletion");
+        assert!(decommission_v2.is_err(), "v2 cannot authorize the sticky per-target decommission fence");
 
-        let (generic_v3, journal_v3) = cross_pool_fence_policy_results(peers, 3);
+        let (generic_v3, journal_v3, decommission_v3) = cross_pool_fence_policy_results(peers.clone(), 3);
         assert!(generic_v3.is_ok());
         assert!(journal_v3.is_ok(), "an all-v3 fleet may authorize journal-v6 deletion");
+        assert!(decommission_v3.is_err(), "v3 members do not understand the per-target decommission fence");
+
+        let (generic_v4, journal_v4, decommission_v4) = cross_pool_fence_policy_results(peers, 4);
+        assert!(generic_v4.is_ok());
+        assert!(journal_v4.is_ok());
+        assert!(decommission_v4.is_ok(), "an all-v4 fleet may create sticky per-target reservations");
     }
 
     #[test]

crates/ecstore/src/services/rebalance/mod.rs

@@ -83,7 +83,6 @@ pub async fn test_store_with_persisted_rebalance_meta(
         decommission_cancelers: tokio::sync::RwLock::new(vec![None]),
         start_gate: tokio::sync::Mutex::new(()),
         pool_meta_save_gate: tokio::sync::Mutex::default(),
-        decommission_capacity_entry_gate: tokio::sync::Mutex::default(),
         ctx,
         bucket_fence_registry: std::sync::Arc::default(),
     });
@@ -233,7 +232,6 @@ async fn test_pool_stores_with_contexts(
             decommission_cancelers: tokio::sync::RwLock::new(vec![None; pool_count]),
             start_gate: tokio::sync::Mutex::new(()),
             pool_meta_save_gate: tokio::sync::Mutex::new(pool_meta_write_state.independent_clone_for_test()),
-            decommission_capacity_entry_gate: tokio::sync::Mutex::default(),
             ctx: store_ctx,
             bucket_fence_registry: std::sync::Arc::default(),
         })

crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs

@@ -3009,7 +3009,6 @@ fn test_store_with_rebalance_meta(meta: RebalanceMeta) -> Arc<crate::store::ECSt
         decommission_cancelers: tokio::sync::RwLock::new(Vec::new()),
         start_gate: tokio::sync::Mutex::new(()),
         pool_meta_save_gate: tokio::sync::Mutex::default(),
-        decommission_capacity_entry_gate: tokio::sync::Mutex::default(),
         ctx: crate::runtime::instance::bootstrap_ctx(),
         bucket_fence_registry: std::sync::Arc::default(),
     })

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

@@ -837,7 +837,7 @@ impl SetDisks {
         orig_bucket: &str,
         error_path: &str,
         root_prefix: &str,
-    ) -> Result<(Vec<Option<DiskStore>>, Vec<String>, usize)> {
+    ) -> Result<(Vec<Option<DiskStore>>, Vec<String>, usize, bool)> {
         let disks = self.disks.read().await.clone();
         if disks.is_empty() {
             return Err(Error::ErasureReadQuorum);
@@ -882,16 +882,24 @@ impl SetDisks {
             return Err(to_object_err(err.into(), vec![orig_bucket, error_path]));
         }
 
+        let mut has_minority_candidate = false;
         let mut candidate_paths = candidate_counts
             .into_iter()
-            .filter_map(|(path, count)| (count >= discovery_quorum).then_some(path))
+            .filter_map(|(path, count)| {
+                if count >= discovery_quorum {
+                    Some(path)
+                } else {
+                    has_minority_candidate = true;
+                    None
+                }
+            })
             .collect::<Vec<_>>();
         candidate_paths.sort_unstable();
-        Ok((disks, candidate_paths, discovery_quorum))
+        Ok((disks, candidate_paths, discovery_quorum, has_minority_candidate))
     }
 
     pub(crate) async fn first_multipart_upload_path_for_decommission(&self, bucket: &str) -> Result<Option<String>> {
-        let (_, paths, _) = self
+        let (_, paths, _, _) = self
             .discover_multipart_upload_paths(bucket, RUSTFS_META_MULTIPART_BUCKET, "")
             .await?;
         Ok(paths.into_iter().next())
@@ -905,7 +913,14 @@ impl SetDisks {
         upload_identity: &str,
     ) -> Result<Vec<String>> {
         let expected_parent = format!("{DATA_MOVEMENT_MULTIPART_PREFIX}/{}", Self::get_multipart_sha_dir(bucket, object));
-        let (_, candidate_paths, _) = self.discover_multipart_upload_paths(bucket, object, &expected_parent).await?;
+        let (_, candidate_paths, _, has_minority_candidate) =
+            self.discover_multipart_upload_paths(bucket, object, &expected_parent).await?;
+        if has_minority_candidate {
+            return Err(Error::DecommissionCapacityBlocked {
+                message: "data movement multipart cleanup found an upload path on fewer than the discovery quorum of disks"
+                    .to_string(),
+            });
+        }
         let mut upload_ids = Vec::new();
         for upload_path in candidate_paths {
             let Some((parent, raw_upload_id)) = upload_path.rsplit_once('/') else {
@@ -921,7 +936,11 @@ impl SetDisks {
             {
                 Ok((file_info, _)) => file_info,
                 Err(err) if crate::error::is_err_invalid_upload_id(&err) || crate::error::is_err_object_not_found(&err) => {
-                    continue;
+                    return Err(Error::DecommissionCapacityBlocked {
+                        message: format!(
+                            "data movement multipart cleanup found quorum-visible upload path {upload_path} without verifiable metadata: {err}"
+                        ),
+                    });
                 }
                 Err(err) => return Err(err),
             };
@@ -1190,7 +1209,7 @@ impl SetDisks {
         max_uploads: usize,
         expected_incarnation_id: Option<Uuid>,
     ) -> Result<ListMultipartsInfo> {
-        let (disks, candidate_paths, discovery_quorum) = self.discover_multipart_upload_paths(bucket, prefix, "").await?;
+        let (disks, candidate_paths, discovery_quorum, _) = self.discover_multipart_upload_paths(bucket, prefix, "").await?;
         let listed_uploads = stream::iter(candidate_paths)
             .map(|upload_path| {
                 let disks = &disks;
@@ -7551,6 +7570,101 @@ mod tests {
         assert_eq!(bucket_wide.uploads[0].object, "blobs/data/layer.bin");
     }
 
+    #[tokio::test]
+    async fn data_movement_cleanup_discovery_rejects_quorum_visible_upload_without_metadata() {
+        let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "data-movement-unverifiable-upload";
+        let object = "staged/object.bin";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let upload_identity = format!("v1:{}:{}", Uuid::new_v4(), OffsetDateTime::now_utc().unix_timestamp_nanos());
+        let mut metadata = HashMap::new();
+        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD, upload_identity.clone());
+        let opts = ObjectOptions {
+            data_movement: true,
+            user_defined: metadata,
+            ..Default::default()
+        };
+        let upload = set_disks
+            .new_multipart_upload(bucket, object, &opts)
+            .await
+            .expect("data movement upload should be created");
+        let upload_path = SetDisks::get_multipart_upload_dir(bucket, object, &upload.upload_id, true);
+        for temp_dir in &temp_dirs {
+            tokio::fs::remove_file(
+                temp_dir
+                    .path()
+                    .join(RUSTFS_META_MULTIPART_BUCKET)
+                    .join(&upload_path)
+                    .join("xl.meta"),
+            )
+            .await
+            .expect("upload metadata should be removable while preserving its quorum-visible directory");
+        }
+
+        let err = set_disks
+            .data_movement_multipart_upload_ids(bucket, object, None, &upload_identity)
+            .await
+            .expect_err("cleanup discovery must fail closed on an unverifiable upload path");
+        assert!(matches!(err, Error::DecommissionCapacityBlocked { .. }));
+    }
+
+    #[tokio::test]
+    async fn data_movement_cleanup_discovery_rejects_minority_upload_until_disks_recover() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "data-movement-minority-upload";
+        let object = "staged/object.bin";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        {
+            let mut disks = set_disks.disks.write().await;
+            disks[3] = None;
+        }
+        let upload_identity = format!("v1:{}:{}", Uuid::new_v4(), OffsetDateTime::now_utc().unix_timestamp_nanos());
+        let mut metadata = HashMap::new();
+        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD, upload_identity.clone());
+        let upload = set_disks
+            .new_multipart_upload(
+                bucket,
+                object,
+                &ObjectOptions {
+                    data_movement: true,
+                    user_defined: metadata,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("write quorum should create the upload while one disk is offline");
+
+        {
+            let mut disks = set_disks.disks.write().await;
+            disks[1] = None;
+            disks[2] = None;
+            disks[3] = Some(disk_stores[3].clone());
+        }
+        let err = set_disks
+            .data_movement_multipart_upload_ids(bucket, object, None, &upload_identity)
+            .await
+            .expect_err("a minority-observed upload must block a destructive absence proof");
+        assert!(matches!(err, Error::DecommissionCapacityBlocked { .. }));
+
+        {
+            let mut disks = set_disks.disks.write().await;
+            disks[1] = Some(disk_stores[1].clone());
+            disks[2] = Some(disk_stores[2].clone());
+        }
+        let recovered = set_disks
+            .data_movement_multipart_upload_ids(bucket, object, None, &upload_identity)
+            .await
+            .expect("recovered quorum should make the staged upload verifiable");
+        assert_eq!(recovered.len(), 1);
+        assert_eq!(upload_uuid_suffix(&recovered[0]), upload_uuid_suffix(&upload.upload_id));
+    }
+
     /// Regression (issue #5716): a single upload directory whose `xl.meta` was
     /// destroyed (crash mid-write, torn disk state) must degrade to that upload
     /// alone. Failing the whole ListMultipartUploads turns one piece of stale

crates/ecstore/src/store/heal.rs

@@ -821,7 +821,6 @@ mod tests {
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
             pool_meta_save_gate: Mutex::default(),
-            decommission_capacity_entry_gate: Mutex::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
         }
@@ -2380,7 +2379,6 @@ mod tests {
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
             pool_meta_save_gate: Mutex::default(),
-            decommission_capacity_entry_gate: Mutex::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
         };

crates/ecstore/src/store/init.rs

@@ -580,7 +580,6 @@ impl ECStore {
             decommission_cancelers,
             start_gate: Mutex::new(()),
             pool_meta_save_gate: Mutex::new(pool_meta_write_state),
-            decommission_capacity_entry_gate: Mutex::default(),
             // Adopt the caller's context (the process bootstrap one on the
             // legacy path) so startup writes (erasure type recorded before
             // this point) and later reads share one cell.
@@ -918,6 +917,35 @@ mod tests {
     use time::OffsetDateTime;
     use tokio::io::AsyncReadExt;
     use tokio_util::sync::CancellationToken;
+
+    fn run_large_stack_async_test<C, F>(name: &str, case: C)
+    where
+        C: FnOnce() -> F + Send + 'static,
+        F: Future<Output = ()> + 'static,
+    {
+        const STACK_SIZE: usize = if cfg!(debug_assertions) {
+            8 * rustfs_config::DEFAULT_THREAD_STACK_SIZE
+        } else if cfg!(target_os = "macos") {
+            2 * rustfs_config::DEFAULT_THREAD_STACK_SIZE
+        } else {
+            rustfs_config::DEFAULT_THREAD_STACK_SIZE
+        };
+        std::thread::Builder::new()
+            .name(name.to_string())
+            .stack_size(STACK_SIZE)
+            .spawn(move || {
+                let runtime = tokio::runtime::Builder::new_multi_thread()
+                    .enable_all()
+                    .worker_threads(2)
+                    .thread_stack_size(STACK_SIZE)
+                    .build()
+                    .expect("large-stack store test runtime should build");
+                runtime.block_on(case());
+            })
+            .expect("large-stack store test thread should spawn")
+            .join()
+            .expect("large-stack store test thread should complete");
+    }
     use uuid::Uuid;
 
     #[test]
@@ -2306,9 +2334,17 @@ mod tests {
     }
 
     #[cfg(feature = "test-util")]
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[test]
     #[serial_test::serial(storage_class_env)]
-    async fn data_movement_multipart_part_staging_holds_no_publication_lock_or_tier_lease() {
+    fn data_movement_multipart_part_staging_holds_no_publication_lock_or_tier_lease() {
+        run_large_stack_async_test(
+            "multipart-part-staging-publication-fence",
+            data_movement_multipart_part_staging_holds_no_publication_lock_or_tier_lease_case,
+        );
+    }
+
+    #[cfg(feature = "test-util")]
+    async fn data_movement_multipart_part_staging_holds_no_publication_lock_or_tier_lease_case() {
         let temp_dir = tempfile::tempdir().expect("create multipart staging-fence store dir");
         let (ctx, store, shutdown) =
             without_storage_class_env(build_isolated_test_store(temp_dir.path(), "mpu-staging-publication", &[4, 4])).await;
@@ -2388,19 +2424,23 @@ mod tests {
         let capacity_owner = test_decommission_capacity_owner(store.as_ref(), 0)
             .await
             .with_mutation_id(Uuid::new_v4());
-        let mut upload_metadata = source.user_defined.as_ref().clone();
-        rustfs_utils::http::insert_str(
-            &mut upload_metadata,
-            rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD,
-            "mpu-staging-test".to_string(),
-        );
+        let upload_metadata = source.user_defined.as_ref().clone();
         let mut staging_opts = ObjectOptions {
             data_movement: true,
             src_pool_idx: 0,
+            versioned: source.version_id.is_some(),
+            version_id: source.version_id.map(|version_id| version_id.to_string()),
+            mod_time: source.mod_time,
             user_defined: upload_metadata,
             expected_bucket_incarnation_id: Some(bucket_incarnation_id),
             ..Default::default()
         };
+        let upload_identity = crate::data_movement::data_movement_upload_identity_from_options(&staging_opts);
+        rustfs_utils::http::insert_str(
+            &mut staging_opts.user_defined,
+            rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD,
+            upload_identity,
+        );
         capacity_owner.apply_to(&mut staging_opts);
         let (upload, target_pool_idx, staged_incarnation_id) = store
             .handle_new_multipart_upload_with_pool_idx(&bucket, object, &staging_opts, None)
@@ -2470,6 +2510,9 @@ mod tests {
         let mut abort_opts = ObjectOptions {
             data_movement: true,
             src_pool_idx: 0,
+            versioned: source.version_id.is_some(),
+            version_id: source.version_id.map(|version_id| version_id.to_string()),
+            mod_time: source.mod_time,
             expected_bucket_incarnation_id: Some(bucket_incarnation_id),
             ..Default::default()
         };
@@ -4972,8 +5015,9 @@ mod tests {
                     let ordinary_faults_for_hook = Arc::clone(&ordinary_faults);
                     let fault_bucket = other_bucket.clone();
                     let _fault_guard = crate::core::pools::DecommissionTestFaultGuard::install(Arc::new(
-                        move |stage, bucket, object, attempt| {
-                            let injected = stage == DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT
+                        move |stage, bucket, object, attempt, succeeded| {
+                            let injected = succeeded
+                                && stage == DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT
                                 && bucket == fault_bucket.as_str()
                                 && object == other_object
                                 && attempt < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS;
@@ -5242,8 +5286,9 @@ mod tests {
                     let fault_calls_for_hook = Arc::clone(&fault_calls);
                     let fault_bucket = bucket.clone();
                     let _fault_guard = crate::core::pools::DecommissionTestFaultGuard::install(Arc::new(
-                        move |stage, called_bucket, called_object, attempt| {
-                            let injected = stage == DECOMMISSION_TEST_FAULT_STAGE_DELETE_MARKER
+                        move |stage, called_bucket, called_object, attempt, succeeded| {
+                            let injected = succeeded
+                                && stage == DECOMMISSION_TEST_FAULT_STAGE_DELETE_MARKER
                                 && called_bucket == fault_bucket.as_str()
                                 && called_object == object
                                 && attempt < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS;
@@ -5362,8 +5407,9 @@ mod tests {
                     let fault_calls_for_hook = Arc::clone(&fault_calls);
                     let fault_bucket = bucket.clone();
                     let _fault_guard = crate::core::pools::DecommissionTestFaultGuard::install(Arc::new(
-                        move |stage, called_bucket, called_object, attempt| {
-                            let injected = stage == DECOMMISSION_TEST_FAULT_STAGE_TIERED
+                        move |stage, called_bucket, called_object, attempt, succeeded| {
+                            let injected = succeeded
+                                && stage == DECOMMISSION_TEST_FAULT_STAGE_TIERED
                                 && called_bucket == fault_bucket.as_str()
                                 && called_object == object
                                 && attempt < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS;
@@ -5468,9 +5514,16 @@ mod tests {
         shutdown.cancel();
     }
 
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[test]
     #[serial_test::serial(storage_class_env)]
-    async fn decommission_outer_fence_loss_blocks_multipart_commits() {
+    fn decommission_outer_fence_loss_blocks_multipart_commits() {
+        run_large_stack_async_test(
+            "decommission-multipart-outer-fence-loss",
+            decommission_outer_fence_loss_blocks_multipart_commits_case,
+        );
+    }
+
+    async fn decommission_outer_fence_loss_blocks_multipart_commits_case() {
         for (object, pause) in [
             ("complete.bin", crate::set_disk::MultipartCommitPause::BeforeLockLost),
             ("new-upload.bin", crate::set_disk::MultipartCommitPause::NewUploadBeforeLockLost),
@@ -10019,6 +10072,29 @@ mod tests {
             .expect("the active decommission should own the checkpoint target");
         assert_eq!(targets.len(), 1);
         let target = targets[0].clone();
+        let consumed_before_checkpoint = store.pool_meta.read().await.pools[0]
+            .decommission
+            .as_ref()
+            .and_then(|info| info.capacity_reservation.as_ref())
+            .expect("checkpoint shutdown capacity reservation should exist")
+            .consumed_target_physical_bytes;
+        let precommit_error = store
+            .run_decommission_capacity_non_growing_replacement_with_capacity_lease(
+                target.target_pool_index,
+                Some(target.capacity_owner),
+                Some(aborting_data.len()),
+                |_| async { Err::<(), Error>(Error::other("injected checkpoint failure before commit")) },
+            )
+            .await
+            .expect_err("the first checkpoint attempt should fail before writing its target")
+            .to_string();
+        assert!(precommit_error.contains("injected checkpoint failure before commit"));
+        assert!(
+            store
+                .has_decommission_capacity_temporary_mutation_state(target.target_pool_index, target.capacity_owner)
+                .await,
+            "a failed checkpoint attempt must retain exact retry state"
+        );
         let barrier = crate::set_disk::PutObjectCommitBarrier::install(
             RUSTFS_META_BUCKET,
             &manifest_name,
@@ -10069,6 +10145,27 @@ mod tests {
                 .await,
             "the admitted target PUT must drain its capacity transaction before releasing the recovery fences"
         );
+        {
+            let pool_meta = store.pool_meta.read().await;
+            let reservation = pool_meta.pools[0]
+                .decommission
+                .as_ref()
+                .and_then(|info| info.capacity_reservation.as_ref())
+                .expect("checkpoint shutdown capacity reservation should remain active");
+            let capacity_target = reservation
+                .targets
+                .iter()
+                .find(|candidate| candidate.pool_index == target.target_pool_index)
+                .expect("checkpoint shutdown capacity target should remain allocated");
+            assert_eq!(
+                reservation.consumed_target_physical_bytes, consumed_before_checkpoint,
+                "a non-growing checkpoint replacement must not consume durable migration capacity"
+            );
+            assert_eq!(reservation.pending_target_physical_bytes, 0);
+            assert_eq!(reservation.inflight_target_physical_bytes, 0);
+            assert_eq!(capacity_target.pending_physical_bytes, 0);
+            assert!(capacity_target.temporary_mutations.is_empty());
+        }
         drop(barrier);
         let mut reloaded_pool_meta = PoolMeta::default();
         reloaded_pool_meta

crates/ecstore/src/store/mod.rs

@@ -468,12 +468,6 @@ pub struct ECStore {
     /// Lock order: acquire `pool_meta_save_gate`, then the distributed
     /// `pool.bin` fence, then clone `pool_meta` under a short read lock.
     pub(crate) pool_meta_save_gate: Mutex<PoolMetaWriteState>,
-    /// Serializes decommission entries while the durable capacity ledger has
-    /// one target mutation intent slot.
-    ///
-    /// Lock order: acquire this gate before object namespaces or
-    /// `pool_meta_save_gate`.
-    pub(crate) decommission_capacity_entry_gate: Mutex<()>,
     /// Per-instance runtime state (Phase 5, backlog#939).
     ///
     /// Carries this instance's identity/runtime out of the process globals so
@@ -1728,7 +1722,6 @@ mod tests {
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
             pool_meta_save_gate: Mutex::default(),
-            decommission_capacity_entry_gate: Mutex::default(),
             ctx,
             bucket_fence_registry: Arc::default(),
         };
@@ -1804,7 +1797,6 @@ mod tests {
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
             pool_meta_save_gate: Mutex::default(),
-            decommission_capacity_entry_gate: Mutex::default(),
             ctx,
             bucket_fence_registry: Arc::default(),
         })

crates/ecstore/src/store/multipart.rs

@@ -17,11 +17,52 @@ use crate::core::pools::{DecommissionCapacityOwner, ensure_decommission_capacity
 use crate::multipart_listing::paginate_multipart_listing;
 use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::multipart::MultipartOperations as _;
+use crate::storage_api_contracts::object::ObjectOperations as _;
 use futures::{StreamExt, stream};
 use std::collections::HashSet;
 
 const MULTIPART_LIST_SET_CONCURRENCY: usize = 4;
 
+#[cfg(test)]
+static DATA_MOVEMENT_MULTIPART_DISCOVERY_COUNTS: std::sync::OnceLock<std::sync::Mutex<std::collections::HashMap<Uuid, usize>>> =
+    std::sync::OnceLock::new();
+
+#[cfg(test)]
+fn data_movement_multipart_discovery_counts() -> &'static std::sync::Mutex<std::collections::HashMap<Uuid, usize>> {
+    DATA_MOVEMENT_MULTIPART_DISCOVERY_COUNTS.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
+}
+
+fn decommission_multipart_target_clear_pending(opts: &ObjectOptions, target: Option<&ObjectInfo>) -> Result<bool> {
+    let expected_mod_time = opts.mod_time.ok_or_else(|| Error::DecommissionCapacityBlocked {
+        message: "multipart cleanup cannot prove exact target absence without a modification time".to_string(),
+    })?;
+    let expected_version_id = opts
+        .version_id
+        .as_deref()
+        .map(Uuid::parse_str)
+        .transpose()
+        .map_err(|err| Error::DecommissionCapacityBlocked {
+            message: format!("multipart cleanup exact target version is invalid: {err}"),
+        })?
+        .filter(|version_id| !version_id.is_nil());
+    let Some(target) = target else {
+        return Ok(true);
+    };
+    if target.version_id.filter(|version_id| !version_id.is_nil()) != expected_version_id
+        || target.mod_time != Some(expected_mod_time)
+    {
+        return Err(Error::DecommissionCapacityBlocked {
+            message: "multipart cleanup found a target but cannot prove the exact staged identity is absent".to_string(),
+        });
+    }
+    if !crate::data_movement::is_owned_data_movement_target(target) {
+        return Err(Error::DecommissionCapacityBlocked {
+            message: "multipart cleanup found an exact target without its ownership proof".to_string(),
+        });
+    }
+    Ok(false)
+}
+
 #[derive(Clone, Debug)]
 pub(super) struct MultipartUploadListRequest {
     pub(super) prefix: String,
@@ -197,6 +238,24 @@ async fn list_pool_multipart_uploads_for_incarnation(
 }
 
 impl ECStore {
+    #[cfg(test)]
+    pub(crate) fn reset_data_movement_multipart_discovery_count_for_test(&self) {
+        data_movement_multipart_discovery_counts()
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(self.id, 0);
+    }
+
+    #[cfg(test)]
+    pub(crate) fn data_movement_multipart_discovery_count_for_test(&self) -> usize {
+        data_movement_multipart_discovery_counts()
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .get(&self.id)
+            .copied()
+            .unwrap_or_default()
+    }
+
     pub(crate) async fn acquire_decommission_multipart_mutation_fence(
         &self,
         owner: DecommissionCapacityOwner,
@@ -728,8 +787,16 @@ impl ECStore {
         upload_id: &str,
         opts: &ObjectOptions,
     ) -> Result<()> {
-        self.abort_multipart_uploads_for_data_movement(target_pool_idx, bucket, object, &[upload_id.to_owned()], None, opts)
-            .await
+        let upload_identity = crate::data_movement::data_movement_upload_identity_from_options(opts);
+        self.abort_multipart_uploads_for_data_movement(
+            target_pool_idx,
+            bucket,
+            object,
+            &[upload_id.to_owned()],
+            &upload_identity,
+            opts,
+        )
+        .await
     }
 
     pub(crate) async fn reconcile_multipart_uploads_for_data_movement(
@@ -740,26 +807,37 @@ impl ECStore {
         upload_identity: &str,
         opts: &ObjectOptions,
     ) -> Result<()> {
-        let pool = self
-            .pools
+        self.pools
             .get(target_pool_idx)
             .ok_or_else(|| Error::other(format!("data movement target pool {target_pool_idx} is out of range")))?;
-        let owner = DecommissionCapacityOwner::from_options(opts);
-        let has_capacity_state = match owner {
-            Some(owner) => {
-                self.has_decommission_capacity_temporary_mutation_state(target_pool_idx, owner)
-                    .await
-            }
-            None => false,
-        };
-        if !has_capacity_state {
+        let owner = DecommissionCapacityOwner::from_options(opts)
+            .ok_or_else(|| Error::other("data movement multipart cleanup is missing its capacity owner"))?;
+        if !self
+            .decommission_capacity_cleanup_target_indices(owner)
+            .await?
+            .contains(&target_pool_idx)
+        {
+            return Err(Error::DecommissionCapacityBlocked {
+                message: format!(
+                    "data movement multipart cleanup target pool {target_pool_idx} is outside its capacity reservation"
+                ),
+            });
+        }
+        if !self
+            .has_decommission_capacity_temporary_mutation_state(target_pool_idx, owner)
+            .await
+        {
             return Ok(());
         }
-        let set = pool.get_disks_by_key(object);
-        let upload_ids = set
-            .data_movement_multipart_upload_ids(bucket, object, opts.expected_bucket_incarnation_id, upload_identity)
-            .await?;
-        self.abort_multipart_uploads_for_data_movement(target_pool_idx, bucket, object, &upload_ids, Some(upload_identity), opts)
+        #[cfg(test)]
+        {
+            *data_movement_multipart_discovery_counts()
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .entry(self.id)
+                .or_default() += 1;
+        }
+        self.abort_multipart_uploads_for_data_movement(target_pool_idx, bucket, object, &[], upload_identity, opts)
             .await
     }
 
@@ -769,7 +847,7 @@ impl ECStore {
         bucket: &str,
         object: &str,
         upload_ids: &[String],
-        expected_upload_identity: Option<&str>,
+        expected_upload_identity: &str,
         opts: &ObjectOptions,
     ) -> Result<()> {
         check_new_multipart_args(bucket, object)?;
@@ -781,42 +859,116 @@ impl ECStore {
         }
         let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
         ensure_decommission_capacity_mutation_id(bucket, object, &mut opts);
+        let capacity_owner = DecommissionCapacityOwner::from_options(&opts);
         let pool = self
             .pools
             .get(target_pool_idx)
             .ok_or_else(|| Error::other(format!("data movement target pool {target_pool_idx} is out of range")))?;
-        let set = pool.get_disks_by_key(object);
-        let mut guards = Vec::with_capacity(upload_ids.len());
-        for upload_id in upload_ids {
-            if let Some(guard) = set
-                .lock_data_movement_multipart_abort(bucket, object, upload_id, expected_upload_identity, &opts)
-                .await?
-            {
-                guard.add_namespace_lock_fence(&mut opts);
-                guards.push(guard);
-            }
-        }
         opts.no_lock = true;
-        let capacity_owner = DecommissionCapacityOwner::from_options(&opts);
-        // Keep every upload namespace guard alive through the final capacity progress save.
-        let result = self
+        let set = pool.get_disks_by_key(object);
+        // Discover and lock uploads only after the target capacity gate is held.
+        // Return the guards so they remain alive through the final capacity save.
+        let (cleanup_decision_error, guards) = self
             .run_decommission_capacity_temporary_release_with_capacity_lease(target_pool_idx, capacity_owner, |capacity_lease| {
                 let mut delete_opts = opts.clone();
-                let guards = &guards;
                 let set = &set;
+                let pool = &pool;
                 async move {
-                    if let Some(capacity_lease) = capacity_lease {
-                        delete_opts.add_namespace_lock_lost_signal(capacity_lease);
+                    if let Some(capacity_lease) = capacity_lease.as_ref() {
+                        delete_opts.add_namespace_lock_lost_signal(Arc::clone(capacity_lease));
                     }
-                    for guard in guards {
-                        guard.delete(set, bucket, object, &delete_opts).await?;
+                    let mut candidate_upload_ids = upload_ids.to_vec();
+                    candidate_upload_ids.extend(
+                        set.data_movement_multipart_upload_ids(
+                            bucket,
+                            object,
+                            delete_opts.expected_bucket_incarnation_id,
+                            expected_upload_identity,
+                        )
+                        .await?,
+                    );
+                    candidate_upload_ids.sort_unstable();
+                    candidate_upload_ids.dedup();
+
+                    let mut guards = Vec::with_capacity(candidate_upload_ids.len());
+                    for upload_id in &candidate_upload_ids {
+                        match set
+                            .lock_data_movement_multipart_abort(
+                                bucket,
+                                object,
+                                upload_id,
+                                Some(expected_upload_identity),
+                                &delete_opts,
+                            )
+                            .await
+                        {
+                            Ok(Some(guard)) => {
+                                guard.add_namespace_lock_fence(&mut delete_opts);
+                                guards.push(guard);
+                            }
+                            Ok(None) => {}
+                            Err(err) => return Err(err),
+                        }
                     }
-                    Ok(())
+                    for guard in &guards {
+                        match guard.delete(set, bucket, object, &delete_opts).await {
+                            Ok(()) => {}
+                            Err(err) if is_err_invalid_upload_id(&err) => {}
+                            Err(err) => return Err(err),
+                        }
+                    }
+
+                    if !set
+                        .data_movement_multipart_upload_ids(
+                            bucket,
+                            object,
+                            delete_opts.expected_bucket_incarnation_id,
+                            expected_upload_identity,
+                        )
+                        .await?
+                        .is_empty()
+                    {
+                        return Err(Error::DecommissionCapacityBlocked {
+                            message: "multipart cleanup could not prove the exact staged uploads are absent".to_string(),
+                        });
+                    }
+
+                    // The target capacity gate makes the exact target proof,
+                    // upload absence proof, and pending-ledger decision one
+                    // critical section with cleanup finalize.
+                    let (clear_pending, cleanup_decision_error) = if capacity_owner.is_some() {
+                        let mut lookup_opts = ObjectOptions {
+                            versioned: delete_opts.versioned,
+                            version_suspended: delete_opts.version_suspended,
+                            version_id: delete_opts.version_id.clone(),
+                            metadata_chg: delete_opts.version_id.is_some(),
+                            no_lock: true,
+                            ..Default::default()
+                        };
+                        if let Some(capacity_lease) = capacity_lease {
+                            lookup_opts.add_namespace_lock_lost_signal(capacity_lease);
+                        }
+                        let target = match pool.get_object_info(bucket, object, &lookup_opts).await {
+                            Ok(target) => Some(target),
+                            Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => None,
+                            Err(err) => return Err(err),
+                        };
+                        match decommission_multipart_target_clear_pending(&delete_opts, target.as_ref()) {
+                            Ok(clear_pending) => (clear_pending, None),
+                            Err(err) => (false, Some(err)),
+                        }
+                    } else {
+                        (true, None)
+                    };
+                    Ok(((cleanup_decision_error, guards), clear_pending))
                 }
             })
-            .await;
+            .await?;
         drop(guards);
-        result
+        if let Some(err) = cleanup_decision_error {
+            return Err(err);
+        }
+        Ok(())
     }
 
     #[instrument(skip(self))]
@@ -1024,6 +1176,66 @@ mod tests {
         }
     }
 
+    #[test]
+    fn decommission_multipart_cleanup_requires_exact_target_evidence() {
+        let version_id = Uuid::new_v4();
+        let mod_time = time::OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(7);
+        let opts = ObjectOptions {
+            versioned: true,
+            version_id: Some(version_id.to_string()),
+            mod_time: Some(mod_time),
+            ..Default::default()
+        };
+
+        assert!(
+            decommission_multipart_target_clear_pending(&opts, None)
+                .expect("an exact target miss should authorize pending cleanup")
+        );
+
+        let mut metadata = HashMap::new();
+        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_DATA_MOVED, "true".to_string());
+        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_DATA_MOVED_TAGS, "v1:".to_string());
+        let owned_target = ObjectInfo {
+            version_id: Some(version_id),
+            mod_time: Some(mod_time),
+            user_defined: Arc::new(metadata),
+            ..Default::default()
+        };
+        assert!(
+            !decommission_multipart_target_clear_pending(&opts, Some(&owned_target))
+                .expect("an exact owned target should preserve pending capacity")
+        );
+
+        let missing_identity = ObjectOptions {
+            mod_time: None,
+            ..opts.clone()
+        };
+        assert!(matches!(
+            decommission_multipart_target_clear_pending(&missing_identity, None),
+            Err(Error::DecommissionCapacityBlocked { .. })
+        ));
+
+        let mismatched_target = ObjectInfo {
+            version_id: Some(Uuid::new_v4()),
+            mod_time: Some(mod_time),
+            ..Default::default()
+        };
+        assert!(matches!(
+            decommission_multipart_target_clear_pending(&opts, Some(&mismatched_target)),
+            Err(Error::DecommissionCapacityBlocked { .. })
+        ));
+
+        let unowned_target = ObjectInfo {
+            version_id: Some(version_id),
+            mod_time: Some(mod_time),
+            ..Default::default()
+        };
+        assert!(matches!(
+            decommission_multipart_target_clear_pending(&opts, Some(&unowned_target)),
+            Err(Error::DecommissionCapacityBlocked { .. })
+        ));
+    }
+
     /// Models a single pool's `list_multipart_uploads`: returns uploads strictly
     /// after the `(key, upload_id)` marker in `(key, upload_id)` order, capped at
     /// `max_uploads` (mirroring the per-pool page cap).
@@ -1171,7 +1383,6 @@ mod tests {
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
             pool_meta_save_gate: Mutex::default(),
-            decommission_capacity_entry_gate: Mutex::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
         }

crates/ecstore/src/store/object.rs

@@ -1531,6 +1531,10 @@ fn data_movement_pool_lookup_opts(opts: &ObjectOptions, no_lock: bool) -> Object
     writer_pool_lookup_opts(opts, no_lock)
 }
 
+fn uses_data_movement_pool_selection(opts: &ObjectOptions) -> bool {
+    opts.data_movement && (opts.version_id.is_some() || DecommissionCapacityOwner::from_options(opts).is_some())
+}
+
 fn writer_pool_lookup_opts(opts: &ObjectOptions, no_lock: bool) -> ObjectOptions {
     let mut lookup_opts = version_aware_lookup_opts(opts, no_lock);
     lookup_opts.skip_decommissioned = true;
@@ -3467,7 +3471,12 @@ impl ECStore {
     }
 
     fn resolve_decommission_tiered_object_result(result: Result<()>, bucket: &str, object: &str) -> Result<()> {
-        result.map_err(|err| Error::other(format!("failed to decommission tiered object for {bucket}/{object}: {err}")))
+        result.map_err(|err| {
+            crate::data_movement::data_movement_context_error(
+                format!("failed to decommission tiered object for {bucket}/{object}: {err}"),
+                err,
+            )
+        })
     }
 
     #[instrument(skip(self, fi, opts))]
@@ -3515,7 +3524,7 @@ impl ECStore {
             );
         }
 
-        let idx = if opts.data_movement && opts.version_id.is_some() {
+        let idx = if uses_data_movement_pool_selection(&opts) {
             Self::resolve_decommission_target_pool_idx_result(
                 self.select_data_movement_pool_idx(bucket, &object, fi.size, &opts, true)
                     .await,
@@ -3713,7 +3722,7 @@ impl ECStore {
             return Ok(0);
         }
 
-        let idx = if opts.data_movement && opts.version_id.is_some() {
+        let idx = if uses_data_movement_pool_selection(opts) {
             self.select_data_movement_pool_idx(bucket, object, size, opts, false).await?
         } else if opts.no_lock {
             self.get_pool_idx_no_lock(bucket, object, size).await?
@@ -7277,6 +7286,23 @@ mod tests {
         assert!(rendered.contains("boom"), "{rendered}");
     }
 
+    #[test]
+    fn resolve_decommission_tiered_object_result_preserves_typed_capacity_error() {
+        let err = ECStore::resolve_decommission_tiered_object_result(
+            Err(Error::DecommissionCapacityBlocked {
+                message: "target gate busy".to_string(),
+            }),
+            "bucket",
+            "object",
+        )
+        .expect_err("expected contextual error");
+
+        assert!(matches!(
+            crate::data_movement::data_movement_stage_source(&err),
+            Some(Error::DecommissionCapacityBlocked { message }) if message == "target gate busy"
+        ));
+    }
+
     #[test]
     fn version_aware_lookup_opts_enables_version_aware_lookup() {
         let opts = ObjectOptions {
@@ -7502,6 +7528,26 @@ mod tests {
         assert!(lookup_opts.skip_rebalancing);
     }
 
+    #[test]
+    fn capacity_owned_unversioned_move_uses_data_movement_pool_selection() {
+        let mut opts = ObjectOptions {
+            data_movement: true,
+            ..Default::default()
+        };
+        assert!(!uses_data_movement_pool_selection(&opts));
+
+        DecommissionCapacityOwner {
+            source_pool_index: 1,
+            operation_id: Uuid::new_v4(),
+            generation: 2,
+            owner_nonce: Uuid::new_v4(),
+            mutation_id: None,
+        }
+        .apply_to(&mut opts);
+
+        assert!(uses_data_movement_pool_selection(&opts));
+    }
+
     #[test]
     fn transition_restore_pool_opts_skips_decommissioned_and_preserves_locking() {
         let lookup_opts = transition_restore_pool_opts(&ObjectOptions {
@@ -7576,7 +7622,6 @@ mod tests {
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
             pool_meta_save_gate: Mutex::default(),
-            decommission_capacity_entry_gate: Mutex::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
         }
@@ -7669,7 +7714,6 @@ mod tests {
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
             pool_meta_save_gate: Mutex::default(),
-            decommission_capacity_entry_gate: Mutex::default(),
             ctx,
             bucket_fence_registry: std::sync::Arc::default(),
         }

docs/operations/snowball-auto-extract.md

@@ -0,0 +1,36 @@
+# Snowball Auto-Extract Limits
+
+RustFS accepts MinIO-compatible Snowball auto-extract uploads. Archive
+members are streamed into objects while RustFS enforces entry-count, path,
+PAX metadata, per-object, cumulative unpacked-size, and decoded-stream
+limits.
+
+## Size limits
+
+The defaults remain compatible with the existing safety policy:
+
+| Environment variable | Default | Hard maximum | Meaning |
+| --- | ---: | ---: | --- |
+| `RUSTFS_SNOWBALL_MAX_ENTRY_BYTES` | 1 GiB | 1 TiB | Maximum unpacked size of one archive member |
+| `RUSTFS_SNOWBALL_MAX_UNPACKED_BYTES` | 10 GiB | 10 TiB | Maximum cumulative unpacked object bytes in one request |
+
+Invalid values use the default. Zero is treated as one byte, values above the
+hard maximum are clamped, and the per-entry limit is never allowed to exceed
+the cumulative request limit. RustFS derives a separate decoded-stream limit
+with bounded room for tar headers and PAX metadata; it cannot be disabled.
+
+Increasing either limit raises the maximum work performed by one admitted
+request. Snowball archive decoder admission remains globally bounded, so a
+larger archive cannot create an unbounded number of concurrent decoders.
+Restart RustFS after changing these environment variables.
+
+## Small-member concurrency
+
+For requests that set Snowball ignore-errors and do not use bucket quota
+accounting, RustFS stages members up to 128 KiB and commits at most 16 at a
+time. Requests that must stop on the first write error and quota-enabled
+requests remain serial so their observable error and accounting behavior does
+not change.
+
+Set `RUSTFS_SNOWBALL_EXTRACT_MAX_INFLIGHT=1` to restore fully serial member
+commits. Values are clamped to the range 1 through 16.

rustfs/src/app/object/extract.rs

@@ -373,10 +373,10 @@ const EXTRACT_MAX_EFFECTIVE_PAX_HEADER_BYTES: usize = 8 * 1024;
 const EXTRACT_MAX_EFFECTIVE_PAX_USER_METADATA_BYTES: usize = 2 * 1024;
 const EXTRACT_MAX_EFFECTIVE_PAX_FIELDS: usize = 4096;
 const EXTRACT_MAX_EXPANDED_PAX_METADATA_BYTES: u64 = 128 * 1024 * 1024;
-const EXTRACT_SMALL_MEMBER_MAX_BYTES: usize = 64 * 1024;
-const EXTRACT_DEFAULT_MAX_INFLIGHT: usize = 1;
+const EXTRACT_SMALL_MEMBER_MAX_BYTES: usize = 128 * 1024;
+const EXTRACT_DEFAULT_MAX_INFLIGHT: usize = 16;
 const EXTRACT_BATCH_MAX_MEMBERS: usize = 16;
-const EXTRACT_BATCH_MAX_STAGING_BYTES: usize = 2 * 1024 * 1024;
+const EXTRACT_BATCH_MAX_STAGING_BYTES: usize = 3 * 1024 * 1024;
 const EXTRACT_MEMBER_CONTEXT_OVERHEAD_BYTES: usize = 512;
 const EXTRACT_METADATA_ENTRY_OVERHEAD_BYTES: usize = 64;
 const ENV_RUSTFS_SNOWBALL_EXTRACT_MAX_INFLIGHT: &str = "RUSTFS_SNOWBALL_EXTRACT_MAX_INFLIGHT";
@@ -1864,7 +1864,34 @@ fn resolve_put_object_extract_options(headers: &HeaderMap) -> S3Result<PutObject
 }
 
 fn put_object_extract_limits() -> ArchiveLimits {
-    ArchiveLimits::default()
+    static LIMITS: OnceLock<ArchiveLimits> = OnceLock::new();
+    *LIMITS.get_or_init(|| {
+        normalize_put_object_extract_limits(
+            rustfs_utils::get_env_u64(
+                rustfs_config::ENV_SNOWBALL_MAX_ENTRY_BYTES,
+                rustfs_config::DEFAULT_SNOWBALL_MAX_ENTRY_BYTES,
+            ),
+            rustfs_utils::get_env_u64(
+                rustfs_config::ENV_SNOWBALL_MAX_UNPACKED_BYTES,
+                rustfs_config::DEFAULT_SNOWBALL_MAX_UNPACKED_BYTES,
+            ),
+        )
+    })
+}
+
+fn normalize_put_object_extract_limits(max_entry_bytes: u64, max_unpacked_bytes: u64) -> ArchiveLimits {
+    let defaults = ArchiveLimits::default();
+    let max_total_unpacked_size = max_unpacked_bytes.clamp(1, rustfs_config::MAX_SNOWBALL_UNPACKED_BYTES);
+    let max_entry_size = max_entry_bytes
+        .clamp(1, rustfs_config::MAX_SNOWBALL_ENTRY_BYTES)
+        .min(max_total_unpacked_size);
+
+    ArchiveLimits {
+        max_entry_size,
+        max_total_unpacked_size,
+        max_decoded_size: max_total_unpacked_size.saturating_add(max_entry_size),
+        ..defaults
+    }
 }
 
 fn build_put_object_extract_archive<R>(decoder: R, limits: ArchiveLimits) -> Archive<R>
@@ -2175,12 +2202,13 @@ impl DefaultObjectUsecase {
             .is_some_and(|result| result.uses_durable_reservations);
         // Without ignore-errors, the legacy contract stops before attempting a
         // later member after the first storage failure. Parallel commits cannot
-        // preserve that boundary, so concurrency requires both ignore-errors
-        // and an explicit max-inflight value above the serial default. Quota
-        // accounting can fail after storage commit, so quota-enabled imports
-        // also remain serial. An opted-in micro-batch is always drained; a
-        // fatal outcome stops later batches but cannot roll back peers that
-        // already committed in the current batch.
+        // preserve that boundary, so only ignore-errors requests use the
+        // configured micro-batch. Quota accounting can fail after storage
+        // commit, so quota-enabled imports also remain serial. Setting
+        // RUSTFS_SNOWBALL_EXTRACT_MAX_INFLIGHT=1 restores serial behavior. A
+        // micro-batch is always drained; a fatal outcome stops later batches
+        // but cannot roll back peers that already committed in the current
+        // batch.
         let max_inflight = select_put_object_extract_max_inflight(
             put_object_extract_max_inflight(),
             extract_options.ignore_errors,
@@ -2856,7 +2884,7 @@ mod tests {
 
     #[test]
     fn snowball_max_inflight_has_a_serial_compatibility_floor_and_bounded_ceiling() {
-        assert_eq!(EXTRACT_DEFAULT_MAX_INFLIGHT, 1);
+        assert_eq!(EXTRACT_DEFAULT_MAX_INFLIGHT, EXTRACT_BATCH_MAX_MEMBERS);
         assert_eq!(normalize_put_object_extract_max_inflight(0), 1);
         assert_eq!(normalize_put_object_extract_max_inflight(1), 1);
         assert_eq!(normalize_put_object_extract_max_inflight(usize::MAX), EXTRACT_BATCH_MAX_MEMBERS);
@@ -2877,6 +2905,35 @@ mod tests {
         );
     }
 
+    #[test]
+    fn snowball_archive_limits_preserve_defaults_and_clamp_operator_overrides() {
+        let defaults = ArchiveLimits::default();
+        assert_eq!(
+            normalize_put_object_extract_limits(
+                rustfs_config::DEFAULT_SNOWBALL_MAX_ENTRY_BYTES,
+                rustfs_config::DEFAULT_SNOWBALL_MAX_UNPACKED_BYTES,
+            ),
+            defaults
+        );
+
+        let minimum = normalize_put_object_extract_limits(0, 0);
+        assert_eq!(minimum.max_entry_size, 1);
+        assert_eq!(minimum.max_total_unpacked_size, 1);
+        assert_eq!(minimum.max_decoded_size, 2);
+
+        let bounded = normalize_put_object_extract_limits(u64::MAX, u64::MAX);
+        assert_eq!(bounded.max_entry_size, rustfs_config::MAX_SNOWBALL_ENTRY_BYTES);
+        assert_eq!(bounded.max_total_unpacked_size, rustfs_config::MAX_SNOWBALL_UNPACKED_BYTES);
+        assert_eq!(
+            bounded.max_decoded_size,
+            rustfs_config::MAX_SNOWBALL_UNPACKED_BYTES + rustfs_config::MAX_SNOWBALL_ENTRY_BYTES
+        );
+
+        let entry_is_bounded_by_the_request_total = normalize_put_object_extract_limits(1024, 512);
+        assert_eq!(entry_is_bounded_by_the_request_total.max_entry_size, 512);
+        assert_eq!(entry_is_bounded_by_the_request_total.max_total_unpacked_size, 512);
+    }
+
     #[test]
     fn snowball_batch_state_flushes_on_duplicates_limits_and_serial_barriers() {
         let mut state = ExtractBatchState::default();

rustfs/src/storage/rpc/node_service.rs

@@ -181,10 +181,11 @@ fn remove_heal_control_replay(
 
 static HEAL_CONTROL_REPLAY_CACHE: OnceLock<tokio::sync::Mutex<HashMap<String, Arc<HealControlReplayEntry>>>> = OnceLock::new();
 static NODE_CAPABILITY_SERVER_EPOCH: LazyLock<Uuid> = LazyLock::new(Uuid::new_v4);
-// v3 additionally promises the v6 tier-delete dispatch-manifest policy. The
+// v3 additionally promises the v6 tier-delete dispatch-manifest policy; v4
+// promises the sticky per-target decommission capacity fence. The
 // existing periodic topology probe carries both capabilities so normal object
 // operations do not add another peer RPC.
-const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 3;
+const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 4;
 
 fn admit_heal_control_replay(
     replay_cache: &mut HashMap<String, Arc<HealControlReplayEntry>>,
@@ -3770,7 +3771,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn cross_pool_fence_probe_authenticates_supported_v3_state() {
+    async fn cross_pool_fence_probe_authenticates_supported_v4_state() {
         let _ = rustfs_credentials::set_global_rpc_secret("cross-pool-fence-node-service-test-secret".to_string());
         let endpoints = heal_control_test_endpoints_with_coordinator("node-0", true);
         assert!(

scripts/error-other-format-baseline.txt

@@ -27,7 +27,7 @@
 6|crates/ecstore/src/config/com.rs
 14|crates/ecstore/src/config/storageclass.rs
 182|crates/ecstore/src/core/pools.rs
-8|crates/ecstore/src/data_movement/mod.rs
+7|crates/ecstore/src/data_movement/mod.rs
 2|crates/ecstore/src/data_usage/local_snapshot.rs
 12|crates/ecstore/src/data_usage/mod.rs
 5|crates/ecstore/src/disk/local.rs
@@ -70,5 +70,5 @@
 12|crates/ecstore/src/store/init.rs
 2|crates/ecstore/src/store/init_format.rs
 3|crates/ecstore/src/store/multipart.rs
-7|crates/ecstore/src/store/object.rs
+6|crates/ecstore/src/store/object.rs
 5|crates/ecstore/src/store/rebalance/support.rs