From 9a22cb85f3b489b08e7608de4a743d9cb9775e88 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Mon, 31 Aug 2026 16:53:14 +0800 Subject: [PATCH] fix(ecstore): complete decommission capacity recovery (#6949) --- crates/ecstore/src/core/pools.rs | 66 +++++++++++++++++------ crates/ecstore/src/core/pools_test.rs | 2 +- crates/ecstore/src/store/init.rs | 35 +++++++++--- crates/ecstore/src/store/object.rs | 78 +++++++++++++++++++++++---- crates/ecstore/src/store/rebalance.rs | 2 +- 5 files changed, 149 insertions(+), 34 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 7a88dddf1..658f0981a 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/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, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub temporary_mutations: Vec, @@ -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( &self, target_pool_index: usize, @@ -9148,17 +9160,6 @@ impl ECStore { idx: usize, generation: OffsetDateTime, ) -> Result> { - 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, - ensure_decommission_target_owner_admission, ensure_external_decommission_target_admission, - is_decommission_capacity_blocked_error, record_decommission_target_consumption, reserve_decommission_target_pending, - resolve_decommission_target_pending, set_decommission_capacity_info_overrides_for_test, + DecommissionCapacityOwner, DecommissionCapacityReservation, DecommissionCapacityTemporaryMutation, + decommission_capacity_mutation_id, ensure_decommission_target_owner_admission, + ensure_external_decommission_target_admission, is_decommission_capacity_blocked_error, + 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); diff --git a/crates/ecstore/src/core/pools_test.rs b/crates/ecstore/src/core/pools_test.rs index 162077c00..1913d8a22 100644 --- a/crates/ecstore/src/core/pools_test.rs +++ b/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; diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 95cb27505..d747c5334 100644 --- a/crates/ecstore/src/store/init.rs +++ b/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, - pool_meta_identity_initialized_for_test, pool_meta_v3_commit_state_for_test, + DecommissionErasureLayout, DecommissionPoolCapacityInfo, POOL_META_IDENTITY_NAME, POOL_META_NAME, POOL_META_VERSION, + 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, 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, pool_idx: usize) { - let mut pool_meta = store.pool_meta.write().await; - pool_meta.pools[pool_idx].decommission = Some(PoolDecommissionInfo { - start_time: Some(OffsetDateTime::now_utc()), - ..Default::default() - }); + set_test_decommission_capacity_override(store, pool_idx); + store + .save_current_pool_meta_for_decommission_start(&[pool_idx], Vec::new()) + .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 diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index b3e1c5008..a56a4b680 100644 --- a/crates/ecstore/src/store/object.rs +++ b/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 where F: FnOnce(ObjectOptions) -> Fut, Fut: std::future::Future>, { - let (capacity_guard, has_active_decommission) = self - .acquire_external_decommission_capacity_fence_with_active_source(&[target_pool_idx], "mutation") - .await?; + self.run_external_decommission_capacity_object_operation( + target_pool_idx, + bucket, + (lock_object, target_object), + opts, + false, + operation, + ) + .await + } + + pub(super) async fn run_external_decommission_capacity_object_delete( + &self, + target_pool_idx: usize, + bucket: &str, + lock_object: &str, + target_object: &str, + opts: ObjectOptions, + operation: F, + ) -> Result + where + F: FnOnce(ObjectOptions) -> Fut, + Fut: std::future::Future>, + { + 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( + &self, + target_pool_idx: usize, + bucket: &str, + objects: (&str, &str), + mut opts: ObjectOptions, + capacity_releasing: bool, + operation: F, + ) -> Result + where + F: FnOnce(ObjectOptions) -> Fut, + Fut: std::future::Future>, + { + 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 - .acquire_external_decommission_capacity_fence(&[target_pool_idx], "mutation") - .await?; + let capacity_guard = if capacity_releasing { + self.acquire_decommission_capacity_release_fence_with_active_source().await?.0 + } 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, diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs index 14de70b66..2bfe36af8 100644 --- a/crates/ecstore/src/store/rebalance.rs +++ b/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,