gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-22 20:36:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(ecstore): fix activation fence synchronization

Browse Source
This commit is contained in:
overtrue
2026-08-22 09:25:08 +08:00
parent 4fdc1c3a85
commit b8e9170cb0
2 changed files with 32 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/core/pools.rs
+18 -3
View File
@@ -2876,8 +2876,6 @@ impl ECStore {
.first()
.cloned()
.ok_or_else(|| Error::other("decommission start rebalance metadata load failed: no storage pools available"))?;
#[cfg(test)]
observe_pool_activation_start_attempt(PoolActivationStartKind::Decommission);
let activation_fence = acquire_pool_rebalance_activation_locks(rebalance_pool.clone()).await?;
let mut rebalance_meta = RebalanceMeta::new();
@@ -4420,6 +4418,8 @@ impl ECStore {
let indices = dedup_indices(&indices);
validate_start_decommission_request(&indices, self.single_pool())?;
#[cfg(test)]
observe_pool_activation_start_attempt(PoolActivationStartKind::Decommission);
self.ensure_decommission_rebalance_idle_after_refresh().await?;
#[cfg(test)]
let endpoints = self.instance_endpoints().unwrap_or_else(|| self.endpoints());
@@ -4775,7 +4775,22 @@ mod tests {
let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await;
let barrier = PoolActivationDurableSaveBarrier::install(&store.pools[0]);
let start_store = Arc::clone(&store);
let start_task = tokio::spawn(async move { start_store.start_decommission(vec![0]).await });
let start_task = tokio::spawn(async move {
start_store
.save_current_pool_meta_for_decommission_start(
&[0],
vec![(
0,
PoolSpaceInfo {
free: 50,
total: 100,
used: 50,
},
)],
Vec::new(),
)
.await
});
barrier.wait_until_paused().await;
let mut replica_disks = Vec::new();
crates/ecstore/src/services/rebalance/control.rs
+14 -6
View File
@@ -1192,6 +1192,17 @@ mod tests {
use crate::object_api::NamespaceLockFence;
use crate::set_disk::{PutObjectCommitBarrier, PutObjectCommitPause, hermetic_set_disks_isolated};
#[tokio::test]
async fn rebalance_stop_wait_probe_matches_run_id() {
let probe = RebalanceStopWaitProbe::install("rebalance-stop-current");
observe_rebalance_stop_wait_attempt(Some("rebalance-stop-stale"));
assert!(!probe.state.attempted.load(std::sync::atomic::Ordering::Acquire));
observe_rebalance_stop_wait_attempt(Some("rebalance-stop-current"));
probe.wait_until_attempted().await;
}
#[tokio::test]
async fn cancel_rebalance_admission_is_id_checked_and_idempotent() {
let rebalance_id = "rebalance-admission-current";
@@ -1340,10 +1351,7 @@ mod tests {
assert!(!admitted_cancel.is_cancelled());
drop(local_meta);
store
.cancel_rebalance_admission_for_id(rebalance_id)
.await
.expect("the installed token should cancel the admitted worker");
admitted_cancel.cancel();
assert!(admitted_cancel.is_cancelled());
let mut persisted = RebalanceMeta::new();
@@ -1401,7 +1409,7 @@ mod tests {
decommission_task = tokio::spawn(async move { store.start_decommission(vec![0]).await });
tokio::time::timeout(std::time::Duration::from_secs(15), probe.wait_until_attempted())
.await
.expect("real decommission start should reach activation lock acquisition");
.expect("real decommission start should reach the activation path");
} else {
let store = Arc::clone(&decommission_store);
decommission_task = tokio::spawn(async move { store.start_decommission(vec![0]).await });
@@ -1412,7 +1420,7 @@ mod tests {
tokio::spawn(async move { store.init_rebalance_start(vec!["bucket".to_string()]).await.map(|_| ()) });
tokio::time::timeout(std::time::Duration::from_secs(15), probe.wait_until_attempted())
.await
.expect("real rebalance start should reach activation lock acquisition");
.expect("real rebalance start should reach the activation path");
}
let mut observed_rebalance_result = None;