From 1aa4fa145f54c31f3e339b085c93078ee4fcd993 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 10 Sep 2026 16:42:21 +0800 Subject: [PATCH] fix(ecstore): use pool lock quorum for set writes (#7630) Use the pool-wide namespace lock client domain for every set in a pool so degraded EC8+4 multi-set writes are gated by node-level lock quorum instead of the narrower per-set endpoint host slice. Keep namespace-lock domain deduplication tied to both the pool namespace and shared clients, and add regression coverage for three-locker degraded writes plus cross-pool domain separation. Co-authored-by: zhi22915 --- crates/ecstore/src/core/sets.rs | 11 ++-- crates/ecstore/src/set_disk/mod.rs | 82 ++++++++++++++++++++++++++++-- crates/ecstore/src/store/object.rs | 56 +++++++++++++------- 3 files changed, 121 insertions(+), 28 deletions(-) diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index 7040776c8..bfe88aae1 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -219,7 +219,10 @@ impl Sets { let mut disk_set = Vec::with_capacity(set_count); - let lock_registry = runtime_sources::lock_registry(); + let pool_lockers = runtime_sources::lock_registry() + .as_ref() + .map(|registry| registry.clients_for_endpoints(endpoints.endpoints.as_ref())) + .unwrap_or_default(); for i in 0..set_count { let mut set_drive = Vec::with_capacity(set_drive_count); @@ -270,10 +273,6 @@ impl Sets { } } - let lockers = lock_registry - .as_ref() - .map(|registry| registry.clients_for_endpoints(&set_endpoints)) - .unwrap_or_default(); let set_disks = SetDisks::new_with_instance_ctx( runtime_sources::local_node_name().await, Arc::new(RwLock::new(set_drive)), @@ -283,7 +282,7 @@ impl Sets { pool_idx, set_endpoints, fm.clone(), - lockers, + pool_lockers.clone(), instance_ctx.clone(), ) .await; diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 5fc4c2e3d..f2cbafc55 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -3852,7 +3852,7 @@ pub struct SetDisks { pub default_parity_count: usize, pub set_index: usize, pub pool_index: usize, - /// Stable namespace shared by every object lock created for this set. + /// Stable namespace shared by every object lock created for this pool. set_lock_namespace: Arc, pub format: FormatV3, #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")] @@ -4491,7 +4491,7 @@ impl SetDisks { instance_ctx: Arc, ) -> Arc { let ctx = instance_ctx; - let set_lock_namespace: Arc = format!("set-{pool_index}-{set_index}").into(); + let set_lock_namespace: Arc = format!("pool-{pool_index}").into(); let shared_lockers = Arc::from(lockers.to_vec()); Arc::new(SetDisks { locker_owner, @@ -4605,7 +4605,9 @@ 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) => same_distributed_lock_domain(&self.lockers, &other.lockers), + (true, true) => { + self.set_lock_namespace == other.set_lock_namespace && same_distributed_lock_domain(&self.lockers, &other.lockers) + } _ => false, } } @@ -7123,7 +7125,7 @@ mod tests { ctx.update_erasure_type(SetupType::Erasure).await; let set = make_test_set_disks_with_ctx(Vec::new(), ctx).await; - assert_eq!(&*set.set_lock_namespace, "set-0-0"); + assert_eq!(&*set.set_lock_namespace, "pool-0"); let before = Arc::strong_count(&set.set_lock_namespace); let lock = set .new_ns_lock("bucket", "object") @@ -8348,6 +8350,78 @@ mod tests { ); } + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn test_new_ns_lock_distributed_write_succeeds_with_three_lockers_one_offline() { + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + + let manager_a = Arc::new(rustfs_lock::GlobalLockManager::new()); + let manager_b = Arc::new(rustfs_lock::GlobalLockManager::new()); + let healthy_a: Arc = Arc::new(LocalClient::with_manager(manager_a)); + let healthy_b: Arc = Arc::new(LocalClient::with_manager(manager_b)); + let failing_client: Arc = Arc::new(FailingClient); + let set_disks = make_test_set_disks(vec![healthy_a, failing_client, healthy_b]).await; + + let guard = set_disks + .new_ns_lock("bucket", "object") + .await + .expect("namespace lock should be created") + .get_write_lock(Duration::from_millis(500)) + .await + .expect("two healthy lockers should satisfy the three-locker write quorum"); + + match guard { + NamespaceLockGuard::Standard(_) => {} + NamespaceLockGuard::Fast(_) => panic!("Expected distributed guard for dist-erasure"), + } + } + + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn namespace_lock_domain_includes_pool_namespace() { + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + + let first: Arc = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new()))); + let second: Arc = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new()))); + let lockers = vec![first, second]; + let same_pool_first_set = make_test_set_disks_with_ctx(lockers.clone(), bootstrap_ctx()).await; + let same_pool_second_set = SetDisks::new_with_instance_ctx( + "test-owner".to_string(), + Arc::new(RwLock::new(vec![None, None])), + 2, + 1, + 1, + 0, + same_pool_first_set.set_endpoints.clone(), + FormatV3::new(2, 2), + lockers.clone(), + bootstrap_ctx(), + ) + .await; + let other_pool_set = SetDisks::new_with_instance_ctx( + "test-owner".to_string(), + Arc::new(RwLock::new(vec![None, None])), + 2, + 1, + 0, + 1, + same_pool_first_set.set_endpoints.clone(), + FormatV3::new(1, 2), + lockers, + bootstrap_ctx(), + ) + .await; + + assert!( + same_pool_first_set.shares_namespace_lock_domain(&same_pool_second_set).await, + "sets in the same pool share the object namespace lock domain" + ); + assert!( + !same_pool_first_set.shares_namespace_lock_domain(&other_pool_set).await, + "different pool namespaces must not be deduplicated solely by identical clients" + ); + } + #[tokio::test(flavor = "multi_thread")] #[serial] async fn streaming_reader_holds_read_lock_until_eof() { diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 59b778da1..cec09eb60 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -50,7 +50,7 @@ use crate::services::notification_sys::{ use crate::services::tier::tier::{TierConfigMgr, TierDestinationId, TierOperationLease, tier_destination_id_from_metadata}; 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, same_distributed_lock_domain, + is_lock_optimization_enabled, is_object_lock_diag_enabled, }; use crate::storage_api_contracts::{ list::ListOperations as _, @@ -3440,10 +3440,15 @@ impl ECStore { for pool in &self.pools { let hashed_set = pool.get_disks_by_key(object); - let lock_domain_already_held = !distributed - || locked_sets - .iter() - .any(|locked_set| same_distributed_lock_domain(&locked_set.lockers, &hashed_set.lockers)); + let mut lock_domain_already_held = !distributed; + if !lock_domain_already_held { + for locked_set in &locked_sets { + if locked_set.shares_namespace_lock_domain(&hashed_set).await { + lock_domain_already_held = true; + break; + } + } + } if lock_domain_already_held { continue; } @@ -3500,10 +3505,15 @@ impl ECStore { let mut locked_sets = vec![fixed_set]; for pool in &self.pools { for set in &pool.disk_set { - let lock_domain_already_held = !distributed - || locked_sets - .iter() - .any(|locked_set| same_distributed_lock_domain(&locked_set.lockers, &set.lockers)); + let mut lock_domain_already_held = !distributed; + if !lock_domain_already_held { + for locked_set in &locked_sets { + if locked_set.shares_namespace_lock_domain(set).await { + lock_domain_already_held = true; + break; + } + } + } if lock_domain_already_held { continue; } @@ -3581,10 +3591,15 @@ impl ECStore { let mut locked_sets = vec![fixed_set]; for pool in &self.pools { for set in &pool.disk_set { - let lock_domain_already_held = !distributed - || locked_sets - .iter() - .any(|locked_set| same_distributed_lock_domain(&locked_set.lockers, &set.lockers)); + let mut lock_domain_already_held = !distributed; + if !lock_domain_already_held { + for locked_set in &locked_sets { + if locked_set.shares_namespace_lock_domain(set).await { + lock_domain_already_held = true; + break; + } + } + } if lock_domain_already_held { continue; } @@ -3667,11 +3682,15 @@ impl ECStore { .get(pool_idx) .ok_or_else(|| Error::other(format!("invalid data movement publication pool {pool_idx}")))?; let set = pool.get_disks_by_key(object); - let lock_domain_already_held = !locked_sets.is_empty() - && (!distributed - || locked_sets.iter().any(|locked_set: &Arc| { - same_distributed_lock_domain(&locked_set.lockers, &set.lockers) - })); + let mut lock_domain_already_held = !locked_sets.is_empty() && !distributed; + if !lock_domain_already_held { + for locked_set in &locked_sets { + if locked_set.shares_namespace_lock_domain(&set).await { + lock_domain_already_held = true; + break; + } + } + } if lock_domain_already_held { continue; } @@ -5717,6 +5736,7 @@ mod tests { GetObjectBodyCacheHook, GetObjectBodyCacheHookLookup, GetObjectBodySource, clear_get_object_body_cache_hook, lookup_get_object_body_cache_hook, register_get_object_body_cache_hook, }; + use crate::set_disk::same_distributed_lock_domain; use crate::set_disk::{SetDisks, disk_call_counters}; use crate::storage_api_contracts::bucket::MakeBucketOptions; use crate::storage_api_contracts::lifecycle::TransitionedObject;