mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-03 18:55:39 +00:00
fix(ecstore): preserve distributed decommission set locks
This commit is contained in:
@@ -3604,6 +3604,15 @@ impl SetDisks {
|
|||||||
&self.ctx
|
&self.ctx
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Whether both sets' namespace-lock implementations cover the same object key.
|
||||||
|
pub(crate) async fn shares_namespace_lock_domain(&self, other: &Self) -> bool {
|
||||||
|
match (self.ctx.is_dist_erasure().await, other.ctx.is_dist_erasure().await) {
|
||||||
|
(false, false) => Arc::ptr_eq(&self.local_lock_manager, &other.local_lock_manager),
|
||||||
|
(true, true) => Arc::ptr_eq(&self.ctx, &other.ctx) && self.set_lock_namespace == other.set_lock_namespace,
|
||||||
|
_ => false,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// The lock manager this set actually uses (test-only; Phase 5 Slice 3).
|
/// The lock manager this set actually uses (test-only; Phase 5 Slice 3).
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
pub(crate) fn local_lock_manager_for_test(&self) -> &Arc<rustfs_lock::GlobalLockManager> {
|
pub(crate) fn local_lock_manager_for_test(&self) -> &Arc<rustfs_lock::GlobalLockManager> {
|
||||||
|
|||||||
@@ -1944,16 +1944,12 @@ impl ECStore {
|
|||||||
};
|
};
|
||||||
|
|
||||||
mutation_fence.add_namespace_lock_fence(opts);
|
mutation_fence.add_namespace_lock_fence(opts);
|
||||||
let distributed = self.ctx.is_dist_erasure().await;
|
|
||||||
let fixed_set = self.pools.first().and_then(|pool| pool.disk_set.first());
|
let fixed_set = self.pools.first().and_then(|pool| pool.disk_set.first());
|
||||||
let target_set = self.pools.get(target_pool_idx).map(|pool| pool.get_disks_by_key(object));
|
let target_set = self.pools.get(target_pool_idx).map(|pool| pool.get_disks_by_key(object));
|
||||||
// Local locks share one manager without set-qualified resource keys.
|
opts.no_lock = match (fixed_set, target_set) {
|
||||||
// Distributed locks overlap when both sets use the same client domain.
|
(Some(fixed), Some(target)) => fixed.shares_namespace_lock_domain(&target).await,
|
||||||
opts.no_lock = matches!(
|
_ => false,
|
||||||
(fixed_set, target_set),
|
};
|
||||||
(Some(fixed), Some(target))
|
|
||||||
if !distributed || same_distributed_lock_domain(&fixed.lockers, &target.lockers)
|
|
||||||
);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) async fn acquire_decommission_source_cleanup_fence(
|
pub(crate) async fn acquire_decommission_source_cleanup_fence(
|
||||||
@@ -1972,9 +1968,8 @@ impl ECStore {
|
|||||||
let test_namespace_lock_fence =
|
let test_namespace_lock_fence =
|
||||||
decommission_mutation_fence_for_test(bucket, object, DecommissionMutationFenceTestPhase::SourceCleanup);
|
decommission_mutation_fence_for_test(bucket, object, DecommissionMutationFenceTestPhase::SourceCleanup);
|
||||||
let object = encode_dir_object(object);
|
let object = encode_dir_object(object);
|
||||||
let distributed = self.ctx.is_dist_erasure().await;
|
|
||||||
let fixed_set = Arc::clone(&self.pools[0].disk_set[0]);
|
let fixed_set = Arc::clone(&self.pools[0].disk_set[0]);
|
||||||
let source_lock_covered = !distributed || same_distributed_lock_domain(&fixed_set.lockers, &source_set.lockers);
|
let source_lock_covered = fixed_set.shares_namespace_lock_domain(source_set).await;
|
||||||
// Lock order: fixed store mutation domain first; source cleanup takes its
|
// Lock order: fixed store mutation domain first; source cleanup takes its
|
||||||
// hashed source-domain lock second only when this guard does not cover it.
|
// hashed source-domain lock second only when this guard does not cover it.
|
||||||
let guard = self
|
let guard = self
|
||||||
@@ -3917,6 +3912,65 @@ mod tests {
|
|||||||
assert!(!same_distributed_lock_domain(&[first, second], &[other]));
|
assert!(!same_distributed_lock_domain(&[first, second], &[other]));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn decommission_fence_keeps_dist_set_locks_when_clients_match_but_namespaces_differ() {
|
||||||
|
let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
|
||||||
|
let (_dirs, original_sets) = make_local_two_set_sets_with_ctx(Arc::clone(&ctx)).await;
|
||||||
|
let mut second_set = (*original_sets.disk_set[1]).clone();
|
||||||
|
second_set.lockers = original_sets.disk_set[0].lockers.clone();
|
||||||
|
let mut sets = (*original_sets).clone();
|
||||||
|
sets.disk_set[1] = Arc::new(second_set);
|
||||||
|
let sets = Arc::new(sets);
|
||||||
|
ctx.update_erasure_type(SetupType::DistErasure).await;
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
sets.disk_set[0]
|
||||||
|
.lockers
|
||||||
|
.iter()
|
||||||
|
.zip(&sets.disk_set[1].lockers)
|
||||||
|
.all(|(fixed, hashed)| Arc::ptr_eq(fixed, hashed)),
|
||||||
|
"the regression requires identical distributed lock clients"
|
||||||
|
);
|
||||||
|
assert_ne!(sets.disk_set[0].set_index, sets.disk_set[1].set_index);
|
||||||
|
|
||||||
|
let pool_config = sets.endpoints.clone();
|
||||||
|
let store = new_prepared_reader_test_store_from_pools(vec![Arc::clone(&sets)], vec![pool_config], ctx);
|
||||||
|
let object = (0..1_000)
|
||||||
|
.map(|index| format!("decommission-dist-domain-{index}.bin"))
|
||||||
|
.find(|candidate| Arc::ptr_eq(&sets.get_disks_by_key(candidate), &sets.disk_set[1]))
|
||||||
|
.expect("a key should hash to the second set namespace");
|
||||||
|
let mutation_fence = store
|
||||||
|
.acquire_decommission_object_mutation_fence("bucket", &object)
|
||||||
|
.await
|
||||||
|
.expect("the fixed distributed mutation fence should be acquired");
|
||||||
|
|
||||||
|
let mut put_opts = ObjectOptions::default();
|
||||||
|
store
|
||||||
|
.apply_decommission_target_mutation_fence(0, &object, &mut put_opts, Some(&mutation_fence))
|
||||||
|
.await;
|
||||||
|
assert!(!put_opts.no_lock, "migration target PUT must retain the hashed-set lock");
|
||||||
|
|
||||||
|
let mut multipart_opts = ObjectOptions::default();
|
||||||
|
store
|
||||||
|
.apply_decommission_target_mutation_fence(0, &object, &mut multipart_opts, Some(&mutation_fence))
|
||||||
|
.await;
|
||||||
|
assert!(!multipart_opts.no_lock, "migration target multipart must retain the hashed-set lock");
|
||||||
|
drop(mutation_fence);
|
||||||
|
|
||||||
|
let cleanup_object = (0..1_000)
|
||||||
|
.map(|index| format!("decommission-dist-cleanup-{index}.bin"))
|
||||||
|
.find(|candidate| Arc::ptr_eq(&sets.get_disks_by_key(candidate), &sets.disk_set[1]))
|
||||||
|
.expect("a cleanup key should hash to the second set namespace");
|
||||||
|
let source_fence = store
|
||||||
|
.acquire_decommission_source_cleanup_fence("bucket", &cleanup_object, sets.disk_set[1].as_ref())
|
||||||
|
.await
|
||||||
|
.expect("the fixed distributed cleanup fence should be acquired");
|
||||||
|
assert!(
|
||||||
|
!source_fence.source_lock_covered(),
|
||||||
|
"source cleanup must retain its distinct distributed set lock"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn select_snapshot_version_matching_normalizes_null_and_uuid_forms() {
|
fn select_snapshot_version_matching_normalizes_null_and_uuid_forms() {
|
||||||
let nil = Uuid::nil();
|
let nil = Uuid::nil();
|
||||||
|
|||||||
Reference in New Issue
Block a user