diff --git a/crates/ecstore/src/services/rebalance/control.rs b/crates/ecstore/src/services/rebalance/control.rs index 98d66a5b2..cc87ce94f 100644 --- a/crates/ecstore/src/services/rebalance/control.rs +++ b/crates/ecstore/src/services/rebalance/control.rs @@ -46,7 +46,44 @@ pub(super) enum RebalanceWorkerActivationFence { } pub(super) struct RebalanceRunGuard { - _guard: tokio::sync::OwnedRwLockReadGuard<()>, + _local_guard: tokio::sync::OwnedRwLockReadGuard<()>, + persisted_guard: rustfs_lock::NamespaceLockGuard, +} + +impl RebalanceRunGuard { + pub(super) fn ensure_held(&self, stage: &str) -> Result<()> { + if self.persisted_guard.is_lock_lost() { + return Err(Error::other(format!("rebalance distributed run fence lost during {stage}"))); + } + Ok(()) + } +} + +async fn acquire_persisted_rebalance_run_guard( + pool: Arc, + expected_id: &str, + stage: &str, +) -> Result +where + S: EcstoreObjectIO + StorageNamespaceLocking, +{ + let ns_lock = pool.new_ns_lock(crate::disk::RUSTFS_META_BUCKET, REBAL_META_NAME).await?; + let guard = ns_lock + .get_read_lock(get_lock_acquire_timeout()) + .await + .map_err(|err| rebalance_meta_lock_error(err, "read"))?; + let mut opts = ObjectOptions { + no_lock: true, + ..Default::default() + }; + opts.add_namespace_lock_guard(&guard); + let mut persisted = RebalanceMeta::new(); + persisted.load_with_opts(pool, opts).await?; + ensure_rebalance_worker_active(Some(&persisted), expected_id, stage)?; + if guard.is_lock_lost() { + return Err(Error::other(format!("rebalance distributed run fence lost during {stage}"))); + } + Ok(guard) } async fn merge_and_save_rebalance_meta_no_lock( @@ -180,6 +217,7 @@ impl ECStore { } pub(super) async fn rebalance_run_guard(&self, expected_id: &str, stage: &str) -> Result { + // Runtime fence order is activation_gate -> rebalance.bin. let activation_gate = { let meta = self.rebalance_meta.read().await; ensure_rebalance_worker_active(meta.as_ref(), expected_id, stage)?; @@ -203,7 +241,12 @@ impl ECStore { ))); } drop(meta); - Ok(RebalanceRunGuard { _guard: guard }) + let pool = clone_first_arc(self.pools.as_slice(), "rebalance run fence: no pools available")?; + let persisted_guard = acquire_persisted_rebalance_run_guard(pool, expected_id, stage).await?; + Ok(RebalanceRunGuard { + _local_guard: guard, + persisted_guard, + }) } pub(super) async fn save_rebalance_meta_with_merge( @@ -247,7 +290,7 @@ impl ECStore { let guard = ns_lock .get_write_lock(get_lock_acquire_timeout()) .await - .map_err(rebalance_meta_lock_error)?; + .map_err(|err| rebalance_meta_lock_error(err, "write"))?; let mut opts = ObjectOptions { no_lock: true, ..Default::default() @@ -984,6 +1027,43 @@ mod tests { assert_eq!(after.pool_stats[0].bytes, 1); } + #[tokio::test] + async fn persisted_rebalance_run_fence_rejects_stale_active_node_after_remote_stop() { + let (_temp_dirs, _disk_stores, set_disks) = hermetic_set_disks_isolated(4).await; + let active = RebalanceMeta { + id: "rebalance-a".to_string(), + cancel: Some(tokio_util::sync::CancellationToken::new()), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + active + .save(set_disks.clone()) + .await + .expect("active rebalance metadata should be saved"); + ensure_rebalance_worker_active(Some(&active), active.id.as_str(), "stale node precondition") + .expect("the stale node snapshot should still look active locally"); + + let mut stopped = active.clone(); + stopped.stopped_at = Some(OffsetDateTime::now_utc()); + stopped.pool_stats[0].info.stopping = true; + stopped + .save(set_disks.clone()) + .await + .expect("remote stop should replace persisted metadata"); + + let err = acquire_persisted_rebalance_run_guard(set_disks, active.id.as_str(), "cross-node stale snapshot") + .await + .expect_err("persisted stop must fence a node that missed stop propagation"); + assert!(err.to_string().contains("inactive rebalance worker rejected")); + } + #[test] fn pool_rebalance_status_ignores_non_participating_pool_state() { let meta = RebalanceMeta { diff --git a/crates/ecstore/src/services/rebalance/entry.rs b/crates/ecstore/src/services/rebalance/entry.rs index 84f0feb90..08cfc680c 100644 --- a/crates/ecstore/src/services/rebalance/entry.rs +++ b/crates/ecstore/src/services/rebalance/entry.rs @@ -56,6 +56,7 @@ impl ECStore { // Persisted stats can complete a pool on restart, so source cleanup must resolve first. let run_guard = self.rebalance_run_guard(expected_id, "rebalance source cleanup").await?; let cleanup_result = cleanup.await; + run_guard.ensure_held("rebalance source cleanup")?; drop(run_guard); let cleanup_result = resolve_rebalance_entry_cleanup_delete_result(cleanup_result, bucket, object); let RebalanceEntryCleanupResult::Completed { warning } = cleanup_result else { @@ -168,7 +169,7 @@ impl ECStore { let run_guard = self .rebalance_run_guard(rebalance_id.as_ref(), "rebalance lifecycle mutation") .await?; - let expired_by_lifecycle = crate::core::pools::should_skip_lifecycle_for_data_movement( + let lifecycle_result = crate::core::pools::should_skip_lifecycle_for_data_movement( self.clone(), &bucket, version, @@ -177,8 +178,9 @@ impl ECStore { true, &crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc::Rebal, ) - .await?; - drop(run_guard); + .await; + run_guard.ensure_held("rebalance lifecycle mutation")?; + let expired_by_lifecycle = lifecycle_result?; if expired_by_lifecycle { expired += 1; // The lifecycle expiry above physically deleted this version from the source set. @@ -196,6 +198,7 @@ impl ECStore { reason = "expired_by_lifecycle", "Skipped rebalance version" ); + drop(run_guard); continue; } @@ -213,6 +216,7 @@ impl ECStore { reason = "last_delete_marker_without_replication", "Skipped rebalance version" ); + drop(run_guard); continue; } @@ -222,7 +226,6 @@ impl ECStore { let store = self.clone(); async move { store - .clone() .rebalance_object(src_pool_idx, bucket, rd, expected_bucket_incarnation_id) .await } @@ -233,9 +236,6 @@ impl ECStore { let store = self.clone(); async move { store.delete_object(&bucket, &object, opts).await } }; - let run_guard = self - .rebalance_run_guard(rebalance_id.as_ref(), "rebalance version migration") - .await?; let result = migrate_entry_version( &RebalanceMigrationBackend::new(set.as_ref(), self.as_ref()), bucket.clone(), @@ -249,6 +249,7 @@ impl ECStore { &mut delete_marker, ) .await; + run_guard.ensure_held("rebalance version migration")?; drop(run_guard); if result.ignored { @@ -678,31 +679,21 @@ mod tests { #[tokio::test] async fn rebalance_stats_wait_for_source_cleanup_result() { - let endpoint_pools: crate::layout::endpoints::EndpointServerPools = Vec::new().into(); - let store = Arc::new(ECStore { - id: uuid::Uuid::new_v4(), - disk_map: std::collections::HashMap::new(), - pools: Vec::new(), - peer_sys: crate::cluster::rpc::S3PeerSys::new(&endpoint_pools), - pool_meta: tokio::sync::RwLock::new(crate::core::pools::PoolMeta::default()), - rebalance_meta: tokio::sync::RwLock::new(Some(RebalanceMeta { - pool_stats: vec![RebalanceStats { - participating: true, - info: RebalanceInfo { - start_time: Some(OffsetDateTime::now_utc()), - status: RebalStatus::Started, - ..Default::default() - }, + let rebalance_id = "rebalance-test"; + let (_temp_dirs, store) = crate::services::rebalance::test_store_with_persisted_rebalance_meta(RebalanceMeta { + id: rebalance_id.to_string(), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + start_time: Some(OffsetDateTime::now_utc()), + status: RebalStatus::Started, ..Default::default() - }], + }, ..Default::default() - })), - decommission_cancelers: tokio::sync::RwLock::new(Vec::new()), - start_gate: tokio::sync::Mutex::new(()), - pool_meta_save_gate: tokio::sync::Mutex::new(()), - ctx: crate::runtime::instance::bootstrap_ctx(), - bucket_fence_registry: std::sync::Arc::default(), - }); + }], + ..Default::default() + }) + .await; let mut version = FileInfo::new("object.bin", 4, 2); version.name = "object.bin".to_string(); version.size = 128; @@ -713,7 +704,7 @@ mod tests { let finish_store = Arc::clone(&store); let finish = tokio::spawn(async move { finish_store - .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&version], "", async move { + .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&version], rebalance_id, async move { cleanup_released.await.expect("cleanup release sender should remain alive"); Ok(ObjectInfo::default()) }) @@ -760,7 +751,7 @@ mod tests { meta.as_mut().expect("rebalance metadata should exist").pool_stats[0].bytes = 0; } let warning_result = store - .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], "", async { + .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], rebalance_id, async { Err(Error::SlowDown.into()) }) .await @@ -777,7 +768,7 @@ mod tests { meta.as_mut().expect("rebalance metadata should exist").pool_stats[0].bytes = 0; } let deferred = store - .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], "", async { + .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], rebalance_id, async { Err(data_movement::SourceCleanupError::SourceChanged) }) .await diff --git a/crates/ecstore/src/services/rebalance/mod.rs b/crates/ecstore/src/services/rebalance/mod.rs index 126cfa81c..f297335c6 100644 --- a/crates/ecstore/src/services/rebalance/mod.rs +++ b/crates/ecstore/src/services/rebalance/mod.rs @@ -53,5 +53,31 @@ pub use types::{ }; use types::{RebalanceBucketConfigs, RebalanceBucketOutcome, RebalanceEntryOutcome}; +#[cfg(test)] +pub(crate) async fn test_store_with_persisted_rebalance_meta( + meta: RebalanceMeta, +) -> (Vec, std::sync::Arc) { + let ctx = std::sync::Arc::new(crate::runtime::instance::InstanceContext::new()); + let (temp_dirs, pool) = crate::core::sets::make_local_two_set_sets_with_ctx(ctx.clone()).await; + meta.save(pool.clone()) + .await + .expect("rebalance test metadata should be persisted"); + let endpoint_pools: crate::layout::endpoints::EndpointServerPools = vec![pool.endpoints.clone()].into(); + let store = std::sync::Arc::new(crate::store::ECStore { + id: uuid::Uuid::new_v4(), + disk_map: std::collections::HashMap::new(), + pools: vec![pool], + peer_sys: crate::cluster::rpc::S3PeerSys::new_with_instance_ctx(&endpoint_pools, ctx.clone()), + pool_meta: tokio::sync::RwLock::new(crate::core::pools::PoolMeta::default()), + rebalance_meta: tokio::sync::RwLock::new(Some(meta)), + decommission_cancelers: tokio::sync::RwLock::new(vec![None]), + start_gate: tokio::sync::Mutex::new(()), + pool_meta_save_gate: tokio::sync::Mutex::new(()), + ctx, + bucket_fence_registry: std::sync::Arc::default(), + }); + (temp_dirs, store) +} + #[cfg(test)] mod rebalance_unit_tests; diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index c4a67d9e8..14ad49285 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs @@ -33,8 +33,8 @@ use super::migration::{ rebalance_delete_marker_opts, }; use super::runtime::{ - RebalanceLocalActivationOutcome, commit_local_rebalance_worker_activation, should_fail_repeated_rebalance_bucket_defer, - source_cleanup_defer_attempt, + RebalanceLocalActivationOutcome, commit_local_rebalance_worker_activation, resolve_rebalance_pre_spawn_result, + should_fail_repeated_rebalance_bucket_defer, source_cleanup_defer_attempt, }; use super::worker::{ RebalanceEntryCleanupResult, ensure_rebalance_listing_disks_available, is_transient_rebalance_error, @@ -2736,6 +2736,54 @@ fn test_stopped_activation_state_prevents_worker_token_commit() { assert!(meta.cancel.is_none(), "stopped rebalance must not receive a worker token"); } +#[test] +fn test_rebalance_start_save_failure_rolls_back_local_worker_token() { + let activation_token = tokio_util::sync::CancellationToken::new(); + let observer = activation_token.clone(); + let mut meta = RebalanceMeta { + id: "rebalance-a".to_string(), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + assert_eq!( + commit_local_rebalance_worker_activation(&mut meta, "rebalance-a", activation_token.clone()) + .expect("active metadata should accept the worker token"), + RebalanceLocalActivationOutcome::Started + ); + + let err = resolve_rebalance_pre_spawn_result( + Some(&mut meta), + "rebalance-a", + &activation_token, + Err(Error::NamespaceLockQuorumUnavailable { + mode: "write", + bucket: crate::disk::RUSTFS_META_BUCKET.to_string(), + object: super::REBAL_META_NAME.to_string(), + required: 3, + achieved: 2, + }), + ) + .expect_err("metadata save failure must abort local worker activation"); + + assert!(matches!( + err, + Error::NamespaceLockQuorumUnavailable { + required: 3, + achieved: 2, + .. + } + )); + assert!(observer.is_cancelled(), "the failed activation token must be canceled"); + assert!(meta.cancel.is_none(), "a retry must not see a phantom active worker token"); +} + #[tokio::test] async fn test_old_worker_cannot_mutate_replacement_rebalance_state() { let meta = RebalanceMeta { @@ -2809,7 +2857,7 @@ async fn test_stop_waits_for_active_rebalance_migration_guard() { }], ..Default::default() }; - let store = test_store_with_rebalance_meta(meta); + let (_temp_dirs, store) = super::test_store_with_persisted_rebalance_meta(meta).await; let run_guard = store .rebalance_run_guard("rebalance-a", "rebalance remote-tier migration") .await @@ -2830,7 +2878,7 @@ async fn test_stop_waits_for_active_rebalance_migration_guard() { drop(run_guard); stop.await - .expect_err("empty test store should fail only after committing the local stop state"); + .expect("stop should persist after the side-effect fence is released"); assert!( store .rebalance_meta diff --git a/crates/ecstore/src/services/rebalance/runtime.rs b/crates/ecstore/src/services/rebalance/runtime.rs index b24b23812..6a5b12a0f 100644 --- a/crates/ecstore/src/services/rebalance/runtime.rs +++ b/crates/ecstore/src/services/rebalance/runtime.rs @@ -64,6 +64,36 @@ pub(super) fn commit_local_rebalance_worker_activation( Ok(RebalanceLocalActivationOutcome::Started) } +pub(super) fn rollback_local_rebalance_worker_activation( + meta: Option<&mut super::RebalanceMeta>, + expected_id: &str, + activation_token: &CancellationToken, +) -> bool { + let Some(meta) = meta else { + return false; + }; + if meta.id != expected_id || meta.cancel.as_ref() != Some(activation_token) { + return false; + } + if let Some(cancel) = meta.cancel.take() { + cancel.cancel(); + return true; + } + false +} + +pub(super) fn resolve_rebalance_pre_spawn_result( + meta: Option<&mut super::RebalanceMeta>, + expected_id: &str, + activation_token: &CancellationToken, + result: Result<()>, +) -> Result<()> { + if result.is_err() { + rollback_local_rebalance_worker_activation(meta, expected_id, activation_token); + } + result +} + impl ECStore { #[tracing::instrument(skip_all)] pub async fn start_rebalance(self: &Arc) -> Result<()> { @@ -84,7 +114,10 @@ impl ECStore { Arc::from(rebalance_meta.as_ref().ok_or(Error::ConfigNotFound)?.id.as_str()) }; let pool = clone_first_arc(self.pools.as_slice(), "start_rebalance: no pools available")?; - let activation_fence = match self.fence_rebalance_worker_activation(pool, expected_id.as_ref()).await? { + let activation_fence = match self + .fence_rebalance_worker_activation(pool.clone(), expected_id.as_ref()) + .await? + { RebalanceWorkerActivationFence::Ready(fence) => fence, RebalanceWorkerActivationFence::NotStartedTerminal => return Ok(()), }; @@ -129,12 +162,15 @@ impl ECStore { drop(activation_fence); if let Some(meta) = meta_to_save { - let pool = clone_first_arc(self.pools.as_slice(), "start_rebalance: no pools available")?; - resolve_rebalance_meta_save_result( + let save_result = resolve_rebalance_meta_save_result( self.save_rebalance_meta_with_merge(pool, &meta, "start_rebalance complete pools at goal") .await, "start_rebalance complete pools at goal", - )?; + ); + if save_result.is_err() { + let mut rebalance_meta = self.rebalance_meta.write().await; + return resolve_rebalance_pre_spawn_result(rebalance_meta.as_mut(), expected_id.as_ref(), &rx, save_result); + } } if activation_outcome != RebalanceLocalActivationOutcome::Started { @@ -156,6 +192,8 @@ impl ECStore { }; if !participants.iter().any(|participating| *participating) { + let mut rebalance_meta = self.rebalance_meta.write().await; + rollback_local_rebalance_worker_activation(rebalance_meta.as_mut(), expected_id.as_ref(), &rx); debug!( event = EVENT_REBALANCE_STATE, component = LOG_COMPONENT_ECSTORE, @@ -225,6 +263,8 @@ impl ECStore { } if workers_started == 0 { + let mut rebalance_meta = self.rebalance_meta.write().await; + rollback_local_rebalance_worker_activation(rebalance_meta.as_mut(), expected_id.as_ref(), &rx); debug!( event = EVENT_REBALANCE_STATE, component = LOG_COMPONENT_ECSTORE, diff --git a/crates/ecstore/src/services/rebalance/worker.rs b/crates/ecstore/src/services/rebalance/worker.rs index 5a6b7abc6..38f15bc78 100644 --- a/crates/ecstore/src/services/rebalance/worker.rs +++ b/crates/ecstore/src/services/rebalance/worker.rs @@ -100,17 +100,17 @@ pub(super) fn resolve_rebalance_meta_save_result(result: Result<()>, stage: &str result.map_err(|err| Error::other(format!("rebalance meta save failed during {stage}: {err}"))) } -pub(super) fn rebalance_meta_lock_error(err: rustfs_lock::LockError) -> Error { +pub(super) fn rebalance_meta_lock_error(err: rustfs_lock::LockError, mode: &'static str) -> Error { match err { rustfs_lock::LockError::QuorumNotReached { required, achieved } => Error::NamespaceLockQuorumUnavailable { - mode: "write", + mode, bucket: crate::disk::RUSTFS_META_BUCKET.to_string(), object: REBAL_META_NAME.to_string(), required, achieved, }, other => Error::other(format!( - "failed to acquire rebalance metadata write lock on {}/{}: {other}", + "failed to acquire rebalance metadata {mode} lock on {}/{}: {other}", crate::disk::RUSTFS_META_BUCKET, REBAL_META_NAME )),