test(e2e): keep multi-set restart graceful

Include the EC8+4 multi-set restart scenario in the graceful interruption lane so the unclean-shutdown marker assertion matches the scenario semantics.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -1823,6 +1823,7 @@ mod tests {
                 scenario,
                 InterruptionScenario::BackgroundTargetRestart
                     | InterruptionScenario::BackgroundTargetRestartEc84
+                    | InterruptionScenario::BackgroundTargetRestartEc84MultiSet
                     | InterruptionScenario::BackgroundCoordinatorRestart
             ) {
                 cluster.stop_node_gracefully(interruption_node).await?;

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;

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<str>,
     pub format: FormatV3,
     #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
@@ -4491,7 +4491,7 @@ impl SetDisks {
         instance_ctx: Arc<InstanceContext>,
     ) -> Arc<Self> {
         let ctx = instance_ctx;
-        let set_lock_namespace: Arc<str> = format!("set-{pool_index}-{set_index}").into();
+        let set_lock_namespace: Arc<str> = 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<dyn LockClient> = Arc::new(LocalClient::with_manager(manager_a));
+        let healthy_b: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(manager_b));
+        let failing_client: Arc<dyn LockClient> = 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<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
+        let second: Arc<dyn LockClient> = 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() {

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<crate::set_disk::SetDisks>| {
-                        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;