fix(ecstore): complete decommission capacity recovery (#6949)

crates/ecstore/src/core/pools.rs

@@ -6977,7 +6977,7 @@ pub struct DecommissionCapacityTarget {
     pub inflight_physical_bytes: usize,
     #[serde(default)]
     pub pending_physical_bytes: usize,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
+    #[serde(default)]
     pub pending_mutation_id: Option<uuid::Uuid>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub temporary_mutations: Vec<DecommissionCapacityTemporaryMutation>,
@@ -8100,6 +8100,18 @@ impl ECStore {
         Ok((pool_meta_guard, has_active_source))
     }
 
+    pub(crate) async fn acquire_decommission_capacity_release_fence_with_active_source(
+        &self,
+    ) -> Result<(rustfs_lock::NamespaceLockGuard, bool)> {
+        let mut save_guard = self.pool_meta_save_gate.lock().await;
+        let (pool_meta_guard, snapshot) = self
+            .acquire_pool_meta_read_guard(&mut save_guard, "capacity release fence failed")
+            .await?;
+        let has_active_source = pool_meta_has_active_decommission(&snapshot);
+        drop(save_guard);
+        Ok((pool_meta_guard, has_active_source))
+    }
+
     pub(crate) async fn run_decommission_capacity_admitted_mutation<T, F, Fut>(
         &self,
         target_pool_index: usize,
@@ -9148,17 +9160,6 @@ impl ECStore {
         idx: usize,
         generation: OffsetDateTime,
     ) -> Result<Option<DecommissionCapacityOwner>> {
-        let active_worker = self
-            .decommission_cancelers
-            .read()
-            .await
-            .get(idx)
-            .and_then(Option::as_ref)
-            .is_some_and(DecommissionCanceler::is_active);
-        if !active_worker {
-            return Ok(None);
-        }
-
         let pool_meta = self.pool_meta.read().await;
         ensure_decommission_generation(&pool_meta, idx, generation)?;
         let reservation = pool_meta
@@ -16459,10 +16460,11 @@ mod pools_tests {
         wait_decommission_worker_drain, with_decommission_entry_context,
     };
     use super::{
-        DecommissionCapacityOwner, DecommissionCapacityReservation, decommission_capacity_mutation_id,
+        DecommissionCapacityOwner, DecommissionCapacityReservation, DecommissionCapacityTemporaryMutation,
-        ensure_decommission_target_owner_admission, ensure_external_decommission_target_admission,
+        decommission_capacity_mutation_id, ensure_decommission_target_owner_admission,
-        is_decommission_capacity_blocked_error, record_decommission_target_consumption, reserve_decommission_target_pending,
+        ensure_external_decommission_target_admission, is_decommission_capacity_blocked_error,
-        resolve_decommission_target_pending, set_decommission_capacity_info_overrides_for_test,
+        record_decommission_target_consumption, reserve_decommission_target_pending, resolve_decommission_target_pending,
+        set_decommission_capacity_info_overrides_for_test,
     };
     use crate::bucket::lifecycle::{
         DurableIlmRecordCheckpoint,
@@ -16483,10 +16485,12 @@ mod pools_tests {
     use crate::storage_api_contracts::{object::ObjectIO, range::HTTPRangeSpec};
     use crate::store::ECStore;
     use byteorder::{ByteOrder, LittleEndian};
+    use rmp_serde::Serializer;
     use rustfs_filemeta::{FileInfo, FileInfoVersions, MetaCacheEntry, ObjectPartInfo};
     use rustfs_filemeta::{MetaCacheEntries, MetadataResolutionParams};
     use rustfs_lock::{GlobalLockManager, LocalClient, LockRequest, LockType, NamespaceLock, ObjectKey};
     use rustfs_rio::Index;
+    use serde::Serialize;
     use std::future::Future;
     use std::io::Cursor;
     use std::sync::{
@@ -20710,6 +20714,36 @@ mod pools_tests {
         }
     }
 
+    #[test]
+    fn decommission_capacity_target_round_trip_preserves_temporary_mutation_without_pending_intent() {
+        let mutation_id = uuid::Uuid::new_v4();
+        let target = DecommissionCapacityTarget {
+            pool_index: 1,
+            layout: DecommissionErasureLayout { data: 2, parity: 2 },
+            physical_total_at_reservation: 200,
+            physical_free_at_reservation: 200,
+            reserved_physical_bytes: 200,
+            consumed_physical_bytes: 0,
+            observed_physical_bytes: 1,
+            inflight_physical_bytes: 1,
+            pending_physical_bytes: 0,
+            pending_mutation_id: None,
+            temporary_mutations: vec![DecommissionCapacityTemporaryMutation {
+                mutation_id,
+                physical_bytes: 1,
+            }],
+        };
+        let mut encoded = Vec::new();
+        target
+            .serialize(&mut Serializer::new(&mut encoded))
+            .expect("capacity target should serialize");
+
+        let restored: DecommissionCapacityTarget =
+            rmp_serde::from_slice(&encoded).expect("capacity target with a released pending intent should deserialize");
+
+        assert_eq!(restored, target);
+    }
+
     #[test]
     fn decommission_capacity_reservation_recovers_expired_lease_after_restart_round_trip() {
         let created_at = OffsetDateTime::UNIX_EPOCH + Duration::hours(1);

crates/ecstore/src/core/pools_test.rs

@@ -1707,7 +1707,6 @@ mod decommission_lock_order_tests {
         set_decommission_capacity_info_overrides_for_test(lossy_store.id, (0..40).map(|_| capacity_snapshot()).collect());
         let part_barrier = MultipartCommitBarrier::install(&bucket, object, MultipartCommitPause::PutPartAfterRename);
         let abort_barrier = data_movement::DataMovementMultipartAbortBarrier::install(&bucket, object);
-        tokio::time::pause();
         let migration = tokio::spawn({
             let migration_store = Arc::clone(&lossy_store);
             let migration_bucket = bucket.clone();
@@ -1725,6 +1724,7 @@ mod decommission_lock_order_tests {
             }
         });
         part_barrier.wait_until_paused().await;
+        tokio::time::pause();
         tokio::task::yield_now().await;
         refresh_calls.arm();
         tokio::time::advance(Duration::from_secs(11)).await;

crates/ecstore/src/store/init.rs

@@ -837,8 +837,9 @@ mod tests {
     use crate::{
         bucket::replication::{ReplicationState, ReplicationStatusType, replication_statuses_map},
         core::pools::{
-            POOL_META_IDENTITY_NAME, POOL_META_NAME, POOL_META_VERSION, PoolDecommissionInfo, PoolMeta, PoolStatus,
+            DecommissionErasureLayout, DecommissionPoolCapacityInfo, POOL_META_IDENTITY_NAME, POOL_META_NAME, POOL_META_VERSION,
-            pool_meta_identity_initialized_for_test, pool_meta_v3_commit_state_for_test,
+            PoolDecommissionInfo, PoolMeta, PoolStatus, pool_meta_identity_initialized_for_test,
+            pool_meta_v3_commit_state_for_test, set_decommission_capacity_info_overrides_for_test,
         },
         disk::endpoint::Endpoint,
         error::{Error, Result, StorageError},
@@ -2163,12 +2164,31 @@ mod tests {
         (source_version, expected_source_versions)
     }
 
+    fn set_test_decommission_capacity_override(store: &Arc<crate::store::ECStore>, pool_idx: usize) {
+        let layout = DecommissionErasureLayout { data: 2, parity: 2 };
+        let source_physical_bytes = 1024 * 1024 * 1024;
+        let target_physical_bytes = source_physical_bytes * 8;
+        let capacity = store
+            .pools
+            .iter()
+            .enumerate()
+            .map(|(index, _)| {
+                if index == pool_idx {
+                    DecommissionPoolCapacityInfo::for_test(index, layout, 0, source_physical_bytes, source_physical_bytes)
+                } else {
+                    DecommissionPoolCapacityInfo::for_test(index, layout, target_physical_bytes, target_physical_bytes, 0)
+                }
+            })
+            .collect();
+        set_decommission_capacity_info_overrides_for_test(store.id, vec![capacity]);
+    }
+
     async fn mark_test_pool_decommissioning(store: &Arc<crate::store::ECStore>, pool_idx: usize) {
-        let mut pool_meta = store.pool_meta.write().await;
+        set_test_decommission_capacity_override(store, pool_idx);
-        pool_meta.pools[pool_idx].decommission = Some(PoolDecommissionInfo {
+        store
-            start_time: Some(OffsetDateTime::now_utc()),
+            .save_current_pool_meta_for_decommission_start(&[pool_idx], Vec::new())
-            ..Default::default()
+            .await
-        });
+            .expect("test decommission capacity reservation should activate");
     }
 
     const DECOMMISSION_TEST_FAULT_STAGE_DELETE_MARKER: &str = "delete_marker_copy";
@@ -7325,6 +7345,7 @@ mod tests {
             .await
             .expect("legacy decommission queue should reload after restart");
         *store.pool_meta.write().await = restarted_pool_meta;
+        set_test_decommission_capacity_override(&store, 0);
         store
             .promote_queued_decommission_for_test(0)
             .await

crates/ecstore/src/store/object.rs

@@ -1204,7 +1204,11 @@ fn delete_pool_lookup_opts(opts: &ObjectOptions, no_lock: bool) -> ObjectOptions
 }
 
 fn should_delete_from_all_pools(opts: &ObjectOptions, pool_count: usize) -> bool {
-    pool_count > 0 && (!opts.versioned && !opts.version_suspended || opts.version_id.is_some())
+    pool_count > 0 && delete_only_releases_capacity(opts)
+}
+
+fn delete_only_releases_capacity(opts: &ObjectOptions) -> bool {
+    !opts.versioned && !opts.version_suspended || opts.version_id.is_some()
 }
 
 fn batch_delete_creates_latest_marker(object: &ObjectToDelete, delete_config_snapshot: &DeleteReplicationConfigSnapshot) -> bool {
@@ -2013,16 +2017,69 @@ impl ECStore {
         bucket: &str,
         lock_object: &str,
         target_object: &str,
-        mut opts: ObjectOptions,
+        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
+        self.run_external_decommission_capacity_object_operation(
-            .acquire_external_decommission_capacity_fence_with_active_source(&[target_pool_idx], "mutation")
+            target_pool_idx,
-            .await?;
+            bucket,
+            (lock_object, target_object),
+            opts,
+            false,
+            operation,
+        )
+        .await
+    }
+
+    pub(super) async fn run_external_decommission_capacity_object_delete<T, F, Fut>(
+        &self,
+        target_pool_idx: usize,
+        bucket: &str,
+        lock_object: &str,
+        target_object: &str,
+        opts: ObjectOptions,
+        operation: F,
+    ) -> Result<T>
+    where
+        F: FnOnce(ObjectOptions) -> Fut,
+        Fut: std::future::Future<Output = Result<T>>,
+    {
+        let capacity_releasing = delete_only_releases_capacity(&opts);
+        self.run_external_decommission_capacity_object_operation(
+            target_pool_idx,
+            bucket,
+            (lock_object, target_object),
+            opts,
+            capacity_releasing,
+            operation,
+        )
+        .await
+    }
+
+    async fn run_external_decommission_capacity_object_operation<T, F, Fut>(
+        &self,
+        target_pool_idx: usize,
+        bucket: &str,
+        objects: (&str, &str),
+        mut opts: ObjectOptions,
+        capacity_releasing: bool,
+        operation: F,
+    ) -> Result<T>
+    where
+        F: FnOnce(ObjectOptions) -> Fut,
+        Fut: std::future::Future<Output = Result<T>>,
+    {
+        let (lock_object, target_object) = objects;
+        let (capacity_guard, has_active_decommission) = if capacity_releasing {
+            self.acquire_decommission_capacity_release_fence_with_active_source().await?
+        } else {
+            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.
@@ -2034,9 +2091,12 @@ impl ECStore {
                 .await?;
             self.apply_decommission_target_mutation_fence(target_pool_idx, target_object, &mut opts, Some(&guard))
                 .await;
-            let capacity_guard = self
+            let capacity_guard = if capacity_releasing {
-                .acquire_external_decommission_capacity_fence(&[target_pool_idx], "mutation")
+                self.acquire_decommission_capacity_release_fence_with_active_source().await?.0
-                .await?;
+            } else {
+                self.acquire_external_decommission_capacity_fence(&[target_pool_idx], "mutation")
+                    .await?
+            };
             (capacity_guard, Some(guard))
         } else {
             (capacity_guard, None)
@@ -3572,7 +3632,7 @@ impl ECStore {
             let pool_idx = pool.pool_idx;
             let pool = pool.clone();
             match self
-                .run_external_decommission_capacity_object_mutation(
+                .run_external_decommission_capacity_object_delete(
                     pool_idx,
                     bucket,
                     object,

crates/ecstore/src/store/rebalance.rs

@@ -929,7 +929,7 @@ impl ECStore {
                 results.push(RebalanceDeletePoolResult {
                     pool_idx: idx,
                     result: self
-                        .run_external_decommission_capacity_object_mutation(
+                        .run_external_decommission_capacity_object_delete(
                             idx,
                             bucket,
                             object,