fix(ecstore): reserve decommission capacity safely (#6917)

crates/ecstore/src/bucket/metadata_sys.rs

@@ -412,8 +412,14 @@ pub(crate) fn require_bucket_metadata_sys_in(
 }
 
 pub(crate) async fn object_store_in(ctx: &crate::runtime::instance::InstanceContext) -> Result<Arc<ECStore>> {
-    let sys = bucket_metadata_sys_of(ctx)?;
-    Ok(sys.read().await.api.clone())
+    object_store_if_initialized_in(ctx)
+        .await
+        .ok_or_else(|| Error::other("bucket metadata sys not initialized for this instance"))
+}
+
+pub(crate) async fn object_store_if_initialized_in(ctx: &crate::runtime::instance::InstanceContext) -> Option<Arc<ECStore>> {
+    let sys = ctx.bucket_metadata_sys().or_else(get_global_bucket_metadata_sys)?;
+    Some(sys.read().await.api.clone())
 }
 
 pub(crate) async fn get_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result<Arc<BucketMetadata>> {

crates/ecstore/src/core/sets.rs

@@ -1351,11 +1351,20 @@ pub(crate) async fn make_local_two_set_sets_with_ctx(ctx: Arc<InstanceContext>)
 pub(crate) async fn make_local_two_set_sets_for_pool_with_ctx(
     ctx: Arc<InstanceContext>,
     pool_idx: usize,
+) -> (Vec<tempfile::TempDir>, Arc<Sets>) {
+    make_local_two_set_sets_for_pool_with_drive_count_and_ctx(ctx, pool_idx, 2).await
+}
+
+#[cfg(any(test, feature = "test-util"))]
+pub(crate) async fn make_local_two_set_sets_for_pool_with_drive_count_and_ctx(
+    ctx: Arc<InstanceContext>,
+    pool_idx: usize,
+    set_drive_count: usize,
 ) -> (Vec<tempfile::TempDir>, Arc<Sets>) {
     use crate::layout::endpoint::Endpoint;
     use rustfs_lock::client::local::LocalClient;
 
-    let format = FormatV3::new(2, 2);
+    let format = FormatV3::new(2, set_drive_count);
     let mut temp_dirs = Vec::new();
     let mut all_endpoints = Vec::new();
     let mut disk_sets = Vec::new();
@@ -1363,7 +1372,7 @@ pub(crate) async fn make_local_two_set_sets_for_pool_with_ctx(
     for set_index in 0..2 {
         let mut endpoints = Vec::new();
         let mut disks = Vec::new();
-        for disk_index in 0..2 {
+        for disk_index in 0..set_drive_count {
             let temp_dir = tempfile::tempdir().expect("tempdir should be created");
             let mut endpoint = Endpoint::try_from(temp_dir.path().to_str().expect("tempdir path should be utf8"))
                 .expect("endpoint should parse");
@@ -1389,7 +1398,7 @@ pub(crate) async fn make_local_two_set_sets_for_pool_with_ctx(
             endpoints.push(endpoint);
             disks.push(Some(disk));
         }
-        let lockers = (0..2)
+        let lockers = (0..set_drive_count)
             .map(|_| {
                 Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::Enabled(Arc::new(
                     rustfs_lock::FastObjectLockManager::new(),
@@ -1400,7 +1409,7 @@ pub(crate) async fn make_local_two_set_sets_for_pool_with_ctx(
             SetDisks::new_with_instance_ctx(
                 "test-owner".to_string(),
                 Arc::new(RwLock::new(disks)),
-                2,
+                set_drive_count,
                 1,
                 set_index,
                 pool_idx,
@@ -1420,7 +1429,7 @@ pub(crate) async fn make_local_two_set_sets_for_pool_with_ctx(
         endpoints: PoolEndpoints {
             legacy: false,
             set_count: 2,
-            drives_per_set: 2,
+            drives_per_set: set_drive_count,
             endpoints: Endpoints::from(all_endpoints),
             cmd_line: String::new(),
             platform: String::new(),
@@ -1428,7 +1437,7 @@ pub(crate) async fn make_local_two_set_sets_for_pool_with_ctx(
         format,
         parity_count: 1,
         set_count: 2,
-        set_drive_count: 2,
+        set_drive_count,
         default_parity_count: 1,
         distribution_algo: DistributionAlgoVersion::V1,
         exit_signal: None,

crates/ecstore/src/data_movement/mod.rs

@@ -16,13 +16,14 @@
 
 pub(crate) mod backpressure;
 
+use crate::core::pools::{DecommissionCapacityOwner, decommission_capacity_mutation_id};
 use crate::error::{
     Error, Result, is_err_data_movement_overwrite, is_err_invalid_upload_id, is_err_object_not_found, is_err_version_not_found,
 };
 use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
 use crate::set_disk::{SetDisks, get_lock_acquire_timeout};
 use crate::storage_api_contracts::{
-    multipart::{CompletePart, MultipartOperations as _},
+    multipart::CompletePart,
     namespace::NamespaceLocking as _,
     object::{HTTPPreconditions, ObjectOperations as _},
 };
@@ -160,6 +161,99 @@ pub fn mark_multipart_upload_completed(flag: &Arc<AtomicBool>) {
     flag.store(false, Ordering::Relaxed);
 }
 
+#[cfg(test)]
+struct DataMovementMultipartAbortBarrierState {
+    bucket: String,
+    object: String,
+    arrived: tokio::sync::Notify,
+    release: tokio::sync::Notify,
+}
+
+#[cfg(test)]
+pub(crate) struct DataMovementMultipartAbortBarrier {
+    state: Arc<DataMovementMultipartAbortBarrierState>,
+}
+
+#[cfg(test)]
+static DATA_MOVEMENT_MULTIPART_ABORT_BARRIER: std::sync::OnceLock<
+    std::sync::Mutex<Option<Arc<DataMovementMultipartAbortBarrierState>>>,
+> = std::sync::OnceLock::new();
+
+#[cfg(test)]
+impl DataMovementMultipartAbortBarrier {
+    pub(crate) fn install(bucket: &str, object: &str) -> Self {
+        let state = Arc::new(DataMovementMultipartAbortBarrierState {
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            arrived: tokio::sync::Notify::new(),
+            release: tokio::sync::Notify::new(),
+        });
+        let mut slot = DATA_MOVEMENT_MULTIPART_ABORT_BARRIER
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("data movement multipart abort barrier mutex should not poison");
+        assert!(slot.is_none(), "data movement multipart abort barrier must be unique");
+        *slot = Some(Arc::clone(&state));
+        Self { state }
+    }
+
+    pub(crate) async fn wait_until_paused(&self) {
+        tokio::time::timeout(StdDuration::from_secs(30), self.state.arrived.notified())
+            .await
+            .expect("data movement multipart failure should reach abort cleanup");
+    }
+}
+
+#[cfg(test)]
+impl Drop for DataMovementMultipartAbortBarrier {
+    fn drop(&mut self) {
+        self.state.release.notify_one();
+        let mut slot = DATA_MOVEMENT_MULTIPART_ABORT_BARRIER
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("data movement multipart abort barrier mutex should not poison");
+        if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) {
+            *slot = None;
+        }
+    }
+}
+
+#[cfg(test)]
+async fn pause_data_movement_multipart_before_abort(bucket: &str, object: &str) {
+    let barrier = DATA_MOVEMENT_MULTIPART_ABORT_BARRIER
+        .get_or_init(|| std::sync::Mutex::new(None))
+        .lock()
+        .expect("data movement multipart abort barrier mutex should not poison")
+        .as_ref()
+        .filter(|barrier| barrier.bucket == bucket && barrier.object == object)
+        .cloned();
+    if let Some(barrier) = barrier {
+        barrier.arrived.notify_one();
+        barrier.release.notified().await;
+    }
+}
+
+fn data_movement_abort_opts(
+    src_pool_idx: usize,
+    expected_bucket_incarnation_id: Option<uuid::Uuid>,
+    lock_lost_signal: Option<&Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
+    capacity_owner: Option<DecommissionCapacityOwner>,
+) -> ObjectOptions {
+    let mut opts = ObjectOptions {
+        data_movement: true,
+        src_pool_idx,
+        expected_bucket_incarnation_id,
+        ..Default::default()
+    };
+    if let Some(capacity_owner) = capacity_owner {
+        capacity_owner.apply_to(&mut opts);
+    }
+    if let Some(signal) = lock_lost_signal {
+        opts.add_namespace_lock_lost_signal(Arc::clone(signal));
+    }
+    opts
+}
+
 fn insert_data_movement_checksum(user_defined: &mut HashMap<String, String>, object_info: &ObjectInfo) {
     rustfs_utils::http::remove_header_map(user_defined, rustfs_utils::http::SUFFIX_REPLICATION_SSEC_CRC);
     if let Some(checksum) = object_info.checksum.as_ref().filter(|checksum| !checksum.is_empty()) {
@@ -192,7 +286,7 @@ fn data_movement_new_multipart_opts(object_info: &ObjectInfo, src_pool_idx: usiz
         preserve_etag: object_info.etag.clone(),
         src_pool_idx,
         data_movement: true,
-        ..Default::default()
+        ..ObjectOptions::with_capacity_expected_data_bytes(usize::try_from(object_info.size).ok())
     }
 }
 
@@ -363,7 +457,7 @@ fn data_movement_complete_multipart_opts(
         preserve_etag: object_info.etag.clone(),
         user_defined,
         src_pool_idx,
-        ..Default::default()
+        ..ObjectOptions::with_capacity_expected_data_bytes(usize::try_from(object_info.size).ok())
     })
 }
 
@@ -533,6 +627,7 @@ fn schedule_data_movement_multipart_abort_cleanup(
     bucket: String,
     object: String,
     upload_id: String,
+    opts: ObjectOptions,
     op_label: &str,
 ) {
     let op_label = op_label.to_string();
@@ -540,23 +635,32 @@ fn schedule_data_movement_multipart_abort_cleanup(
         for attempt in 1..=DATA_MOVEMENT_MULTIPART_ABORT_RETRY_ATTEMPTS {
             tokio::time::sleep(StdDuration::from_secs(DATA_MOVEMENT_MULTIPART_ABORT_RETRY_DELAY_SECS)).await;
 
-            let Some(pool) = store.pools.get(target_pool_idx).cloned() else {
+            if store.pools.get(target_pool_idx).is_none() {
                 error!(
                     "{op_label}: background abort_multipart_upload cleanup skipped for {bucket}/{object} upload {upload_id}: target pool {target_pool_idx} is out of range"
                 );
                 return;
+            }
+
+            let mut cleanup_opts = opts.clone();
+            let _multipart_mutation_fence = match DecommissionCapacityOwner::from_options(&cleanup_opts) {
+                Some(owner) => match store.acquire_decommission_multipart_mutation_fence(owner).await {
+                    Ok(fence) => {
+                        fence.add_namespace_lock_fence(&mut cleanup_opts);
+                        Some(fence)
+                    }
+                    Err(err) => {
+                        error!(
+                            "{op_label}: background abort_multipart_upload cleanup could not fence {bucket}/{object} upload {upload_id} on attempt {attempt}: {err:?}"
+                        );
+                        continue;
+                    }
+                },
+                None => None,
             };
 
-            match pool
-                .abort_multipart_upload(
-                    &bucket,
-                    &object,
-                    &upload_id,
-                    &ObjectOptions {
-                        data_movement: true,
-                        ..Default::default()
-                    },
-                )
+            match store
+                .abort_multipart_upload_for_data_movement(target_pool_idx, &bucket, &object, &upload_id, &cleanup_opts)
                 .await
             {
                 Ok(()) => {
@@ -1334,27 +1438,43 @@ fn resolve_data_movement_overwrite_resume_result_for(
     Ok(matches!(err, Error::PreconditionFailed) && is_superseding_unversioned_data_movement_object(source, &target))
 }
 
+#[derive(Clone, Copy)]
+struct DataMovementOverwriteCapacity {
+    owner: Option<DecommissionCapacityOwner>,
+    expected_data_bytes: Option<usize>,
+}
+
 async fn should_treat_data_movement_overwrite_as_complete(
     store: &ECStore,
-    src_pool_idx: usize,
-    target_pool_idx: usize,
+    pool_indices: (usize, usize),
     bucket: &str,
     object_info: &ObjectInfo,
     err: &Error,
     compare_part_checksums: bool,
+    capacity: DataMovementOverwriteCapacity,
 ) -> Result<bool> {
     if !should_check_data_movement_overwrite_resume(err) {
         return Ok(false);
     }
+    let (src_pool_idx, target_pool_idx) = pool_indices;
 
-    resolve_data_movement_overwrite_resume_result_for(
+    let equivalent = resolve_data_movement_overwrite_resume_result_for(
         err,
         find_data_movement_target_info(store, target_pool_idx, bucket, object_info).await,
         object_info,
         src_pool_idx,
         target_pool_idx,
         compare_part_checksums,
-    )
+    )?;
+    if equivalent && let Some(owner) = capacity.owner {
+        let expected_data_bytes = capacity
+            .expected_data_bytes
+            .ok_or_else(|| Error::other("equivalent data-movement target cannot reconcile unknown committed data size"))?;
+        store
+            .reconcile_decommission_capacity_after_equivalent_target(owner, target_pool_idx, expected_data_bytes)
+            .await?;
+    }
+    Ok(equivalent)
 }
 
 fn data_movement_part_stage_error(
@@ -1395,6 +1515,7 @@ pub(crate) async fn migrate_decommission_object(
     rd: GetObjectReader,
     source_bucket_incarnation_id: Option<uuid::Uuid>,
     op_label: &str,
+    capacity_owner: Option<DecommissionCapacityOwner>,
 ) -> Result<()> {
     let source = rd.object_info.clone();
     let _mutation_fence = store
@@ -1415,6 +1536,7 @@ pub(crate) async fn migrate_decommission_object(
         source_bucket_incarnation_id,
         op_label,
         None,
+        capacity_owner,
         Some(&_mutation_fence),
     )
     .await
@@ -1451,6 +1573,7 @@ pub(crate) async fn migrate_object_with_lock_lost_signal(
         op_label,
         lock_lost_signal,
         None,
+        None,
     )
     .await
 }
@@ -1464,22 +1587,98 @@ async fn migrate_object_inner(
     source_bucket_incarnation_id: Option<uuid::Uuid>,
     op_label: &str,
     lock_lost_signal: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
+    capacity_owner: Option<DecommissionCapacityOwner>,
     mutation_fence: Option<&ObjectLockDiagGuard>,
 ) -> Result<()> {
     let object_info = rd.object_info.clone();
+    let capacity_owner = capacity_owner.map(|owner| {
+        let version_id = object_info.version_id.map(|version_id| version_id.to_string());
+        let mutation_id = owner.mutation_id.unwrap_or_else(|| {
+            decommission_capacity_mutation_id(
+                owner,
+                &bucket,
+                &object_info.name,
+                version_id.as_deref(),
+                object_info.delete_marker,
+                object_info.mod_time,
+            )
+        });
+        owner.with_mutation_id(mutation_id)
+    });
     let has_part_checksums = object_info
         .parts
         .iter()
         .any(|part| part.checksums.as_ref().is_some_and(|checksums| !checksums.is_empty()));
 
     let preserve_part_checksums = data_movement_part_checksum_writer_enabled();
+    let capacity_expected_data_bytes = usize::try_from(object_info.size).ok();
 
     if should_use_multipart_data_movement(&object_info, has_part_checksums) {
+        let multipart_mutation_fence = match capacity_owner {
+            Some(owner) => Some(store.acquire_decommission_multipart_mutation_fence(owner).await?),
+            None => None,
+        };
         let mut new_multipart_opts = data_movement_new_multipart_opts(&object_info, pool_idx);
+        if let Some(capacity_owner) = capacity_owner {
+            capacity_owner.apply_to(&mut new_multipart_opts);
+        }
         new_multipart_opts.expected_bucket_incarnation_id = source_bucket_incarnation_id;
         if let Some(signal) = lock_lost_signal.as_ref() {
             new_multipart_opts.add_namespace_lock_lost_signal(Arc::clone(signal));
         }
+        if let Some(fence) = multipart_mutation_fence.as_ref() {
+            fence.add_namespace_lock_fence(&mut new_multipart_opts);
+        }
+        if let Some(owner) = capacity_owner {
+            let existing_target_pool_idx = store
+                .select_data_movement_pool_idx(&bucket, &object_info.name, -1, &new_multipart_opts, false)
+                .await?;
+            if existing_target_pool_idx != pool_idx
+                && let Some(target) =
+                    find_data_movement_target_info(store.as_ref(), existing_target_pool_idx, &bucket, &object_info).await?
+                && is_equivalent_data_movement_object_identity(&object_info, &target, true, preserve_part_checksums)
+            {
+                let expected_data_bytes = capacity_expected_data_bytes
+                    .ok_or_else(|| Error::other("equivalent multipart target cannot reconcile unknown committed data size"))?;
+                store
+                    .reconcile_decommission_capacity_after_equivalent_target(owner, existing_target_pool_idx, expected_data_bytes)
+                    .await?;
+                info!(
+                    "{op_label}: multipart upload restart reconciled equivalent target for {}/{}",
+                    bucket.as_str(),
+                    object_info.name.as_str()
+                );
+                return Ok(());
+            }
+            let mut cleanup_opts =
+                data_movement_abort_opts(pool_idx, source_bucket_incarnation_id, lock_lost_signal.as_ref(), capacity_owner);
+            if let Some(fence) = mutation_fence {
+                fence.add_namespace_lock_fence(&mut cleanup_opts);
+            }
+            if let Some(fence) = multipart_mutation_fence.as_ref() {
+                fence.add_namespace_lock_fence(&mut cleanup_opts);
+            }
+            for target_pool_idx in store.decommission_capacity_cleanup_target_indices(owner).await? {
+                store
+                    .reconcile_multipart_uploads_for_data_movement(
+                        target_pool_idx,
+                        &bucket,
+                        &object_info.name,
+                        &data_movement_upload_identity(&object_info),
+                        &cleanup_opts,
+                    )
+                    .await
+                    .map_err(|err| {
+                        data_movement_stage_error(
+                            op_label,
+                            "reconcile_multipart_upload",
+                            bucket.as_str(),
+                            object_info.name.as_str(),
+                            err,
+                        )
+                    })?;
+            }
+        }
         let (res, target_pool_idx, expected_bucket_incarnation_id) = match store
             .handle_new_multipart_upload_with_pool_idx(&bucket, &object_info.name, &new_multipart_opts, mutation_fence)
             .await
@@ -1532,9 +1731,15 @@ async fn migrate_object_inner(
                     expected_bucket_incarnation_id,
                     ..Default::default()
                 };
+                if let Some(capacity_owner) = capacity_owner {
+                    capacity_owner.apply_to(&mut part_opts);
+                }
                 if let Some(signal) = lock_lost_signal.as_ref() {
                     part_opts.add_namespace_lock_lost_signal(Arc::clone(signal));
                 }
+                if let Some(fence) = multipart_mutation_fence.as_ref() {
+                    fence.add_namespace_lock_fence(&mut part_opts);
+                }
                 let pi = match store
                     .put_object_part_for_data_movement(
                         target_pool_idx,
@@ -1578,10 +1783,16 @@ async fn migrate_object_inner(
                         err,
                     )
                 })?;
+            if let Some(capacity_owner) = capacity_owner {
+                capacity_owner.apply_to(&mut complete_multipart_opts);
+            }
             complete_multipart_opts.expected_bucket_incarnation_id = expected_bucket_incarnation_id;
             if let Some(signal) = lock_lost_signal.as_ref() {
                 complete_multipart_opts.add_namespace_lock_lost_signal(Arc::clone(signal));
             }
+            if let Some(fence) = multipart_mutation_fence.as_ref() {
+                fence.add_namespace_lock_fence(&mut complete_multipart_opts);
+            }
             if let Err(err) = store
                 .clone()
                 .complete_multipart_upload_for_data_movement(
@@ -1596,12 +1807,15 @@ async fn migrate_object_inner(
             {
                 if should_treat_data_movement_overwrite_as_complete(
                     store.as_ref(),
-                    pool_idx,
-                    target_pool_idx,
+                    (pool_idx, target_pool_idx),
                     bucket.as_str(),
                     &object_info,
                     &err,
                     preserve_part_checksums,
+                    DataMovementOverwriteCapacity {
+                        owner: capacity_owner,
+                        expected_data_bytes: capacity_expected_data_bytes,
+                    },
                 )
                 .await?
                 {
@@ -1629,31 +1843,37 @@ 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);
+            if let Some(fence) = mutation_fence {
+                fence.add_namespace_lock_fence(&mut abort_opts);
+            }
+            if let Some(fence) = multipart_mutation_fence.as_ref() {
+                fence.add_namespace_lock_fence(&mut abort_opts);
+            }
             let abort_result = store
-                .abort_multipart_upload_for_data_movement(target_pool_idx, &bucket, &object_info.name, &res.upload_id, &{
-                    let mut opts = ObjectOptions {
-                        data_movement: true,
-                        src_pool_idx: pool_idx,
-                        expected_bucket_incarnation_id,
-                        ..Default::default()
-                    };
-                    if let Some(signal) = lock_lost_signal.as_ref() {
-                        opts.add_namespace_lock_lost_signal(Arc::clone(signal));
-                    }
-                    opts
-                })
+                .abort_multipart_upload_for_data_movement(
+                    target_pool_idx,
+                    &bucket,
+                    &object_info.name,
+                    &res.upload_id,
+                    &abort_opts,
+                )
                 .await;
             match abort_result {
                 Ok(()) => return Ok(()),
                 Err(abort_err) if is_err_invalid_upload_id(&abort_err) => {
                     if should_treat_data_movement_overwrite_as_complete(
                         store.as_ref(),
-                        pool_idx,
-                        target_pool_idx,
+                        (pool_idx, target_pool_idx),
                         bucket.as_str(),
                         &object_info,
                         &abort_err,
                         preserve_part_checksums,
+                        DataMovementOverwriteCapacity {
+                            owner: capacity_owner,
+                            expected_data_bytes: capacity_expected_data_bytes,
+                        },
                     )
                     .await?
                     {
@@ -1683,6 +1903,7 @@ async fn migrate_object_inner(
                         bucket.clone(),
                         object_info.name.clone(),
                         res.upload_id.clone(),
+                        abort_opts,
                         op_label,
                     );
                     return Err(data_movement_stage_error(
@@ -1698,19 +1919,24 @@ async fn migrate_object_inner(
 
         if let Err(primary_err) = multipart_result {
             if should_abort_multipart_upload(&abort_multipart_flag) {
+                #[cfg(test)]
+                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);
+                if let Some(fence) = mutation_fence {
+                    fence.add_namespace_lock_fence(&mut abort_opts);
+                }
+                if let Some(fence) = multipart_mutation_fence.as_ref() {
+                    fence.add_namespace_lock_fence(&mut abort_opts);
+                }
                 return match store
-                    .abort_multipart_upload_for_data_movement(target_pool_idx, &bucket, &object_info.name, &res.upload_id, &{
-                        let mut opts = ObjectOptions {
-                            data_movement: true,
-                            src_pool_idx: pool_idx,
-                            expected_bucket_incarnation_id,
-                            ..Default::default()
-                        };
-                        if let Some(signal) = lock_lost_signal.as_ref() {
-                            opts.add_namespace_lock_lost_signal(Arc::clone(signal));
-                        }
-                        opts
-                    })
+                    .abort_multipart_upload_for_data_movement(
+                        target_pool_idx,
+                        &bucket,
+                        &object_info.name,
+                        &res.upload_id,
+                        &abort_opts,
+                    )
                     .await
                 {
                     Ok(()) => Err(primary_err),
@@ -1722,6 +1948,7 @@ async fn migrate_object_inner(
                             bucket.clone(),
                             object_info.name.clone(),
                             res.upload_id.clone(),
+                            abort_opts,
                             op_label,
                         );
                         Err(resolve_data_movement_abort_result(
@@ -1744,6 +1971,9 @@ async fn migrate_object_inner(
     let mut data = data_movement_put_object_reader(bucket.as_str(), &object_info, rd, op_label)?;
 
     let mut put_opts = data_movement_put_object_opts(&object_info, pool_idx);
+    if let Some(capacity_owner) = capacity_owner {
+        capacity_owner.apply_to(&mut put_opts);
+    }
     put_opts.expected_bucket_incarnation_id = source_bucket_incarnation_id;
     if let Some(signal) = lock_lost_signal {
         put_opts.add_namespace_lock_lost_signal(signal);
@@ -1755,12 +1985,15 @@ async fn migrate_object_inner(
     if let Err(err) = put_result {
         if should_treat_data_movement_overwrite_as_complete(
             store.as_ref(),
-            pool_idx,
-            target_pool_idx,
+            (pool_idx, target_pool_idx),
             bucket.as_str(),
             &object_info,
             &err,
             preserve_part_checksums,
+            DataMovementOverwriteCapacity {
+                owner: capacity_owner,
+                expected_data_bytes: capacity_expected_data_bytes,
+            },
         )
         .await?
         {

crates/ecstore/src/error/mod.rs

@@ -183,6 +183,10 @@ pub enum StorageError {
     DecommissionNotStarted,
     #[error("Decommission already running")]
     DecommissionAlreadyRunning,
+    #[error("Decommission capacity error: {0}")]
+    DecommissionCapacity(String),
+    #[error("decommission_capacity_blocked: Storage reached its minimum free drive threshold.: {message}")]
+    DecommissionCapacityBlocked { message: String },
     #[error("Rebalance already running")]
     RebalanceAlreadyRunning,
     #[error("{operation}: stale pool metadata update rejected for pool {pool_index}; {reason}")]
@@ -569,6 +573,10 @@ impl Clone for StorageError {
             StorageError::EntityTooLarge(a, b) => StorageError::EntityTooLarge(*a, *b),
             StorageError::DoneForNow => StorageError::DoneForNow,
             StorageError::DecommissionAlreadyRunning => StorageError::DecommissionAlreadyRunning,
+            StorageError::DecommissionCapacity(message) => StorageError::DecommissionCapacity(message.clone()),
+            StorageError::DecommissionCapacityBlocked { message } => StorageError::DecommissionCapacityBlocked {
+                message: message.clone(),
+            },
             StorageError::RebalanceAlreadyRunning => StorageError::RebalanceAlreadyRunning,
             StorageError::StalePoolMetadataUpdate {
                 operation,
@@ -681,6 +689,8 @@ impl StorageError {
             StorageError::InvalidPart(_, _, _) => StorageErrorCode::InvalidPart,
             StorageError::DoneForNow => StorageErrorCode::DoneForNow,
             StorageError::DecommissionAlreadyRunning => StorageErrorCode::DecommissionAlreadyRunning,
+            StorageError::DecommissionCapacity(_) => StorageErrorCode::InvalidArgument,
+            StorageError::DecommissionCapacityBlocked { .. } => StorageErrorCode::StorageFull,
             StorageError::RebalanceAlreadyRunning => StorageErrorCode::RebalanceAlreadyRunning,
             StorageError::StalePoolMetadataUpdate { .. } => StorageErrorCode::InvalidArgument,
             StorageError::OperationCanceled => StorageErrorCode::OperationCanceled,

crates/ecstore/src/object_api/types.rs

@@ -681,6 +681,16 @@ impl Drop for ScannerPublicationCommitScopeInner {
     }
 }
 
+#[derive(Clone, Default)]
+#[doc(hidden)]
+pub struct DecommissionCapacityOptions {
+    pub(crate) expected_data_bytes: Option<usize>,
+    pub(crate) operation_id: Option<Uuid>,
+    pub(crate) generation: Option<u64>,
+    pub(crate) owner_nonce: Option<Uuid>,
+    pub(crate) mutation_id: Option<Uuid>,
+}
+
 #[derive(Default, Clone)]
 pub struct ObjectOptions {
     // Use the maximum parity (N/2), used when saving server configuration files
@@ -725,6 +735,12 @@ pub struct ObjectOptions {
 
     pub data_movement: bool,
     pub raw_data_movement_read: bool,
+    /// Durable reservation identity carried only by decommission writes. Other
+    /// data-movement users, including rebalance, leave it unset. Keep this
+    /// context boxed because `ObjectOptions` is passed by value through deep
+    /// storage futures.
+    #[doc(hidden)]
+    pub decommission_capacity: Option<Box<DecommissionCapacityOptions>>,
     /// Materialize the data-movement per-part checksum sidecar for APIs that
     /// return part checksums. Ordinary object reads leave it encoded.
     pub include_part_checksums: bool,
@@ -788,6 +804,36 @@ pub struct ObjectOptions {
     /// Storage-owned journal writer used by the atomic delete path. This is
     /// populated only by the `ECStore` wrapper that holds the namespace locks.
     pub tier_delete_journal_api: Option<Arc<crate::store::ECStore>>,
+    /// Internal staged-mutation admission supplied by `ECStore`; each local
+    /// publish is fenced namespace-first and then by decommission capacity.
+    #[doc(hidden)]
+    pub decommission_capacity_admission: Option<Arc<crate::store::ECStore>>,
+}
+
+impl ObjectOptions {
+    pub(crate) fn with_capacity_expected_data_bytes(expected_data_bytes: Option<usize>) -> Self {
+        Self {
+            decommission_capacity: expected_data_bytes.map(|expected_data_bytes| {
+                Box::new(DecommissionCapacityOptions {
+                    expected_data_bytes: Some(expected_data_bytes),
+                    ..Default::default()
+                })
+            }),
+            ..Default::default()
+        }
+    }
+
+    pub(crate) fn capacity_expected_data_bytes(&self) -> Option<usize> {
+        self.decommission_capacity
+            .as_deref()
+            .and_then(|capacity| capacity.expected_data_bytes)
+    }
+
+    pub(crate) fn has_decommission_capacity_reservation(&self) -> bool {
+        self.decommission_capacity
+            .as_deref()
+            .is_some_and(|capacity| capacity.operation_id.is_some())
+    }
 }
 
 impl std::fmt::Debug for ObjectOptions {

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

@@ -1336,7 +1336,11 @@ mod tests {
     use crate::set_disk::{PutObjectCommitBarrier, PutObjectCommitPause, hermetic_set_disks_isolated};
 
     async fn persist_initialized_identity_then_remove_pool_meta(store: &Arc<ECStore>) {
-        let mut write_state = PoolMetaWriteState::for_startup(store.id, false);
+        let deployment_id = store
+            .ctx
+            .deployment_id()
+            .expect("test store should have a deployment identity");
+        let mut write_state = PoolMetaWriteState::for_startup(deployment_id, false);
         persist_pool_meta_identity_for_startup(store.pools.clone(), &mut write_state, true)
             .await
             .expect("initialized pool metadata identity should persist");

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

@@ -97,7 +97,7 @@ pub(crate) async fn test_two_pool_stores(
     std::sync::Arc<crate::store::ECStore>,
     std::sync::Arc<crate::store::ECStore>,
 ) {
-    test_two_pool_stores_with_contexts(rebalance_meta, false).await
+    test_pool_stores_with_contexts(rebalance_meta, false, 2, 2).await
 }
 
 #[cfg(test)]
@@ -108,7 +108,7 @@ pub(crate) async fn test_two_pool_stores_with_isolated_node_contexts(
     std::sync::Arc<crate::store::ECStore>,
     std::sync::Arc<crate::store::ECStore>,
 ) {
-    test_two_pool_stores_with_contexts(rebalance_meta, true).await
+    test_pool_stores_with_contexts(rebalance_meta, true, 2, 2).await
 }
 
 #[cfg(test)]
@@ -123,26 +123,58 @@ pub(crate) async fn promote_test_pool_meta_to_v2(store: &std::sync::Arc<crate::s
 }
 
 #[cfg(test)]
-async fn test_two_pool_stores_with_contexts(
+pub(crate) async fn test_three_pool_stores_with_isolated_node_contexts(
+    rebalance_meta: Option<RebalanceMeta>,
+) -> (
+    Vec<tempfile::TempDir>,
+    std::sync::Arc<crate::store::ECStore>,
+    std::sync::Arc<crate::store::ECStore>,
+) {
+    test_pool_stores_with_contexts(rebalance_meta, true, 3, 2).await
+}
+
+#[cfg(test)]
+pub(crate) async fn test_three_pool_stores_with_three_disk_sets_with_isolated_node_contexts(
+    rebalance_meta: Option<RebalanceMeta>,
+) -> (
+    Vec<tempfile::TempDir>,
+    std::sync::Arc<crate::store::ECStore>,
+    std::sync::Arc<crate::store::ECStore>,
+) {
+    test_pool_stores_with_contexts(rebalance_meta, true, 3, 3).await
+}
+
+#[cfg(test)]
+async fn test_pool_stores_with_contexts(
     rebalance_meta: Option<RebalanceMeta>,
     isolate_node_contexts: bool,
+    pool_count: usize,
+    set_drive_count: usize,
 ) -> (
     Vec<tempfile::TempDir>,
     std::sync::Arc<crate::store::ECStore>,
     std::sync::Arc<crate::store::ECStore>,
 ) {
     crate::services::notification_sys::install_cross_pool_fence_fleet_proof_for_test();
-    use crate::core::pools::PoolMeta;
+    use crate::core::pools::{POOL_META_VERSION, PoolMeta, PoolMetaWriteState, persist_pool_meta_identity_for_startup};
     use crate::layout::endpoints::{EndpointServerPools, SetupType};
 
     let ctx = std::sync::Arc::new(crate::runtime::instance::InstanceContext::new());
     ctx.update_erasure_type(SetupType::DistErasure).await;
-    let (mut temp_dirs, first_pool) =
-        crate::core::sets::make_local_two_set_sets_for_pool_with_ctx(std::sync::Arc::clone(&ctx), 0).await;
-    let (second_temp_dirs, second_pool) =
-        crate::core::sets::make_local_two_set_sets_for_pool_with_ctx(std::sync::Arc::clone(&ctx), 1).await;
-    temp_dirs.extend(second_temp_dirs);
-    let pools = vec![first_pool, second_pool];
+    let deployment_id = uuid::Uuid::new_v4();
+    ctx.set_deployment_id(deployment_id);
+    let mut temp_dirs = Vec::new();
+    let mut pools = Vec::with_capacity(pool_count);
+    for pool_index in 0..pool_count {
+        let (pool_temp_dirs, pool) = crate::core::sets::make_local_two_set_sets_for_pool_with_drive_count_and_ctx(
+            std::sync::Arc::clone(&ctx),
+            pool_index,
+            set_drive_count,
+        )
+        .await;
+        temp_dirs.extend(pool_temp_dirs);
+        pools.push(pool);
+    }
     {
         let local_disk_map = ctx.local_disk_map();
         let mut local_disk_map = local_disk_map.write().await;
@@ -154,11 +186,24 @@ async fn test_two_pool_stores_with_contexts(
             }
         }
     }
-    let pool_meta = PoolMeta::new(&pools, &PoolMeta::default());
+    let mut pool_meta = PoolMeta::new(&pools, &PoolMeta::default());
+    pool_meta.version = POOL_META_VERSION;
     pool_meta
         .save_for_startup(pools.clone())
         .await
         .expect("baseline pool metadata should be persisted");
+    let mut pool_meta_write_state = PoolMetaWriteState::for_startup(deployment_id, true);
+    persist_pool_meta_identity_for_startup(pools.clone(), &mut pool_meta_write_state, false)
+        .await
+        .expect("pending pool metadata identity should be persisted");
+    let replica_state = pool_meta
+        .load_no_lock_from_replicas_observing(pools.clone(), &mut pool_meta_write_state)
+        .await
+        .expect("baseline pool metadata should remain readable");
+    pool_meta_write_state.observe_replicas(replica_state);
+    persist_pool_meta_identity_for_startup(pools.clone(), &mut pool_meta_write_state, true)
+        .await
+        .expect("initialized pool metadata identity should be persisted");
     if let Some(meta) = rebalance_meta.as_ref() {
         meta.save(pools[0].clone())
             .await
@@ -169,6 +214,7 @@ async fn test_two_pool_stores_with_contexts(
     let other_ctx = if isolate_node_contexts {
         let other_ctx = std::sync::Arc::new(crate::runtime::instance::InstanceContext::new());
         other_ctx.update_erasure_type(SetupType::DistErasure).await;
+        other_ctx.set_deployment_id(deployment_id);
         *other_ctx.local_disk_map().write().await = ctx.local_disk_map().read().await.clone();
         other_ctx.set_endpoints(endpoint_pools.clone());
         other_ctx
@@ -183,9 +229,9 @@ async fn test_two_pool_stores_with_contexts(
             peer_sys: crate::cluster::rpc::S3PeerSys::new_with_instance_ctx(&endpoint_pools, std::sync::Arc::clone(&store_ctx)),
             pool_meta: tokio::sync::RwLock::new(pool_meta.clone()),
             rebalance_meta: tokio::sync::RwLock::new(rebalance_meta.clone()),
-            decommission_cancelers: tokio::sync::RwLock::new(vec![None, None]),
+            decommission_cancelers: tokio::sync::RwLock::new(vec![None; pool_count]),
             start_gate: tokio::sync::Mutex::new(()),
-            pool_meta_save_gate: tokio::sync::Mutex::default(),
+            pool_meta_save_gate: tokio::sync::Mutex::new(pool_meta_write_state.independent_clone_for_test()),
             ctx: store_ctx,
             bucket_fence_registry: std::sync::Arc::default(),
         })
@@ -195,6 +241,8 @@ async fn test_two_pool_stores_with_contexts(
     if isolate_node_contexts {
         crate::bucket::metadata_sys::init_bucket_metadata_sys(std::sync::Arc::clone(&store), Vec::new()).await;
         crate::bucket::metadata_sys::init_bucket_metadata_sys(std::sync::Arc::clone(&other_store), Vec::new()).await;
+    } else {
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(std::sync::Arc::clone(&store), Vec::new()).await;
     }
     (temp_dirs, store, other_store)
 }

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -475,7 +475,7 @@ const ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE: &str = "RUSTFS_GET_METADATA
 const ENV_RUSTFS_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS: &str = "RUSTFS_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS";
 const DEFAULT_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS: u64 = 200;
 const METRIC_GET_METADATA_READ_VERSION_COALESCER_TOTAL: &str = "rustfs_get_metadata_read_version_coalescer_total";
-pub(in crate::set_disk) const ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE: &str = "RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE";
+pub(crate) const ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE: &str = "RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE";
 /// Default reader-setup strategy for the GET read path (rustfs/backlog#1215,
 /// #1159, #923).
 ///
@@ -6849,7 +6849,7 @@ pub(in crate::set_disk) mod rename_fanout_barrier_phase {
 /// Cross-process/black-box fault injection (toxiproxy, blackhole peers, 2-pool)
 /// is a later cluster-harness block, not this one.
 #[cfg(test)]
-pub(in crate::set_disk) mod rename_fanout_barrier {
+pub(crate) mod rename_fanout_barrier {
     use std::collections::HashMap;
     use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
     use std::sync::{Arc, Mutex, OnceLock};

crates/ecstore/src/set_disk/mod.rs

@@ -857,7 +857,7 @@ static OBJECT_LOCK_DIAG_ENABLED: OnceLock<bool> = OnceLock::new();
 
 mod core;
 #[cfg(test)]
-pub(crate) use core::io_primitives::disk_call_counters;
+pub(crate) use core::io_primitives::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, disk_call_counters, rename_fanout_barrier};
 mod ctx;
 mod metadata;
 mod ops;

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

@@ -27,18 +27,18 @@ use super::super::MetadataCacheInvalidationProbe;
 #[cfg(test)]
 use super::super::capacity_scope_from_disks;
 use super::super::{
-    AMZ_STORAGE_CLASS, Arc, Bytes, CompletePart, Cursor, DiskError, DiskStore, EVENT_SET_DISK_MULTIPART, Error, FileInfo,
-    GLOBAL_MIN_PART_SIZE, HashAlgorithm, HashMap, HashReader, HashSet, HealChannelPriority, Instant, LOG_COMPONENT_ECSTORE,
-    LOG_SUBSYSTEM_SET_DISK, ListMultipartsInfo, ListPartsInfo, MAX_PARTS_COUNT, MULTIPART_WRITE_QUORUM_RENAME_PART,
-    MULTIPART_WRITE_QUORUM_UPLOAD_METADATA, MULTIPART_WRITE_QUORUM_WRITER_SETUP, MultipartInfo, MultipartUploadResult,
-    MultipartWriteQuorumContext, NamespaceLockFence, OBJECT_OP_IGNORED_ERRS, ObjectInfo, ObjectLockDiagGuard, ObjectOptions,
-    ObjectPartInfo, OffsetDateTime, PartInfo, PutObjReader, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET,
-    RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY, Result, SLASH_SEPARATOR, SUFFIX_ACTUAL_OBJECT_SIZE_CAP,
-    SUFFIX_ACTUAL_SIZE, SUFFIX_BUCKET_INCARNATION_ID, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_SSEC_CRC,
-    SUFFIX_RESTORE_OPERATION_ID, SetDisks, SmallWritePath, StorageError, Uuid, WriteLayout,
-    check_object_lock_for_deletion_with_state, classify_multipart_part_write_path, coding, complete_multipart_part_error,
-    complete_multipart_part_error_result, complete_part_checksum, completed_multipart_object_part, contains_key_str,
-    create_bitrot_writer, debug, disk, error, get_complete_multipart_md5, get_header_map, get_str, insert_str,
+    AMZ_STORAGE_CLASS, Arc, Bytes, CompletePart, Cursor, DATA_MOVEMENT_MULTIPART_PREFIX, DiskError, DiskStore,
+    EVENT_SET_DISK_MULTIPART, Error, FileInfo, GLOBAL_MIN_PART_SIZE, HashAlgorithm, HashMap, HashReader, HashSet,
+    HealChannelPriority, Instant, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_SET_DISK, ListMultipartsInfo, ListPartsInfo,
+    MAX_PARTS_COUNT, MULTIPART_WRITE_QUORUM_RENAME_PART, MULTIPART_WRITE_QUORUM_UPLOAD_METADATA,
+    MULTIPART_WRITE_QUORUM_WRITER_SETUP, MultipartInfo, MultipartUploadResult, MultipartWriteQuorumContext, NamespaceLockFence,
+    OBJECT_OP_IGNORED_ERRS, ObjectInfo, ObjectLockDiagGuard, ObjectOptions, ObjectPartInfo, OffsetDateTime, PartInfo,
+    PutObjReader, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET, RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY,
+    Result, SLASH_SEPARATOR, SUFFIX_ACTUAL_OBJECT_SIZE_CAP, SUFFIX_ACTUAL_SIZE, SUFFIX_BUCKET_INCARNATION_ID,
+    SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_RESTORE_OPERATION_ID, SetDisks, SmallWritePath, StorageError,
+    Uuid, WriteLayout, check_object_lock_for_deletion_with_state, classify_multipart_part_write_path, coding,
+    complete_multipart_part_error, complete_multipart_part_error_result, complete_part_checksum, completed_multipart_object_part,
+    contains_key_str, create_bitrot_writer, debug, disk, error, get_complete_multipart_md5, get_header_map, get_str, insert_str,
     is_err_object_not_found, is_err_version_not_found, is_min_allowed_part_size, log_multipart_write_quorum_failure,
     parts_after_marker, path_join_buf, record_compression_total_memory, reduce_read_quorum_errs, reduce_write_quorum_errs,
     remove_header_map, resolve_write_layout, restore_commit_operation_id_from_metadata, should_persist_encryption_original_size,
@@ -183,6 +183,45 @@ pub(crate) struct StaleMultipartCleanupGuard {
     lock_guard: ObjectLockDiagGuard,
 }
 
+pub(crate) struct DataMovementMultipartAbortGuard {
+    upload_path: String,
+    write_quorum: Option<usize>,
+    lock_guard: ObjectLockDiagGuard,
+}
+
+impl DataMovementMultipartAbortGuard {
+    pub(crate) fn add_namespace_lock_fence(&self, opts: &mut ObjectOptions) {
+        opts.add_namespace_lock_guard(&self.lock_guard.guard);
+    }
+
+    pub(crate) async fn delete(&self, set: &SetDisks, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> {
+        #[cfg(any(test, feature = "test-util"))]
+        pause_multipart_commit(bucket, object, MultipartCommitPause::AbortBeforeDelete).await;
+        fence_commit_on_lock_loss(Some(&self.lock_guard), "abort_multipart_upload_commit", &self.upload_path)?;
+        if opts
+            .namespace_lock_fence
+            .as_ref()
+            .is_some_and(NamespaceLockFence::is_lock_lost)
+        {
+            return Err(StorageError::NamespaceLockQuorumUnavailable {
+                mode: "abort_multipart_upload_outer_lock",
+                bucket: bucket.to_string(),
+                object: object.to_string(),
+                required: 1,
+                achieved: 0,
+            });
+        }
+        ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
+        if let Some(write_quorum) = self.write_quorum {
+            set.delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &self.upload_path, write_quorum)
+                .await?;
+            #[cfg(any(test, feature = "test-util"))]
+            pause_multipart_commit(bucket, object, MultipartCommitPause::AbortAfterDelete).await;
+        }
+        Ok(())
+    }
+}
+
 impl StaleMultipartCleanupGuard {
     pub(crate) fn file_info(&self) -> &FileInfo {
         &self.file_info
@@ -210,7 +249,10 @@ pub enum MultipartCommitPause {
     NewUploadBeforeLockLost,
     PutPartBeforeLockAcquire,
     PutPartBeforeLockLost,
+    PutPartAfterCapacityAdmission,
     PutPartAfterRename,
+    AbortBeforeDelete,
+    AbortAfterDelete,
     BeforeLockLost,
     BeforeQuotaRename,
     BeforeTransactionEpochVerify,
@@ -673,12 +715,28 @@ fn is_corrupt_upload_metadata_error(err: &DiskError) -> bool {
     )
 }
 
-async fn multipart_upload_paths_on_disk(disk: DiskStore, bucket: &str) -> disk::error::Result<Vec<String>> {
+async fn multipart_upload_paths_on_disk(disk: DiskStore, bucket: &str, root_prefix: &str) -> disk::error::Result<Vec<String>> {
     if !disk.is_online().await {
         return Err(DiskError::DiskNotFound);
     }
 
-    let sha_dirs = match disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, "", -1).await {
+    if !root_prefix.is_empty() {
+        let upload_dirs = match disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, root_prefix, -1).await {
+            Ok(entries) => entries,
+            Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => return Ok(Vec::new()),
+            Err(err) => return Err(err),
+        };
+        return Ok(upload_dirs
+            .into_iter()
+            .filter_map(|upload_dir| {
+                let upload_dir = upload_dir.trim_end_matches('/');
+                (!upload_dir.is_empty() && upload_dir != "." && upload_dir != ".." && !upload_dir.contains(['/', '\\']))
+                    .then(|| format!("{root_prefix}/{upload_dir}"))
+            })
+            .collect());
+    }
+
+    let sha_dirs = match disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, root_prefix, -1).await {
         Ok(entries) => entries,
         Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => return Ok(Vec::new()),
         Err(err) => return Err(err),
@@ -778,6 +836,7 @@ impl SetDisks {
         &self,
         orig_bucket: &str,
         error_path: &str,
+        root_prefix: &str,
     ) -> Result<(Vec<Option<DiskStore>>, Vec<String>, usize)> {
         let disks = self.disks.read().await.clone();
         if disks.is_empty() {
@@ -794,9 +853,10 @@ impl SetDisks {
         for (index, disk) in disks.iter().enumerate() {
             let disk = disk.clone();
             let orig_bucket = orig_bucket.to_string();
+            let root_prefix = root_prefix.to_string();
             discovery_tasks.spawn(async move {
                 let result = match disk {
-                    Some(disk) => multipart_upload_paths_on_disk(disk, &orig_bucket).await,
+                    Some(disk) => multipart_upload_paths_on_disk(disk, &orig_bucket, &root_prefix).await,
                     None => Err(DiskError::DiskNotFound),
                 };
                 (index, result)
@@ -832,11 +892,64 @@ impl SetDisks {
 
     pub(crate) async fn first_multipart_upload_path_for_decommission(&self, bucket: &str) -> Result<Option<String>> {
         let (_, paths, _) = self
-            .discover_multipart_upload_paths(bucket, RUSTFS_META_MULTIPART_BUCKET)
+            .discover_multipart_upload_paths(bucket, RUSTFS_META_MULTIPART_BUCKET, "")
             .await?;
         Ok(paths.into_iter().next())
     }
 
+    pub(crate) async fn data_movement_multipart_upload_ids(
+        &self,
+        bucket: &str,
+        object: &str,
+        expected_incarnation_id: Option<Uuid>,
+        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 mut upload_ids = Vec::new();
+        for upload_path in candidate_paths {
+            let Some((parent, raw_upload_id)) = upload_path.rsplit_once('/') else {
+                continue;
+            };
+            if parent != expected_parent || raw_upload_id.is_empty() {
+                continue;
+            }
+            let upload_id = runtime_sources::deployment_upload_id(raw_upload_id);
+            let file_info = match self
+                .check_multipart_upload_path_exists(bucket, object, &upload_id, &upload_path, false)
+                .await
+            {
+                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;
+                }
+                Err(err) => return Err(err),
+            };
+            if file_info.metadata.get(RUSTFS_MULTIPART_BUCKET_KEY).map(String::as_str) != Some(bucket)
+                || file_info.metadata.get(RUSTFS_MULTIPART_OBJECT_KEY).map(String::as_str) != Some(object)
+            {
+                return Err(Error::other("data movement multipart upload target metadata is inconsistent"));
+            }
+            if expected_incarnation_id
+                .is_some_and(|expected| !multipart_bucket_incarnation_matches(&file_info.metadata, expected))
+            {
+                return Err(Error::other("data movement multipart upload bucket incarnation is inconsistent"));
+            }
+            let Some(actual_upload_identity) =
+                rustfs_utils::http::get_consistent_str(&file_info.metadata, rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD)
+            else {
+                return Err(Error::other("data movement multipart upload identity is inconsistent"));
+            };
+            if actual_upload_identity != upload_identity {
+                continue;
+            }
+            upload_ids.push(upload_id);
+        }
+        upload_ids.sort_unstable();
+        upload_ids.dedup();
+        Ok(upload_ids)
+    }
+
     async fn acquire_multipart_upload_read_lock(
         &self,
         op: &'static str,
@@ -873,6 +986,55 @@ impl SetDisks {
             .map(Some)
     }
 
+    pub(crate) async fn lock_data_movement_multipart_abort(
+        &self,
+        bucket: &str,
+        object: &str,
+        upload_id: &str,
+        expected_upload_identity: Option<&str>,
+        opts: &ObjectOptions,
+    ) -> Result<Option<DataMovementMultipartAbortGuard>> {
+        let upload_path = Self::get_multipart_upload_dir(bucket, object, upload_id, true);
+        let lock_guard = self
+            .acquire_write_lock_diag("abort_data_movement_multipart", RUSTFS_META_MULTIPART_BUCKET, &upload_path)
+            .await?;
+        let file_info = match self
+            .check_multipart_upload_path_exists(bucket, object, upload_id, &upload_path, true)
+            .await
+        {
+            Ok((file_info, _)) => file_info,
+            Err(err) if crate::error::is_err_invalid_upload_id(&err) || crate::error::is_err_object_not_found(&err) => {
+                return Ok(Some(DataMovementMultipartAbortGuard {
+                    upload_path,
+                    write_quorum: None,
+                    lock_guard,
+                }));
+            }
+            Err(err) => return Err(err),
+        };
+        ensure_data_movement_upload_access(&file_info, bucket, object, upload_id, opts)?;
+        let upload_identity =
+            rustfs_utils::http::get_consistent_str(&file_info.metadata, rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD);
+        if upload_identity.is_none() || expected_upload_identity.is_some_and(|expected| upload_identity != Some(expected)) {
+            return Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()));
+        }
+        ensure_multipart_bucket_incarnation(
+            &self.ctx,
+            &file_info,
+            bucket,
+            object,
+            upload_id,
+            opts.expected_bucket_incarnation_id,
+        )
+        .await?;
+        ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
+        Ok(Some(DataMovementMultipartAbortGuard {
+            upload_path,
+            write_quorum: Some(file_info.write_quorum(self.default_write_quorum())),
+            lock_guard,
+        }))
+    }
+
     pub(super) async fn list_parts(
         disks: &[Option<DiskStore>],
         part_path: &str,
@@ -1028,7 +1190,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;
@@ -1624,7 +1786,27 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 admitted_multipart_size(current_size, candidate_size, limit)?;
             }
 
-            let _ = self
+            let decommission_capacity_guard = if let Some(store) = opts.decommission_capacity_admission.as_ref() {
+                Some(
+                    store
+                        .acquire_external_decommission_capacity_fence(&[self.pool_index], "mutation")
+                        .await?,
+                )
+            } else {
+                None
+            };
+            #[cfg(test)]
+            pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterCapacityAdmission).await;
+            if decommission_capacity_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
+                return Err(StorageError::NamespaceLockQuorumUnavailable {
+                    mode: "put_object_part_decommission_capacity",
+                    bucket: bucket.to_string(),
+                    object: object.to_string(),
+                    required: 1,
+                    achieved: 0,
+                });
+            }
+            let rename_result = self
                 .rename_part(
                     &shuffle_disks,
                     RUSTFS_META_TMP_BUCKET,
@@ -1641,7 +1823,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                         part_number: Some(part_id),
                     }),
                 )
-                .await?;
+                .await;
+            drop(decommission_capacity_guard);
+            let _ = rename_result?;
             #[cfg(test)]
             observe_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
 
@@ -2112,6 +2296,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         let upload_id_path = Self::get_multipart_upload_dir(bucket, object, upload_id, opts.data_movement);
         let range_seek_rollout_enabled = crate::object_api::legacy_encrypted_range_seek_enabled() && !opts.no_lock;
         let mut object_lock_guard = None;
+        let mut decommission_object_lock_guard = None;
+        let mut decommission_target_lock_covered = false;
+        let mut decommission_capacity_guard = None;
 
         if opts.http_preconditions.is_some() {
             if !opts.no_lock {
@@ -2126,7 +2313,25 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             }
         }
 
-        if !opts.no_lock && object_lock_guard.is_none() {
+        if let Some(store) = opts.decommission_capacity_admission.as_ref() {
+            #[cfg(test)]
+            {
+                crate::core::pools::notify_decommission_external_object_commit_phase_started(store.id);
+                crate::core::pools::wait_for_decommission_external_object_commit_phase_release(store.id).await;
+            }
+            let (object_guard, target_lock_covered, capacity_guard) = store
+                .acquire_external_decommission_commit_guards(
+                    self.pool_index,
+                    bucket,
+                    object,
+                    opts.no_lock || object_lock_guard.is_some(),
+                )
+                .await?;
+            decommission_object_lock_guard = object_guard;
+            decommission_target_lock_covered = target_lock_covered;
+            decommission_capacity_guard = capacity_guard;
+        }
+        if !opts.no_lock && object_lock_guard.is_none() && !decommission_target_lock_covered {
             object_lock_guard = Some(
                 self.acquire_write_lock_diag("complete_multipart_upload_commit", bucket, object)
                     .await?,
@@ -2728,7 +2933,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         // so a lost lock leaves the upload intact and retryable.
         #[cfg(test)]
         pause_multipart_commit(bucket, object, MultipartCommitPause::BeforeLockLost).await;
-        if object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
+        if object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+            || decommission_object_lock_guard
+                .as_ref()
+                .is_some_and(|guard| guard.is_lock_lost())
+        {
             return Err(StorageError::NamespaceLockQuorumUnavailable {
                 mode: "complete_multipart_upload_commit",
                 bucket: bucket.to_string(),
@@ -2805,7 +3014,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 Err(err) => return Err(err),
             }
         }
-        if object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
+        if object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+            || decommission_object_lock_guard
+                .as_ref()
+                .is_some_and(|guard| guard.is_lock_lost())
+        {
             return Err(StorageError::NamespaceLockQuorumUnavailable {
                 mode: "complete_multipart_upload_commit",
                 bucket: bucket.to_string(),
@@ -2838,6 +3051,19 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         }
         ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
+        // Complete has already acquired the object and upload-id namespaces.
+        // Recheck decommission capacity only after those locks, and retain the
+        // guard through the durable rename below.
+        if decommission_capacity_guard.is_none()
+            && let Some(store) = opts.decommission_capacity_admission.as_ref()
+        {
+            decommission_capacity_guard = Some(
+                store
+                    .acquire_external_decommission_capacity_fence(&[self.pool_index], "mutation")
+                    .await?,
+            );
+        }
+
         let transaction_fencing_proof = object_transaction_fencing_fleet_proof();
         if object_transaction_fencing_requested() && transaction_fencing_proof.is_none() {
             return Err(Error::other("object transaction fencing requires a live fleet capability proof"));
@@ -2896,11 +3122,16 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         let commit_is_versioned = opts.versioned || opts.version_suspended;
         let commit_capacity_scope_token = opts.capacity_scope_token;
         let commit_object_lock_guard = object_lock_guard.take();
-        let commit_allows_early_ack = commit_object_lock_guard.is_some();
+        let commit_decommission_object_lock_guard = decommission_object_lock_guard.take();
+        let commit_decommission_capacity_guard = decommission_capacity_guard.take();
+        let commit_allows_early_ack = !(opts.data_movement && opts.has_decommission_capacity_reservation())
+            && (commit_object_lock_guard.is_some() || commit_decommission_object_lock_guard.is_some());
         let detach_commit_owner = commit_allows_early_ack || upload_guard.is_some() || quota_mutation_fence;
         let commit = async move {
             let mut _object_lock_guard = commit_object_lock_guard;
+            let mut _decommission_object_lock_guard = commit_decommission_object_lock_guard;
             let mut _upload_guard = upload_guard;
+            let mut _decommission_capacity_guard = commit_decommission_capacity_guard;
             let mut quota_reservation = quota_reservation;
             let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
 
@@ -2919,6 +3150,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 if quota_reservation.is_lock_lost()
                     || !quota_reservation.capability_proof_matches()
                     || _object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                    || _decommission_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())
                     || commit_namespace_lock_fence
                         .as_ref()
@@ -2926,6 +3160,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                     || commit_bucket_lifecycle_lock_fence
                         .as_ref()
                         .is_some_and(NamespaceLockFence::is_lock_lost)
+                    || _decommission_capacity_guard
+                        .as_ref()
+                        .is_some_and(|guard| guard.is_lock_lost())
                 {
                     return Err(StorageError::NamespaceLockQuorumUnavailable {
                         mode: "quota_reservation",
@@ -2967,6 +3204,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 if quota_reservation.is_lock_lost()
                     || !quota_reservation.capability_proof_matches()
                     || _object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                    || _decommission_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())
                     || commit_namespace_lock_fence
                         .as_ref()
@@ -2974,6 +3214,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                     || commit_bucket_lifecycle_lock_fence
                         .as_ref()
                         .is_some_and(NamespaceLockFence::is_lock_lost)
+                    || _decommission_capacity_guard
+                        .as_ref()
+                        .is_some_and(|guard| guard.is_lock_lost())
                 {
                     return Err(StorageError::NamespaceLockQuorumUnavailable {
                         mode: "quota_reservation",
@@ -3037,6 +3280,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                         .map(|version_id| version_id.to_string());
                     let object_lock_guard = _object_lock_guard.take();
                     let upload_guard = _upload_guard.take();
+                    let decommission_object_lock_guard = _decommission_object_lock_guard.take();
+                    let decommission_capacity_guard = _decommission_capacity_guard.take();
                     let cleanup_bucket = commit_bucket.clone();
                     let cleanup_object = commit_object.clone();
                     let heal_set = commit_set.clone();
@@ -3054,7 +3299,12 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                     tokio::spawn(finish_rename_tail_heal(
                         rename_tail_drain,
                         guard_release_rx,
-                        (object_lock_guard, upload_guard),
+                        (
+                            object_lock_guard,
+                            upload_guard,
+                            decommission_object_lock_guard,
+                            decommission_capacity_guard,
+                        ),
                         request,
                         move || async move {
                             if quota_mutation_fence {
@@ -3068,7 +3318,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                                 .await;
                             }
                         },
-                        move |(object_lock_guard, upload_guard), targets| async move {
+                        move |(object_lock_guard, upload_guard, decommission_object_lock_guard, decommission_capacity_guard),
+                              targets| async move {
                             drop(object_lock_guard);
                             cleanup_set.cleanup_multipart_path(&cleanup_parts).await;
                             cleanup_set
@@ -3093,11 +3344,16 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                                 );
                             }
                             drop(upload_guard);
+                            drop(decommission_object_lock_guard);
+                            drop(decommission_capacity_guard);
                         },
                         |request| async move { heal_set.submit_rename_tail_heal(request).await },
                     ));
                 }
             }
+            if !tail_owns_staging_cleanup {
+                drop(_decommission_capacity_guard.take());
+            }
             if quota_mutation_fence && !tail_owns_staging_cleanup {
                 let _ = SetDisks::release_quota_mutation_fences(
                     &commit_disks,

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

@@ -2760,6 +2760,9 @@ impl SetDisks {
         let disks = self.get_disks_internal().await;
 
         let mut object_lock_guard = None;
+        let mut decommission_object_lock_guard = None;
+        let mut decommission_target_lock_covered = false;
+        let mut decommission_capacity_guard = None;
         let mut bucket_lifecycle_guard = None;
 
         // This pre-body check is advisory fast-fail only: the authoritative
@@ -3212,7 +3215,25 @@ impl SetDisks {
                 .await?;
             }
 
-            if !opts.no_lock && object_lock_guard.is_none() {
+            if let Some(store) = opts.decommission_capacity_admission.as_ref() {
+                #[cfg(test)]
+                {
+                    crate::core::pools::notify_decommission_external_object_commit_phase_started(store.id);
+                    crate::core::pools::wait_for_decommission_external_object_commit_phase_release(store.id).await;
+                }
+                let (object_guard, target_lock_covered, capacity_guard) = store
+                    .acquire_external_decommission_commit_guards(
+                        self.pool_index,
+                        bucket,
+                        object,
+                        opts.no_lock || object_lock_guard.is_some(),
+                    )
+                    .await?;
+                decommission_object_lock_guard = object_guard;
+                decommission_target_lock_covered = target_lock_covered;
+                decommission_capacity_guard = capacity_guard;
+            }
+            if !opts.no_lock && object_lock_guard.is_none() && !decommission_target_lock_covered {
                 #[cfg(any(test, feature = "test-util"))]
                 pause_put_object_commit(bucket, object, PutObjectCommitPause::BeforeNamespace).await;
                 if let Some(expected_incarnation_id) = opts.expected_bucket_incarnation_id
@@ -3389,6 +3410,33 @@ impl SetDisks {
                 });
             }
 
+            if decommission_capacity_guard.is_none()
+                && let Some(store) = opts.decommission_capacity_admission.as_ref()
+            {
+                decommission_capacity_guard = Some(
+                    store
+                        .acquire_external_decommission_capacity_fence(&[self.pool_index], "mutation")
+                        .await?,
+                );
+            }
+
+            // The object namespace is acquired above, after the input stream
+            // has been fully staged. Only then admit the local publication
+            // against the decommission capacity ledger; holding this guard
+            // through rename_data keeps the namespace -> capacity order.
+            if decommission_object_lock_guard
+                .as_ref()
+                .is_some_and(|guard| guard.is_lock_lost())
+            {
+                return Err(StorageError::NamespaceLockQuorumUnavailable {
+                    mode: "put_object_external_namespace",
+                    bucket: bucket.to_string(),
+                    object: object.to_string(),
+                    required: 1,
+                    achieved: 0,
+                });
+            }
+
             let transaction_fencing_proof = object_transaction_fencing_fleet_proof();
             if object_transaction_fencing_requested() && transaction_fencing_proof.is_none() {
                 return Err(Error::other("object transaction fencing requires a live fleet capability proof"));
@@ -3523,13 +3571,17 @@ impl SetDisks {
             let commit_object = object.to_owned();
             let commit_tmp_dir = tmp_dir.clone();
             let commit_object_lock_guard = object_lock_guard.take();
+            let commit_decommission_object_lock_guard = decommission_object_lock_guard.take();
             let commit_bucket_lifecycle_guard = bucket_lifecycle_guard.take();
+            let commit_decommission_capacity_guard = decommission_capacity_guard.take();
             let commit_scanner_publication_scope = opts.scanner_publication_commit_scope.clone();
             // A scanner publication scope owns the movement permit until the
             // complete rename fan-out drains. Keep this path synchronous so
             // its terminal state is known before the coordinator releases
             // remote leases.
-            let commit_allows_early_ack = commit_object_lock_guard.is_some() && commit_scanner_publication_scope.is_none();
+            let commit_allows_early_ack = !(opts.data_movement && opts.has_decommission_capacity_reservation())
+                && (commit_object_lock_guard.is_some() || commit_decommission_object_lock_guard.is_some())
+                && commit_scanner_publication_scope.is_none();
             let detach_commit_owner = commit_scanner_publication_scope.is_some()
                 || commit_allows_early_ack
                 || commit_bucket_lifecycle_guard.is_some()
@@ -3549,7 +3601,9 @@ impl SetDisks {
 
             let commit = move |cancellation: Option<CancellationToken>| async move {
                 let mut _object_lock_guard = commit_object_lock_guard;
+                let mut _decommission_object_lock_guard = commit_decommission_object_lock_guard;
                 let mut _bucket_lifecycle_guard = commit_bucket_lifecycle_guard;
+                let mut _decommission_capacity_guard = commit_decommission_capacity_guard;
                 let mut quota_reservation = quota_reservation;
                 let rename_stage_start = Instant::now();
                 let pre_rename = async {
@@ -3561,6 +3615,9 @@ impl SetDisks {
                     if quota_reservation.is_lock_lost()
                         || !quota_reservation.capability_proof_matches()
                         || _object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                        || _decommission_object_lock_guard
+                            .as_ref()
+                            .is_some_and(|guard| guard.is_lock_lost())
                         || commit_namespace_lock_fence
                             .as_ref()
                             .is_some_and(NamespaceLockFence::is_lock_lost)
@@ -3568,6 +3625,9 @@ impl SetDisks {
                             .as_ref()
                             .is_some_and(NamespaceLockFence::is_lock_lost)
                         || _bucket_lifecycle_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                        || _decommission_capacity_guard
+                            .as_ref()
+                            .is_some_and(|guard| guard.is_lock_lost())
                     {
                         return Err(StorageError::NamespaceLockQuorumUnavailable {
                             mode: "quota_reservation",
@@ -3611,6 +3671,9 @@ impl SetDisks {
                     if quota_reservation.is_lock_lost()
                         || !quota_reservation.capability_proof_matches()
                         || _object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                        || _decommission_object_lock_guard
+                            .as_ref()
+                            .is_some_and(|guard| guard.is_lock_lost())
                         || commit_namespace_lock_fence
                             .as_ref()
                             .is_some_and(NamespaceLockFence::is_lock_lost)
@@ -3618,6 +3681,9 @@ impl SetDisks {
                             .as_ref()
                             .is_some_and(NamespaceLockFence::is_lock_lost)
                         || _bucket_lifecycle_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                        || _decommission_capacity_guard
+                            .as_ref()
+                            .is_some_and(|guard| guard.is_lock_lost())
                     {
                         return Err(StorageError::NamespaceLockQuorumUnavailable {
                             mode: "quota_reservation",
@@ -3727,6 +3793,8 @@ impl SetDisks {
                             .map(|version_id| version_id.to_string());
                         let object_lock_guard = _object_lock_guard.take();
                         let bucket_lifecycle_guard = _bucket_lifecycle_guard.take();
+                        let decommission_object_lock_guard = _decommission_object_lock_guard.take();
+                        let decommission_capacity_guard = _decommission_capacity_guard.take();
                         let cleanup_bucket = commit_bucket.clone();
                         let cleanup_object = commit_object.clone();
                         let heal_set = commit_set.clone();
@@ -3741,7 +3809,12 @@ impl SetDisks {
                         tokio::spawn(finish_rename_tail_heal(
                             rename_tail_drain,
                             guard_release_rx,
-                            (object_lock_guard, bucket_lifecycle_guard),
+                            (
+                                object_lock_guard,
+                                bucket_lifecycle_guard,
+                                decommission_object_lock_guard,
+                                decommission_capacity_guard,
+                            ),
                             request,
                             move || async move {
                                 if quota_mutation_fence {
@@ -3755,7 +3828,13 @@ impl SetDisks {
                                     .await;
                                 }
                             },
-                            move |(object_lock_guard, bucket_lifecycle_guard), targets| async move {
+                            move |(
+                                object_lock_guard,
+                                bucket_lifecycle_guard,
+                                decommission_object_lock_guard,
+                                decommission_capacity_guard,
+                            ),
+                                  targets| async move {
                                 drop(object_lock_guard);
                                 drop(bucket_lifecycle_guard);
                                 cleanup_set
@@ -3776,11 +3855,16 @@ impl SetDisks {
                                         "issue3031_put_object_tmp_cleanup_done"
                                     );
                                 }
+                                drop(decommission_object_lock_guard);
+                                drop(decommission_capacity_guard);
                             },
                             |request| async move { heal_set.submit_rename_tail_heal(request).await },
                         ));
                     }
                 }
+                if !tail_owns_tmp_cleanup {
+                    drop(_decommission_capacity_guard.take());
+                }
                 #[cfg(any(test, feature = "test-util"))]
                 if rename_result.is_ok() {
                     pause_put_object_commit(&commit_bucket, &commit_object, PutObjectCommitPause::AfterRenameHandoff).await;
@@ -5687,6 +5771,7 @@ fn notify_put_object_commit_namespace_acquired(bucket: &str, object: &str) {
 struct DeleteObjectCommitBarrierState {
     bucket: String,
     object: String,
+    pause_after_publish: bool,
     arrived: tokio::sync::Notify,
     release: tokio::sync::Notify,
 }
@@ -5703,9 +5788,18 @@ static DELETE_OBJECT_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option
 #[cfg(test)]
 impl DeleteObjectCommitBarrier {
     pub(crate) fn install(bucket: &str, object: &str) -> Self {
+        Self::install_with_mode(bucket, object, false)
+    }
+
+    pub(crate) fn install_after_publish(bucket: &str, object: &str) -> Self {
+        Self::install_with_mode(bucket, object, true)
+    }
+
+    fn install_with_mode(bucket: &str, object: &str, pause_after_publish: bool) -> Self {
         let state = Arc::new(DeleteObjectCommitBarrierState {
             bucket: bucket.to_string(),
             object: object.to_string(),
+            pause_after_publish,
             arrived: tokio::sync::Notify::new(),
             release: tokio::sync::Notify::new(),
         });
@@ -5750,7 +5844,22 @@ async fn pause_delete_object_commit(bucket: &str, object: &str) {
         .lock()
         .expect("delete object commit barrier mutex should not poison")
         .as_ref()
-        .filter(|barrier| barrier.bucket == bucket && barrier.object == object)
+        .filter(|barrier| barrier.bucket == bucket && barrier.object == object && !barrier.pause_after_publish)
+        .cloned();
+    if let Some(barrier) = barrier {
+        barrier.arrived.notify_one();
+        barrier.release.notified().await;
+    }
+}
+
+#[cfg(test)]
+async fn pause_delete_object_commit_after_publish(bucket: &str, object: &str) {
+    let barrier = DELETE_OBJECT_COMMIT_BARRIER
+        .get_or_init(|| std::sync::Mutex::new(None))
+        .lock()
+        .expect("delete object commit barrier mutex should not poison")
+        .as_ref()
+        .filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause_after_publish)
         .cloned();
     if let Some(barrier) = barrier {
         barrier.arrived.notify_one();
@@ -7527,6 +7636,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             self.delete_object_version(bucket, object, &fi, should_force_delete_marker_for_missing_version(&opts))
                 .await
                 .map_err(|e| to_object_err(e, vec![bucket, object]))?;
+            #[cfg(test)]
+            pause_delete_object_commit_after_publish(bucket, object).await;
 
             let disks = self.disk_inventory().await;
             self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
@@ -7567,6 +7678,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         self.delete_object_version(bucket, object, &dfi, opts.delete_marker)
             .await
             .map_err(|e| to_object_err(e, vec![bucket, object]))?;
+        #[cfg(test)]
+        pause_delete_object_commit_after_publish(bucket, object).await;
 
         let disks = self.disk_inventory().await;
         self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
@@ -8319,19 +8432,15 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 part_checksums.to_string(),
             );
         }
-        // The restore copy-back re-writes this same object via put_object /
-        // new_multipart_upload / complete_multipart_upload, each of which takes
-        // the object write lock in its commit phase. The caller
-        // (handle_restore_transitioned_object, #4877) already holds that write
-        // lock for the whole restore and forwards no_lock=true, so the inner
-        // writes must inherit it or they self-deadlock on the lock we already
-        // hold and time out. put_restore_opts builds fresh options that default
-        // no_lock=false, so propagate it explicitly here.
+        // Keep the public ECStore capacity admission attached to each local
+        // commit. The tier reads below must remain outside the object write
+        // lock so HEAD/GET do not wait for a slow remote copy-back.
         ropts.no_lock = opts.no_lock;
         ropts.expected_bucket_incarnation_id = opts.expected_bucket_incarnation_id;
         ropts.bucket_lifecycle_lock_fence = opts.bucket_lifecycle_lock_fence.clone();
         ropts.namespace_lock_fence = opts.namespace_lock_fence.clone();
         ropts.object_lock_config_snapshot = opts.object_lock_config_snapshot.clone();
+        ropts.decommission_capacity_admission = opts.decommission_capacity_admission.clone();
         if oi.parts.len() == 1 {
             let mut opts = opts.clone();
             opts.part_number = Some(1);
@@ -8462,25 +8571,19 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                     .expect("multipart restore must contain at least one uploaded part")
                     .etag = Some("injected-invalid-complete-etag".to_string());
             }
+            let complete_opts = ObjectOptions {
+                mod_time: oi.mod_time,
+                version_id: oi.version_id.map(|version| version.to_string()),
+                expected_bucket_incarnation_id: opts.expected_bucket_incarnation_id,
+                bucket_lifecycle_lock_fence: opts.bucket_lifecycle_lock_fence.clone(),
+                user_defined: restore_commit_metadata,
+                no_lock: opts.no_lock,
+                decommission_capacity_admission: opts.decommission_capacity_admission.clone(),
+                ..Default::default()
+            };
             self_
                 .clone()
-                .complete_multipart_upload(
-                    bucket,
-                    object,
-                    &res.upload_id,
-                    uploaded_parts,
-                    &ObjectOptions {
-                        mod_time: oi.mod_time,
-                        version_id: oi.version_id.map(|version| version.to_string()),
-                        expected_bucket_incarnation_id: opts.expected_bucket_incarnation_id,
-                        bucket_lifecycle_lock_fence: opts.bucket_lifecycle_lock_fence.clone(),
-                        user_defined: restore_commit_metadata,
-                        // Inherit the restore write lock (see ropts.no_lock above):
-                        // the commit phase re-acquires this object's write lock.
-                        no_lock: opts.no_lock,
-                        ..Default::default()
-                    },
-                )
+                .complete_multipart_upload(bucket, object, &res.upload_id, uploaded_parts, &complete_opts)
                 .await
         }
         .await;

crates/ecstore/src/set_disk/replication.rs

@@ -157,7 +157,15 @@ impl SetDisks {
             .clone()
             .unwrap_or_else(|| get_raw_etag(obj_info.user_defined.as_ref()));
         let version_id = expected.version_id.map(|v| v.to_string());
-        let lock_guard = if !opts.no_lock {
+        let (decommission_object_lock_guard, decommission_target_lock_covered, mut decommission_capacity_guard) =
+            if let Some(store) = opts.decommission_capacity_admission.as_ref() {
+                store
+                    .acquire_external_decommission_commit_guards(self.pool_index, bucket, object, opts.no_lock)
+                    .await?
+            } else {
+                (None, false, None)
+            };
+        let lock_guard = if !opts.no_lock && !decommission_target_lock_covered {
             Some(
                 self.acquire_write_lock_diag("restore_finalize_metadata", bucket, object)
                     .await?,
@@ -165,6 +173,15 @@ impl SetDisks {
         } else {
             None
         };
+        if decommission_capacity_guard.is_none()
+            && let Some(store) = opts.decommission_capacity_admission.as_ref()
+        {
+            decommission_capacity_guard = Some(
+                store
+                    .acquire_external_decommission_capacity_fence(&[self.pool_index], "mutation")
+                    .await?,
+            );
+        }
         let read_opts = ObjectOptions {
             version_id,
             versioned: opts.versioned,
@@ -198,7 +215,12 @@ impl SetDisks {
         );
         self.invalidate_get_object_metadata_cache(bucket, object).await;
         ensure_restore_metadata_lock_held(bucket, object, opts, "restore_finalize_metadata")?;
-        if lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
+        if lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+            || decommission_object_lock_guard
+                .as_ref()
+                .is_some_and(|guard| guard.is_lock_lost())
+            || decommission_capacity_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+        {
             return Err(Error::other("restore finalization lock lost before metadata update"));
         }
         self.update_object_meta_with_opts(
@@ -233,7 +255,15 @@ impl SetDisks {
             .clone()
             .unwrap_or_else(|| get_raw_etag(obj_info.user_defined.as_ref()));
         let version_id = expected.version_id.map(|v| v.to_string());
-        let lock_guard = if !opts.no_lock {
+        let (decommission_object_lock_guard, decommission_target_lock_covered, mut decommission_capacity_guard) =
+            if let Some(store) = opts.decommission_capacity_admission.as_ref() {
+                store
+                    .acquire_external_decommission_commit_guards(self.pool_index, bucket, object, opts.no_lock)
+                    .await?
+            } else {
+                (None, false, None)
+            };
+        let lock_guard = if !opts.no_lock && !decommission_target_lock_covered {
             Some(
                 self.acquire_write_lock_diag("restore_cleanup_metadata", bucket, object)
                     .await?,
@@ -241,6 +271,15 @@ impl SetDisks {
         } else {
             None
         };
+        if decommission_capacity_guard.is_none()
+            && let Some(store) = opts.decommission_capacity_admission.as_ref()
+        {
+            decommission_capacity_guard = Some(
+                store
+                    .acquire_external_decommission_capacity_fence(&[self.pool_index], "mutation")
+                    .await?,
+            );
+        }
         let read_opts = ObjectOptions {
             version_id,
             versioned: opts.versioned,
@@ -269,7 +308,12 @@ impl SetDisks {
             &mut fi.metadata,
             rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID,
         );
-        if lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
+        if lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+            || decommission_object_lock_guard
+                .as_ref()
+                .is_some_and(|guard| guard.is_lock_lost())
+            || decommission_capacity_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+        {
             return Err(Error::other("restore cleanup lock lost before metadata update".to_string()));
         }
         self.invalidate_get_object_metadata_cache(bucket, object).await;

crates/ecstore/src/store/heal.rs

@@ -470,6 +470,27 @@ impl ECStore {
         let object = encode_dir_object(object);
 
         let pools = self.get_pools_for_heal_object(opts)?;
+        if let Some(set_idx) = opts.set {
+            for pool in &pools {
+                if set_idx >= pool.disk_set.len() {
+                    let err = StorageError::InvalidArgument(
+                        "heal".to_string(),
+                        "set".to_string(),
+                        format!(
+                            "invalid heal set index {set_idx} for pool {} with {} sets",
+                            pool.pool_idx,
+                            pool.disk_set.len()
+                        ),
+                    );
+                    if opts.pool.is_some() {
+                        return Err(err);
+                    }
+                    return Ok((HealResultItem::default(), Some(err)));
+                }
+            }
+        }
+        #[cfg(test)]
+        let store_id = self.id;
 
         let mut futures = Vec::with_capacity(pools.len());
         for pool in pools.iter() {
@@ -499,7 +520,17 @@ impl ECStore {
                 }
                 continue;
             }
-            futures.push(pool.heal_object(bucket, &object, version_id, opts));
+            let pool_idx = pool.pool_idx;
+            let pool = Arc::clone(pool);
+            let pool_object = object.clone();
+            let opts = *opts;
+            futures.push(
+                self.run_external_decommission_capacity_heal(pool_idx, bucket, &object, opts, move |opts| async move {
+                    #[cfg(test)]
+                    crate::core::pools::notify_decommission_external_heal_operation_started(store_id);
+                    pool.heal_object(bucket, &pool_object, version_id, &opts).await
+                }),
+            );
         }
         let results = join_all(futures).await;
 
@@ -594,14 +625,18 @@ mod tests {
     use crate::cluster::rpc::PeerS3Client;
     use crate::config::com::{delete_config, read_config_no_lock_preserve_empty_with_metadata, save_config};
     use crate::core::pools::{
-        POOL_META_IDENTITY_NAME, PoolDecommissionInfo, PoolMetaReplicaState, PoolStatus, initialized_pool_meta_identity_for_test,
+        DecommissionCapacityLockOrderBarrier, DecommissionErasureLayout, DecommissionPoolCapacityInfo, POOL_META_IDENTITY_NAME,
+        PoolDecommissionInfo, PoolMetaReplicaState, PoolStatus, initialized_pool_meta_identity_for_test,
+        set_decommission_capacity_info_overrides_for_test,
     };
     use crate::core::sets::HealFormatAfterSaveBarrier;
     use crate::disk::error::Result as DiskResult;
     use crate::disk::{DeleteOptions, DiskOption, FORMAT_CONFIG_FILE, format::FormatV3, new_disk};
     use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
     use crate::runtime::instance::InstanceContext;
-    use crate::services::rebalance::{RebalanceInfo, RebalanceStats};
+    use crate::services::rebalance::{
+        RebalanceInfo, RebalanceStats, test_three_pool_stores_with_isolated_node_contexts, test_two_pool_stores,
+    };
     use crate::storage_api_contracts::bucket::{
         BucketInfo, BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions,
     };
@@ -1207,6 +1242,292 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn targeted_heal_is_blocked_by_exact_fit_decommission_reservation() {
+        let (_temp_dirs, store, _other_store) = test_two_pool_stores(None).await;
+        let bucket = format!("heal-capacity-{}", Uuid::new_v4().simple());
+        let object = "targeted-heal.bin";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("heal capacity bucket should be created");
+
+        let target_set = store.pools[1].get_disks(0);
+        let mut reader = PutObjReader::from_vec(b"targeted heal body".to_vec());
+        target_set
+            .put_object(&bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("targeted heal fixture should be written");
+        let missing_disk = target_set.disks.read().await[0]
+            .clone()
+            .expect("targeted heal fixture disk should be online");
+        missing_disk
+            .delete(
+                &bucket,
+                object,
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("targeted heal fixture shard should be removed");
+        assert!(
+            missing_disk.read_xl(&bucket, object, false).await.is_err(),
+            "targeted heal fixture must start with one missing metadata copy"
+        );
+
+        let layout = DecommissionErasureLayout { data: 1, parity: 0 };
+        set_decommission_capacity_info_overrides_for_test(
+            store.id,
+            vec![vec![
+                DecommissionPoolCapacityInfo::for_test(0, layout, 0, 30, 30),
+                DecommissionPoolCapacityInfo::for_test(1, layout, 60, 60, 0),
+            ]],
+        );
+        store
+            .save_current_pool_meta_for_decommission_start(&[0], Vec::new())
+            .await
+            .expect("the exact-fit decommission reservation should activate");
+        {
+            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("the active decommission reservation should be durable");
+            assert_eq!(reservation.targets.len(), 1);
+            assert_eq!(reservation.targets[0].pool_index, 1);
+            assert_eq!(reservation.targets[0].reserved_physical_bytes, 60);
+        }
+
+        let (_, err) = store
+            .handle_heal_object(
+                &bucket,
+                object,
+                "",
+                &HealOpts {
+                    pool: Some(1),
+                    set: Some(0),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("targeted heal should return a mapped capacity result");
+        assert!(matches!(err, Some(Error::SlowDown)), "targeted heal must be capacity-blocked: {err:?}");
+        assert!(
+            missing_disk.read_xl(&bucket, object, false).await.is_err(),
+            "capacity-blocked targeted heal must not rewrite the missing shard"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn targeted_heal_keeps_target_lock_for_different_lock_domain() {
+        let (_temp_dirs, store, other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
+        let bucket = format!("heal-lock-domain-{}", Uuid::new_v4().simple());
+        let object = "different-domain-heal.bin";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("heal lock-domain bucket should be created");
+
+        let target_set = other_store.pools[1].get_disks(0);
+        let mut reader = PutObjReader::from_vec(b"different lock domain body".to_vec());
+        target_set
+            .put_object(&bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("heal lock-domain fixture should be written");
+        let missing_disk = target_set.disks.read().await[0]
+            .clone()
+            .expect("heal lock-domain fixture disk should be online");
+        missing_disk
+            .delete(
+                &bucket,
+                object,
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("heal lock-domain fixture shard should be removed");
+        assert!(
+            missing_disk.read_xl(&bucket, object, false).await.is_err(),
+            "heal lock-domain fixture must start with one missing metadata copy"
+        );
+
+        let layout = DecommissionErasureLayout { data: 1, parity: 0 };
+        set_decommission_capacity_info_overrides_for_test(
+            store.id,
+            vec![vec![
+                DecommissionPoolCapacityInfo::for_test(0, layout, 0, 30, 30),
+                DecommissionPoolCapacityInfo::for_test(1, layout, 0, 100, 100),
+                DecommissionPoolCapacityInfo::for_test(2, layout, 60, 60, 0),
+            ]],
+        );
+        store
+            .save_current_pool_meta_for_decommission_start(&[0], Vec::new())
+            .await
+            .expect("heal lock-domain reservation should activate");
+        {
+            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("heal lock-domain reservation should be durable");
+            assert_eq!(reservation.targets.len(), 1);
+            assert_eq!(reservation.targets[0].pool_index, 2);
+        }
+        let pool_meta = store.pool_meta.read().await.clone();
+        *other_store.pool_meta.write().await = pool_meta;
+
+        let fixed_set = other_store.pools[0].disk_set[0].clone();
+        assert!(
+            !fixed_set.shares_namespace_lock_domain(&target_set).await,
+            "heal fixture must use different fixed and target lock domains"
+        );
+        let barrier = DecommissionCapacityLockOrderBarrier::install(store.id, other_store.id);
+        let heal_store = Arc::clone(&other_store);
+        let heal_bucket = bucket.clone();
+        let heal_object = object.to_string();
+        let mut heal = tokio::spawn(async move {
+            heal_store
+                .handle_heal_object(
+                    &heal_bucket,
+                    &heal_object,
+                    "",
+                    &HealOpts {
+                        pool: Some(1),
+                        set: Some(0),
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+
+        barrier.wait_until_external_capacity_released().await;
+        barrier.wait_until_external_heal_target_lock_attempted().await;
+
+        let (_, err) = tokio::time::timeout(std::time::Duration::from_secs(30), &mut heal)
+            .await
+            .expect("targeted heal should finish after target lock release")
+            .expect("targeted heal task should not panic")
+            .expect("targeted heal should complete");
+        assert!(err.is_none(), "targeted heal should repair after target lock release: {err:?}");
+        assert!(
+            missing_disk.read_xl(&bucket, object, false).await.is_ok(),
+            "targeted heal should rewrite the missing shard after the target lock is released"
+        );
+        drop(barrier);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn targeted_heal_reuses_shared_target_lock_without_reentrant_lock() {
+        let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
+        let bucket = format!("heal-shared-lock-domain-{}", Uuid::new_v4().simple());
+        let object = "shared-domain-heal.bin";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("shared heal lock-domain bucket should be created");
+
+        let target_set = store.pools[0].get_disks(0);
+        let mut reader = PutObjReader::from_vec(b"shared lock domain body".to_vec());
+        target_set
+            .put_object(&bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("shared heal lock-domain fixture should be written");
+        let missing_disk = target_set.disks.read().await[0]
+            .clone()
+            .expect("shared heal lock-domain fixture disk should be online");
+        missing_disk
+            .delete(
+                &bucket,
+                object,
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("shared heal lock-domain fixture shard should be removed");
+        assert!(
+            missing_disk.read_xl(&bucket, object, false).await.is_err(),
+            "shared heal lock-domain fixture must start with one missing metadata copy"
+        );
+
+        let layout = DecommissionErasureLayout { data: 1, parity: 0 };
+        set_decommission_capacity_info_overrides_for_test(
+            store.id,
+            vec![vec![
+                DecommissionPoolCapacityInfo::for_test(0, layout, 0, 100, 100),
+                DecommissionPoolCapacityInfo::for_test(1, layout, 60, 60, 0),
+                DecommissionPoolCapacityInfo::for_test(2, layout, 0, 30, 30),
+            ]],
+        );
+        store
+            .save_current_pool_meta_for_decommission_start(&[2], Vec::new())
+            .await
+            .expect("shared heal lock-domain reservation should activate");
+        {
+            let pool_meta = store.pool_meta.read().await;
+            let reservation = pool_meta.pools[2]
+                .decommission
+                .as_ref()
+                .and_then(|info| info.capacity_reservation.as_ref())
+                .expect("shared heal lock-domain reservation should be durable");
+            assert_eq!(reservation.targets.len(), 1);
+            assert_eq!(reservation.targets[0].pool_index, 1);
+            assert_eq!(reservation.targets[0].reserved_physical_bytes, 60);
+        }
+
+        let fixed_set = store.pools[0].disk_set[0].clone();
+        assert!(
+            fixed_set.shares_namespace_lock_domain(&target_set).await,
+            "shared heal fixture must use one fixed and target lock domain"
+        );
+
+        let barrier = DecommissionCapacityLockOrderBarrier::install(store.id, store.id);
+        let heal_store = Arc::clone(&store);
+        let heal_bucket = bucket.clone();
+        let heal_object = object.to_string();
+        let mut heal = tokio::spawn(async move {
+            heal_store
+                .handle_heal_object(
+                    &heal_bucket,
+                    &heal_object,
+                    "",
+                    &HealOpts {
+                        pool: Some(0),
+                        set: Some(0),
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+
+        barrier.wait_until_external_capacity_released().await;
+        barrier.wait_until_external_heal_operation_started().await;
+        let (_, err) = tokio::time::timeout(std::time::Duration::from_secs(5), &mut heal)
+            .await
+            .expect("shared-domain heal must not reenter the fixed namespace lock")
+            .expect("shared-domain heal task should not panic")
+            .expect("shared-domain heal should complete");
+        assert!(err.is_none(), "shared-domain heal should repair after admission: {err:?}");
+        assert!(
+            missing_disk.read_xl(&bucket, object, false).await.is_ok(),
+            "shared-domain heal should rewrite the missing shard"
+        );
+        drop(barrier);
+    }
+
     #[tokio::test]
     async fn scoped_heal_object_defers_when_requested_pool_is_suspended() {
         let mut store = minimal_heal_store().await;
@@ -1697,8 +2018,10 @@ mod tests {
             .expect("quorum boundary heal should return a mapped result");
         *store.pools[0].disk_set[0].disks.write().await = original_quorum_disks;
         assert!(
-            matches!(quorum_err, Some(Error::ErasureReadQuorum)),
-            "quorum-boundary heal must preserve quorum error, got {quorum_err:?}"
+            quorum_err.as_ref().is_some_and(|err| err
+                .to_string()
+                .contains("pool metadata writes remain blocked after a recovery-required replica state")),
+            "heal must fail closed when capacity admission cannot verify pool metadata, got {quorum_err:?}"
         );
         shutdown.cancel();
     }

crates/ecstore/src/store/init.rs

@@ -2142,8 +2142,16 @@ mod tests {
                     },
                 )
                 .await?;
-            crate::data_movement::migrate_decommission_object(migration_store, 0, migration_bucket, source_reader, None, op_label)
-                .await
+            crate::data_movement::migrate_decommission_object(
+                migration_store,
+                0,
+                migration_bucket,
+                source_reader,
+                None,
+                op_label,
+                None,
+            )
+            .await
         });
         barrier.wait_until_paused().await;
         barrier.release();

crates/ecstore/src/store/multipart.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use super::*;
+use crate::core::pools::{DecommissionCapacityOwner, ensure_decommission_capacity_mutation_id};
 use crate::multipart_listing::paginate_multipart_listing;
 use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::multipart::MultipartOperations as _;
@@ -196,6 +197,21 @@ async fn list_pool_multipart_uploads_for_incarnation(
 }
 
 impl ECStore {
+    pub(crate) async fn acquire_decommission_multipart_mutation_fence(
+        &self,
+        owner: DecommissionCapacityOwner,
+    ) -> Result<ObjectLockDiagGuard> {
+        let mutation_id = owner
+            .mutation_id
+            .ok_or_else(|| Error::other("decommission multipart mutation identity is missing"))?;
+        let object = format!(
+            "decommission-multipart/{}/{}/{}/{}",
+            owner.source_pool_index, owner.operation_id, owner.generation, mutation_id
+        );
+        self.acquire_object_write_lock("decommission_multipart_mutation", crate::disk::RUSTFS_META_MULTIPART_BUCKET, &object)
+            .await
+    }
+
     async fn existing_multipart_pool_order(&self) -> Vec<usize> {
         // A draining source must not hide a valid UploadID in an active target,
         // while physical order within each phase preserves fail-closed errors.
@@ -215,6 +231,24 @@ impl ECStore {
         active
     }
 
+    async fn multipart_upload_pool_idx(
+        &self,
+        bucket: &str,
+        object: &str,
+        upload_id: &str,
+        opts: &ObjectOptions,
+    ) -> Result<usize> {
+        for pool_idx in self.existing_multipart_pool_order().await {
+            match self.pools[pool_idx].get_multipart_info(bucket, object, upload_id, opts).await {
+                Ok(_) => return Ok(pool_idx),
+                Err(err) if is_err_invalid_upload_id(&err) => continue,
+                Err(err) => return Err(err),
+            }
+        }
+
+        Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()))
+    }
+
     #[allow(clippy::too_many_arguments)]
     pub async fn list_multipart_uploads_for_bucket_incarnation(
         &self,
@@ -433,8 +467,10 @@ impl ECStore {
         if self.single_pool() {
             self.apply_decommission_target_mutation_fence(0, object, &mut opts, mutation_fence)
                 .await;
-            return self.pools[0]
-                .new_multipart_upload(bucket, object, &opts)
+            return self
+                .run_decommission_capacity_admitted_mutation(0, None, None, || async {
+                    self.pools[0].new_multipart_upload(bucket, object, &opts).await
+                })
                 .await
                 .map(|res| (res, 0, opts.expected_bucket_incarnation_id));
         }
@@ -450,7 +486,14 @@ impl ECStore {
             }
             self.apply_decommission_target_mutation_fence(idx, object, &mut opts, mutation_fence)
                 .await;
-            let res = self.pools[idx].new_multipart_upload(bucket, object, &opts).await?;
+            let res = self
+                .run_decommission_capacity_temporary_mutation(
+                    idx,
+                    DecommissionCapacityOwner::from_options(&opts),
+                    None,
+                    || async { self.pools[idx].new_multipart_upload(bucket, object, &opts).await },
+                )
+                .await?;
             return Ok((res, idx, opts.expected_bucket_incarnation_id));
         }
 
@@ -475,8 +518,19 @@ impl ECStore {
             if !res.uploads.is_empty() {
                 self.apply_decommission_target_mutation_fence(idx, object, &mut opts, mutation_fence)
                     .await;
-                let res = self.pools[idx].new_multipart_upload(bucket, object, &opts).await?;
-                return Ok((res, idx, opts.expected_bucket_incarnation_id));
+                let expected_bucket_incarnation_id = opts.expected_bucket_incarnation_id;
+                let lock_object = encode_dir_object(object);
+                let res = self
+                    .run_external_decommission_capacity_object_mutation(
+                        idx,
+                        bucket,
+                        &lock_object,
+                        object,
+                        opts,
+                        |opts| async move { self.pools[idx].new_multipart_upload(bucket, object, &opts).await },
+                    )
+                    .await?;
+                return Ok((res, idx, expected_bucket_incarnation_id));
             }
         }
         let idx = self.get_pool_idx(bucket, object, -1).await?;
@@ -490,8 +544,23 @@ impl ECStore {
 
         self.apply_decommission_target_mutation_fence(idx, object, &mut opts, mutation_fence)
             .await;
-        let res = self.pools[idx].new_multipart_upload(bucket, object, &opts).await?;
-        Ok((res, idx, opts.expected_bucket_incarnation_id))
+        let expected_bucket_incarnation_id = opts.expected_bucket_incarnation_id;
+        let res = if opts.data_movement {
+            self.run_decommission_capacity_temporary_mutation(
+                idx,
+                DecommissionCapacityOwner::from_options(&opts),
+                None,
+                || async { self.pools[idx].new_multipart_upload(bucket, object, &opts).await },
+            )
+            .await?
+        } else {
+            let lock_object = encode_dir_object(object);
+            self.run_external_decommission_capacity_object_mutation(idx, bucket, &lock_object, object, opts, |opts| async move {
+                self.pools[idx].new_multipart_upload(bucket, object, &opts).await
+            })
+            .await?
+        };
+        Ok((res, idx, expected_bucket_incarnation_id))
     }
 
     #[instrument(skip(self))]
@@ -529,7 +598,8 @@ impl ECStore {
         opts: &ObjectOptions,
     ) -> Result<PartInfo> {
         check_put_object_part_args(bucket, object, upload_id)?;
-        let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
         let opts = &opts;
 
         if self.single_pool() {
@@ -538,26 +608,10 @@ impl ECStore {
                 .await;
         }
 
-        for pool_idx in self.existing_multipart_pool_order().await {
-            let pool = &self.pools[pool_idx];
-            let err = match pool.put_object_part(bucket, object, upload_id, part_id, data, opts).await {
-                Ok(res) => return Ok(res),
-                Err(err) => {
-                    if is_err_invalid_upload_id(&err) {
-                        None
-                    } else {
-                        Some(err)
-                    }
-                }
-            };
-
-            if let Some(err) = err {
-                error!("put_object_part err: {:?}", err);
-                return Err(err);
-            }
-        }
-
-        Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()))
+        let pool_idx = self.multipart_upload_pool_idx(bucket, object, upload_id, opts).await?;
+        self.pools[pool_idx]
+            .put_object_part(bucket, object, upload_id, part_id, data, opts)
+            .await
     }
 
     pub(crate) async fn put_object_part_for_data_movement(
@@ -576,12 +630,24 @@ impl ECStore {
         if !opts.data_movement {
             return Err(Error::other("targeted multipart upload requires data_movement options"));
         }
-        let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
-        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")))?;
-        pool.put_object_part(bucket, object, upload_id, part_id, data, &opts).await
+        let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        ensure_decommission_capacity_mutation_id(bucket, object, &mut opts);
+        let pool = self.pools.get(target_pool_idx).ok_or_else(|| {
+            Error::InvalidArgument("data-movement".to_string(), "target-pool".to_string(), target_pool_idx.to_string())
+        })?;
+        let expected_data_bytes = usize::try_from(data.size()).ok();
+        self.run_decommission_capacity_temporary_mutation_with_capacity_lease(
+            target_pool_idx,
+            DecommissionCapacityOwner::from_options(&opts),
+            expected_data_bytes,
+            |capacity_lease| async move {
+                if let Some(capacity_lease) = capacity_lease {
+                    opts.add_namespace_lock_lost_signal(capacity_lease);
+                }
+                pool.put_object_part(bucket, object, upload_id, part_id, data, &opts).await
+            },
+        )
+        .await
     }
 
     #[instrument(skip(self))]
@@ -662,16 +728,95 @@ impl ECStore {
         upload_id: &str,
         opts: &ObjectOptions,
     ) -> Result<()> {
-        check_abort_multipart_args(bucket, object, upload_id)?;
-        if !opts.data_movement {
-            return Err(Error::other("targeted multipart abort requires data_movement options"));
-        }
-        let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        self.abort_multipart_uploads_for_data_movement(target_pool_idx, bucket, object, &[upload_id.to_owned()], None, opts)
+            .await
+    }
+
+    pub(crate) async fn reconcile_multipart_uploads_for_data_movement(
+        &self,
+        target_pool_idx: usize,
+        bucket: &str,
+        object: &str,
+        upload_identity: &str,
+        opts: &ObjectOptions,
+    ) -> Result<()> {
         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")))?;
-        pool.abort_multipart_upload(bucket, object, upload_id, &opts).await
+        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 {
+            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)
+            .await
+    }
+
+    async fn abort_multipart_uploads_for_data_movement(
+        &self,
+        target_pool_idx: usize,
+        bucket: &str,
+        object: &str,
+        upload_ids: &[String],
+        expected_upload_identity: Option<&str>,
+        opts: &ObjectOptions,
+    ) -> Result<()> {
+        check_new_multipart_args(bucket, object)?;
+        for upload_id in upload_ids {
+            check_abort_multipart_args(bucket, object, upload_id)?;
+        }
+        if !opts.data_movement {
+            return Err(Error::other("targeted multipart abort requires data_movement options"));
+        }
+        let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        ensure_decommission_capacity_mutation_id(bucket, object, &mut 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
+            .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;
+                async move {
+                    if let Some(capacity_lease) = capacity_lease {
+                        delete_opts.add_namespace_lock_lost_signal(capacity_lease);
+                    }
+                    for guard in guards {
+                        guard.delete(set, bucket, object, &delete_opts).await?;
+                    }
+                    Ok(())
+                }
+            })
+            .await;
+        drop(guards);
+        result
     }
 
     #[instrument(skip(self))]
@@ -684,7 +829,8 @@ impl ECStore {
         opts: &ObjectOptions,
     ) -> Result<ObjectInfo> {
         check_complete_multipart_args(bucket, object, upload_id)?;
-        let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
         let opts = &opts;
 
         if self.single_pool() {
@@ -694,27 +840,10 @@ impl ECStore {
                 .await;
         }
 
-        for pool_idx in self.existing_multipart_pool_order().await {
-            let pool = &self.pools[pool_idx];
-
-            let pool = pool.clone();
-            let err = match pool
-                .complete_multipart_upload(bucket, object, upload_id, uploaded_parts.clone(), opts)
-                .await
-            {
-                Ok(res) => return Ok(res),
-                Err(err) => {
-                    //
-                    if is_err_invalid_upload_id(&err) { None } else { Some(err) }
-                }
-            };
-
-            if let Some(er) = err {
-                return Err(er);
-            }
-        }
-
-        Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()))
+        let pool_idx = self.multipart_upload_pool_idx(bucket, object, upload_id, opts).await?;
+        let pool = self.pools[pool_idx].clone();
+        pool.complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
+            .await
     }
 
     pub(crate) async fn complete_multipart_upload_for_data_movement(
@@ -732,6 +861,7 @@ impl ECStore {
             return Err(Error::other("targeted multipart completion requires data_movement options"));
         }
         let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        ensure_decommission_capacity_mutation_id(bucket, object, &mut opts);
         if opts.overwrites_existing_version() && !is_meta_bucketname(bucket) {
             let expected_incarnation_id = opts
                 .expected_bucket_incarnation_id
@@ -764,12 +894,24 @@ impl ECStore {
             .get(target_pool_idx)
             .ok_or_else(|| Error::other(format!("data movement target pool {target_pool_idx} is out of range")))?
             .clone();
-        let result = enqueue_transition_after_write(
-            pool.complete_multipart_upload(bucket, object, upload_id, uploaded_parts, &opts)
-                .await,
-            LcEventSrc::S3CompleteMultipartUpload,
-        )
-        .await;
+        // Data movement already owns the pool-meta write lease. Forward its
+        // loss signal into SetDisks so commit fencing observes the same lease
+        // without trying to reacquire the namespace.
+        let result = self
+            .run_decommission_capacity_admitted_mutation_with_capacity_lease(
+                target_pool_idx,
+                DecommissionCapacityOwner::from_options(&opts),
+                opts.capacity_expected_data_bytes(),
+                |capacity_lease| async move {
+                    if let Some(capacity_lease) = capacity_lease {
+                        opts.add_namespace_lock_lost_signal(capacity_lease);
+                    }
+                    pool.complete_multipart_upload(bucket, object, upload_id, uploaded_parts, &opts)
+                        .await
+                },
+            )
+            .await;
+        let result = enqueue_transition_after_write(result, LcEventSrc::S3CompleteMultipartUpload).await;
         if result.is_ok() {
             list_objects::observe_list_objects_mutation(self.as_ref(), bucket).await;
         }

crates/ecstore/src/store/object.rs

@@ -34,6 +34,7 @@ use crate::bucket::object_lock::objectlock_sys::{
 };
 use crate::bucket::replication::{DeleteReplicationConfigSnapshot, ReplicationObjectBridge};
 use crate::bucket::versioning::VersioningApi;
+use crate::core::pools::{DecommissionCapacityOwner, ensure_decommission_capacity_mutation_id};
 use crate::disk::OldCurrentSize;
 use crate::object_api::{NamespaceLockFence, ObjectLockConfigSnapshot};
 use crate::set_disk::{
@@ -1865,7 +1866,12 @@ impl ECStore {
         }
     }
 
-    async fn acquire_object_write_lock(&self, op: &'static str, bucket: &str, object: &str) -> Result<ObjectLockDiagGuard> {
+    pub(super) async fn acquire_object_write_lock(
+        &self,
+        op: &'static str,
+        bucket: &str,
+        object: &str,
+    ) -> Result<ObjectLockDiagGuard> {
         let diag_enabled = is_object_lock_diag_enabled();
         let ns_lock = self.handle_new_ns_lock(bucket, object).await?;
         let acquire_start = Instant::now();
@@ -2001,6 +2007,150 @@ impl ECStore {
         )))
     }
 
+    pub(super) async fn run_external_decommission_capacity_object_mutation<T, F, Fut>(
+        &self,
+        target_pool_idx: usize,
+        bucket: &str,
+        lock_object: &str,
+        target_object: &str,
+        mut opts: ObjectOptions,
+        operation: F,
+    ) -> Result<T>
+    where
+        F: FnOnce(ObjectOptions) -> Fut,
+        Fut: std::future::Future<Output = Result<T>>,
+    {
+        let (capacity_guard, has_active_decommission) = self
+            .acquire_external_decommission_capacity_fence_with_active_source(&[target_pool_idx], "mutation")
+            .await?;
+        let (capacity_guard, object_guard) = if has_active_decommission && !opts.no_lock {
+            // Active migration acquires the object namespace before its capacity
+            // write. Match that order, then recheck capacity admission.
+            drop(capacity_guard);
+            #[cfg(test)]
+            crate::core::pools::notify_decommission_external_object_capacity_released(self.id);
+            let guard = self
+                .acquire_object_write_lock("external_capacity_order", bucket, lock_object)
+                .await?;
+            self.apply_decommission_target_mutation_fence(target_pool_idx, target_object, &mut opts, Some(&guard))
+                .await;
+            let capacity_guard = self
+                .acquire_external_decommission_capacity_fence(&[target_pool_idx], "mutation")
+                .await?;
+            (capacity_guard, Some(guard))
+        } else {
+            (capacity_guard, None)
+        };
+
+        let result = operation(opts).await;
+        drop(capacity_guard);
+        drop(object_guard);
+        result
+    }
+
+    /// Finish an external object mutation's staged phase by acquiring the
+    /// fixed namespace before the target-side commit lock. The caller must
+    /// recheck capacity only after all applicable namespaces are held.
+    ///
+    /// Callers must invoke this only after consuming and staging their input
+    /// stream. The returned namespace guard remains owned by the caller until
+    /// its local commit completes; `target_lock_covered` tells the set layer
+    /// whether that guard also covers its target namespace. When no
+    /// decommission is active, the returned capacity guard is the admission
+    /// probe and must likewise remain held until the commit completes.
+    pub(crate) async fn acquire_external_decommission_commit_guards(
+        &self,
+        target_pool_idx: usize,
+        bucket: &str,
+        object: &str,
+        no_lock: bool,
+    ) -> Result<(Option<ObjectLockDiagGuard>, bool, Option<rustfs_lock::NamespaceLockGuard>)> {
+        let (capacity_guard, has_active_decommission) = self
+            .acquire_external_decommission_capacity_fence_with_active_source(&[target_pool_idx], "mutation")
+            .await?;
+        if !has_active_decommission {
+            // Keep the read probe through the staged commit. This closes the
+            // activation gap between admission and publication.
+            #[cfg(test)]
+            {
+                crate::core::pools::notify_decommission_external_object_capacity_probe_acquired(self.id);
+                crate::core::pools::wait_for_decommission_external_object_capacity_probe_release(self.id).await;
+            }
+            return Ok((None, false, Some(capacity_guard)));
+        }
+        drop(capacity_guard);
+        if no_lock {
+            return Ok((None, false, None));
+        }
+
+        // Active migration holds the fixed object namespace before taking its
+        // capacity write fence. Drop the probe and acquire that namespace
+        // before rechecking capacity, so the staged external commit cannot
+        // form capacity-read -> object-write against the migration path.
+        let object_guard = self
+            .acquire_object_write_lock("external_capacity_order", bucket, object)
+            .await?;
+        let fixed_set = self.pools.first().and_then(|pool| pool.disk_set.first());
+        let target_set = self.pools.get(target_pool_idx).map(|pool| pool.get_disks_by_key(object));
+        let target_lock_covered = match (fixed_set, target_set) {
+            (Some(fixed), Some(target)) => fixed.shares_namespace_lock_domain(&target).await,
+            _ => false,
+        };
+        Ok((Some(object_guard), target_lock_covered, None))
+    }
+
+    pub(super) async fn run_external_decommission_capacity_heal<T, F, Fut>(
+        &self,
+        target_pool_idx: usize,
+        bucket: &str,
+        lock_object: &str,
+        mut opts: HealOpts,
+        operation: F,
+    ) -> Result<T>
+    where
+        F: FnOnce(HealOpts) -> Fut,
+        Fut: std::future::Future<Output = Result<T>>,
+    {
+        let (capacity_guard, has_active_decommission) = self
+            .acquire_external_decommission_capacity_fence_with_active_source(&[target_pool_idx], "heal")
+            .await?;
+        let (capacity_guard, object_guard) = if has_active_decommission && !opts.no_lock {
+            // Active migration acquires the object namespace before its capacity
+            // write. Match that order, then recheck capacity admission.
+            drop(capacity_guard);
+            #[cfg(test)]
+            crate::core::pools::notify_decommission_external_object_capacity_released(self.id);
+            let guard = self
+                .acquire_object_write_lock("external_capacity_order", bucket, lock_object)
+                .await?;
+            let target_set = self
+                .pools
+                .get(target_pool_idx)
+                .ok_or_else(|| Error::other("heal target pool is unavailable"))?
+                .get_disks_for_heal_object(lock_object, &opts)?;
+            let target_lock_covered = match self.pools.first().and_then(|pool| pool.disk_set.first()) {
+                Some(fixed_set) => fixed_set.shares_namespace_lock_domain(&target_set).await,
+                None => false,
+            };
+            let capacity_guard = self
+                .acquire_external_decommission_capacity_fence(&[target_pool_idx], "heal")
+                .await?;
+            opts.no_lock = target_lock_covered;
+            (capacity_guard, Some(guard))
+        } else {
+            (capacity_guard, None)
+        };
+
+        #[cfg(test)]
+        if !opts.no_lock {
+            crate::core::pools::notify_decommission_external_heal_target_lock_attempted();
+        }
+        let result = operation(opts).await;
+        drop(capacity_guard);
+        drop(object_guard);
+        result
+    }
+
     pub(crate) async fn acquire_decommission_object_mutation_fence(
         &self,
         bucket: &str,
@@ -2239,7 +2389,7 @@ impl ECStore {
         resolve_latest_object_access(bucket, object, info, idx, opts)
     }
 
-    pub(super) async fn select_data_movement_pool_idx(
+    pub(crate) async fn select_data_movement_pool_idx(
         &self,
         bucket: &str,
         object: &str,
@@ -2257,6 +2407,16 @@ impl ECStore {
                     return Err(err);
                 }
 
+                if let Some(owner) = DecommissionCapacityOwner::from_options(opts) {
+                    let expected_data_bytes = opts
+                        .capacity_expected_data_bytes()
+                        .or_else(|| usize::try_from(size).ok())
+                        .unwrap_or_default();
+                    return self
+                        .select_decommission_capacity_target_pool(owner, expected_data_bytes)
+                        .await;
+                }
+
                 self.get_available_pool_idx(bucket, object, size).await.ok_or(Error::DiskFull)
             }
         }
@@ -2293,13 +2453,18 @@ impl ECStore {
         opts: &ObjectOptions,
         target_pool_idx: usize,
     ) -> Result<bool> {
-        resolve_data_movement_delete_marker_resume_result(
+        let equivalent = resolve_data_movement_delete_marker_resume_result(
             self.find_data_movement_target_info(bucket, object, target_pool_idx, opts)
                 .await,
             source,
             opts.src_pool_idx,
             target_pool_idx,
-        )
+        )?;
+        if equivalent && let Some(owner) = DecommissionCapacityOwner::from_options(opts) {
+            self.reconcile_decommission_capacity_after_equivalent_target(owner, target_pool_idx, 0)
+                .await?;
+        }
+        Ok(equivalent)
     }
 
     async fn has_equivalent_data_movement_tiered_object(
@@ -2310,13 +2475,19 @@ impl ECStore {
         opts: &ObjectOptions,
         target_pool_idx: usize,
     ) -> Result<bool> {
-        resolve_data_movement_tiered_resume_result(
+        let equivalent = resolve_data_movement_tiered_resume_result(
             self.find_data_movement_target_info(bucket, object, target_pool_idx, opts)
                 .await,
             source,
             opts.src_pool_idx,
             target_pool_idx,
-        )
+        )?;
+        if equivalent && let Some(owner) = DecommissionCapacityOwner::from_options(opts) {
+            let expected_data_bytes = usize::try_from(source.size).unwrap_or_default();
+            self.reconcile_decommission_capacity_after_equivalent_target(owner, target_pool_idx, expected_data_bytes)
+                .await?;
+        }
+        Ok(equivalent)
     }
 
     async fn has_equivalent_data_movement_tier_free_version(
@@ -2331,9 +2502,15 @@ impl ECStore {
             .pools
             .get(target_pool_idx)
             .ok_or_else(|| Error::other(format!("invalid tiered data movement target pool {target_pool_idx}")))?;
-        pool.get_disks_by_key(object)
+        let equivalent = pool
+            .get_disks_by_key(object)
             .has_decommission_tier_free_version_write_quorum(bucket, object, source, opts)
-            .await
+            .await?;
+        if equivalent && let Some(owner) = DecommissionCapacityOwner::from_options(opts) {
+            self.reconcile_decommission_capacity_after_equivalent_target(owner, target_pool_idx, 0)
+                .await?;
+        }
+        Ok(equivalent)
     }
 
     fn resolve_decommission_target_pool_idx_result(result: Result<usize>, bucket: &str, object: &str) -> Result<usize> {
@@ -2374,6 +2551,7 @@ impl ECStore {
 
         let logical_object = object;
         let object = encode_dir_object(logical_object);
+        ensure_decommission_capacity_mutation_id(bucket, &object, &mut opts);
         if self.single_pool() {
             return Self::resolve_decommission_tiered_object_result(
                 Err(Error::other("single pool deployments cannot decommission tiered objects")),
@@ -2428,9 +2606,15 @@ impl ECStore {
             crate::data_movement::prepare_tiered_data_movement_file_info(&mut fi)?;
         }
         if opts.data_movement && idx == opts.src_pool_idx {
-            let resume_target_pool_idx = self
-                .get_available_pool_idx_excluding(bucket, &object, fi.size, opts.src_pool_idx)
-                .await;
+            let resume_target_pool_idx = if let Some(owner) = DecommissionCapacityOwner::from_options(&opts) {
+                Some(
+                    self.select_decommission_capacity_target_pool(owner, usize::try_from(fi.size).unwrap_or_default())
+                        .await?,
+                )
+            } else {
+                self.get_available_pool_idx_excluding(bucket, &object, fi.size, opts.src_pool_idx)
+                    .await
+            };
             let target_pool_idx = resolve_data_movement_resume_target_pool(idx, resume_target_pool_idx, opts.src_pool_idx);
             if is_free_version && target_pool_idx == opts.src_pool_idx {
                 return Err(Error::DiskFull);
@@ -2449,17 +2633,21 @@ impl ECStore {
             return Err(decommission_free_version_overwrite_error(bucket, &object, fi.version_id));
         }
 
-        let result = if is_free_version {
-            self.pools[idx]
-                .get_disks_by_key(&object)
-                .decommission_tier_free_version(bucket, &object, &fi, &opts)
-                .await
-        } else {
-            self.pools[idx]
-                .get_disks_by_key(&object)
-                .decommission_tiered_object(bucket, &object, &fi, &opts)
-                .await
-        };
+        let result = self
+            .run_decommission_capacity_admitted_mutation(idx, DecommissionCapacityOwner::from_options(&opts), None, || async {
+                if is_free_version {
+                    self.pools[idx]
+                        .get_disks_by_key(&object)
+                        .decommission_tier_free_version(bucket, &object, &fi, &opts)
+                        .await
+                } else {
+                    self.pools[idx]
+                        .get_disks_by_key(&object)
+                        .decommission_tiered_object(bucket, &object, &fi, &opts)
+                        .await
+                }
+            })
+            .await;
         if matches!(result, Err(Error::PreconditionFailed)) {
             if self
                 .has_equivalent_data_movement_tiered_object(bucket, &object, &fi, &opts, idx)
@@ -2611,15 +2799,29 @@ impl ECStore {
             return Err(Error::other("data movement PUT requires data_movement options"));
         }
         let (object, mut opts) = self.prepare_put_object(bucket, object, opts).await?;
+        ensure_decommission_capacity_mutation_id(bucket, &object, &mut opts);
         let idx = self
             .select_put_object_pool_idx(bucket, object.as_str(), data.size(), &opts)
             .await?;
         self.apply_decommission_target_mutation_fence(idx, object.as_str(), &mut opts, mutation_fence)
             .await;
-        let result = self.pools[idx]
-            .put_object_with_old_current_size(bucket, &object, data, &opts)
-            .await
-            .map(|(object_info, _)| object_info);
+        let expected_data_bytes = usize::try_from(data.size()).ok();
+        let result = self
+            .run_decommission_capacity_admitted_mutation_with_capacity_lease(
+                idx,
+                DecommissionCapacityOwner::from_options(&opts),
+                expected_data_bytes,
+                |capacity_lease| async move {
+                    if let Some(capacity_lease) = capacity_lease {
+                        opts.add_namespace_lock_lost_signal(capacity_lease);
+                    }
+                    self.pools[idx]
+                        .put_object_with_old_current_size(bucket, &object, data, &opts)
+                        .await
+                        .map(|(object_info, _)| object_info)
+                },
+            )
+            .await;
         let result = enqueue_transition_after_write(result, LcEventSrc::S3PutObject).await;
         if result.is_ok() {
             list_objects::observe_list_objects_mutation(self, bucket).await;
@@ -2640,11 +2842,10 @@ impl ECStore {
         let idx = self
             .select_put_object_pool_idx(bucket, object.as_str(), data.size(), &opts)
             .await?;
-
-        // Keep PUT atomic-read friendly: SetDisks takes the object write lock only
-        // around precondition checks and the final rename/commit.
+        let mut opts = opts;
+        opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
         self.pools[idx]
-            .put_object_with_old_current_size(bucket, &object, data, &opts)
+            .put_object_with_old_current_size(bucket, object.as_str(), data, &opts)
             .await
     }
 
@@ -2761,8 +2962,20 @@ impl ECStore {
                 && let (Some(src_vid), Some(dst_vid)) = (&src_opts.version_id, &dst_opts.version_id)
                 && src_vid == dst_vid
             {
-                return self.pools[pool_idx]
-                    .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, &dst_opts)
+                let capacity_object = dst_object.clone();
+                return self
+                    .run_external_decommission_capacity_object_mutation(
+                        pool_idx,
+                        dst_bucket,
+                        &capacity_object,
+                        &capacity_object,
+                        dst_opts.clone(),
+                        |opts| async move {
+                            self.pools[pool_idx]
+                                .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, &opts)
+                                .await
+                        },
+                    )
                     .await;
             }
 
@@ -2777,12 +2990,24 @@ impl ECStore {
                     // metadata_only decision in rustfs/src/app/object_usecase.rs); the sibling
                     // versioned branch below resolves the same risk by rewriting through
                     // put_object (issue #4238).
-                    return self.pools[pool_idx]
-                        .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, &dst_opts)
+                    let capacity_object = dst_object.clone();
+                    return self
+                        .run_external_decommission_capacity_object_mutation(
+                            pool_idx,
+                            dst_bucket,
+                            &capacity_object,
+                            &capacity_object,
+                            dst_opts.clone(),
+                            |opts| async move {
+                                self.pools[pool_idx]
+                                    .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, &opts)
+                                    .await
+                            },
+                        )
                         .await;
                 }
                 // Transitioned object self-copy: restore from tier into the same pool.
-                let put_opts = ObjectOptions {
+                let mut put_opts = ObjectOptions {
                     user_defined: (*src_info.user_defined).clone(),
                     versioned: dst_opts.versioned,
                     version_id: dst_opts.version_id.clone(),
@@ -2796,6 +3021,8 @@ impl ECStore {
                     object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
                     ..Default::default()
                 };
+                put_opts.decommission_capacity_admission =
+                    crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
                 return if let Some(reader) = src_info.put_object_reader.as_mut() {
                     self.pools[pool_idx]
                         .put_object(dst_bucket, &dst_object, reader, &put_opts)
@@ -2816,7 +3043,7 @@ impl ECStore {
                 // stays consistent with the new version's metadata. Sharing the source data_dir via
                 // a metadata-only version copy would corrupt SSE/compressed objects (issue #4238).
                 if let Some(reader) = src_info.put_object_reader.as_mut() {
-                    let put_opts = ObjectOptions {
+                    let mut put_opts = ObjectOptions {
                         user_defined: (*src_info.user_defined).clone(),
                         versioned: dst_opts.versioned,
                         version_id: dst_opts.version_id.clone(),
@@ -2830,13 +3057,27 @@ impl ECStore {
                         object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
                         ..Default::default()
                     };
+                    put_opts.decommission_capacity_admission =
+                        crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
                     return self.pools[pool_idx]
                         .put_object(dst_bucket, &dst_object, reader, &put_opts)
                         .await;
                 }
                 src_info.version_only = true;
-                return self.pools[pool_idx]
-                    .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, &dst_opts)
+                let capacity_object = dst_object.clone();
+                return self
+                    .run_external_decommission_capacity_object_mutation(
+                        pool_idx,
+                        dst_bucket,
+                        &capacity_object,
+                        &capacity_object,
+                        dst_opts.clone(),
+                        |opts| async move {
+                            self.pools[pool_idx]
+                                .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, &opts)
+                                .await
+                        },
+                    )
                     .await;
             }
         }
@@ -2846,8 +3087,9 @@ impl ECStore {
         } else {
             self.get_pool_idx(dst_bucket, &dst_object, src_info.size).await?
         };
+        let dst_object_name = dst_object.as_str();
 
-        let put_opts = ObjectOptions {
+        let mut put_opts = ObjectOptions {
             user_defined: (*src_info.user_defined).clone(),
             versioned: dst_opts.versioned,
             version_id: dst_opts.version_id.clone(),
@@ -2861,10 +3103,11 @@ impl ECStore {
             object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
             ..Default::default()
         };
+        put_opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
 
         if let Some(put_object_reader) = src_info.put_object_reader.as_mut() {
             return self.pools[pool_idx]
-                .put_object(dst_bucket, &dst_object, put_object_reader, &put_opts)
+                .put_object(dst_bucket, dst_object_name, put_object_reader, &put_opts)
                 .await;
         }
 
@@ -2982,6 +3225,7 @@ impl ECStore {
         };
         let object = object.as_str();
         let mut opts = opts;
+        ensure_decommission_capacity_mutation_id(bucket, object, &mut opts);
         let delete_all_configs = if opts.lifecycle_delete_all.is_some() {
             Some(get_expiry_configs(self, bucket).await?)
         } else {
@@ -3132,11 +3376,21 @@ impl ECStore {
             let selected_target_pool_idx =
                 match select_data_movement_target_pool(existing_pool_idx, opts.src_pool_idx, opts.delete_marker)? {
                     Some(pool_idx) => pool_idx,
-                    None => self.get_pool_idx_no_lock(bucket, object, 0).await?,
+                    None => {
+                        if let Some(owner) = DecommissionCapacityOwner::from_options(&opts) {
+                            self.select_decommission_capacity_target_pool(owner, 0).await?
+                        } else {
+                            self.get_pool_idx_no_lock(bucket, object, 0).await?
+                        }
+                    }
                 };
             let resume_target_pool_idx = if selected_target_pool_idx == opts.src_pool_idx {
-                self.get_available_pool_idx_excluding(bucket, object, 0, opts.src_pool_idx)
-                    .await
+                if let Some(owner) = DecommissionCapacityOwner::from_options(&opts) {
+                    Some(self.select_decommission_capacity_target_pool(owner, 0).await?)
+                } else {
+                    self.get_available_pool_idx_excluding(bucket, object, 0, opts.src_pool_idx)
+                        .await
+                }
             } else {
                 None
             };
@@ -3174,6 +3428,10 @@ impl ECStore {
                     .await?;
                 if let Some(target) = target {
                     if is_equivalent_data_movement_delete_marker(&source, &target) {
+                        if let Some(owner) = DecommissionCapacityOwner::from_options(&target_opts) {
+                            self.reconcile_decommission_capacity_after_equivalent_target(owner, target_pool_idx, 0)
+                                .await?;
+                        }
                         let mut target = target;
                         target.name = decode_dir_object(object);
                         return Ok(target);
@@ -3214,7 +3472,20 @@ impl ECStore {
             }
 
             let target_opts = delete_marker_target_opts.unwrap_or(opts);
-            let mut obj = self.pools[target_pool_idx].delete_object(bucket, object, target_opts).await?;
+            let mut obj = self
+                .run_decommission_capacity_admitted_mutation_with_capacity_lease(
+                    target_pool_idx,
+                    DecommissionCapacityOwner::from_options(&target_opts),
+                    None,
+                    |capacity_lease| async move {
+                        let mut target_opts = target_opts;
+                        if let Some(capacity_lease) = capacity_lease {
+                            target_opts.add_namespace_lock_lost_signal(capacity_lease);
+                        }
+                        self.pools[target_pool_idx].delete_object(bucket, object, target_opts).await
+                    },
+                )
+                .await?;
             obj.name = decode_dir_object(obj.name.as_str());
             return Ok(obj);
         }
@@ -3225,7 +3496,17 @@ impl ECStore {
             Err(err) if is_err_read_quorum(&err) => return Err(StorageError::ErasureWriteQuorum),
             Err(err) if is_err_object_not_found(&err) && should_create_delete_marker_for_missing_object(&opts) => {
                 let target_pool_idx = self.get_pool_idx_no_lock(bucket, object, 0).await?;
-                let mut obj = self.pools[target_pool_idx].delete_object(bucket, object, opts).await?;
+                let pool = self.pools[target_pool_idx].clone();
+                let mut obj = self
+                    .run_external_decommission_capacity_object_mutation(
+                        target_pool_idx,
+                        bucket,
+                        object,
+                        object,
+                        opts,
+                        |opts| async move { pool.delete_object(bucket, object, opts).await },
+                    )
+                    .await?;
                 #[cfg(test)]
                 pause_versioned_delete_marker_after_commit(bucket, object).await;
                 obj.name = decode_dir_object(object);
@@ -3288,7 +3569,19 @@ impl ECStore {
                 continue;
             }
 
-            match pool.delete_object(bucket, object, opts.clone()).await {
+            let pool_idx = pool.pool_idx;
+            let pool = pool.clone();
+            match self
+                .run_external_decommission_capacity_object_mutation(
+                    pool_idx,
+                    bucket,
+                    object,
+                    object,
+                    opts.clone(),
+                    |opts| async move { pool.delete_object(bucket, object, opts).await },
+                )
+                .await
+            {
                 Ok(res) => {
                     #[cfg(test)]
                     pause_versioned_delete_marker_after_commit(bucket, object).await;
@@ -3474,6 +3767,19 @@ impl ECStore {
             }
         }
 
+        let _capacity_fence = if latest_marker_objects.iter().any(|creates_marker| *creates_marker) {
+            let target_pool_indices = (0..self.pools.len()).collect::<Vec<_>>();
+            match self
+                .acquire_external_decommission_capacity_fence(&target_pool_indices, "batch_delete")
+                .await
+            {
+                Ok(fence) => Some(fence),
+                Err(err) => return return_batch_delete_lock_error_with_accounting(objects.as_slice(), err),
+            }
+        } else {
+            None
+        };
+
         let mut futures = Vec::with_capacity(self.pools.len());
         for pool in self.pools.iter() {
             if self.is_pool_rebalancing(pool.pool_idx).await {
@@ -3777,20 +4083,23 @@ impl ECStore {
         // re-check is tracked separately. Holding the lock here instead
         // (#4877) blocked HEAD/get_object_info for the whole copy-back and
         // self-deadlocked on the inner commits.
+        let object_name = object.as_str();
         if self.single_pool() {
+            opts.decommission_capacity_admission = Some(Arc::clone(&self));
             return self.pools[0]
                 .clone()
-                .restore_transitioned_object(bucket, &object, &opts)
+                .restore_transitioned_object(bucket, object_name, &opts)
                 .await;
         }
 
         let (_, idx) = self
-            .get_latest_accessible_object_info_with_idx(bucket, object.as_str(), &writer_pool_lookup_opts(&opts, opts.no_lock))
+            .get_latest_accessible_object_info_with_idx(bucket, object_name, &writer_pool_lookup_opts(&opts, opts.no_lock))
             .await?;
 
+        opts.decommission_capacity_admission = Some(Arc::clone(&self));
         self.pools[idx]
             .clone()
-            .restore_transitioned_object(bucket, &object, &opts)
+            .restore_transitioned_object(bucket, object_name, &opts)
             .await
     }
 
@@ -3824,14 +4133,36 @@ impl ECStore {
         };
 
         if self.single_pool() {
-            return self.pools[0].put_object_metadata(bucket, object.as_str(), &opts).await;
+            let pool = self.pools[0].clone();
+            let capacity_object = object.clone();
+            return self
+                .run_external_decommission_capacity_object_mutation(
+                    0,
+                    bucket,
+                    capacity_object.as_str(),
+                    capacity_object.as_str(),
+                    opts,
+                    |opts| async move { pool.put_object_metadata(bucket, object.as_str(), &opts).await },
+                )
+                .await;
         }
 
         let (_, idx) = self
             .get_latest_accessible_object_info_with_idx(bucket, object.as_str(), &writer_pool_lookup_opts(&opts, opts.no_lock))
             .await?;
 
-        let result = self.pools[idx].put_object_metadata(bucket, object.as_str(), &opts).await;
+        let pool = self.pools[idx].clone();
+        let capacity_object = object.clone();
+        let result = self
+            .run_external_decommission_capacity_object_mutation(
+                idx,
+                bucket,
+                capacity_object.as_str(),
+                capacity_object.as_str(),
+                opts,
+                |opts| async move { pool.put_object_metadata(bucket, object.as_str(), &opts).await },
+            )
+            .await;
         drop(bucket_lifecycle_guard);
         result
     }
@@ -3857,16 +4188,34 @@ impl ECStore {
         opts: &ObjectOptions,
     ) -> Result<ObjectInfo> {
         let object = encode_dir_object(object);
+        let object_name = object.as_str();
 
         if self.single_pool() {
-            return self.pools[0].put_object_tags(bucket, object.as_str(), tags, opts).await;
+            return self
+                .run_external_decommission_capacity_object_mutation(
+                    0,
+                    bucket,
+                    object_name,
+                    object_name,
+                    opts.clone(),
+                    |opts| async move { self.pools[0].put_object_tags(bucket, object_name, tags, &opts).await },
+                )
+                .await;
         }
 
         let (_, idx) = self
-            .get_latest_accessible_object_info_with_idx(bucket, object.as_str(), &writer_pool_lookup_opts(opts, opts.no_lock))
+            .get_latest_accessible_object_info_with_idx(bucket, object_name, &writer_pool_lookup_opts(opts, opts.no_lock))
             .await?;
 
-        self.pools[idx].put_object_tags(bucket, object.as_str(), tags, opts).await
+        self.run_external_decommission_capacity_object_mutation(
+            idx,
+            bucket,
+            object_name,
+            object_name,
+            opts.clone(),
+            |opts| async move { self.pools[idx].put_object_tags(bucket, object_name, tags, &opts).await },
+        )
+        .await
     }
 
     #[instrument(skip(self))]
@@ -3892,16 +4241,34 @@ impl ECStore {
     #[instrument(skip(self))]
     pub(super) async fn handle_delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
         let object = encode_dir_object(object);
+        let object_name = object.as_str();
 
         if self.single_pool() {
-            return self.pools[0].delete_object_tags(bucket, object.as_str(), opts).await;
+            return self
+                .run_external_decommission_capacity_object_mutation(
+                    0,
+                    bucket,
+                    object_name,
+                    object_name,
+                    opts.clone(),
+                    |opts| async move { self.pools[0].delete_object_tags(bucket, object_name, &opts).await },
+                )
+                .await;
         }
 
         let (_, idx) = self
-            .get_latest_accessible_object_info_with_idx(bucket, object.as_str(), &writer_pool_lookup_opts(opts, opts.no_lock))
+            .get_latest_accessible_object_info_with_idx(bucket, object_name, &writer_pool_lookup_opts(opts, opts.no_lock))
             .await?;
 
-        self.pools[idx].delete_object_tags(bucket, object.as_str(), opts).await
+        self.run_external_decommission_capacity_object_mutation(
+            idx,
+            bucket,
+            object_name,
+            object_name,
+            opts.clone(),
+            |opts| async move { self.pools[idx].delete_object_tags(bucket, object_name, &opts).await },
+        )
+        .await
     }
 
     pub(super) async fn handle_verify_object_integrity(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> {

crates/ecstore/src/store/rebalance.rs

@@ -925,9 +925,19 @@ impl ECStore {
             }
 
             if let Some(idx) = pe.index {
+                let pool = self.pools[idx].clone();
                 results.push(RebalanceDeletePoolResult {
                     pool_idx: idx,
-                    result: self.pools[idx].delete_object(bucket, object, opts.clone()).await,
+                    result: self
+                        .run_external_decommission_capacity_object_mutation(
+                            idx,
+                            bucket,
+                            object,
+                            object,
+                            opts.clone(),
+                            |opts| async move { pool.delete_object(bucket, object, opts).await },
+                        )
+                        .await,
                 });
             }
         }

crates/log-analyzer/src/rules/seed/capacity.rs

@@ -75,13 +75,13 @@ pub(super) fn rules() -> Vec<Rule> {
         },
         Rule {
             evidence_fields: strings(["required", "target_free"]),
-            anchors: strings(["insufficient target pool capacity"]),
+            anchors: strings(["insufficient reserved physical target capacity"]),
             ..base(
                 "decom-capacity-insufficient",
                 P1Unavailable,
                 "capacity",
                 "pool 下线目标容量不足",
-                contains("insufficient target pool capacity"),
+                contains("insufficient reserved physical target capacity"),
                 "pool 下线迁移目标容量不足,decommission 无法开始。",
                 "先扩容目标 pool。",
             )

crates/log-analyzer/src/rules/seed/tests.rs

@@ -262,7 +262,9 @@ fn every_rule_has_a_positive_sample() {
         ),
         (
             "decom-capacity-insufficient",
-            msg("failed to start decommission: insufficient target pool capacity: required 100 bytes available 50 bytes"),
+            msg(
+                "failed to start decommission: insufficient reserved physical target capacity: required 100 bytes available 50 bytes",
+            ),
         ),
         // ops
         ("decom-object-failed", msg("Decommission object migration failed")),