diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 6d1b777be..b0fc9d452 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -2952,17 +2952,6 @@ impl ECStore { ) .await?; - data_movement::ensure_source_cleanup_versions_unchanged( - set.clone(), - bucket.as_str(), - entry.name.as_str(), - &fivs, - &cleanup_preflight_allowed_missing, - "decommission", - ) - .await - .map_err(|err| with_decommission_entry_context("cleanup_preflight", bucket.as_str(), entry.name.as_str(), err))?; - self.save_decommission_entry_progress_stage( idx, bucket.as_str(), @@ -2971,18 +2960,15 @@ impl ECStore { ) .await?; - let cleanup_result = set - .delete_object( - bucket.as_str(), - &encode_dir_object(&entry.name), - ObjectOptions { - delete_prefix: true, - delete_prefix_object: true, - - ..Default::default() - }, - ) - .await; + let cleanup_result = data_movement::cleanup_source_entry_if_unchanged( + set.clone(), + bucket.as_str(), + entry.name.as_str(), + &fivs, + &cleanup_preflight_allowed_missing, + "decommission", + ) + .await; resolve_decommission_entry_cleanup_delete_result(cleanup_result, bucket.as_str(), entry.name.as_str())? } else if decommissioned != fivs.versions.len() || expired > 0 { warn!( diff --git a/crates/ecstore/src/data_movement/mod.rs b/crates/ecstore/src/data_movement/mod.rs index a9fcbe4c4..099f73e62 100644 --- a/crates/ecstore/src/data_movement/mod.rs +++ b/crates/ecstore/src/data_movement/mod.rs @@ -19,9 +19,10 @@ pub(crate) mod backpressure; use crate::error::{Error, Result, is_err_data_movement_overwrite, is_err_object_not_found, is_err_version_not_found}; use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; -use crate::set_disk::SetDisks; +use crate::set_disk::{SetDisks, get_lock_acquire_timeout}; use crate::storage_api_contracts::{ multipart::{CompletePart, MultipartOperations as _}, + namespace::NamespaceLocking as _, object::{ObjectIO as _, ObjectOperations as _}, }; use crate::store::ECStore; @@ -492,6 +493,33 @@ pub(crate) async fn ensure_source_cleanup_versions_unchanged( )) } +pub(crate) async fn cleanup_source_entry_if_unchanged( + set: Arc, + bucket: &str, + object: &str, + expected: &FileInfoVersions, + allowed_missing: &[SourceCleanupVersionIdentity], + op_label: &str, +) -> Result { + let cleanup_key = encode_dir_object(object); + let ns_lock = set.new_ns_lock(bucket, cleanup_key.as_str()).await?; + let _guard = ns_lock.get_write_lock(get_lock_acquire_timeout()).await?; + + ensure_source_cleanup_versions_unchanged(set.clone(), bucket, object, expected, allowed_missing, op_label).await?; + + set.delete_object( + bucket, + cleanup_key.as_str(), + ObjectOptions { + delete_prefix: true, + delete_prefix_object: true, + no_lock: true, + ..Default::default() + }, + ) + .await +} + fn should_check_data_movement_resume_target(src_pool_idx: usize, target_pool_idx: usize) -> bool { target_pool_idx != src_pool_idx } diff --git a/crates/ecstore/src/services/rebalance/entry.rs b/crates/ecstore/src/services/rebalance/entry.rs index f585e52bd..ff048ab66 100644 --- a/crates/ecstore/src/services/rebalance/entry.rs +++ b/crates/ecstore/src/services/rebalance/entry.rs @@ -33,12 +33,10 @@ use crate::core::pools::ListCallback; use crate::data_movement; use crate::data_movement::backpressure::{self, DataMovementOperation}; use crate::error::{Error, Result}; -use crate::object_api::{GetObjectReader, ObjectOptions}; +use crate::object_api::GetObjectReader; use crate::set_disk::SetDisks; -use crate::storage_api_contracts::object::ObjectOperations as _; use crate::store::ECStore; use rustfs_filemeta::MetaCacheEntry; -use rustfs_utils::path::encode_dir_object; use std::sync::Arc; use time::OffsetDateTime; use tokio_util::sync::CancellationToken; @@ -259,15 +257,13 @@ impl ECStore { if should_cleanup_rebalance_source_entry(rebalanced, fivs.versions.len()) { let cleanup_warning = resolve_rebalance_entry_cleanup_delete_result( - set.delete_object( + data_movement::cleanup_source_entry_if_unchanged( + set.clone(), bucket.as_str(), - &encode_dir_object(&entry.name), - ObjectOptions { - delete_prefix: true, - delete_prefix_object: true, - - ..Default::default() - }, + entry.name.as_str(), + &fivs, + &[], + "rebalance", ) .await, bucket.as_str(),