fix(ecstore): reconcile pending capacity before exact delete (#7016)

fix(ecstore): reconcile capacity before exact delete

crates/ecstore/src/core/pools.rs

@@ -49,7 +49,7 @@ use crate::error::{
     is_err_version_not_found,
 };
 use crate::layout::endpoints::EndpointServerPools;
-use crate::object_api::{DecommissionCapacityOptions, GetObjectReader, ObjectOptions};
+use crate::object_api::{DecommissionCapacityOptions, GetObjectReader, ObjectInfo, ObjectOptions};
 use crate::runtime::sources as runtime_sources;
 use crate::services::notification_sys::{
     acquire_tier_delete_journal_fleet_proof, tier_delete_journal_fleet_proof_matches, tier_delete_journal_topology_generation,
@@ -78,7 +78,9 @@ use rmp_serde::Serializer;
 use rustfs_filemeta::{FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
 use rustfs_heal_contracts::heal_channel::HealOpts;
 use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum};
-use rustfs_utils::path::{encode_dir_object, path_join, path_to_bucket_object, path_to_bucket_object_with_base_path};
+use rustfs_utils::path::{
+    decode_dir_object, encode_dir_object, path_join, path_to_bucket_object, path_to_bucket_object_with_base_path,
+};
 use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, ReplicationConfiguration};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
@@ -7190,6 +7192,170 @@ pub(crate) fn decommission_capacity_mutation_id(
     uuid::Uuid::from_bytes(bytes)
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct ExactDeleteCapacityReconciliation {
+    source_pool_index: usize,
+    target_pool_index: usize,
+    mutation_id: uuid::Uuid,
+    expected_data_bytes: usize,
+    expected_target_physical_bytes: usize,
+}
+
+fn plan_exact_delete_capacity_reconciliations(
+    meta: &PoolMeta,
+    object: &str,
+    exact: &ObjectInfo,
+) -> Result<Vec<ExactDeleteCapacityReconciliation>> {
+    let version_id = exact.version_id.map(|version_id| version_id.to_string());
+    let mut matches = Vec::new();
+
+    for (source_pool_index, pool) in meta.pools.iter().enumerate() {
+        let Some(reservation) = pool
+            .decommission
+            .as_ref()
+            .and_then(|info| info.capacity_reservation.as_ref())
+            .filter(|reservation| reservation.active())
+        else {
+            continue;
+        };
+        let owner = DecommissionCapacityOwner {
+            source_pool_index,
+            operation_id: reservation.operation_id,
+            generation: reservation.generation,
+            owner_nonce: reservation.owner_nonce,
+            mutation_id: None,
+        };
+        let logical_mutation_id = decommission_capacity_mutation_id(
+            owner,
+            &exact.bucket,
+            &exact.name,
+            version_id.as_deref(),
+            exact.delete_marker,
+            exact.mod_time,
+        );
+        // Existing data-movement producers persist directory-key intents using
+        // either the logical name or its internal `__XLDIR__` representation.
+        // Accept both while retaining the exact persisted identity for CAS.
+        let internal_mutation_id = if object == exact.name {
+            logical_mutation_id
+        } else {
+            decommission_capacity_mutation_id(
+                owner,
+                &exact.bucket,
+                object,
+                version_id.as_deref(),
+                exact.delete_marker,
+                exact.mod_time,
+            )
+        };
+        let mut source_match = None;
+
+        for target in &reservation.targets {
+            if target.pending_physical_bytes == 0 {
+                continue;
+            }
+            let Some(pending_mutation_id) = target.pending_mutation_id else {
+                return Err(decommission_capacity_blocked_error(format!(
+                    "source pool {source_pool_index} target pool {} has pending capacity without an object identity",
+                    target.pool_index
+                )));
+            };
+            if pending_mutation_id != logical_mutation_id && pending_mutation_id != internal_mutation_id {
+                continue;
+            }
+            if source_match.is_some() {
+                return Err(decommission_capacity_blocked_error(format!(
+                    "source pool {source_pool_index} has the same exact-delete capacity intent on multiple targets"
+                )));
+            }
+            source_match = Some((target.pool_index, target.layout, target.pending_physical_bytes, pending_mutation_id));
+        }
+
+        let Some((target_pool_index, target_layout, pending_physical_bytes, mutation_id)) = source_match else {
+            continue;
+        };
+        if exact.version_id.is_none() && exact.mod_time.is_none() {
+            return Err(decommission_capacity_blocked_error(
+                "unversioned exact delete cannot identify pending capacity without a modification time",
+            ));
+        }
+        let expected_data_bytes = if exact.delete_marker {
+            0
+        } else {
+            usize::try_from(exact.size).map_err(|_| {
+                decommission_capacity_blocked_error("exact delete cannot reconcile a negative or overflowing object size")
+            })?
+        };
+        let expected_target_physical_bytes = capacity_target_physical_bytes(expected_data_bytes.max(1), target_layout)?;
+        if pending_physical_bytes != expected_target_physical_bytes {
+            return Err(decommission_capacity_blocked_error(format!(
+                "source pool {source_pool_index} target pool {target_pool_index} pending capacity does not match the exact object size"
+            )));
+        }
+        let remaining_target_physical_bytes = reservation
+            .targets
+            .iter()
+            .find(|target| target.pool_index == target_pool_index)
+            .map(|target| {
+                target.remaining_reserved_physical_bytes(reservation.temporary_copies)
+                    / 1usize.saturating_add(reservation.temporary_copies)
+            })
+            .unwrap_or_default();
+        let remaining_total_physical_bytes = reservation
+            .predicted_physical_bytes
+            .saturating_sub(reservation.consumed_target_physical_bytes);
+        let remaining_data_bytes = reservation
+            .source_data_equivalent_bytes
+            .saturating_sub(reservation.committed_data_bytes);
+        if expected_target_physical_bytes > remaining_target_physical_bytes
+            || expected_target_physical_bytes > remaining_total_physical_bytes
+            || expected_data_bytes > remaining_data_bytes
+        {
+            return Err(decommission_capacity_blocked_error(format!(
+                "source pool {source_pool_index} target pool {target_pool_index} lacks reservation capacity for the exact object"
+            )));
+        }
+        matches.push(ExactDeleteCapacityReconciliation {
+            source_pool_index,
+            target_pool_index,
+            mutation_id,
+            expected_data_bytes,
+            expected_target_physical_bytes,
+        });
+    }
+
+    Ok(matches)
+}
+
+fn ensure_exact_delete_capacity_namespace_fences(opts: &ObjectOptions, bucket: &str, object: &str) -> Result<()> {
+    let object_fence = opts.namespace_lock_fence.as_ref().ok_or_else(|| {
+        decommission_capacity_blocked_error("exact delete capacity reconciliation requires an object namespace fence")
+    })?;
+    if object_fence.is_lock_lost() {
+        return Err(StorageError::NamespaceLockQuorumUnavailable {
+            mode: "exact_delete_capacity_reconciliation",
+            bucket: bucket.to_string(),
+            object: decode_dir_object(object),
+            required: 1,
+            achieved: 0,
+        });
+    }
+    if opts
+        .bucket_lifecycle_lock_fence
+        .as_ref()
+        .is_some_and(crate::object_api::NamespaceLockFence::is_lock_lost)
+    {
+        return Err(StorageError::NamespaceLockQuorumUnavailable {
+            mode: "exact_delete_capacity_bucket_generation",
+            bucket: bucket.to_string(),
+            object: decode_dir_object(object),
+            required: 1,
+            achieved: 0,
+        });
+    }
+    Ok(())
+}
+
 pub(crate) fn ensure_decommission_capacity_mutation_id(bucket: &str, object: &str, opts: &mut ObjectOptions) {
     if opts
         .decommission_capacity
@@ -8253,6 +8419,117 @@ impl ECStore {
             .await
     }
 
+    pub(crate) async fn reconcile_decommission_capacity_before_exact_delete(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+        exact: &ObjectInfo,
+    ) -> Result<()> {
+        if exact.bucket != bucket || exact.name != decode_dir_object(object) {
+            return Err(decommission_capacity_blocked_error(
+                "exact delete object identity changed before capacity reconciliation",
+            ));
+        }
+
+        let reconciliations = {
+            let mut save_guard = self.pool_meta_save_gate.lock().await;
+            let (_read_guard, snapshot) = self
+                .acquire_pool_meta_read_guard(&mut save_guard, "exact delete capacity reconciliation failed")
+                .await?;
+            plan_exact_delete_capacity_reconciliations(&snapshot, object, exact)?
+        };
+        if reconciliations.is_empty() {
+            return Ok(());
+        }
+        ensure_exact_delete_capacity_namespace_fences(opts, bucket, object)?;
+
+        let target_lookup_options = ObjectOptions {
+            versioned: opts.versioned,
+            version_suspended: opts.version_suspended,
+            version_id: opts.version_id.clone(),
+            metadata_chg: opts.version_id.is_some(),
+            no_lock: true,
+            ..Default::default()
+        };
+        for reconciliation in &reconciliations {
+            let target_pool = self.pools.get(reconciliation.target_pool_index).ok_or_else(|| {
+                decommission_capacity_blocked_error(format!(
+                    "source pool {} exact-delete capacity target pool {} is out of range",
+                    reconciliation.source_pool_index, reconciliation.target_pool_index
+                ))
+            })?;
+            let target = target_pool
+                .get_object_info(bucket, object, &target_lookup_options)
+                .await
+                .map_err(|err| {
+                    decommission_capacity_blocked_error(format!(
+                        "source pool {} target pool {} exact object evidence could not be read: {err}",
+                        reconciliation.source_pool_index, reconciliation.target_pool_index
+                    ))
+                })?;
+            if !Self::is_equivalent_decommission_capacity_target(exact, &target) {
+                return Err(decommission_capacity_blocked_error(format!(
+                    "source pool {} target pool {} does not contain an equivalent exact object for its pending capacity intent",
+                    reconciliation.source_pool_index, reconciliation.target_pool_index
+                )));
+            }
+        }
+        ensure_exact_delete_capacity_namespace_fences(opts, bucket, object)?;
+
+        let mut save_guard = self.pool_meta_save_gate.lock().await;
+        let (write_guard, mut snapshot) = self
+            .acquire_pool_meta_write_guard(&mut save_guard, "exact delete capacity reconciliation failed")
+            .await?;
+        let current_reconciliations = plan_exact_delete_capacity_reconciliations(&snapshot, object, exact)?;
+        if current_reconciliations != reconciliations {
+            return Err(decommission_capacity_blocked_error(
+                "pending capacity changed while exact target evidence was being verified",
+            ));
+        }
+        ensure_exact_delete_capacity_namespace_fences(opts, bucket, object)?;
+
+        let now = OffsetDateTime::now_utc();
+        let mut source_pool_indices = Vec::with_capacity(current_reconciliations.len());
+        for reconciliation in current_reconciliations {
+            resolve_decommission_target_pending(
+                &mut snapshot,
+                reconciliation.source_pool_index,
+                reconciliation.target_pool_index,
+                reconciliation.expected_target_physical_bytes,
+                reconciliation.mutation_id,
+            )?;
+            record_decommission_target_consumption(
+                &mut snapshot,
+                reconciliation.source_pool_index,
+                reconciliation.target_pool_index,
+                DecommissionTargetConsumption {
+                    committed_data_bytes: reconciliation.expected_data_bytes,
+                    target_physical_bytes: reconciliation.expected_target_physical_bytes,
+                    observed_physical_bytes: 0,
+                },
+                reconciliation.mutation_id,
+                now,
+            )?;
+            source_pool_indices.push(reconciliation.source_pool_index);
+        }
+        source_pool_indices.sort_unstable();
+        source_pool_indices.dedup();
+        ensure_exact_delete_capacity_namespace_fences(opts, bucket, object)?;
+
+        let outcome = snapshot
+            .save_no_lock_armed(self.pools.clone(), &mut save_guard, write_guard.lock_lost_signal(), &source_pool_indices)
+            .await?;
+        ensure_pool_meta_write_fence(&write_guard, "exact delete capacity reconciliation save failed")?;
+        {
+            let mut pool_meta = self.pool_meta.write().await;
+            publish_pool_meta_updates(&mut pool_meta, &outcome.committed, &source_pool_indices);
+        }
+        ensure_pool_meta_write_fence(&write_guard, "exact delete capacity reconciliation save failed")?;
+        outcome.disarm();
+        Ok(())
+    }
+
     pub(crate) async fn reconcile_decommission_capacity_after_equivalent_target(
         &self,
         owner: DecommissionCapacityOwner,
@@ -17180,7 +17457,8 @@ mod pools_tests {
     use super::{
         DecommissionCapacityOwner, DecommissionCapacityReservation, DecommissionCapacityTemporaryMutation,
         decommission_capacity_mutation_id, ensure_decommission_target_owner_admission,
-        ensure_external_decommission_target_admission, is_decommission_capacity_blocked_error,
+        ensure_exact_delete_capacity_namespace_fences, ensure_external_decommission_target_admission,
+        is_decommission_capacity_blocked_error, plan_exact_delete_capacity_reconciliations,
         record_decommission_target_consumption, reserve_decommission_target_pending, resolve_decommission_target_pending,
         set_decommission_capacity_info_overrides_for_test,
     };
@@ -17195,7 +17473,7 @@ mod pools_tests {
     use crate::disk::{STORAGE_FORMAT_FILE, endpoint::Endpoint};
     use crate::error::{Error, StorageError};
     use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
-    use crate::object_api::ObjectOptions;
+    use crate::object_api::{ObjectInfo, ObjectOptions};
     use crate::runtime::instance::InstanceContext;
     use crate::services::rebalance::{RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats};
     use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
@@ -21316,6 +21594,135 @@ mod pools_tests {
         assert_eq!(first, recovered, "a lease nonce rotation must not change the mutation identity");
     }
 
+    #[test]
+    fn exact_delete_capacity_plan_requires_identity_and_exact_size() {
+        let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(2);
+        let layout = DecommissionErasureLayout { data: 1, parity: 0 };
+        let capacity_infos = vec![
+            DecommissionPoolCapacityInfo::for_test(0, layout, 0, 30, 30),
+            DecommissionPoolCapacityInfo::for_test(1, layout, 60, 60, 0),
+        ];
+        let mut meta = PoolMeta {
+            version: POOL_META_VERSION,
+            pools: vec![decommission_test_pool_status(0, None), decommission_test_pool_status(1, None)],
+            ..Default::default()
+        };
+        meta.decommission(0, capacity_infos[0].space).unwrap();
+        reserve_decommission_start_target_capacity(&mut meta, &[0], &capacity_infos, uuid::Uuid::new_v4(), 1, now)
+            .expect("the exact-delete test reservation should fit");
+        let exact = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            version_id: Some(uuid::Uuid::from_u128(7)),
+            mod_time: Some(now),
+            size: 10,
+            ..Default::default()
+        };
+        let owner = {
+            let reservation = meta.pools[0]
+                .decommission
+                .as_ref()
+                .and_then(|info| info.capacity_reservation.as_ref())
+                .expect("the exact-delete test reservation should exist");
+            DecommissionCapacityOwner {
+                source_pool_index: 0,
+                operation_id: reservation.operation_id,
+                generation: reservation.generation,
+                owner_nonce: reservation.owner_nonce,
+                mutation_id: None,
+            }
+        };
+        let version_id = exact.version_id.map(|version_id| version_id.to_string());
+        let mutation_id = decommission_capacity_mutation_id(
+            owner,
+            &exact.bucket,
+            &exact.name,
+            version_id.as_deref(),
+            exact.delete_marker,
+            exact.mod_time,
+        );
+        reserve_decommission_target_pending(&mut meta, 0, 1, 10, mutation_id, now + Duration::seconds(1))
+            .expect("the exact-delete test intent should be reserved");
+
+        let plan = plan_exact_delete_capacity_reconciliations(&meta, &exact.name, &exact)
+            .expect("the exact identity should match the pending intent");
+        assert_eq!(plan.len(), 1);
+        assert_eq!(plan[0].source_pool_index, 0);
+        assert_eq!(plan[0].target_pool_index, 1);
+        assert_eq!(plan[0].expected_data_bytes, 10);
+        assert_eq!(plan[0].expected_target_physical_bytes, 10);
+
+        let mismatched_size = ObjectInfo {
+            size: 9,
+            ..exact.clone()
+        };
+        let mismatched_size = plan_exact_delete_capacity_reconciliations(&meta, &mismatched_size.name, &mismatched_size)
+            .expect_err("a different exact size must not consume the pending intent");
+        assert!(mismatched_size.to_string().contains("does not match the exact object size"));
+
+        let directory_exact = ObjectInfo {
+            name: "directory/".to_string(),
+            ..exact.clone()
+        };
+        let internal_directory = rustfs_utils::path::encode_dir_object(&directory_exact.name);
+        let internal_directory_mutation_id = decommission_capacity_mutation_id(
+            owner,
+            &directory_exact.bucket,
+            &internal_directory,
+            version_id.as_deref(),
+            directory_exact.delete_marker,
+            directory_exact.mod_time,
+        );
+        meta.pools[0]
+            .decommission
+            .as_mut()
+            .and_then(|info| info.capacity_reservation.as_mut())
+            .expect("the exact-delete test reservation should exist")
+            .targets[0]
+            .pending_mutation_id = Some(internal_directory_mutation_id);
+        let directory_plan = plan_exact_delete_capacity_reconciliations(&meta, &internal_directory, &directory_exact)
+            .expect("an internally encoded directory intent should match its logical exact object");
+        assert_eq!(directory_plan[0].mutation_id, internal_directory_mutation_id);
+
+        let logical_directory_mutation_id = decommission_capacity_mutation_id(
+            owner,
+            &directory_exact.bucket,
+            &directory_exact.name,
+            version_id.as_deref(),
+            directory_exact.delete_marker,
+            directory_exact.mod_time,
+        );
+        meta.pools[0]
+            .decommission
+            .as_mut()
+            .and_then(|info| info.capacity_reservation.as_mut())
+            .expect("the exact-delete test reservation should exist")
+            .targets[0]
+            .pending_mutation_id = Some(logical_directory_mutation_id);
+        let directory_plan = plan_exact_delete_capacity_reconciliations(&meta, &internal_directory, &directory_exact)
+            .expect("a logical directory intent should match its internally encoded delete path");
+        assert_eq!(directory_plan[0].mutation_id, logical_directory_mutation_id);
+
+        meta.pools[0]
+            .decommission
+            .as_mut()
+            .and_then(|info| info.capacity_reservation.as_mut())
+            .expect("the exact-delete test reservation should exist")
+            .targets[0]
+            .pending_mutation_id = None;
+        let unidentified = plan_exact_delete_capacity_reconciliations(&meta, &exact.name, &exact)
+            .expect_err("an unidentified pending intent must fail closed");
+        assert!(unidentified.to_string().contains("without an object identity"));
+
+        let mut opts = ObjectOptions::default();
+        let unfenced = ensure_exact_delete_capacity_namespace_fences(&opts, &exact.bucket, &exact.name)
+            .expect_err("capacity reconciliation must reject a missing object namespace fence");
+        assert!(unfenced.to_string().contains("requires an object namespace fence"));
+        opts.ensure_namespace_lock_fence();
+        ensure_exact_delete_capacity_namespace_fences(&opts, &exact.bucket, &exact.name)
+            .expect("a live object namespace fence should admit capacity reconciliation");
+    }
+
     #[test]
     fn ordinary_write_admission_cannot_race_into_a_reserved_target() {
         let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(2);

crates/ecstore/src/core/pools_test.rs

@@ -196,7 +196,7 @@ mod decommission_lock_order_tests {
     use crate::bucket::lifecycle::lifecycle::TRANSITION_PENDING;
     use crate::core::pools::{
         DecommissionCapacityLockOrderBarrier, DecommissionCapacityOwner, DecommissionErasureLayout, DecommissionPoolCapacityInfo,
-        POOL_META_NAME, set_decommission_capacity_info_overrides_for_test,
+        POOL_META_NAME, decommission_capacity_mutation_id, set_decommission_capacity_info_overrides_for_test,
     };
     use crate::data_movement;
     use crate::disk::RUSTFS_META_BUCKET;
@@ -3047,6 +3047,198 @@ mod decommission_lock_order_tests {
         );
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn exact_delete_reconciles_pending_capacity_before_removing_replicas() {
+        let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
+        let bucket = test_bucket("exact-delete-capacity");
+        let object = "published-before-exact-delete.bin";
+        let body = vec![0x55; 64 * 1024];
+        let version_id = uuid::Uuid::new_v4().to_string();
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create the exact-delete reconciliation bucket");
+        let incarnation = store
+            .bucket_incarnation_id(&bucket)
+            .await
+            .expect("load the exact-delete bucket incarnation");
+
+        let mut source_data = PutObjReader::from_vec(body.clone());
+        let source = store.pools[0]
+            .put_object(
+                &bucket,
+                object,
+                &mut source_data,
+                &ObjectOptions {
+                    versioned: true,
+                    version_id: Some(version_id.clone()),
+                    expected_bucket_incarnation_id: Some(incarnation),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("seed the exact source version");
+        let layout = DecommissionErasureLayout { data: 1, parity: 0 };
+        let target_total = body.len().saturating_mul(4);
+        set_decommission_capacity_info_overrides_for_test(
+            store.id,
+            vec![vec![
+                DecommissionPoolCapacityInfo::for_test(0, layout, 0, body.len(), body.len()),
+                DecommissionPoolCapacityInfo::for_test(1, layout, 0, target_total, target_total),
+                DecommissionPoolCapacityInfo::for_test(2, layout, target_total, target_total, 0),
+            ]],
+        );
+        store
+            .save_current_pool_meta_for_decommission_start(&[0], Vec::new())
+            .await
+            .expect("activate the exact-delete capacity reservation");
+        let owner = decommission_capacity_owner(&*store.pool_meta.read().await);
+        let target_pool_index = store.pool_meta.read().await.pools[0]
+            .decommission
+            .as_ref()
+            .and_then(|info| info.capacity_reservation.as_ref())
+            .expect("the exact-delete capacity reservation should exist")
+            .targets[0]
+            .pool_index;
+        assert_eq!(target_pool_index, 2);
+
+        let target_options = ObjectOptions {
+            versioned: true,
+            version_id: Some(version_id.clone()),
+            mod_time: source.mod_time,
+            preserve_etag: source.etag.clone(),
+            user_defined: (*source.user_defined).clone(),
+            data_movement: true,
+            src_pool_idx: 0,
+            expected_bucket_incarnation_id: Some(incarnation),
+            ..Default::default()
+        };
+        let mut target_data = PutObjReader::from_vec(body.clone());
+        let target = store.pools[target_pool_index]
+            .put_object(&bucket, object, &mut target_data, &target_options)
+            .await
+            .expect("publish the target version before capacity progress");
+        assert_eq!(target.version_id, source.version_id);
+        assert_eq!(target.mod_time, source.mod_time);
+        assert_eq!(target.size, source.size);
+
+        let source_version_id = source.version_id.map(|version_id| version_id.to_string());
+        let mutation_id = decommission_capacity_mutation_id(
+            owner,
+            &source.bucket,
+            &source.name,
+            source_version_id.as_deref(),
+            source.delete_marker,
+            source.mod_time,
+        );
+        {
+            let mut pool_meta = store.pool_meta.write().await;
+            let source_pool = &mut pool_meta.pools[0];
+            let reservation = source_pool
+                .decommission
+                .as_mut()
+                .and_then(|info| info.capacity_reservation.as_mut())
+                .expect("the exact-delete capacity reservation should remain active");
+            let target = reservation
+                .targets
+                .iter_mut()
+                .find(|target| target.pool_index == target_pool_index)
+                .expect("the exact-delete target allocation should exist");
+            target.pending_physical_bytes = body.len();
+            target.pending_mutation_id = Some(mutation_id);
+            reservation.pending_target_physical_bytes = body.len();
+            source_pool.last_update = time::OffsetDateTime::now_utc();
+        }
+        store
+            .save_current_pool_meta_for_test(&[0])
+            .await
+            .expect("persist the simulated post-commit capacity intent");
+
+        let exact_delete_options = ObjectOptions {
+            versioned: true,
+            version_id: Some(version_id.clone()),
+            expected_bucket_incarnation_id: Some(incarnation),
+            ..Default::default()
+        };
+        store.pools[target_pool_index]
+            .delete_object(&bucket, object, exact_delete_options.clone())
+            .await
+            .expect("remove the target evidence before the fail-closed exact delete");
+        let delete_err = store
+            .delete_object(&bucket, object, exact_delete_options.clone())
+            .await
+            .expect_err("exact delete must fail while its pending target evidence is absent");
+        assert!(matches!(delete_err, crate::error::Error::DecommissionCapacityBlocked { .. }));
+        store.pools[0]
+            .get_object_info(
+                &bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    version_id: Some(version_id.clone()),
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("a failed reconciliation must preserve the source evidence");
+        let mut failed = crate::core::pools::PoolMeta::default();
+        failed
+            .load_no_lock_from_replicas(store.pools.clone())
+            .await
+            .expect("the failed exact delete must preserve readable capacity metadata");
+        let failed_reservation = failed.pools[0]
+            .decommission
+            .as_ref()
+            .and_then(|info| info.capacity_reservation.as_ref())
+            .expect("the failed exact delete reservation should remain present");
+        assert_eq!(failed_reservation.pending_target_physical_bytes, body.len());
+        assert_eq!(failed_reservation.consumed_target_physical_bytes, 0);
+
+        let mut replacement_target_data = PutObjReader::from_vec(body.clone());
+        store.pools[target_pool_index]
+            .put_object(&bucket, object, &mut replacement_target_data, &target_options)
+            .await
+            .expect("restore the equivalent target evidence for the exact-delete retry");
+
+        store
+            .delete_object(&bucket, object, exact_delete_options)
+            .await
+            .expect("the exact delete should reconcile capacity before removing replicas");
+
+        let mut persisted = crate::core::pools::PoolMeta::default();
+        persisted
+            .load_no_lock_from_replicas(store.pools.clone())
+            .await
+            .expect("the exact-delete reconciliation should remain durable");
+        let reservation = persisted.pools[0]
+            .decommission
+            .as_ref()
+            .and_then(|info| info.capacity_reservation.as_ref())
+            .expect("the reconciled reservation should remain present");
+        assert_eq!(reservation.pending_target_physical_bytes, 0);
+        assert_eq!(reservation.consumed_target_physical_bytes, body.len());
+        assert_eq!(reservation.committed_data_bytes, body.len());
+
+        for pool_index in [0, target_pool_index] {
+            let err = store.pools[pool_index]
+                .get_object_info(
+                    &bucket,
+                    object,
+                    &ObjectOptions {
+                        versioned: true,
+                        version_id: Some(version_id.clone()),
+                        no_lock: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect_err("the exact version should be absent after reconciliation and delete");
+            assert!(crate::error::is_err_object_not_found(&err) || crate::error::is_err_version_not_found(&err));
+        }
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn data_movement_equivalent_target_reconciles_published_capacity_after_restart() {

crates/ecstore/src/store/object.rs

@@ -2353,6 +2353,16 @@ impl ECStore {
         })
     }
 
+    pub(crate) fn is_equivalent_decommission_capacity_target(source: &ObjectInfo, target: &ObjectInfo) -> bool {
+        source.bucket == target.bucket
+            && source.name == decode_dir_object(&target.name)
+            && if source.delete_marker {
+                is_equivalent_data_movement_delete_marker(source, target)
+            } else {
+                crate::data_movement::is_equivalent_data_movement_object_identity(source, target, true, false)
+            }
+    }
+
     /// Captures Object Lock state once for a batch of PUTs to the same bucket.
     /// `handle_put_object` only reuses the token for the same store, bucket,
     /// bucket incarnation, and Object Lock configuration revision.
@@ -4440,7 +4450,9 @@ impl ECStore {
         }
 
         if should_delete_from_all_pools(&opts, errs.len()) {
-            let mut obj = self.delete_object_from_all_pools(bucket, object, &opts, errs).await?;
+            let mut obj = self
+                .delete_object_from_all_pools(bucket, object, &opts, &pinfo.object_info, errs)
+                .await?;
             obj.name = decode_dir_object(object);
             return Ok(obj);
         }
@@ -5823,6 +5835,36 @@ mod tests {
         );
     }
 
+    #[test]
+    fn exact_delete_capacity_target_requires_matching_namespace_and_object_identity() {
+        let source = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "directory/".to_string(),
+            version_id: Some(Uuid::from_u128(1)),
+            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
+            size: 10,
+            etag: Some("source-etag".to_string()),
+            ..Default::default()
+        };
+        let target = ObjectInfo {
+            name: rustfs_utils::path::encode_dir_object(&source.name),
+            ..source.clone()
+        };
+        assert!(ECStore::is_equivalent_decommission_capacity_target(&source, &target));
+
+        let mismatched_identity = ObjectInfo {
+            etag: Some("different-etag".to_string()),
+            ..target.clone()
+        };
+        assert!(!ECStore::is_equivalent_decommission_capacity_target(&source, &mismatched_identity));
+
+        let wrong_bucket = ObjectInfo {
+            bucket: "other-bucket".to_string(),
+            ..target
+        };
+        assert!(!ECStore::is_equivalent_decommission_capacity_target(&source, &wrong_bucket));
+    }
+
     #[tokio::test]
     async fn generic_data_movement_put_rejects_transition_ownership_without_capability() {
         let (_dirs, set) = make_local_set_disks(4, 2).await;

crates/ecstore/src/store/rebalance.rs

@@ -939,8 +939,12 @@ impl ECStore {
         bucket: &str,
         object: &str,
         opts: &ObjectOptions,
+        exact: &ObjectInfo,
         errs: Vec<PoolErr>,
     ) -> Result<ObjectInfo> {
+        self.reconcile_decommission_capacity_before_exact_delete(bucket, object, opts, exact)
+            .await?;
+
         let mut results = Vec::with_capacity(errs.len());
 
         for pe in errs.iter() {