diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index c7dbf5686..1cae6c65b 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -444,6 +444,7 @@ pub mod rebalance { #[cfg(feature = "test-util")] pub mod test_util { pub use crate::services::rebalance::entry::test_util::PausedRebalanceEntryTestFixture; + pub use crate::services::rebalance::test_store_with_persisted_rebalance_meta; } } diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index e7d274cad..67370594a 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -28,7 +28,9 @@ use crate::bucket::{ metadata_sys, }; use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw}; -use crate::config::com::{CONFIG_PREFIX, read_config, read_config_no_lock, save_config, save_config_with_opts}; +use crate::config::com::{ + CONFIG_PREFIX, read_config, read_config_no_lock, save_config, save_config_with_opts, save_config_with_opts_quiet, +}; use crate::data_movement; use crate::data_movement::backpressure::{self, DataMovementOperation}; use crate::data_usage::DATA_USAGE_CACHE_NAME; @@ -1997,9 +1999,10 @@ impl PoolMeta { // Pool zero is canonical. Once its save succeeds, later writes only // replicate committed state and must not reuse the admission fence. + // Failed replicas remain repairable by the next full pool metadata save. drop(activation_fence); - for pool in pools { - save_config_with_opts( + for (pool_index, pool) in pools.enumerate() { + if let Err(err) = save_config_with_opts_quiet( pool, POOL_META_NAME, data.clone(), @@ -2009,7 +2012,18 @@ impl PoolMeta { ..Default::default() }, ) - .await?; + .await + { + warn!( + event = EVENT_DECOMMISSION_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + pool_index = pool_index + 1, + state = "activation_replica_repair_pending", + error = %err, + "Decommission activation replica repair pending" + ); + } } Ok(()) @@ -4755,6 +4769,65 @@ mod tests { } } + #[tokio::test] + #[serial_test::serial] + async fn decommission_activation_adopts_canonical_commit_after_replica_failure() { + 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 }); + + barrier.wait_until_paused().await; + let mut replica_disks = Vec::new(); + for set in &store.pools[1].disk_set { + let mut disks = set.disks.write().await; + let saved = std::mem::take(&mut *disks); + *disks = vec![None; saved.len()]; + replica_disks.push(saved); + } + barrier.release_after_fence_loss(); + + tokio::time::timeout(std::time::Duration::from_secs(30), start_task) + .await + .expect("decommission activation should finish after its canonical commit") + .expect("decommission activation task should not panic") + .expect("a replica save failure must not report the committed activation as failed"); + + for (set, disks) in store.pools[1].disk_set.iter().zip(replica_disks) { + *set.disks.write().await = disks; + } + + let local = store.pool_meta.read().await; + assert!(pool_meta_has_active_decommission(&local)); + drop(local); + + let mut canonical = PoolMeta::default(); + canonical + .load_no_lock(store.pools[0].clone()) + .await + .expect("the canonical committed decommission metadata should remain readable"); + assert!(pool_meta_has_active_decommission(&canonical)); + + let mut replica = PoolMeta::default(); + replica + .load_no_lock(store.pools[1].clone()) + .await + .expect("the stale replica metadata should remain readable after disks recover"); + assert!(!pool_meta_has_active_decommission(&replica)); + + let worker_cancel = CancellationToken::new(); + store + .spawn_decommission_routines(Arc::clone(&store), worker_cancel.clone(), vec![0]) + .await + .expect("the committed activation should admit its decommission worker"); + let admitted_cancel = store.decommission_cancelers.read().await[0] + .clone() + .expect("the admitted decommission worker should have a cancellation token"); + assert!(!admitted_cancel.is_cancelled()); + worker_cancel.cancel(); + assert!(admitted_cancel.is_cancelled()); + } + #[test] fn ensure_pool_not_left_in_cmdline_after_decommission_allows_active_pool() { assert!(ensure_pool_not_left_in_cmdline_after_decommission(0, "http://node{1...4}/disk{1...4}", false).is_ok()); diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index fc94a4dda..5b9bada4c 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -119,8 +119,8 @@ pub(crate) fn endpoint_erasure_set_count() -> Option { endpoint_pools().map(|endpoints| endpoints.es_count()) } -pub(crate) fn endpoint_pool_is_local(pool_index: usize) -> bool { - get_global_endpoints() +pub(crate) fn endpoint_pool_is_local(endpoints: &EndpointServerPools, pool_index: usize) -> bool { + endpoints .as_ref() .get(pool_index) .is_some_and(|pool| pool.endpoints.as_ref().first().is_some_and(|endpoint| endpoint.is_local)) diff --git a/crates/ecstore/src/services/rebalance/control.rs b/crates/ecstore/src/services/rebalance/control.rs index 3b9304617..fdb24a1da 100644 --- a/crates/ecstore/src/services/rebalance/control.rs +++ b/crates/ecstore/src/services/rebalance/control.rs @@ -536,6 +536,41 @@ impl ECStore { self.load_rebalance_meta_under_start_gate().await } + /// Cancels local admission before refreshing the persisted stop target under the start gate. + pub async fn prepare_rebalance_stop(&self) -> Result> { + let _start_guard = self.start_gate.lock().await; + + { + let mut rebalance_meta = self.rebalance_meta.write().await; + if let Some(meta) = rebalance_meta.as_mut() + && is_rebalance_conflicting_with_decommission(meta) + { + meta.cancel + .get_or_insert_with(tokio_util::sync::CancellationToken::new) + .cancel(); + #[cfg(any(test, feature = "test-util"))] + observe_rebalance_stop_wait_attempt(Some(meta.id.as_str())); + } + } + + self.load_rebalance_meta_under_start_gate().await?; + + let mut rebalance_meta = self.rebalance_meta.write().await; + let Some(meta) = rebalance_meta.as_mut() else { + return Ok(None); + }; + if !is_rebalance_conflicting_with_decommission(meta) { + return Ok(None); + } + if meta.id.is_empty() { + return Err(Error::other("active rebalance metadata has no activation id")); + } + meta.cancel + .get_or_insert_with(tokio_util::sync::CancellationToken::new) + .cancel(); + Ok(Some(meta.id.clone())) + } + pub(crate) async fn load_rebalance_meta_under_start_gate(&self) -> Result<()> { let mut meta = RebalanceMeta::new(); debug!( @@ -1298,11 +1333,19 @@ mod tests { .expect("the committed worker candidate should remain installed locally"); assert_eq!(local.id, rebalance_id); assert_eq!(local.pool_stats[0].info.status, RebalStatus::Completed); - if let Some(cancel) = local.cancel.as_ref() { - cancel.cancel(); - } + let admitted_cancel = local + .cancel + .clone() + .expect("the committed worker candidate should install its cancellation token"); + 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"); + assert!(admitted_cancel.is_cancelled()); + let mut persisted = RebalanceMeta::new(); persisted .load(store.pools[0].clone()) diff --git a/crates/ecstore/src/services/rebalance/mod.rs b/crates/ecstore/src/services/rebalance/mod.rs index bc3fdeb3b..960041f74 100644 --- a/crates/ecstore/src/services/rebalance/mod.rs +++ b/crates/ecstore/src/services/rebalance/mod.rs @@ -54,7 +54,7 @@ pub use types::{ use types::{RebalanceBucketConfigs, RebalanceBucketOutcome, RebalanceEntryOutcome}; #[cfg(any(test, feature = "test-util"))] -pub(crate) async fn test_store_with_persisted_rebalance_meta( +pub 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()); diff --git a/crates/ecstore/src/services/rebalance/runtime.rs b/crates/ecstore/src/services/rebalance/runtime.rs index ed8da595f..62d06799f 100644 --- a/crates/ecstore/src/services/rebalance/runtime.rs +++ b/crates/ecstore/src/services/rebalance/runtime.rs @@ -272,6 +272,11 @@ impl ECStore { return Ok(()); } + #[cfg(test)] + let endpoints = self.instance_endpoints().unwrap_or_else(|| self.endpoints()); + #[cfg(not(test))] + let endpoints = self.endpoints(); + let mut workers_started = 0usize; for (idx, participating) in participants.iter().enumerate() { if !*participating { @@ -287,7 +292,7 @@ impl ECStore { continue; } - if !runtime_sources::endpoint_pool_is_local(idx) { + if !runtime_sources::endpoint_pool_is_local(&endpoints, idx) { debug!( event = EVENT_REBALANCE_STATE, component = LOG_COMPONENT_ECSTORE, diff --git a/rustfs/src/admin/handlers/rebalance.rs b/rustfs/src/admin/handlers/rebalance.rs index f0edb8fec..63d8ce87f 100644 --- a/rustfs/src/admin/handlers/rebalance.rs +++ b/rustfs/src/admin/handlers/rebalance.rs @@ -872,6 +872,13 @@ impl Operation for RebalanceStatus { } } +async fn rebalance_stop_target_id(store: &Arc) -> S3Result> { + store + .prepare_rebalance_stop() + .await + .map_err(|e| s3_error!(InternalError, "failed to prepare rebalance metadata for stop: {}", e)) +} + async fn stop_rebalance_admission_first( store: &Arc, notification_sys: Option<&NotificationSys>, @@ -947,14 +954,9 @@ impl Operation for RebalanceStop { return Err(s3_error!(InternalError, "object layer is not initialized")); }; - let expected_rebalance_id = store.current_rebalance_id().await; - - if !store.is_rebalance_conflicting_with_decommission().await { + let Some(expected_rebalance_id) = rebalance_stop_target_id(&store).await? else { log_rebalance_request_rejected("stop", "rebalance_not_started", &request_id, &actor, &remote_addr); return Err(s3_error!(NoSuchResource, "pool rebalance is not started")); - } - let Some(expected_rebalance_id) = expected_rebalance_id else { - return Err(s3_error!(InternalError, "active rebalance metadata has no activation id")); }; let notification_sys = current_notification_system(); @@ -1096,8 +1098,8 @@ mod rebalance_handler_tests { RebalPoolProgress, RebalanceAdminStatus, RebalancePoolStatus, RebalanceStartStep, RebalanceStopPropagationStatus, build_rebalance_admin_status, build_rebalance_pool_statuses, build_rebalance_stop_propagation_status, rebalance_pool_used, rebalance_query_present, rebalance_remaining_buckets, rebalance_rollback_failure_message, - rebalance_rollback_stop_failure_message, rebalance_start_rollback_error, rebalance_start_steps, rebalance_used_pct, - rollback_result_label, stop_rebalance_admission_first, + rebalance_rollback_stop_failure_message, rebalance_start_rollback_error, rebalance_start_steps, rebalance_stop_target_id, + rebalance_used_pct, rollback_result_label, stop_rebalance_admission_first, }; use crate::admin::storage_api::rebalance::{ DiskStat, RebalStatus, RebalanceCleanupWarningEntry, RebalanceCleanupWarnings, RebalanceInfo, RebalanceMeta, @@ -1105,6 +1107,22 @@ mod rebalance_handler_tests { }; use time::OffsetDateTime; + fn started_rebalance_meta(id: &str) -> RebalanceMeta { + RebalanceMeta { + id: id.to_string(), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + start_time: Some(OffsetDateTime::now_utc()), + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + } + } + #[tokio::test] #[serial_test::serial] async fn real_admin_stop_cancels_paused_entry_before_waiting_for_activation_gate() { @@ -1113,7 +1131,13 @@ mod rebalance_handler_tests { fixture.wait_until_entry_paused().await; let stop_store = fixture.store(); - let mut stop_task = tokio::spawn(async move { stop_rebalance_admission_first(&stop_store, None, REBALANCE_ID).await }); + let mut stop_task = tokio::spawn(async move { + let expected_rebalance_id = rebalance_stop_target_id(&stop_store) + .await + .expect("admin stop target resolution should succeed") + .expect("the active rebalance should remain stoppable"); + stop_rebalance_admission_first(&stop_store, None, expected_rebalance_id.as_str()).await + }); fixture.wait_until_admission_cancelled().await; fixture.wait_until_stop_waiting_for_entry().await; assert!(!stop_task.is_finished(), "admin stop must wait for the paused entry guard to drain"); @@ -1129,6 +1153,75 @@ mod rebalance_handler_tests { assert!(!fixture.store().is_rebalance_conflicting_with_decommission().await); } + #[tokio::test] + #[serial_test::serial] + async fn real_admin_stop_loads_persisted_active_rebalance_from_cold_memory() { + const REBALANCE_ID: &str = "admin-stop-cold-memory"; + let (_temp_dirs, store) = rustfs_ecstore::api::rebalance::test_util::test_store_with_persisted_rebalance_meta( + started_rebalance_meta(REBALANCE_ID), + ) + .await; + *store.rebalance_meta.write().await = None; + assert!(store.current_rebalance_id().await.is_none()); + + let expected_rebalance_id = rebalance_stop_target_id(&store) + .await + .expect("admin stop should load persisted rebalance metadata") + .expect("persisted active rebalance should be stoppable"); + assert_eq!(expected_rebalance_id, REBALANCE_ID); + + let stop_failures = stop_rebalance_admission_first(&store, None, expected_rebalance_id.as_str()) + .await + .expect("admin stop should persist the terminal state after a cold load"); + assert!(stop_failures.is_empty()); + assert!(!store.is_rebalance_conflicting_with_decommission().await); + + *store.rebalance_meta.write().await = None; + store + .load_rebalance_meta() + .await + .expect("the persisted terminal rebalance metadata should remain readable"); + assert_eq!(store.current_rebalance_id().await.as_deref(), Some(REBALANCE_ID)); + assert!(!store.is_rebalance_conflicting_with_decommission().await); + } + + #[tokio::test] + #[serial_test::serial] + async fn real_admin_stop_refreshes_persisted_active_over_stale_inactive_memory() { + const PERSISTED_REBALANCE_ID: &str = "admin-stop-persisted-active"; + const STALE_REBALANCE_ID: &str = "admin-stop-stale-terminal"; + let (_temp_dirs, store) = rustfs_ecstore::api::rebalance::test_util::test_store_with_persisted_rebalance_meta( + started_rebalance_meta(PERSISTED_REBALANCE_ID), + ) + .await; + *store.rebalance_meta.write().await = Some(RebalanceMeta { + id: STALE_REBALANCE_ID.to_string(), + stopped_at: Some(OffsetDateTime::now_utc()), + ..Default::default() + }); + assert_eq!(store.current_rebalance_id().await.as_deref(), Some(STALE_REBALANCE_ID)); + assert!(!store.is_rebalance_conflicting_with_decommission().await); + + let expected_rebalance_id = rebalance_stop_target_id(&store) + .await + .expect("admin stop should refresh stale inactive local metadata") + .expect("persisted active rebalance should replace the stale local terminal state"); + assert_eq!(expected_rebalance_id, PERSISTED_REBALANCE_ID); + + let stop_failures = stop_rebalance_admission_first(&store, None, expected_rebalance_id.as_str()) + .await + .expect("admin stop should persist the refreshed run's terminal state"); + assert!(stop_failures.is_empty()); + + *store.rebalance_meta.write().await = None; + store + .load_rebalance_meta() + .await + .expect("the refreshed run's persisted terminal metadata should remain readable"); + assert_eq!(store.current_rebalance_id().await.as_deref(), Some(PERSISTED_REBALANCE_ID)); + assert!(!store.is_rebalance_conflicting_with_decommission().await); + } + #[test] fn test_calculate_rebalance_progress_running() { let start = OffsetDateTime::from_unix_timestamp(1_000).unwrap();