diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 658f0981a..61cf84ec2 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -128,6 +128,7 @@ const METRIC_DECOMMISSION_CAPACITY_PREDICTION_ABSOLUTE_ERROR_BYTES: &str = const DECOMMISSION_LISTING_MAX_ATTEMPTS: usize = 3; const DECOMMISSION_LISTING_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(5); pub(crate) const DECOMMISSION_ENTRY_MAX_ATTEMPTS: usize = 3; +const DECOMMISSION_CAPACITY_INTENT_CONFLICT_MAX_ATTEMPTS: usize = 12; const DECOMMISSION_SOURCE_CLEANUP_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(100); pub(crate) const DECOMMISSION_VERSION_COPY_ATTEMPTS: usize = 3; const DECOMMISSION_COPY_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(50); @@ -927,6 +928,16 @@ fn is_decommission_capacity_blocked_error(err: &Error) -> bool { data_movement::data_movement_stage_source(err).is_some_and(is_decommission_capacity_blocked_error) } +fn is_decommission_capacity_intent_conflict(err: &Error) -> bool { + if matches!(err, Error::DecommissionCapacityBlocked { message } if message.contains("unresolved target capacity intent")) { + return true; + } + if data_movement::data_movement_stage_source(err).is_some_and(is_decommission_capacity_intent_conflict) { + return true; + } + err.to_string().contains("unresolved target capacity intent") +} + fn validate_decommission_capacity_reservation(reservation: Option<&DecommissionCapacityReservation>) -> Result<()> { let Some(reservation) = reservation else { return Ok(()); @@ -1608,9 +1619,9 @@ fn reserve_decommission_target_pending( pool.last_update = now; return Ok(additional); } - _ => { + pending_mutation_id => { return Err(decommission_capacity_blocked_error(format!( - "source pool {source_pool_index} target pool {target_pool_index} has an unresolved target capacity intent" + "source pool {source_pool_index} target pool {target_pool_index} has an unresolved target capacity intent {pending_mutation_id:?} while mutation {mutation_id} is waiting" ))); } } @@ -7789,6 +7800,19 @@ fn decommission_remote_tiered_opts( } } +fn decommission_capacity_version_mutation_id( + owner: DecommissionCapacityOwner, + bucket: &str, + version: &rustfs_filemeta::FileInfo, +) -> uuid::Uuid { + let version_id = if version.deleted && version.version_id.is_none() { + Some(uuid::Uuid::nil().to_string()) + } else { + version.version_id.map(|version_id| version_id.to_string()) + }; + decommission_capacity_mutation_id(owner, bucket, &version.name, version_id.as_deref(), version.deleted, version.mod_time) +} + fn decommission_capacity_owned_opts(mut opts: ObjectOptions, capacity_owner: Option) -> ObjectOptions { if let Some(capacity_owner) = capacity_owner { capacity_owner.apply_to(&mut opts); @@ -9162,13 +9186,15 @@ impl ECStore { ) -> Result> { let pool_meta = self.pool_meta.read().await; ensure_decommission_generation(&pool_meta, idx, generation)?; - let reservation = pool_meta + let Some(reservation) = pool_meta .pools .get(idx) .and_then(|pool| pool.decommission.as_ref()) .and_then(|info| info.capacity_reservation.as_ref()) .filter(|reservation| reservation.lease_active_at(OffsetDateTime::now_utc())) - .ok_or_else(|| decommission_capacity_blocked_error(format!("source pool {idx} has no active reservation")))?; + else { + return Ok(None); + }; Ok(Some(DecommissionCapacityOwner { source_pool_index: idx, operation_id: reservation.operation_id, @@ -10320,29 +10346,65 @@ impl ECStore { expected_bucket_incarnation_id: Option, source_changed_exhaustions: Arc, ) -> Result<()> { + let uses_capacity_ledger = self + .pool_meta + .read() + .await + .pools + .get(idx) + .and_then(|pool| pool.decommission.as_ref()) + .and_then(|info| info.capacity_reservation.as_ref()) + .is_some_and(DecommissionCapacityReservation::active); let mut counted_versions = HashSet::new(); for entry_attempt in 1..=DECOMMISSION_ENTRY_MAX_ATTEMPTS { - match self - .decommission_entry_attempt( - rx.clone(), - idx, - generation, - entry.clone(), - bucket.clone(), - Arc::clone(&set), - lifecycle_config.clone(), - object_lock_config.clone(), - replication_config.clone(), - expected_bucket_incarnation_id, - entry_attempt, - source_changed_exhaustions.as_ref(), - &mut counted_versions, - ) - .await? - { - DecommissionEntryAttemptOutcome::Complete => return Ok(()), - DecommissionEntryAttemptOutcome::SourceChanged => { + let attempt_result = { + let mut conflict_attempt = 0; + loop { + let result = { + let _capacity_entry_guard = if uses_capacity_ledger { + Some(tokio::select! { + biased; + _ = rx.cancelled() => return decommission_cancel_signal_result(true), + guard = self.decommission_capacity_entry_gate.lock() => guard, + }) + } else { + None + }; + self.decommission_entry_attempt( + rx.clone(), + idx, + generation, + entry.clone(), + bucket.clone(), + Arc::clone(&set), + lifecycle_config.clone(), + object_lock_config.clone(), + replication_config.clone(), + expected_bucket_incarnation_id, + entry_attempt, + source_changed_exhaustions.as_ref(), + &mut counted_versions, + ) + .await + }; + if result.as_ref().is_err_and(is_decommission_capacity_intent_conflict) + && conflict_attempt < DECOMMISSION_CAPACITY_INTENT_CONFLICT_MAX_ATTEMPTS + { + conflict_attempt += 1; + let retry_delay = + decommission_retry_backoff_delay(DECOMMISSION_SOURCE_CLEANUP_RETRY_DELAY, conflict_attempt); + if wait_decommission_retry_backoff(&rx, retry_delay).await { + decommission_cancel_signal_result(rx.is_cancelled())?; + } + continue; + } + break result; + } + }; + match attempt_result { + Ok(DecommissionEntryAttemptOutcome::Complete) => return Ok(()), + Ok(DecommissionEntryAttemptOutcome::SourceChanged) => { let retry_delay = decommission_retry_backoff_delay(DECOMMISSION_SOURCE_CLEANUP_RETRY_DELAY, entry_attempt); warn!( event = EVENT_DECOMMISSION_ENTRY, @@ -10361,6 +10423,7 @@ impl ECStore { decommission_cancel_signal_result(rx.is_cancelled())?; } } + Err(err) => return Err(err), } } @@ -10434,8 +10497,34 @@ impl ECStore { let mut fivs = load_decommission_entry_exact_versions(&set, &entry, &bucket, "file_info_versions").await?; - fivs.versions - .sort_by_key(|v| (v.mod_time.is_none(), std::cmp::Reverse(v.mod_time))); + let pending_mutations = if let Some(owner) = capacity_owner { + self.pool_meta + .read() + .await + .pools + .get(owner.source_pool_index) + .and_then(|pool| pool.decommission.as_ref()) + .and_then(|info| info.capacity_reservation.as_ref()) + .filter(|reservation| reservation.admits_cleanup_owner(owner)) + .map(|reservation| { + reservation + .targets + .iter() + .filter_map(|target| target.pending_mutation_id) + .collect::>() + }) + .unwrap_or_default() + } else { + HashSet::new() + }; + fivs.versions.sort_by_key(|version| { + let mutation_id = capacity_owner.map(|owner| decommission_capacity_version_mutation_id(owner, &bucket, version)); + ( + mutation_id.is_none_or(|mutation_id| !pending_mutations.contains(&mutation_id)), + version.mod_time.is_none(), + std::cmp::Reverse(version.mod_time), + ) + }); let mut decommissioned: usize = 0; let mut expired: usize = 0; @@ -15353,6 +15442,18 @@ mod tests { assert!(is_decommission_target_capacity_error(&storage_full)); } + #[test] + fn decommission_capacity_intent_conflict_accepts_context_wrapped_error() { + let err = with_decommission_entry_context( + "migrate_object", + "bucket", + "object", + decommission_capacity_blocked_error("target has an unresolved target capacity intent"), + ); + + assert!(is_decommission_capacity_intent_conflict(&err)); + } + #[test] fn decommission_target_capacity_error_rejects_unrelated_errors() { assert!(!is_decommission_target_capacity_error(&Error::SlowDown)); @@ -15429,6 +15530,7 @@ mod tests { #[test] fn decommission_delete_marker_opts_preserves_suspended_null_version() { let version = rustfs_filemeta::FileInfo { + name: "object".to_string(), deleted: true, ..Default::default() }; @@ -15437,6 +15539,25 @@ mod tests { assert!(!opts.versioned); assert!(opts.version_suspended); assert_eq!(opts.version_id.as_deref(), Some(uuid::Uuid::nil().to_string().as_str())); + + let owner = DecommissionCapacityOwner { + source_pool_index: 7, + operation_id: uuid::Uuid::new_v4(), + generation: 1, + owner_nonce: uuid::Uuid::new_v4(), + mutation_id: None, + }; + assert_eq!( + decommission_capacity_version_mutation_id(owner, "bucket", &version), + decommission_capacity_mutation_id( + owner, + "bucket", + &version.name, + opts.version_id.as_deref(), + opts.delete_marker, + opts.mod_time, + ) + ); } #[test] @@ -16549,6 +16670,7 @@ mod pools_tests { decommission_cancelers: tokio::sync::RwLock::new(cancelers), start_gate: tokio::sync::Mutex::new(()), pool_meta_save_gate: tokio::sync::Mutex::new(super::PoolMetaWriteState::for_test_bootstrap()), + decommission_capacity_entry_gate: tokio::sync::Mutex::default(), ctx, bucket_fence_registry: Arc::default(), }) diff --git a/crates/ecstore/src/core/pools_test.rs b/crates/ecstore/src/core/pools_test.rs index 1913d8a22..5cd5dc714 100644 --- a/crates/ecstore/src/core/pools_test.rs +++ b/crates/ecstore/src/core/pools_test.rs @@ -422,6 +422,7 @@ mod decommission_lock_order_tests { pool_meta_save_gate: tokio::sync::Mutex::new( other_store.pool_meta_save_gate.lock().await.independent_clone_for_test(), ), + decommission_capacity_entry_gate: tokio::sync::Mutex::default(), ctx, bucket_fence_registry: Arc::default(), }); diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index b829208ea..c144435a6 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -2964,6 +2964,7 @@ mod tests { decommission_cancelers: RwLock::new(Vec::new()), start_gate: TokioMutex::new(()), pool_meta_save_gate: TokioMutex::default(), + decommission_capacity_entry_gate: TokioMutex::default(), ctx, bucket_fence_registry: Arc::default(), }) diff --git a/crates/ecstore/src/services/rebalance/mod.rs b/crates/ecstore/src/services/rebalance/mod.rs index 352c1138e..96d02eedc 100644 --- a/crates/ecstore/src/services/rebalance/mod.rs +++ b/crates/ecstore/src/services/rebalance/mod.rs @@ -83,6 +83,7 @@ pub async fn test_store_with_persisted_rebalance_meta( decommission_cancelers: tokio::sync::RwLock::new(vec![None]), start_gate: tokio::sync::Mutex::new(()), pool_meta_save_gate: tokio::sync::Mutex::default(), + decommission_capacity_entry_gate: tokio::sync::Mutex::default(), ctx, bucket_fence_registry: std::sync::Arc::default(), }); @@ -232,6 +233,7 @@ async fn test_pool_stores_with_contexts( decommission_cancelers: tokio::sync::RwLock::new(vec![None; pool_count]), start_gate: tokio::sync::Mutex::new(()), pool_meta_save_gate: tokio::sync::Mutex::new(pool_meta_write_state.independent_clone_for_test()), + decommission_capacity_entry_gate: tokio::sync::Mutex::default(), ctx: store_ctx, bucket_fence_registry: std::sync::Arc::default(), }) diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index a544ed733..d280d403a 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs @@ -3009,6 +3009,7 @@ fn test_store_with_rebalance_meta(meta: RebalanceMeta) -> Arc, bucket: &str, object: &str) { - let source_versions = store.pools[0] - .get_disks_by_key(object) - .load_file_info_versions_exact(bucket, object) - .await - .expect("source versions should remain readable after suspended convergence"); - assert!( - source_versions.is_none_or(|versions| versions.versions.is_empty()), - "worker convergence must remove only the decommissioned source null version" - ); - - let target_versions = store.pools[1] - .get_disks_by_key(object) - .load_file_info_versions_exact(bucket, object) - .await - .expect("active target versions should be readable") - .expect("active target must retain the suspended DELETE marker"); - assert!( - matches!(target_versions.versions.as_slice(), [marker] if marker.deleted && marker.version_id.is_none_or(|version_id| version_id.is_nil())), - "active target must contain only its null delete marker: {target_versions:?}" - ); - - let err = store - .get_object_info( - bucket, - object, - &ObjectOptions { - version_suspended: true, - ..Default::default() - }, - ) - .await - .expect_err("the active null delete marker must hide the migrated source generation"); - assert!( - matches!(err, StorageError::ObjectNotFound(_, _)), - "unexpected suspended latest-object result: {err:?}" - ); - } - #[tokio::test] #[serial_test::serial(storage_class_env)] async fn tag_updates_skip_active_rebalance_source_pool() { @@ -4216,13 +4178,7 @@ mod tests { .put_object(&bucket, &object, &mut source, &ObjectOptions::default()) .await .expect("write source object to the pool being decommissioned"); - { - let mut pool_meta = store.pool_meta.write().await; - pool_meta.pools[0].decommission = Some(PoolDecommissionInfo { - start_time: Some(OffsetDateTime::now_utc()), - ..Default::default() - }); - } + mark_test_pool_decommissioning(&store, 0).await; assert!(store.is_suspended(0).await, "pool 0 must be a suspended decommission source"); let barrier = crate::set_disk::PutObjectCommitBarrier::install( @@ -4958,9 +4914,10 @@ mod tests { .then(|| crate::set_disk::NewMultipartUploadCommitObservation::install(&bucket, object)); let barrier = crate::set_disk::MultipartCommitBarrier::install(&bucket, object, pause); let source_set = store.pools[0].get_disks_by_key(object); + let recovery_source_set = Arc::clone(&source_set); let worker_store = Arc::clone(&store); let worker_bucket = bucket.clone(); - let worker = tokio::spawn(async move { + let mut worker = tokio::spawn(async move { worker_store .decommission_entry_for_test( 0, @@ -4974,7 +4931,10 @@ mod tests { .await }); - barrier.wait_until_paused().await; + tokio::select! { + () = barrier.wait_until_paused() => {} + result = &mut worker => panic!("decommission multipart worker exited before the commit barrier: {result:?}"), + } loss_hook.mark_lost(); barrier.release(); drop(barrier); @@ -4996,6 +4956,19 @@ mod tests { .await .expect("list target multipart uploads after fenced migration"); assert!(uploads.uploads.is_empty(), "fenced multipart migration must not retain target staging"); + drop(loss_hook); + store + .decommission_entry_for_test( + 0, + MetaCacheEntry { + name: object.to_string(), + ..Default::default() + }, + bucket.clone(), + recovery_source_set, + ) + .await + .expect("same-mutation retry should recover the durable capacity intent"); } shutdown.cancel(); @@ -5204,6 +5177,12 @@ mod tests { .expect("self-copy should keep using the committed active target"); assert_eq!(active_copy_result.data_dir, active_copy_data_dir); + { + let mut pool_meta = store.pool_meta.write().await; + pool_meta.pools[1].decommission = None; + } + mark_test_pool_decommissioning(&store, 1).await; + let cleanup_barrier = crate::data_movement::SourceCleanupDeleteBarrier::install(&bucket, object); let commit_barrier = crate::set_disk::PutObjectCommitBarrier::install( &bucket, @@ -6217,28 +6196,27 @@ mod tests { write_suspended_decommission_source(&store, &bucket, object).await; mark_test_pool_decommissioning(&store, 0).await; - let delete_barrier = crate::store::object::VersionedDeleteMarkerCommitBarrier::install(&bucket, object); - let delete_store = Arc::clone(&store); - let delete_bucket = bucket.clone(); - let delete = tokio::spawn(async move { - delete_store - .delete_object( - &delete_bucket, - object, - ObjectOptions { - version_suspended: true, - ..Default::default() - }, - ) - .await - }); - delete_barrier.wait_until_paused().await; + let delete_err = store + .delete_object( + &bucket, + object, + ObjectOptions { + version_suspended: true, + ..Default::default() + }, + ) + .await + .expect_err("capacity-reserved target must reject a concurrent suspended DELETE"); + assert!( + matches!(delete_err, Error::SlowDown), + "unexpected suspended DELETE result: {delete_err:?}" + ); assert_suspended_null_source_present(&store, &bucket, object).await; let source_set = store.pools[0].get_disks_by_key(object); let worker_store = Arc::clone(&store); let worker_bucket = bucket.clone(); - let worker = tokio::spawn(async move { + tokio::spawn(async move { worker_store .decommission_entry_for_test( 0, @@ -6250,25 +6228,25 @@ mod tests { source_set, ) .await - }); + }) + .await + .expect("suspended decommission worker should join") + .expect("worker must migrate the fenced suspended source"); - delete_barrier.release(); - let marker = delete - .await - .expect("suspended DELETE task should join") - .expect("suspended DELETE should commit its active-pool marker"); - drop(delete_barrier); - assert!(marker.delete_marker, "suspended DELETE must create a marker"); - assert!( - marker.version_id.is_none_or(|version_id| version_id.is_nil()), - "suspended DELETE marker must keep the null version identity" + assert_decommission_source_absent( + &store, + &bucket, + object, + &ObjectOptions { + version_suspended: true, + ..Default::default() + }, + ) + .await; + assert_eq!( + read_decommission_target_body(&store, &bucket, object, &ObjectOptions::default()).await, + b"suspended source generation" ); - worker - .await - .expect("suspended decommission worker should join") - .expect("worker must treat the newer active null marker as a completed migration"); - - assert_suspended_decommission_converged(&store, &bucket, object).await; shutdown.cancel(); } @@ -6301,31 +6279,29 @@ mod tests { }, None, )); - let delete_barrier = crate::store::object::VersionedDeleteMarkerCommitBarrier::install(&bucket, object); - let delete_store = Arc::clone(&store); - let delete_bucket = bucket.clone(); - let delete = tokio::spawn(async move { - delete_store - .delete_objects( - &delete_bucket, - vec![ObjectToDelete { - object_name: object.to_string(), - ..Default::default() - }], - ObjectOptions { - delete_replication_config_snapshot: Some(delete_config_snapshot), - ..Default::default() - }, - ) - .await - }); - delete_barrier.wait_until_paused().await; + let (_deleted, errors) = store + .delete_objects( + &bucket, + vec![ObjectToDelete { + object_name: object.to_string(), + ..Default::default() + }], + ObjectOptions { + delete_replication_config_snapshot: Some(delete_config_snapshot), + ..Default::default() + }, + ) + .await; + assert!( + matches!(errors.as_slice(), [Some(Error::SlowDown)]), + "unexpected suspended batch DELETE result: {errors:?}" + ); assert_suspended_null_source_present(&store, &bucket, object).await; let source_set = store.pools[0].get_disks_by_key(object); let worker_store = Arc::clone(&store); let worker_bucket = bucket.clone(); - let worker = tokio::spawn(async move { + tokio::spawn(async move { worker_store .decommission_entry_for_test( 0, @@ -6337,22 +6313,25 @@ mod tests { source_set, ) .await - }); + }) + .await + .expect("suspended batch decommission worker should join") + .expect("worker must migrate the batch-fenced suspended source"); - delete_barrier.release(); - let (deleted, errors) = delete.await.expect("suspended batch DELETE task should join"); - drop(delete_barrier); - assert!(errors.iter().all(Option::is_none), "suspended batch DELETE should succeed: {errors:?}"); - assert!( - matches!(deleted.as_slice(), [marker] if marker.delete_marker && marker.delete_marker_version_id.is_none_or(|version_id| version_id.is_nil())), - "suspended batch DELETE must create one null marker: {deleted:?}" + assert_decommission_source_absent( + &store, + &bucket, + object, + &ObjectOptions { + version_suspended: true, + ..Default::default() + }, + ) + .await; + assert_eq!( + read_decommission_target_body(&store, &bucket, object, &ObjectOptions::default()).await, + b"suspended source generation" ); - worker - .await - .expect("suspended batch decommission worker should join") - .expect("worker must treat the newer batch null marker as a completed migration"); - - assert_suspended_decommission_converged(&store, &bucket, object).await; shutdown.cancel(); } diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index c3dd77605..a7a9dc4ee 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -260,6 +260,12 @@ pub struct ECStore { /// Lock order: acquire `pool_meta_save_gate`, then the distributed /// `pool.bin` fence, then clone `pool_meta` under a short read lock. pub(crate) pool_meta_save_gate: Mutex, + /// Serializes decommission entries while the durable capacity ledger has + /// one target mutation intent slot. + /// + /// Lock order: acquire this gate before object namespaces or + /// `pool_meta_save_gate`. + pub(crate) decommission_capacity_entry_gate: Mutex<()>, /// Per-instance runtime state (Phase 5, backlog#939). /// /// Carries this instance's identity/runtime out of the process globals so @@ -1514,6 +1520,7 @@ mod tests { decommission_cancelers: RwLock::new(Vec::new()), start_gate: Mutex::new(()), pool_meta_save_gate: Mutex::default(), + decommission_capacity_entry_gate: Mutex::default(), ctx, bucket_fence_registry: Arc::default(), }; @@ -1589,6 +1596,7 @@ mod tests { decommission_cancelers: RwLock::new(Vec::new()), start_gate: Mutex::new(()), pool_meta_save_gate: Mutex::default(), + decommission_capacity_entry_gate: Mutex::default(), ctx, bucket_fence_registry: Arc::default(), }) diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index a2eed1658..0c74cad5f 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -1103,6 +1103,7 @@ mod tests { decommission_cancelers: RwLock::new(Vec::new()), start_gate: Mutex::new(()), pool_meta_save_gate: Mutex::default(), + decommission_capacity_entry_gate: Mutex::default(), ctx: crate::runtime::instance::bootstrap_ctx(), bucket_fence_registry: std::sync::Arc::default(), } diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index a56a4b680..c1fbf3b3f 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -2693,20 +2693,26 @@ impl ECStore { return Err(decommission_free_version_overwrite_error(bucket, &object, fi.version_id)); } + let expected_data_bytes = usize::try_from(fi.size).ok(); 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 - } - }) + .run_decommission_capacity_admitted_mutation( + idx, + DecommissionCapacityOwner::from_options(&opts), + expected_data_bytes, + || 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 @@ -5883,6 +5889,7 @@ mod tests { decommission_cancelers: RwLock::new(Vec::new()), start_gate: Mutex::new(()), pool_meta_save_gate: Mutex::default(), + decommission_capacity_entry_gate: Mutex::default(), ctx: crate::runtime::instance::bootstrap_ctx(), bucket_fence_registry: std::sync::Arc::default(), } @@ -5946,6 +5953,7 @@ mod tests { decommission_cancelers: RwLock::new(Vec::new()), start_gate: Mutex::new(()), pool_meta_save_gate: Mutex::default(), + decommission_capacity_entry_gate: Mutex::default(), ctx, bucket_fence_registry: std::sync::Arc::default(), }