diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index fb983dc0c..b2b7a9dfe 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -840,7 +840,7 @@ fn resolve_start_decommission_pool_meta_reload_result(result: Result<()>) -> Res resolve_decommission_pool_meta_reload_result(result, "start_decommission") } -fn decommission_rebalance_meta_lock_error(err: rustfs_lock::LockError) -> Error { +fn activation_rebalance_meta_lock_error(err: rustfs_lock::LockError) -> Error { match err { rustfs_lock::LockError::QuorumNotReached { required, achieved } => Error::NamespaceLockQuorumUnavailable { mode: "write", @@ -850,12 +850,12 @@ fn decommission_rebalance_meta_lock_error(err: rustfs_lock::LockError) -> Error achieved, }, other => Error::other(format!( - "failed to acquire rebalance metadata write lock before decommission start on {RUSTFS_META_BUCKET}/{REBAL_META_NAME}: {other}" + "failed to acquire rebalance activation lock on {RUSTFS_META_BUCKET}/{REBAL_META_NAME}: {other}" )), } } -fn decommission_pool_meta_lock_error(err: rustfs_lock::LockError) -> Error { +fn activation_pool_meta_lock_error(err: rustfs_lock::LockError) -> Error { match err { rustfs_lock::LockError::QuorumNotReached { required, achieved } => Error::NamespaceLockQuorumUnavailable { mode: "write", @@ -865,11 +865,35 @@ fn decommission_pool_meta_lock_error(err: rustfs_lock::LockError) -> Error { achieved, }, other => Error::other(format!( - "failed to acquire pool metadata write lock before decommission start on {RUSTFS_META_BUCKET}/{POOL_META_NAME}: {other}" + "failed to acquire pool activation lock on {RUSTFS_META_BUCKET}/{POOL_META_NAME}: {other}" )), } } +pub(crate) async fn acquire_pool_rebalance_activation_locks( + pool: Arc, +) -> Result<(rustfs_lock::NamespaceLockGuard, rustfs_lock::NamespaceLockGuard)> +where + S: crate::storage_api_contracts::namespace::NamespaceLocking< + Error = Error, + NamespaceLock = rustfs_lock::NamespaceLockWrapper, + >, +{ + // Activation lock order is always pool.bin -> rebalance.bin. + let pool_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?; + let pool_meta_guard = pool_meta_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(activation_pool_meta_lock_error)?; + let rebalance_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?; + let rebalance_meta_guard = rebalance_meta_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(activation_rebalance_meta_lock_error)?; + + Ok((pool_meta_guard, rebalance_meta_guard)) +} + fn rollback_decommission_pool_meta(pool_meta: &mut PoolMeta, previous_pool_meta: PoolMeta) { *pool_meta = previous_pool_meta; } @@ -1675,7 +1699,7 @@ impl PoolMeta { self.load_no_lock(pool).await } - async fn load_no_lock(&mut self, pool: Arc) -> Result<()> + pub(crate) async fn load_no_lock(&mut self, pool: Arc) -> Result<()> where S: EcstoreObjectIO, { @@ -2501,16 +2525,7 @@ impl ECStore { .first() .cloned() .ok_or_else(|| Error::other("decommission start rebalance metadata load failed: no storage pools available"))?; - let pool_meta_lock = rebalance_pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?; - let _pool_meta_guard = pool_meta_lock - .get_write_lock(get_lock_acquire_timeout()) - .await - .map_err(decommission_pool_meta_lock_error)?; - let ns_lock = rebalance_pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?; - let _guard = ns_lock - .get_write_lock(get_lock_acquire_timeout()) - .await - .map_err(decommission_rebalance_meta_lock_error)?; + let (_pool_meta_guard, _rebalance_meta_guard) = acquire_pool_rebalance_activation_locks(rebalance_pool.clone()).await?; let mut rebalance_meta = RebalanceMeta::new(); match rebalance_meta @@ -5265,8 +5280,9 @@ pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usi mod pools_tests { use super::{ DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD, DecomBucketInfo, - DecommissionStartPoolState, DecommissionTerminalState, ListCallback, PoolDecommissionInfo, PoolMeta, PoolSpaceInfo, - PoolStatus, apply_decommission_status_space_info, bind_decommission_cancelers, bind_missing_decommission_cancelers, + DecommissionStartPoolState, DecommissionTerminalState, ListCallback, POOL_META_NAME, PoolDecommissionInfo, PoolMeta, + PoolSpaceInfo, PoolStatus, REBAL_META_NAME, acquire_pool_rebalance_activation_locks, + apply_decommission_status_space_info, bind_decommission_cancelers, bind_missing_decommission_cancelers, cancel_decommission_canceler, classify_decommission_terminal_state, count_decommission_item, decommission_cancel_signal_result, decommission_item_size, decommission_meta_bucket_options, decommission_start_pool_state, dedup_indices, default_decommission_bucket_concurrency, @@ -5304,15 +5320,45 @@ mod pools_tests { use crate::services::rebalance::{RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats}; use rustfs_filemeta::{FileInfo, FileInfoVersions, MetaCacheEntry, ObjectPartInfo}; use rustfs_rio::Index; + use std::future::Future; use std::sync::{ - Arc, + Arc, Mutex, atomic::{AtomicBool, AtomicUsize, Ordering}, }; + use std::task::{Context, Poll}; use std::time::Duration as StdDuration; use time::{Duration, OffsetDateTime}; use tokio::sync::Semaphore; use tokio_util::sync::CancellationToken; + #[derive(Debug)] + struct ActivationLockRecorder { + lock_manager: Arc, + owner: &'static str, + resources: Mutex>, + } + + #[async_trait::async_trait] + impl crate::storage_api_contracts::namespace::NamespaceLocking for ActivationLockRecorder { + type Error = Error; + type NamespaceLock = rustfs_lock::NamespaceLockWrapper; + + async fn new_ns_lock(&self, bucket: &str, object: &str) -> crate::error::Result { + self.resources + .lock() + .expect("activation lock recorder should not be poisoned") + .push(object.to_string()); + Ok(rustfs_lock::NamespaceLockWrapper::new( + rustfs_lock::NamespaceLock::with_local_manager( + "activation-lock-test".to_string(), + Arc::clone(&self.lock_manager), + ), + rustfs_lock::ObjectKey::new(bucket, object), + self.owner.to_string(), + )) + } + } + fn noop_decommission_list_callback() -> ListCallback { Arc::new(|_| Box::pin(async {})) } @@ -5343,6 +5389,48 @@ mod pools_tests { } } + #[tokio::test] + async fn test_activation_fence_uses_one_lock_order_and_serializes_callers() { + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let first = Arc::new(ActivationLockRecorder { + lock_manager: Arc::clone(&manager), + owner: "first", + resources: Mutex::new(Vec::new()), + }); + let second = Arc::new(ActivationLockRecorder { + lock_manager: manager, + owner: "second", + resources: Mutex::new(Vec::new()), + }); + + let first_guards = acquire_pool_rebalance_activation_locks(first.clone()) + .await + .expect("first activation should acquire both locks"); + assert_eq!( + *first + .resources + .lock() + .expect("activation lock recorder should not be poisoned"), + vec![POOL_META_NAME.to_string(), REBAL_META_NAME.to_string()] + ); + + let mut second_acquire = Box::pin(acquire_pool_rebalance_activation_locks(second.clone())); + let mut context = Context::from_waker(futures::task::noop_waker_ref()); + assert!(matches!(second_acquire.as_mut().poll(&mut context), Poll::Pending)); + + drop(first_guards); + second_acquire + .await + .expect("second activation should acquire both locks after the first releases them"); + assert_eq!( + *second + .resources + .lock() + .expect("activation lock recorder should not be poisoned"), + vec![POOL_META_NAME.to_string(), REBAL_META_NAME.to_string()] + ); + } + #[test] fn test_apply_decommission_status_space_info_adds_idle_pool_usage() { let status = apply_decommission_status_space_info( diff --git a/crates/ecstore/src/services/rebalance/control.rs b/crates/ecstore/src/services/rebalance/control.rs index 09e1993e7..0083fd61b 100644 --- a/crates/ecstore/src/services/rebalance/control.rs +++ b/crates/ecstore/src/services/rebalance/control.rs @@ -16,6 +16,7 @@ use super::{ RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats, RebalanceStopPropagationRecord, encode_rebalance_stop_propagation_record, }; +use crate::core::pools::{PoolMeta, acquire_pool_rebalance_activation_locks, pool_meta_has_active_decommission}; use crate::error::{Error, Result}; use crate::object_api::ObjectOptions; use crate::set_disk::get_lock_acquire_timeout; @@ -29,6 +30,38 @@ use time::OffsetDateTime; use tracing::{debug, info}; use uuid::Uuid; +fn ensure_rebalance_activation_pool_meta_allowed(meta: &PoolMeta) -> Result<()> { + if pool_meta_has_active_decommission(meta) { + return Err(Error::DecommissionAlreadyRunning); + } + + Ok(()) +} + +async fn merge_and_save_rebalance_meta_no_lock(pool: Arc, local_snapshot: &RebalanceMeta, stage: &str) -> Result<()> +where + S: EcstoreObjectIO, +{ + let opts = ObjectOptions { + no_lock: true, + ..Default::default() + }; + let mut merged = RebalanceMeta::new(); + match merged.load_with_opts(pool.clone(), opts.clone()).await { + Ok(()) => { + if merge_rebalance_meta(&mut merged, local_snapshot) == RebalanceMetaMergeOutcome::RejectedActiveConflict { + return Err(Error::RebalanceAlreadyRunning); + } + } + Err(Error::ConfigNotFound) => { + merged = local_snapshot.clone(); + } + Err(err) => return Err(Error::other(format!("rebalance meta load before save failed during {stage}: {err}"))), + } + + merged.save_with_opts(pool, opts).await +} + pub(super) fn validate_rebalance_disk_stats_coverage(disk_stats: &[DiskStat]) -> Result<()> { for (idx, disk_stat) in disk_stats.iter().enumerate() { if disk_stat.total_space == 0 { @@ -90,24 +123,53 @@ impl ECStore { .await .map_err(rebalance_meta_lock_error)?; - let opts = ObjectOptions { - no_lock: true, - ..Default::default() - }; - let mut merged = RebalanceMeta::new(); - match merged.load_with_opts(pool.clone(), opts.clone()).await { - Ok(()) => { - if merge_rebalance_meta(&mut merged, local_snapshot) == RebalanceMetaMergeOutcome::RejectedActiveConflict { - return Err(Error::RebalanceAlreadyRunning); - } - } - Err(Error::ConfigNotFound) => { - merged = local_snapshot.clone(); - } - Err(err) => return Err(Error::other(format!("rebalance meta load before save failed during {stage}: {err}"))), + merge_and_save_rebalance_meta_no_lock(pool, local_snapshot, stage).await + } + + async fn save_rebalance_activation_meta_with_merge( + &self, + pool: Arc, + local_snapshot: &RebalanceMeta, + stage: &str, + ) -> Result<()> + where + S: EcstoreObjectIO + StorageNamespaceLocking, + { + let (_pool_meta_guard, _rebalance_meta_guard) = acquire_pool_rebalance_activation_locks(pool.clone()).await?; + let mut pool_meta = PoolMeta::default(); + pool_meta.load_no_lock(pool.clone()).await?; + ensure_rebalance_activation_pool_meta_allowed(&pool_meta)?; + + merge_and_save_rebalance_meta_no_lock(pool, local_snapshot, stage).await + } + + pub(super) async fn fence_rebalance_worker_activation(&self, pool: Arc, expected_id: &str) -> Result + where + S: EcstoreObjectIO + StorageNamespaceLocking, + { + let (_pool_meta_guard, _rebalance_meta_guard) = acquire_pool_rebalance_activation_locks(pool.clone()).await?; + let mut pool_meta = PoolMeta::default(); + pool_meta.load_no_lock(pool.clone()).await?; + ensure_rebalance_activation_pool_meta_allowed(&pool_meta)?; + + let mut persisted = RebalanceMeta::new(); + persisted + .load_with_opts( + pool, + ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await?; + if persisted.id != expected_id { + return Err(Error::other(format!( + "rebalance metadata changed before worker activation: expected {expected_id}, found {}", + persisted.id + ))); } - merged.save_with_opts(pool, opts).await + Ok(is_rebalance_conflicting_with_decommission(&persisted)) } #[tracing::instrument(skip_all)] @@ -301,7 +363,8 @@ impl ECStore { let pool = clone_first_arc(&self.pools, "init_rebalance_meta: no pools available")?; resolve_rebalance_meta_save_result( - self.save_rebalance_meta_with_merge(pool, &meta, "init_rebalance_meta").await, + self.save_rebalance_activation_meta_with_merge(pool, &meta, "init_rebalance_meta") + .await, "init_rebalance_meta", )?; @@ -379,7 +442,7 @@ impl ECStore { } } - self.start_rebalance().await + self.start_rebalance_under_gate().await } pub async fn rollback_rebalance_start_for_id(self: &Arc, expected_id: Option<&str>, start_error: String) -> Result<()> { @@ -620,6 +683,31 @@ impl ECStore { mod tests { use super::*; + #[test] + fn rebalance_activation_rejects_persisted_decommission_despite_idle_local_snapshot() { + let persisted = PoolMeta { + pools: vec![crate::core::pools::PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(crate::core::pools::PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }), + }], + ..Default::default() + }; + + let err = ensure_rebalance_activation_pool_meta_allowed(&persisted) + .expect_err("persisted decommission must block a stale rebalance admission"); + assert!(matches!(err, Error::DecommissionAlreadyRunning)); + } + + #[test] + fn rebalance_activation_allows_persisted_idle_pool_meta() { + assert!(ensure_rebalance_activation_pool_meta_allowed(&PoolMeta::default()).is_ok()); + } + #[test] fn pool_rebalance_status_ignores_non_participating_pool_state() { let meta = RebalanceMeta { diff --git a/crates/ecstore/src/services/rebalance/runtime.rs b/crates/ecstore/src/services/rebalance/runtime.rs index 906e92922..e349afbe2 100644 --- a/crates/ecstore/src/services/rebalance/runtime.rs +++ b/crates/ecstore/src/services/rebalance/runtime.rs @@ -42,6 +42,11 @@ pub(super) fn source_cleanup_defer_attempt(deferred_attempts: &mut HashMap) -> Result<()> { + let _start_guard = self.start_gate.lock().await; + self.start_rebalance_under_gate().await + } + + pub(super) async fn start_rebalance_under_gate(self: &Arc) -> Result<()> { info!( event = EVENT_REBALANCE_STATE, component = LOG_COMPONENT_ECSTORE, @@ -49,8 +54,16 @@ impl ECStore { state = "starting", "Starting rebalance" ); + let expected_id = { + let rebalance_meta = self.rebalance_meta.read().await; + rebalance_meta.as_ref().ok_or(Error::ConfigNotFound)?.id.clone() + }; + let pool = clone_first_arc(self.pools.as_slice(), "start_rebalance: no pools available")?; + if !self.fence_rebalance_worker_activation(pool, &expected_id).await? { + return Ok(()); + } + let decommission_running = self.is_decommission_running().await; - // let rebalance_meta = self.rebalance_meta.read().await; let cancel_tx = CancellationToken::new(); let rx = cancel_tx.clone();