gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-15 17:43:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(ecstore): isolate rename publication hooks (#6106)

Browse Source
This commit is contained in:
cxymds
2026-08-14 19:07:34 +08:00
committed by GitHub
parent 5a4c063d16
commit 6f29431a65
4 changed files with 27 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/disk/local.rs
+17 -15
View File
@@ -2028,14 +2028,17 @@ static RENAME_DATA_REMOVE_DST_BASE_BEFORE_COMMIT: std::sync::Mutex<Option<(Strin
#[cfg(test)]
type InlinePreparationHook = Box<dyn FnOnce() + Send>;
#[cfg(test)]
type RenameDataPublicationHookKey = (PathBuf, String, String);
#[cfg(test)]
static INLINE_PREPARATION_BEFORE_BACKUP: std::sync::LazyLock<std::sync::Mutex<HashMap<String, InlinePreparationHook>>> =
std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new()));
#[cfg(test)]
static INLINE_BEFORE_FILE_SYNC_ADMISSION: std::sync::LazyLock<std::sync::Mutex<HashMap<String, InlinePreparationHook>>> =
std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new()));
#[cfg(test)]
static RENAME_DATA_AFTER_FIRST_PUBLICATION: std::sync::LazyLock<std::sync::Mutex<HashMap<String, InlinePreparationHook>>> =
std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new()));
static RENAME_DATA_AFTER_FIRST_PUBLICATION: std::sync::LazyLock<
std::sync::Mutex<HashMap<RenameDataPublicationHookKey, InlinePreparationHook>>,
> = std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new()));
#[cfg(test)]
static OWNED_FILE_WRITE_BEFORE_OPEN: std::sync::LazyLock<std::sync::Mutex<HashMap<PathBuf, InlinePreparationHook>>> =
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");
});
crates/ecstore/src/set_disk/ops/multipart.rs
+1 -1
View File
@@ -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 {
crates/scanner/src/scanner_io.rs
+8 -4
View File
@@ -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());
crates/scanner/src/storage_api.rs
+1 -1
View File
@@ -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 {