From d9080ae77fa1cf5a4ad2e11179e49ed34b0d4684 Mon Sep 17 00:00:00 2001 From: hector <42570491+majinghe@users.noreply.github.com> Date: Thu, 27 Aug 2026 16:18:39 +0800 Subject: [PATCH] test(pool): cover rebalance retry and cold-start recovery (#6720) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(pool): fix warp log path and retry rebalance start - warp writes now use a unique mktemp log file instead of a fixed /tmp/rustfs-warp.log: the runner user could not write the stale root-owned file, which made the background warp process die instantly (warp never ran). The workflow uploads /tmp/rustfs-warp.*.log. - rebalance start is retried (6x, 20s apart): nightly builds gate rebalance activation on a live cross-pool fence fleet capability proof that takes ~10-20s to re-establish after a pool joins. Verified live: attempt 1 fails with 500 'pool activation requires a live fleet capability proof', attempt 2 succeeds. * test(pool): annotate known server-side issues in failure output When a node fails to start, grab the rustfs journal tail and match known server-side error signatures (e.g. the fleet capability proof cold-start regression, rustfs/backlog#2031), printing a hint with the tracking issue. Also annotate the rebalance-start retry exhaustion and the rc.3 decommission metacache-listing failure with actionable guidance. * fix(ecstore): defer rebalance activation without fleet proof --------- Co-authored-by: 马登山 Co-authored-by: cxymds --- .github/actionlint.yaml | 1 + .github/workflows/rustfs-pool-expand-test.yml | 4 +- crates/ecstore/src/core/pools.rs | 32 ++- .../ecstore/src/services/notification_sys.rs | 33 +++ .../ecstore/src/services/rebalance/control.rs | 74 +++++- crates/ecstore/src/services/rebalance/meta.rs | 8 + crates/ecstore/src/services/rebalance/mod.rs | 2 +- .../rebalance/rebalance_unit_tests.rs | 56 ++++- crates/ecstore/src/store/init.rs | 212 ++++++++++++++++-- scripts/test/rustfs_pool_expand.sh | 97 ++++++-- 10 files changed, 473 insertions(+), 46 deletions(-) diff --git a/.github/actionlint.yaml b/.github/actionlint.yaml index a309de9d0..5f07463dc 100644 --- a/.github/actionlint.yaml +++ b/.github/actionlint.yaml @@ -5,3 +5,4 @@ self-hosted-runner: - sm-standard-2 - sm-standard-4 - dind-sm-standard-2 + - smoke-testing diff --git a/.github/workflows/rustfs-pool-expand-test.yml b/.github/workflows/rustfs-pool-expand-test.yml index 98d6cdd15..05262b2ca 100644 --- a/.github/workflows/rustfs-pool-expand-test.yml +++ b/.github/workflows/rustfs-pool-expand-test.yml @@ -91,7 +91,7 @@ jobs: - name: Install RustFS package & start first pool run: | - ARGS=(--steps 1,2,3 -y --endpoint "${{ env.RUSTFS_API_ENDPOINT }}") + ARGS=(--steps "1,2,3" -y --endpoint "${{ env.RUSTFS_API_ENDPOINT }}") if [ -n "${{ inputs.package_url }}" ]; then ARGS+=(--package-url "${{ inputs.package_url }}") elif [ -n "${{ inputs.rustfs_version }}" ]; then @@ -138,7 +138,7 @@ jobs: name: rustfs-pool-test-${{ github.run_id }} path: | /tmp/rustfs-pool-test.log - /tmp/rustfs-warp.log + /tmp/rustfs-warp.*.log if-no-files-found: warn - name: Reset test environment (after) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 365448ec8..f25b42f82 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -96,6 +96,8 @@ const LOG_SUBSYSTEM_POOLS: &str = "pools"; const EVENT_DECOMMISSION_STATE: &str = "decommission_state"; const EVENT_DECOMMISSION_BUCKET: &str = "decommission_bucket"; const EVENT_DECOMMISSION_ENTRY: &str = "decommission_entry"; +const POOL_ACTIVATION_FLEET_PROOF_REQUIRED: &str = "pool activation requires a live fleet capability proof"; +const POOL_ACTIVATION_FLEET_PROOF_EXPIRED: &str = "pool activation fleet capability proof expired before commit"; const DECOMMISSION_STAGE_MIGRATE_OBJECT: &str = "migrate_object"; const DECOMMISSION_STAGE_CLEANUP_PREFLIGHT: &str = "cleanup_preflight"; const DECOMMISSION_STAGE_SOURCE_CLEANUP: &str = "source_cleanup"; @@ -1832,6 +1834,13 @@ pub(crate) struct PoolRebalanceActivationFence { } impl PoolRebalanceActivationFence { + pub(crate) fn set_fleet_proof( + &mut self, + fleet_proof: Option, + ) { + self.fleet_proof = fleet_proof; + } + pub(crate) fn ensure_held(&self) -> Result<()> { #[cfg(test)] let forced_lost = self.forced_lost.load(Ordering::Acquire); @@ -1845,7 +1854,7 @@ impl PoolRebalanceActivationFence { .as_ref() .is_some_and(|proof| !crate::services::notification_sys::cross_pool_fence_fleet_proof_matches(proof)) { - return Err(Error::other("pool activation fleet capability proof expired before commit")); + return Err(Error::other(POOL_ACTIVATION_FLEET_PROOF_EXPIRED)); } Ok(()) @@ -1901,7 +1910,17 @@ pub(crate) async fn acquire_pool_activation_fleet_proof( } crate::services::notification_sys::acquire_cross_pool_fence_fleet_proof() .map(Some) - .ok_or_else(|| Error::other("pool activation requires a live fleet capability proof")) + .ok_or_else(|| Error::other(POOL_ACTIVATION_FLEET_PROOF_REQUIRED)) +} + +pub(crate) fn is_pool_activation_fleet_proof_error(err: &Error) -> bool { + // Save-stage helpers add context by formatting the original error, so the + // marker may be nested in the display string. Restrict matching to the + // `Error::other` I/O shape used by this activation path. + matches!(err, Error::Io(io_error) if io_error.kind() == std::io::ErrorKind::Other && { + let message = io_error.to_string(); + message.contains(POOL_ACTIVATION_FLEET_PROOF_REQUIRED) || message.contains(POOL_ACTIVATION_FLEET_PROOF_EXPIRED) + }) } #[cfg(test)] @@ -10828,6 +10847,15 @@ mod tests { use crate::bucket::replication::{ReplicationState, ReplicationStatusType}; use serde::Serialize; + #[test] + fn pool_activation_fleet_proof_error_classifier_matches_only_retryable_proof_failures() { + assert!(is_pool_activation_fleet_proof_error(&Error::other(POOL_ACTIVATION_FLEET_PROOF_REQUIRED))); + assert!(is_pool_activation_fleet_proof_error(&Error::other(POOL_ACTIVATION_FLEET_PROOF_EXPIRED))); + let wrapped = format!("rebalance meta save failed during start_rebalance: {POOL_ACTIVATION_FLEET_PROOF_EXPIRED}"); + assert!(is_pool_activation_fleet_proof_error(&Error::other(wrapped))); + assert!(!is_pool_activation_fleet_proof_error(&Error::ConfigNotFound)); + } + #[tokio::test] #[serial_test::serial] async fn decommission_activation_fence_loss_after_durable_save_blocks_publication() { diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index 8c36304a1..0c78007c2 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -234,6 +234,39 @@ pub(crate) fn install_cross_pool_fence_fleet_proof_for_test() { }); } +#[cfg(test)] +pub(crate) struct CrossPoolFenceFleetProofGuard { + previous_proof: Option, + previous_topology_conflict: bool, +} + +#[cfg(test)] +impl Drop for CrossPoolFenceFleetProofGuard { + fn drop(&mut self) { + let mut state = cross_pool_fence_fleet_proof_slot() + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.proof = self.previous_proof.take(); + state.topology_conflict = self.previous_topology_conflict; + } +} + +/// Temporarily revoke the test proof so activation paths can exercise their +/// fail-closed behavior without changing the process-wide topology binding. +#[cfg(test)] +pub(crate) fn without_cross_pool_fence_fleet_proof_for_test() -> CrossPoolFenceFleetProofGuard { + let mut state = cross_pool_fence_fleet_proof_slot() + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let guard = CrossPoolFenceFleetProofGuard { + previous_proof: state.proof.clone(), + previous_topology_conflict: state.topology_conflict, + }; + state.proof = None; + state.topology_conflict = true; + guard +} + #[cfg(any(test, feature = "test-util"))] pub fn rotate_cross_pool_fence_fleet_proof_for_test() -> bool { let mut state = cross_pool_fence_fleet_proof_slot() diff --git a/crates/ecstore/src/services/rebalance/control.rs b/crates/ecstore/src/services/rebalance/control.rs index a4a65ae86..aeb382803 100644 --- a/crates/ecstore/src/services/rebalance/control.rs +++ b/crates/ecstore/src/services/rebalance/control.rs @@ -570,10 +570,13 @@ impl ECStore { where S: EcstoreObjectIO + StorageNamespaceLocking, { - let fleet_proof = acquire_pool_activation_fleet_proof(&self.ctx).await?; + // Lock order: pool_meta_save_gate -> pool.bin -> rebalance.bin. let mut pool_meta_guard = self.pool_meta_save_gate.lock().await; pool_meta_guard.ensure_write_safe("rebalance worker activation")?; - let activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), fleet_proof).await?; + // Classify the durable rebalance record while holding both namespace + // fences. A terminal record is a no-op and must not depend on the + // notification subsystem having published a fleet proof yet. + let mut activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), None).await?; let pool_meta = self .load_runtime_pool_meta_under_activation_fence(&mut pool_meta_guard, &activation_fence, "rebalance worker activation") .await?; @@ -597,10 +600,17 @@ impl ECStore { } activation_fence.ensure_held()?; - if !is_rebalance_conflicting_with_decommission(&persisted) { + if !crate::services::rebalance::rebalance_requires_worker_activation(&persisted) { return Ok(RebalanceWorkerActivationFence::NotStartedTerminal); } + // Active worker admission still requires the fail-closed fleet proof. + // Attach it immediately before the final fence validation so expiry or + // topology changes are checked again at every later commit boundary. + let fleet_proof = acquire_pool_activation_fleet_proof(&self.ctx).await?; + activation_fence.set_fleet_proof(fleet_proof); + activation_fence.ensure_held()?; + Ok(RebalanceWorkerActivationFence::Ready(Box::new(activation_fence))) } @@ -1476,6 +1486,64 @@ mod tests { assert_activation_locks_released(&store).await; } + #[tokio::test] + #[serial_test::serial] + async fn rebalance_worker_skips_terminal_metadata_without_fleet_proof() { + let rebalance_id = "terminal-metadata-without-proof"; + let completed = RebalanceMeta { + id: rebalance_id.to_string(), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Completed, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(Some(completed)).await; + let _proof_guard = crate::services::notification_sys::without_cross_pool_fence_fleet_proof_for_test(); + + let activation = store + .fence_rebalance_worker_activation(store.pools[0].clone(), rebalance_id) + .await + .expect("terminal metadata should not require a fleet proof"); + assert!(matches!(activation, RebalanceWorkerActivationFence::NotStartedTerminal)); + } + + #[tokio::test] + #[serial_test::serial] + async fn rebalance_worker_still_requires_fleet_proof_for_active_metadata() { + let rebalance_id = "active-metadata-without-proof"; + let active = RebalanceMeta { + id: rebalance_id.to_string(), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(Some(active)).await; + let _proof_guard = crate::services::notification_sys::without_cross_pool_fence_fleet_proof_for_test(); + + let err = match store + .fence_rebalance_worker_activation(store.pools[0].clone(), rebalance_id) + .await + { + Ok(_) => panic!("active metadata must not be admitted without a fleet proof"), + Err(err) => err, + }; + assert!( + err.to_string() + .contains("pool activation requires a live fleet capability proof") + ); + } + #[tokio::test] #[serial_test::serial] async fn rebalance_activation_adopts_commit_after_post_save_fence_loss() { diff --git a/crates/ecstore/src/services/rebalance/meta.rs b/crates/ecstore/src/services/rebalance/meta.rs index 0d21e7e8c..678be730a 100644 --- a/crates/ecstore/src/services/rebalance/meta.rs +++ b/crates/ecstore/src/services/rebalance/meta.rs @@ -214,6 +214,14 @@ pub(super) fn is_rebalance_in_progress(meta: &RebalanceMeta) -> bool { meta.pool_stats.iter().any(is_rebalance_pool_active) } +/// Persisted rebalance metadata requires worker activation only while it has +/// not reached a durable terminal marker and at least one pool is still marked +/// active. Merely finding `rebalance.bin` is not evidence that admission is +/// required: terminal metadata is retained for status reporting. +pub(crate) fn rebalance_requires_worker_activation(meta: &RebalanceMeta) -> bool { + meta.stopped_at.is_none() && is_rebalance_in_progress(meta) +} + pub(crate) fn is_rebalance_conflicting_with_decommission(meta: &RebalanceMeta) -> bool { is_rebalance_in_progress(meta) } diff --git a/crates/ecstore/src/services/rebalance/mod.rs b/crates/ecstore/src/services/rebalance/mod.rs index 4701b5c7f..d7ffeb7cd 100644 --- a/crates/ecstore/src/services/rebalance/mod.rs +++ b/crates/ecstore/src/services/rebalance/mod.rs @@ -49,8 +49,8 @@ mod worker; #[cfg(feature = "test-util")] pub use entry::test_util::PausedRebalanceEntryTestFixture; -pub(crate) use meta::is_rebalance_conflicting_with_decommission; pub use meta::{decode_rebalance_stop_propagation_record, encode_rebalance_stop_propagation_record}; +pub(crate) use meta::{is_rebalance_conflicting_with_decommission, rebalance_requires_worker_activation}; pub use types::{ DiskStat, RebalSaveOpt, RebalStatus, RebalanceCleanupWarningEntry, RebalanceCleanupWarnings, RebalanceInfo, RebalanceMeta, RebalanceStats, RebalanceStopPropagationRecord, diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index e0e50f4f8..a544ed733 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs @@ -22,11 +22,11 @@ use super::meta::{ is_rebalance_in_progress, is_rebalance_meta_replaceable_for_new_id, is_rebalance_stopped_terminal_event, mark_rebalance_bucket_done, merge_rebalance_bucket_lists, merge_rebalance_meta, next_rebal_bucket_from_stat, percent_free_ratio, rebalance_goal_reached, rebalance_meta_load_no_data_error, rebalance_meta_load_unknown_format_error, - rebalance_meta_load_unknown_version_error, record_rebalance_cleanup_warning_in_meta, remove_rebalanced_buckets_from_queue, - resolve_next_rebalance_bucket, resolve_rebalance_participants, should_accept_rebalance_stats_update, - should_ignore_rebalance_data_usage_cache, should_pool_participate, should_preserve_rebalance_stopped_state, - should_skip_start_rebalance, stop_rebalance_meta_snapshot, stop_rebalance_state, take_bucket_from_rebalance_queue, - validate_init_rebalance_state, validate_start_rebalance_state, + rebalance_meta_load_unknown_version_error, rebalance_requires_worker_activation, record_rebalance_cleanup_warning_in_meta, + remove_rebalanced_buckets_from_queue, resolve_next_rebalance_bucket, resolve_rebalance_participants, + should_accept_rebalance_stats_update, should_ignore_rebalance_data_usage_cache, should_pool_participate, + should_preserve_rebalance_stopped_state, should_skip_start_rebalance, stop_rebalance_meta_snapshot, stop_rebalance_state, + take_bucket_from_rebalance_queue, validate_init_rebalance_state, validate_start_rebalance_state, }; use super::migration::{ MigrationBackend, MigrationVersionResult, migrate_entry_version, migrate_entry_version_with_retry_wait, @@ -3386,6 +3386,52 @@ fn test_is_rebalance_in_progress_only_started_participants() { assert!(is_rebalance_in_progress(&meta)); } +#[test] +fn test_rebalance_requires_worker_activation_only_for_active_non_stopped_metadata() { + let now = OffsetDateTime::now_utc(); + let active = RebalanceMeta { + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + let stopped_active = RebalanceMeta { + stopped_at: Some(now), + pool_stats: active.pool_stats.clone(), + ..Default::default() + }; + + assert!(rebalance_requires_worker_activation(&active)); + for status in [ + RebalStatus::Completed, + RebalStatus::Stopped, + RebalStatus::Failed, + RebalStatus::None, + ] { + let terminal = RebalanceMeta { + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + assert!( + !rebalance_requires_worker_activation(&terminal), + "terminal status {status:?} must not resume" + ); + } + assert!(!rebalance_requires_worker_activation(&stopped_active)); +} + #[test] fn test_is_rebalance_conflicting_with_decommission_true_when_in_progress() { let now = OffsetDateTime::now_utc(); diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 257dd7bcb..7e0c24a1c 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -99,6 +99,8 @@ fn preflight_startup_rpc_secret_with( const LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES: usize = 6; const LOCAL_DECOMMISSION_INITIAL_RESUME_DELAY: Duration = Duration::from_secs(60 * 3); const LOCAL_DECOMMISSION_RESUME_RETRY_DELAY: Duration = Duration::from_secs(30); +const REBALANCE_INITIAL_RESUME_DELAY: Duration = Duration::from_secs(10); +const REBALANCE_RESUME_RETRY_DELAY: Duration = Duration::from_secs(10); fn should_retry_local_decommission_resume(err: &Error, attempt: usize) -> bool { matches!(err, Error::ConfigNotFound) && attempt < LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES @@ -108,8 +110,12 @@ fn should_retry_format_load(err: &Error) -> bool { !matches!(err, Error::CorruptedFormat) } -fn should_auto_start_rebalance_after_init(decommission_running: bool, rebalance_meta_loaded: bool) -> bool { - rebalance_meta_loaded && !decommission_running +fn should_auto_start_rebalance_after_init(decommission_running: bool, rebalance_resume_required: bool) -> bool { + rebalance_resume_required && !decommission_running +} + +fn should_defer_rebalance_auto_start(distributed: bool, fleet_proof_available: bool) -> bool { + distributed && !fleet_proof_available } fn should_schedule_local_decommission_resume( @@ -127,6 +133,17 @@ async fn wait_for_local_decommission_resume_delay(rx: &CancellationToken, delay: } } +async fn wait_for_rebalance_resume_delay(rx: &CancellationToken, delay: Duration) -> bool { + tokio::select! { + _ = rx.cancelled() => false, + _ = tokio::time::sleep(delay) => true, + } +} + +async fn wait_for_rebalance_resume_retry(rx: &CancellationToken) -> bool { + wait_for_rebalance_resume_delay(rx, REBALANCE_RESUME_RETRY_DELAY).await +} + fn resolve_store_init_stage_result(result: Result<()>, stage: &str) -> Result<()> { result.map_err(|err| Error::other(format!("store init failed during {stage}: {err}"))) } @@ -283,6 +300,71 @@ async fn resume_local_decommission_after_init(store: Arc, rx: Cancellat } } +async fn resume_rebalance_after_init(store: Arc, rx: CancellationToken) { + if !wait_for_rebalance_resume_delay(&rx, REBALANCE_INITIAL_RESUME_DELAY).await { + return; + } + + loop { + if rx.is_cancelled() { + return; + } + + let resume_required = store + .rebalance_meta + .read() + .await + .as_ref() + .is_some_and(crate::services::rebalance::rebalance_requires_worker_activation); + if !resume_required { + return; + } + + if should_defer_rebalance_auto_start( + store.ctx.is_dist_erasure().await, + crate::services::notification_sys::acquire_cross_pool_fence_fleet_proof().is_some(), + ) { + if !wait_for_rebalance_resume_retry(&rx).await { + return; + } + continue; + } + + match store.start_rebalance().await { + Ok(()) => return, + Err(err) if crate::core::pools::is_pool_activation_fleet_proof_error(&err) => { + warn!( + event = EVENT_ECSTORE_INIT_STATUS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + stage = "start_rebalance", + state = "retrying", + reason = "fleet_capability_proof_unavailable", + error = %err, + retry_delay_secs = REBALANCE_RESUME_RETRY_DELAY.as_secs(), + "Retrying deferred rebalance auto-start" + ); + if !wait_for_rebalance_resume_retry(&rx).await { + return; + } + } + Err(err) => { + error!( + event = EVENT_ECSTORE_INIT_STATUS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + stage = "start_rebalance", + state = "failed", + reason = "deferred_resume_failed", + error = %err, + "Failed to resume rebalance after store initialization" + ); + return; + } + } + } +} + impl ECStore { /// Validate topology and process storage-class overrides before any disk is opened. pub fn validate_startup_storage_class(endpoint_pools: &EndpointServerPools) -> Result<()> { @@ -574,12 +656,49 @@ impl ECStore { } resolve_store_init_stage_result(self.load_rebalance_meta().await, "load_rebalance_meta")?; - let rebalance_meta_loaded = self.rebalance_meta.read().await.is_some(); + let rebalance_resume_required = { + let rebalance_meta = self.rebalance_meta.read().await; + rebalance_meta + .as_ref() + .is_some_and(crate::services::rebalance::rebalance_requires_worker_activation) + }; let decommission_running = pool_meta_has_active_decommission(&installed_pool_meta) || self.is_decommission_running().await; - if should_auto_start_rebalance_after_init(decommission_running, rebalance_meta_loaded) { - resolve_store_init_stage_result(self.start_rebalance().await, "start_rebalance")?; - } else if decommission_running && rebalance_meta_loaded { + let distributed = self.ctx.is_dist_erasure().await; + let fleet_proof_available = crate::services::notification_sys::acquire_cross_pool_fence_fleet_proof().is_some(); + let mut rebalance_auto_start_deferred = false; + if should_auto_start_rebalance_after_init(decommission_running, rebalance_resume_required) { + if should_defer_rebalance_auto_start(distributed, fleet_proof_available) { + rebalance_auto_start_deferred = true; + warn!( + event = EVENT_ECSTORE_INIT_STATUS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + stage = "start_rebalance", + state = "deferred", + reason = "fleet_capability_proof_unavailable", + retry_delay_secs = REBALANCE_RESUME_RETRY_DELAY.as_secs(), + "Deferred rebalance auto-start until a live fleet capability proof is available" + ); + } else if let Err(err) = self.start_rebalance().await { + if crate::core::pools::is_pool_activation_fleet_proof_error(&err) { + rebalance_auto_start_deferred = true; + warn!( + event = EVENT_ECSTORE_INIT_STATUS, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + stage = "start_rebalance", + state = "deferred", + reason = "fleet_capability_proof_changed", + error = %err, + retry_delay_secs = REBALANCE_RESUME_RETRY_DELAY.as_secs(), + "Deferred rebalance auto-start after the fleet capability proof changed" + ); + } else { + return resolve_store_init_stage_result(Err(err), "start_rebalance"); + } + } + } else if decommission_running && rebalance_resume_required { warn!( event = EVENT_ECSTORE_INIT_STATUS, component = LOG_COMPONENT_ECSTORE, @@ -616,12 +735,13 @@ impl ECStore { .is_ok(); if should_schedule_local_decommission_resume(&local_pool_indices, pool_meta_replica_state, pool_meta_write_safe) { let store = self.clone(); + let decommission_rx = rx.clone(); tokio::spawn(async move { - if !wait_for_local_decommission_resume_delay(&rx, LOCAL_DECOMMISSION_INITIAL_RESUME_DELAY).await { + if !wait_for_local_decommission_resume_delay(&decommission_rx, LOCAL_DECOMMISSION_INITIAL_RESUME_DELAY).await { return; } - resume_local_decommission_after_init(store, rx, local_pool_indices).await; + resume_local_decommission_after_init(store, decommission_rx, local_pool_indices).await; }); } else if !local_pool_indices.is_empty() { error!( @@ -648,6 +768,11 @@ impl ECStore { info!("TierConfigMgr init error: {}", err); } + if rebalance_auto_start_deferred { + let store = self.clone(); + tokio::spawn(resume_rebalance_after_init(store, rx)); + } + Ok(()) } @@ -665,7 +790,8 @@ mod tests { load_pool_meta_for_startup, persist_pool_meta_for_startup_if_safe, pool_first_endpoint_is_local, pool_meta_has_active_decommission, preflight_startup_rpc_secret_with, resolve_startup_pool_defaults_with, resolve_store_init_stage_result, save_validated_pool_meta_for_startup, should_auto_start_rebalance_after_init, - should_retry_format_load, should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay, + should_defer_rebalance_auto_start, should_retry_format_load, should_retry_local_decommission_resume, + wait_for_local_decommission_resume_delay, }; #[cfg(feature = "test-util")] use crate::disk::DiskAPI; @@ -1450,7 +1576,7 @@ mod tests { } #[test] - fn test_should_auto_start_rebalance_after_init_allows_loaded_rebalance_without_decommission() { + fn test_should_auto_start_rebalance_after_init_allows_active_rebalance_without_decommission() { assert!(should_auto_start_rebalance_after_init(false, true)); } @@ -1460,10 +1586,17 @@ mod tests { } #[test] - fn test_should_auto_start_rebalance_after_init_rejects_missing_rebalance_meta() { + fn test_should_auto_start_rebalance_after_init_rejects_terminal_or_missing_rebalance() { assert!(!should_auto_start_rebalance_after_init(false, false)); } + #[test] + fn test_should_defer_rebalance_auto_start_only_without_distributed_fleet_proof() { + assert!(should_defer_rebalance_auto_start(true, false)); + assert!(!should_defer_rebalance_auto_start(true, true)); + assert!(!should_defer_rebalance_auto_start(false, false)); + } + #[test] fn test_store_init_recovery_skips_rebalance_when_decommission_metadata_is_active() { let pool_meta = init_test_pool_meta(Some(PoolDecommissionInfo { @@ -1473,22 +1606,69 @@ mod tests { canceled: false, ..Default::default() })); - let rebalance_meta = Some(RebalanceMeta::default()); + let rebalance_meta = Some(RebalanceMeta { + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }); assert!(!should_auto_start_rebalance_after_init( pool_meta_has_active_decommission(&pool_meta), - rebalance_meta.is_some() + rebalance_meta + .as_ref() + .is_some_and(crate::services::rebalance::rebalance_requires_worker_activation) )); } #[test] - fn test_store_init_recovery_allows_rebalance_when_only_rebalance_metadata_exists() { + fn test_store_init_recovery_allows_active_rebalance_without_decommission() { let pool_meta = init_test_pool_meta(None); - let rebalance_meta = Some(RebalanceMeta::default()); + let rebalance_meta = Some(RebalanceMeta { + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }); assert!(should_auto_start_rebalance_after_init( pool_meta_has_active_decommission(&pool_meta), - rebalance_meta.is_some() + rebalance_meta + .as_ref() + .is_some_and(crate::services::rebalance::rebalance_requires_worker_activation) + )); + } + + #[test] + fn test_store_init_recovery_skips_completed_rebalance_metadata() { + let pool_meta = init_test_pool_meta(None); + let rebalance_meta = Some(RebalanceMeta { + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Completed, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }); + + assert!(!should_auto_start_rebalance_after_init( + pool_meta_has_active_decommission(&pool_meta), + rebalance_meta + .as_ref() + .is_some_and(crate::services::rebalance::rebalance_requires_worker_activation) )); } diff --git a/scripts/test/rustfs_pool_expand.sh b/scripts/test/rustfs_pool_expand.sh index 49ec8835e..2f5f744d7 100755 --- a/scripts/test/rustfs_pool_expand.sh +++ b/scripts/test/rustfs_pool_expand.sh @@ -93,6 +93,9 @@ WARP_BUCKET="test-10mb" WARP_OBJ_SIZE="100MiB" WARP_CONCURRENT=32 WARP_DURATION="5m" +# Warp log path; empty = auto-created unique temp file (the runner user may +# not be able to write a shared /tmp path owned by another user). +WARP_LOG_FILE="${RUSTFS_WARP_LOG_FILE:-}" STORAGE_THRESHOLD=85 # stop writing when usage reaches N% (note suggests 80-85) POLL_INTERVAL=30 # status polling interval (seconds) @@ -102,6 +105,8 @@ DECOMMISSION_TIMEOUT=86400 SERVICE_TIMEOUT=300 DECOMMISSION_RETRIES=3 # auto clear+retry attempts after a failed decommission DECOMMISSION_RETRY_DELAY=30 # delay between retries (seconds) +REBALANCE_START_RETRIES=6 # rebalance start retries (fleet proof may take ~10-20s after a topology change) +REBALANCE_START_RETRY_DELAY=20 # delay between rebalance start retries (seconds) # Pool to decommission (zero-based; 0 in the note) DECOMMISSION_POOL_ID=0 @@ -325,9 +330,45 @@ wait_service_active() { sleep 5 waited=$((waited + 5)) done + diagnose_node_start_failure "${node}" die "${node}: timed out waiting for ${RUSTFS_SERVICE} (${SERVICE_TIMEOUT}s)" } +# Known server-side issues the test can hit. Format: +# "||" +KNOWN_SERVER_ISSUES=( + "pool activation requires a live fleet capability proof|rustfs/backlog#2031|server-side cold-start recovery is covered by this PR; if this appears, collect node journals and treat it as a regression" +) + +# Print a hint when $1 matches a known server-side issue signature. +hint_server_issue() { + local text="$1" entry sig tracking hint + for entry in "${KNOWN_SERVER_ISSUES[@]}"; do + sig="${entry%%|*}" + tracking="${entry#*|}" + hint="${tracking#*|}" + tracking="${tracking%%|*}" + if printf '%s' "${text}" | grep -qiF "${sig}"; then + printf '\033[1;33m[KNOWN SERVER ISSUE]\033[0m %s (%s): %s\n' "${sig}" "${tracking}" "${hint}" >&2 + return 0 + fi + done + return 1 +} + +# Fetch the journal tail from a node whose service failed to start and +# annotate known server-side issues. +diagnose_node_start_failure() { + local node="$1" journal + if ! journal="$(ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" \ + "SUDO=\"\"; [ \"\$(id -u)\" -ne 0 ] && SUDO=\"sudo -n\"; \${SUDO} journalctl -u ${RUSTFS_SERVICE} --no-pager -n 60 2>/dev/null || true")"; then + journal="unable to collect journal (SSH command failed)" + fi + printf '%s\n' "--- ${node}: ${RUSTFS_SERVICE} journal (last 60 lines) ---" >&2 + printf '%s\n' "${journal}" >&2 + hint_server_issue "${journal}" || true +} + # Generate the /etc/default/rustfs content rustfs_config_body() { local volumes="$1" @@ -406,9 +447,13 @@ service_action() { local action="$1" node="$2" log "${node}: systemctl ${action} ${RUSTFS_SERVICE}" if [ "${DRY_RUN}" -eq 1 ]; then return 0; fi - ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" \ - "if [ \"\$(id -u)\" -ne 0 ]; then sudo -n systemctl ${action} ${RUSTFS_SERVICE}; else systemctl ${action} ${RUSTFS_SERVICE}; fi" \ - || die "${node}: systemctl ${action} failed" + if ! ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" \ + "if [ \"\$(id -u)\" -ne 0 ]; then sudo -n systemctl ${action} ${RUSTFS_SERVICE}; else systemctl ${action} ${RUSTFS_SERVICE}; fi"; then + if [ "${action}" = "start" ]; then + diagnose_node_start_failure "${node}" + fi + die "${node}: systemctl ${action} failed" + fi } service_action_all() { @@ -587,6 +632,29 @@ wait_rebalance() { die "timed out waiting for rebalance (${REBALANCE_TIMEOUT}s)" } +# Start rebalance via the admin API. Nightly builds gate rebalance activation +# on a live cross-pool fence fleet capability proof that is re-established +# shortly after a pool joins, so retry a few times before failing. +start_rebalance_with_retry() { + local attempts="${REBALANCE_START_RETRIES}" delay="${REBALANCE_START_RETRY_DELAY}" attempt=1 body code id + while :; do + body="$(admin_api POST /rustfs/admin/v3/rebalance/start "")" + code="$(admin_api_code)" + if [ "${code}" = "200" ]; then + id="$(printf '%s' "${body}" | jq -r '.id // empty')" + log "rebalance started: id=${id}" + return 0 + fi + warn "rebalance start attempt ${attempt}/${attempts} failed (HTTP ${code}): ${body}" + if [ "${attempt}" -ge "${attempts}" ]; then + hint_server_issue "${body}" || true + die "rebalance start failed after ${attempts} attempts (see last error above)" + fi + attempt=$((attempt + 1)) + sleep "${delay}" + done +} + # Print a detailed decommission failure/progress report for one pool print_decommission_detail() { local body="$1" pool_id="$2" label="$3" @@ -817,16 +885,18 @@ step4_write_data() { log "DRY-RUN: monitoring storage usage until ${STORAGE_THRESHOLD}%" return 0 fi - log "starting warp writes (background)..." + local warp_log warp_pid + warp_log="${WARP_LOG_FILE:-$(mktemp "${TMPDIR:-/tmp}/rustfs-warp.XXXXXX.log")}" + log "starting warp writes (background), log: ${warp_log}" warp put --host "${API_ENDPOINT#http://}" \ --bucket "${WARP_BUCKET}" \ --access-key "${ACCESS_KEY}" \ --secret-key "${SECRET_KEY}" \ --obj.size "${WARP_OBJ_SIZE}" \ --concurrent "${WARP_CONCURRENT}" \ - --noprefix --duration "${WARP_DURATION}" --noclear >/tmp/rustfs-warp.log 2>&1 & - local warp_pid=$! - log "warp PID=${warp_pid}, log /tmp/rustfs-warp.log" + --noprefix --duration "${WARP_DURATION}" --noclear >"${warp_log}" 2>&1 & + warp_pid=$! + log "warp PID=${warp_pid}" trap 'kill "${warp_pid:-}" 2>/dev/null || true' EXIT monitor_storage "${STORAGE_THRESHOLD}" "${warp_pid}" kill "${warp_pid}" 2>/dev/null || true @@ -866,12 +936,8 @@ step5_expand_pool2() { step6_rebalance() { log "step 6: start data rebalance (admin API)" confirm "About to start rebalance (POST ${API_ENDPOINT}/rustfs/admin/v3/rebalance/start). Continue?" - local body id if [ "${DRY_RUN}" -eq 0 ]; then - body="$(admin_api POST /rustfs/admin/v3/rebalance/start "")" - [ "$(admin_api_code)" = "200" ] || die "rebalance start failed (HTTP $(admin_api_code)): ${body}" - id="$(printf '%s' "${body}" | jq -r '.id // empty')" - log "rebalance started: id=${id}" + start_rebalance_with_retry fi wait_rebalance } @@ -894,12 +960,8 @@ step7_expand_pool3() { step8_rebalance() { log "step 8: start data rebalance (admin API)" confirm "About to start rebalance (POST ${API_ENDPOINT}/rustfs/admin/v3/rebalance/start). Continue?" - local body id if [ "${DRY_RUN}" -eq 0 ]; then - body="$(admin_api POST /rustfs/admin/v3/rebalance/start "")" - [ "$(admin_api_code)" = "200" ] || die "rebalance start failed (HTTP $(admin_api_code)): ${body}" - id="$(printf '%s' "${body}" | jq -r '.id // empty')" - log "rebalance started: id=${id}" + start_rebalance_with_retry fi wait_rebalance } @@ -928,6 +990,7 @@ step9_decommission() { break fi if [ "${attempt}" -ge "${DECOMMISSION_RETRIES}" ]; then + warn "if the source bucket has many objects and the tested version is 1.0.0-rc.3, this is the known metacache-listing decommission bug; remove the test bucket (rc rb --force rustfs/${WARP_BUCKET}) or lower --storage-threshold, then re-run step 9" die "pool ${DECOMMISSION_POOL_ID} still failed after ${attempt} attempts; investigate manually (POST ${API_ENDPOINT}/rustfs/admin/v3/pools/clear?by-id=true&pool=${DECOMMISSION_POOL_ID} to reset)" fi warn "attempt ${attempt} failed; clearing metadata and retrying in ${DECOMMISSION_RETRY_DELAY}s"