From 856abfb46f3ee58a93d0ef9adf3102b6b217c026 Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 22 Aug 2026 05:34:48 +0800 Subject: [PATCH] fix(ecstore): match decommission lock backend domain --- crates/ecstore/src/set_disk/mod.rs | 12 ++++++++- crates/ecstore/src/store/object.rs | 41 +++++++++++++++++------------- 2 files changed, 34 insertions(+), 19 deletions(-) diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 0054b62bc..c65ee4917 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -3029,6 +3029,16 @@ pub struct SetDisks { storage_class_config_override: Arc>>>, } +// DistributedLock sends the raw ObjectKey to its clients; LockRegistry clones +// each endpoint's canonical Arc, so an exact Arc set identifies the lock domain. +pub(crate) fn same_distributed_lock_domain(left: &[Arc], right: &[Arc]) -> bool { + left.iter() + .all(|left_client| right.iter().any(|right_client| Arc::ptr_eq(left_client, right_client))) + && right + .iter() + .all(|right_client| left.iter().any(|left_client| Arc::ptr_eq(left_client, right_client))) +} + const ERASURE_CACHE_MAX_ENTRIES: usize = 32; #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] @@ -3608,7 +3618,7 @@ impl SetDisks { 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, + (true, true) => same_distributed_lock_domain(&self.lockers, &other.lockers), _ => false, } } diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 24c85f6f9..002fea448 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -38,7 +38,7 @@ use crate::disk::OldCurrentSize; use crate::object_api::{NamespaceLockFence, ObjectLockConfigSnapshot}; use crate::set_disk::{ SetDisks, get_lock_acquire_timeout, get_object_lock_diag_slow_acquire_threshold, get_object_lock_diag_slow_hold_threshold, - is_lock_optimization_enabled, is_object_lock_diag_enabled, + is_lock_optimization_enabled, is_object_lock_diag_enabled, same_distributed_lock_domain, }; use crate::storage_api_contracts::{ namespace::NamespaceLocking as _, @@ -801,16 +801,6 @@ impl SelectObjectSnapshotLockLossWake { } } -// LockRegistry clones its canonical client Arc for each endpoint host, so an -// exact Arc set identifies one distributed namespace-lock quorum domain. -fn same_distributed_lock_domain(left: &[Arc], right: &[Arc]) -> bool { - left.iter() - .all(|left_client| right.iter().any(|right_client| Arc::ptr_eq(left_client, right_client))) - && right - .iter() - .all(|right_client| left.iter().any(|left_client| Arc::ptr_eq(left_client, right_client))) -} - impl AsyncRead for SelectObjectSnapshotReader { fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { if self.lock_loss_wake.poll_lost(cx) || self.lease.is_lost() { @@ -3913,7 +3903,7 @@ mod tests { } #[tokio::test] - async fn decommission_fence_keeps_dist_set_locks_when_clients_match_but_namespaces_differ() { + async fn decommission_fence_covers_dist_sets_with_same_clients_despite_different_namespaces() { 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(); @@ -3943,18 +3933,27 @@ mod tests { .acquire_decommission_object_mutation_fence("bucket", &object) .await .expect("the fixed distributed mutation fence should be acquired"); + let target_lock = sets.disk_set[1] + .new_ns_lock("bucket", &object) + .await + .expect("the hashed-set namespace lock should be created"); + let target_err = target_lock + .get_write_lock(Duration::from_millis(50)) + .await + .expect_err("the fixed read fence must conflict through the shared clients"); + assert!(matches!(target_err, rustfs_lock::LockError::Timeout { .. })); 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"); + assert!(put_opts.no_lock, "migration target PUT must reuse the covering fixed fence"); 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"); + assert!(multipart_opts.no_lock, "migration target multipart must reuse the covering fixed fence"); drop(mutation_fence); let cleanup_object = (0..1_000) @@ -3965,10 +3964,16 @@ mod tests { .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" - ); + assert!(source_fence.source_lock_covered(), "source cleanup must reuse the covering fixed fence"); + let source_lock = sets.disk_set[1] + .new_ns_lock("bucket", &cleanup_object) + .await + .expect("the source-set namespace lock should be created"); + let source_err = source_lock + .get_read_lock(Duration::from_millis(50)) + .await + .expect_err("the fixed write fence must conflict through the shared clients"); + assert!(matches!(source_err, rustfs_lock::LockError::Timeout { .. })); } #[test]