From 6f29431a65a179cb0ef03f2f3846b84d4457fd6c Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 14 Aug 2026 19:07:34 +0800 Subject: [PATCH] test(ecstore): isolate rename publication hooks (#6106) --- crates/ecstore/src/disk/local.rs | 32 +++++++++++--------- crates/ecstore/src/set_disk/ops/multipart.rs | 2 +- crates/scanner/src/scanner_io.rs | 12 +++++--- crates/scanner/src/storage_api.rs | 2 +- 4 files changed, 27 insertions(+), 21 deletions(-) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index c6ff03604..e8059319f 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -2028,14 +2028,17 @@ static RENAME_DATA_REMOVE_DST_BASE_BEFORE_COMMIT: std::sync::Mutex; #[cfg(test)] +type RenameDataPublicationHookKey = (PathBuf, String, String); +#[cfg(test)] static INLINE_PREPARATION_BEFORE_BACKUP: std::sync::LazyLock>> = std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new())); #[cfg(test)] static INLINE_BEFORE_FILE_SYNC_ADMISSION: std::sync::LazyLock>> = std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new())); #[cfg(test)] -static RENAME_DATA_AFTER_FIRST_PUBLICATION: std::sync::LazyLock>> = - std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new())); +static RENAME_DATA_AFTER_FIRST_PUBLICATION: std::sync::LazyLock< + std::sync::Mutex>, +> = std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new())); #[cfg(test)] static OWNED_FILE_WRITE_BEFORE_OPEN: std::sync::LazyLock>> = std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new())); @@ -2107,11 +2110,11 @@ fn set_inline_before_file_sync_admission(dst_path: &str, hook: impl FnOnce() + S } #[cfg(test)] -fn set_rename_data_after_first_publication(dst_path: &str, hook: impl FnOnce() + Send + 'static) { +fn set_rename_data_after_first_publication(root: &Path, dst_volume: &str, dst_path: &str, hook: impl FnOnce() + Send + 'static) { RENAME_DATA_AFTER_FIRST_PUBLICATION .lock() .expect("test publication hook lock should not be poisoned") - .insert(dst_path.to_string(), Box::new(hook)); + .insert((root.to_path_buf(), dst_volume.to_string(), dst_path.to_string()), Box::new(hook)); } #[cfg(test)] @@ -2263,11 +2266,11 @@ fn run_inline_before_file_sync_admission(dst_path: &str) { } #[cfg(test)] -fn run_rename_data_after_first_publication(dst_path: &str) { +fn run_rename_data_after_first_publication(root: &Path, dst_volume: &str, dst_path: &str) { let hook = RENAME_DATA_AFTER_FIRST_PUBLICATION .lock() .expect("test publication hook lock should not be poisoned") - .remove(dst_path); + .remove(&(root.to_path_buf(), dst_volume.to_string(), dst_path.to_string())); if let Some(hook) = hook { hook(); } @@ -2365,9 +2368,6 @@ async fn remove_dst_base_before_commit( #[cfg(not(test))] fn run_inline_preparation_before_backup(_dst_path: &str) {} -#[cfg(not(test))] -fn run_rename_data_after_first_publication(_dst_path: &str) {} - #[cfg(not(test))] fn should_fail_after_delete_data_staged(_path: &str) -> bool { false @@ -9015,8 +9015,9 @@ impl DiskAPI for LocalDisk { .await?; return Err(err); } + #[cfg(test)] if has_data_dir_path.is_some() { - run_rename_data_after_first_publication(dst_path); + run_rename_data_after_first_publication(&self.root, dst_volume, dst_path); } // Crash-consistency injection: hard power loss after the data dir @@ -9449,7 +9450,8 @@ impl DiskAPI for LocalDisk { let _ = remove_file_if_exists(staged_backup); return Err(err); } - run_rename_data_after_first_publication(dst_path); + #[cfg(test)] + run_rename_data_after_first_publication(&self.root, dst_volume, dst_path); if sync { file_sync_admission = Some( os::acquire_file_sync_admission(self.file_sync_permits.clone()) @@ -13133,7 +13135,7 @@ mod test { let replacement_staging_parent_for_hook = replacement_staging_parent.clone(); let staged_metadata_for_hook = staged_metadata.clone(); let replacement_staged_metadata_for_hook = replacement_staged_metadata.clone(); - set_rename_data_after_first_publication(object, move || { + set_rename_data_after_first_publication(&disk.root, bucket, object, move || { std::fs::rename(&object_dir_for_hook, &replacement_dir_for_hook) .expect_err("the destination object identity must remain pinned until xl.meta commits"); std::fs::rename(&staging_parent_for_hook, &replacement_staging_parent_for_hook) @@ -13400,7 +13402,7 @@ mod test { let replacement_dir_for_hook = replacement_dir.clone(); let staged_metadata_for_hook = staged_metadata.clone(); let replacement_staged_metadata_for_hook = replacement_staged_metadata.clone(); - set_rename_data_after_first_publication(object, move || { + set_rename_data_after_first_publication(&disk.root, bucket, object, move || { std::fs::rename(&object_dir_for_hook, &replacement_dir_for_hook) .expect_err("the destination object identity must remain pinned after publishing its rollback backup"); std::fs::rename(&staged_metadata_for_hook, &replacement_staged_metadata_for_hook) @@ -13766,7 +13768,7 @@ mod test { let (entered_tx, entered_rx) = mpsc::channel(); let (release_tx, release_rx) = mpsc::channel(); - set_rename_data_after_first_publication(object, move || { + set_rename_data_after_first_publication(&disk.root, bucket, object, move || { entered_tx.send(()).expect("signal first publication"); release_rx.recv().expect("wait while delete_volume is blocked"); }); @@ -14360,7 +14362,7 @@ mod test { let (published_tx, published_rx) = mpsc::channel(); let (release_tx, release_rx) = mpsc::channel(); - set_rename_data_after_first_publication(object, move || { + set_rename_data_after_first_publication(&disk.root, bucket, object, move || { published_tx.send(()).expect("signal backup publication"); release_rx.recv().expect("wait for lock-order assertion"); }); diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index a11c0030a..008542b8e 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -2034,7 +2034,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } object_size = object_size.checked_add(ext_part.size).ok_or(Error::PartMissingOrCorrupt)?; - if ext_part.actual_size < 0 && (!opts.replication_request || quota_context.is_enforced()) { + if ext_part.actual_size < 0 && (quota_context.is_enforced() || (!opts.replication_request && !opts.data_movement)) { return Err(Error::PartMissingOrCorrupt); } let normalized_actual_size = if ext_part.actual_size >= 0 && !transformed_object { diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 019a1fe64..6b0586ecf 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -4023,7 +4023,7 @@ mod tests { use crate::scanner_budget::ScannerCycleBudgetConfig; use crate::scanner_folder::ScannerItem; use crate::storage_api::owner::{EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats}; - use crate::storage_api::scan::{BucketOperations as _, MakeBucketOptions, ObjectIO as _}; + use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _}; use crate::{ DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerObjectOptions, ScannerPutObjReader, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, @@ -4248,16 +4248,20 @@ mod tests { async fn multi_pool_scanner_cycle_zero_fills_bucket_absent_from_first_pool() { let (_temp_dir, store) = setup_two_pool_scanner_store().await; let bucket = format!("scanner-second-pool-{}", Uuid::new_v4().simple()); - store.pools[1].disk_set[0] + store .make_bucket(&bucket, &MakeBucketOptions::default()) .await - .expect("bucket should be created only in the second pool"); + .expect("bucket and its authoritative metadata should be created"); let body = b"second-only"; let mut reader = ScannerPutObjReader::from_vec(body.to_vec()); - store.pools[1].disk_set[0] + store.pools[1] .put_object(&bucket, "pool-b", &mut reader, &ScannerObjectOptions::default()) .await .expect("object should be written only to the second pool"); + store.pools[0] + .delete_bucket(&bucket, &DeleteBucketOptions::default()) + .await + .expect("bucket should be removed from the first pool only"); let ctx = CancellationToken::new(); let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()); diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index 634f20e45..e77033aa2 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -278,7 +278,7 @@ pub(crate) mod scan { SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, }; #[cfg(test)] - pub(crate) use super::storage_contracts::{MakeBucketOptions, ObjectIO}; + pub(crate) use super::storage_contracts::{DeleteBucketOptions, MakeBucketOptions, ObjectIO}; } pub(crate) mod scanner_io {