From eeab9d201bd545eef825f8744bf1555bb1bc1f7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E7=99=BB=E5=B1=B1?= Date: Sat, 22 Aug 2026 00:25:32 +0800 Subject: [PATCH] fix(heal): fence format repair during pool transitions --- crates/ecstore/src/core/pools.rs | 2 +- crates/ecstore/src/store/heal.rs | 332 +++++++++++++++++- crates/heal/src/heal/task/heal_erasure_set.rs | 11 +- crates/heal/src/heal/task/tests.rs | 28 ++ crates/test-utils/src/lib.rs | 12 + 5 files changed, 380 insertions(+), 5 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index fb983dc0c..674efa496 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -1675,7 +1675,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, { diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index d10abe740..a1c76d177 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -13,7 +13,12 @@ // limitations under the License. use super::*; +use crate::core::pools::POOL_META_NAME; +use crate::services::rebalance::{REBAL_META_NAME, RebalStatus}; +use crate::set_disk::get_lock_acquire_timeout; use crate::storage_api_contracts::heal::HealOperations as _; +use crate::storage_api_contracts::namespace::NamespaceLocking as _; +use rustfs_lock::NamespaceLockGuard; use tracing::trace; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; @@ -30,7 +35,119 @@ fn invalid_heal_pool_index(pool_idx: usize, pool_count: usize) -> Error { ) } +#[derive(Debug, Clone, Copy)] +enum HealFormatPoolSkip { + Completed, + Retryable, +} + +fn classify_heal_format_pool( + pool_idx: usize, + pool_cmd_line: &str, + pool_meta: &PoolMeta, + rebalance_meta: Option<&RebalanceMeta>, +) -> Option { + let Some(pool) = pool_meta.pools.get(pool_idx) else { + return Some(HealFormatPoolSkip::Retryable); + }; + + if pool.id != pool_idx || pool_cmd_line.is_empty() || pool.cmd_line.is_empty() || pool.cmd_line != pool_cmd_line { + return Some(HealFormatPoolSkip::Retryable); + } + + if let Some(decommission) = pool.decommission.as_ref() { + if decommission.complete { + return Some(HealFormatPoolSkip::Completed); + } + if decommission.failed || decommission.canceled || decommission.queued || pool_meta.is_suspended(pool_idx) { + return Some(HealFormatPoolSkip::Retryable); + } + } + + if let Some(meta) = rebalance_meta { + let Some(pool_stats) = meta.pool_stats.get(pool_idx) else { + return Some(HealFormatPoolSkip::Retryable); + }; + if pool_stats.info.stopping || (pool_stats.participating && pool_stats.info.status == RebalStatus::Started) { + return Some(HealFormatPoolSkip::Retryable); + } + } + + None +} + +fn heal_format_pool_skip_error(skip: HealFormatPoolSkip) -> Error { + match skip { + HealFormatPoolSkip::Completed => StorageError::NoHealRequired, + HealFormatPoolSkip::Retryable => StorageError::SlowDown, + } +} + +fn heal_format_fence_lost_error() -> Error { + StorageError::SlowDown +} + impl ECStore { + async fn acquire_heal_format_fence( + &self, + ) -> Result<(NamespaceLockGuard, NamespaceLockGuard, PoolMeta, Option)> { + let metadata_pool = self + .pools + .first() + .cloned() + .ok_or_else(|| Error::other("heal format requires at least one storage pool"))?; + + // Metadata fence order is part of the decommission/rebalance protocol: + // pool.bin must always be acquired before rebalance.bin. + let pool_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?; + let pool_guard = pool_lock.get_write_lock(get_lock_acquire_timeout()).await?; + let rebalance_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?; + let rebalance_guard = rebalance_lock.get_write_lock(get_lock_acquire_timeout()).await?; + + if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() { + return Err(heal_format_fence_lost_error()); + } + + let mut pool_meta = PoolMeta::default(); + pool_meta.load_no_lock(metadata_pool.clone()).await?; + if pool_meta.pools.len() != self.pools.len() + || pool_meta.pools.iter().enumerate().any(|(pool_idx, pool)| { + pool.id != pool_idx || pool.cmd_line.is_empty() || pool.cmd_line != self.pools[pool_idx].endpoints.cmd_line + }) + { + return Err(heal_format_fence_lost_error()); + } + + let mut rebalance_meta = RebalanceMeta::new(); + let rebalance_meta = match rebalance_meta + .load_with_opts( + metadata_pool, + ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + { + Ok(()) => Some(rebalance_meta), + Err(Error::ConfigNotFound) => None, + Err(err) => return Err(err), + }; + + if rebalance_meta + .as_ref() + .is_some_and(|meta| meta.pool_stats.len() != self.pools.len()) + { + return Err(heal_format_fence_lost_error()); + } + + if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() { + return Err(heal_format_fence_lost_error()); + } + + Ok((pool_guard, rebalance_guard, pool_meta, rebalance_meta)) + } + fn get_pools_for_heal_object(&self, opts: &HealOpts) -> Result>> { match opts.pool { Some(pool_idx) => Ok(vec![ @@ -52,8 +169,24 @@ impl ECStore { }; let mut count_no_heal = 0; + let mut count_completed = 0; let mut first_error = None; - for pool in self.pools.iter() { + for (pool_idx, pool) in self.pools.iter().enumerate() { + let (pool_guard, rebalance_guard, pool_meta, rebalance_meta) = self.acquire_heal_format_fence().await?; + if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() { + first_error.get_or_insert(heal_format_fence_lost_error()); + break; + } + if let Some(skip) = classify_heal_format_pool(pool_idx, &pool.endpoints.cmd_line, &pool_meta, rebalance_meta.as_ref()) + { + if matches!(skip, HealFormatPoolSkip::Completed) { + count_completed += 1; + } else { + first_error.get_or_insert(heal_format_pool_skip_error(skip)); + } + continue; + } + let (mut result, err) = pool.heal_format(dry_run).await?; if let Some(err) = err { match err { @@ -69,11 +202,18 @@ impl ECStore { r.set_count += result.set_count; r.before.drives.append(&mut result.before.drives); r.after.drives.append(&mut result.after.drives); + + // Sets::heal_format cannot observe this guard before each disk write; + // fail closed after the call if the lease was lost during format IO. + if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() { + first_error.get_or_insert(heal_format_fence_lost_error()); + break; + } } if let Some(err) = first_error { return Ok((r, Some(err))); } - if count_no_heal == self.pools.len() { + if count_no_heal + count_completed == self.pools.len() { info!( event = EVENT_HEAL_FORMAT_COMPLETED, component = LOG_COMPONENT_ECSTORE, @@ -300,6 +440,7 @@ mod tests { use crate::core::pools::{PoolDecommissionInfo, PoolStatus}; use crate::disk::{DiskOption, format::FormatV3, new_disk}; use crate::layout::endpoints::{Endpoints, PoolEndpoints}; + use crate::services::rebalance::{RebalanceInfo, RebalanceStats}; use crate::store::init_format::{load_format_erasure, save_format_file}; async fn minimal_heal_pool(pool_idx: usize) -> Arc { @@ -347,6 +488,164 @@ mod tests { } } + fn pool_meta_with_decommission(info: PoolDecommissionInfo) -> PoolMeta { + PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(info), + }], + ..Default::default() + } + } + + #[test] + fn heal_format_pool_state_barriers_are_classified() { + let active = pool_meta_with_decommission(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }); + assert!(matches!( + classify_heal_format_pool(0, "pool-0", &active, None), + Some(HealFormatPoolSkip::Retryable) + )); + + for info in [ + PoolDecommissionInfo { + failed: true, + ..Default::default() + }, + PoolDecommissionInfo { + canceled: true, + ..Default::default() + }, + ] { + assert!(matches!( + classify_heal_format_pool(0, "pool-0", &pool_meta_with_decommission(info), None), + Some(HealFormatPoolSkip::Retryable) + )); + } + + let completed = pool_meta_with_decommission(PoolDecommissionInfo { + complete: true, + ..Default::default() + }); + assert!(matches!( + classify_heal_format_pool(0, "pool-0", &completed, None), + Some(HealFormatPoolSkip::Completed) + )); + } + + #[test] + fn heal_format_pool_rebalance_barriers_and_identity_are_fail_closed() { + let identity_meta = pool_meta_with_decommission(PoolDecommissionInfo::default()); + let rebalance = RebalanceMeta { + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + assert!(matches!( + classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&rebalance)), + Some(HealFormatPoolSkip::Retryable) + )); + + let stopping = RebalanceMeta { + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + stopping: true, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + assert!(matches!( + classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping)), + Some(HealFormatPoolSkip::Retryable) + )); + + let identity = pool_meta_with_decommission(PoolDecommissionInfo::default()); + assert!(matches!( + classify_heal_format_pool(0, "pool-new", &identity, None), + Some(HealFormatPoolSkip::Retryable) + )); + + let identity_without_decommission = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: None, + }], + ..Default::default() + }; + assert!(matches!( + classify_heal_format_pool(0, "pool-new", &identity_without_decommission, None), + Some(HealFormatPoolSkip::Retryable) + )); + + assert!(matches!( + classify_heal_format_pool(0, "", &identity_meta, None), + Some(HealFormatPoolSkip::Retryable) + )); + + assert!(matches!( + classify_heal_format_pool(0, "pool-0", &PoolMeta::default(), None), + Some(HealFormatPoolSkip::Retryable) + )); + + let stopped = RebalanceMeta { + stopped_at: Some(OffsetDateTime::UNIX_EPOCH), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Stopped, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + assert!(classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopped)).is_none()); + + let stopping_after_stop = RebalanceMeta { + stopped_at: Some(OffsetDateTime::UNIX_EPOCH), + pool_stats: vec![RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + stopping: true, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + assert!(matches!( + classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping_after_stop)), + Some(HealFormatPoolSkip::Retryable) + )); + } + + #[test] + fn skipped_heal_format_pool_is_never_reported_as_success() { + assert!(matches!( + heal_format_pool_skip_error(HealFormatPoolSkip::Retryable), + StorageError::SlowDown + )); + assert!(matches!( + heal_format_pool_skip_error(HealFormatPoolSkip::Completed), + StorageError::NoHealRequired + )); + } + #[tokio::test] async fn heal_object_pool_scope_selects_only_requested_pool() { let store = minimal_heal_store().await; @@ -615,6 +914,18 @@ mod tests { bucket_fence_registry: std::sync::Arc::default(), }; + let err = store + .handle_heal_format(false) + .await + .expect_err("missing pool metadata must fail closed before format writes"); + assert!(matches!(err, StorageError::SlowDown)); + + let pool_meta = PoolMeta::new(&store.pools, &PoolMeta::default()); + pool_meta + .save(store.pools.clone()) + .await + .expect("pool metadata should be persisted before format heal"); + let (result, err) = store .handle_heal_format(false) .await @@ -628,5 +939,22 @@ mod tests { .await .expect("the later pool should be healed despite the first pool error"); assert_eq!(healed.erasure.this, recoverable_format.erasure.sets[0][2]); + + let mut completed_meta = PoolMeta::new(&store.pools, &PoolMeta::default()); + for status in &mut completed_meta.pools { + status.decommission = Some(PoolDecommissionInfo { + complete: true, + ..Default::default() + }); + } + completed_meta + .save(store.pools.clone()) + .await + .expect("completed pool metadata should be persisted"); + let (_, err) = store + .handle_heal_format(false) + .await + .expect("completed pools should be reported as a no-op"); + assert!(matches!(err, Some(StorageError::NoHealRequired))); } } diff --git a/crates/heal/src/heal/task/heal_erasure_set.rs b/crates/heal/src/heal/task/heal_erasure_set.rs index e0cfe5a90..7b25bcba2 100644 --- a/crates/heal/src/heal/task/heal_erasure_set.rs +++ b/crates/heal/src/heal/task/heal_erasure_set.rs @@ -231,6 +231,10 @@ impl HealTask { "Heal erasure set format repair skipped because no format heal was required" ); } else { + let error = e; + if error.is_recoverable_heal() { + return Err(error); + } error!( target: "rustfs::heal::task", event = EVENT_HEAL_ERASURE_SET_RESULT, @@ -239,7 +243,7 @@ impl HealTask { task_id = %self.id, set_disk_id, result = "format_failed", - error = %e, + error = %error, "Heal erasure set failed" ); { @@ -247,7 +251,7 @@ impl HealTask { progress.update_progress(4, 4, 0, 0); } return Err(Error::TaskExecutionFailed { - message: format!("Failed to heal disk format for {set_disk_id}: {e}"), + message: format!("Failed to heal disk format for {set_disk_id}: {error}"), }); } } else { @@ -284,6 +288,9 @@ impl HealTask { Err(Error::TaskCancelled) => return Err(Error::TaskCancelled), Err(Error::TaskTimeout) => return Err(Error::TaskTimeout), Err(e) => { + if e.is_recoverable_heal() { + return Err(e); + } error!( target: "rustfs::heal::task", event = EVENT_HEAL_ERASURE_SET_RESULT, diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 464ff3606..f2b442205 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -547,6 +547,7 @@ struct MockStorage { heal_object_outcome: Mutex>, heal_object_outcomes: Mutex>>, format_no_heal_required: Mutex, + format_error: Mutex>, global_format_calls: Mutex, replacement_format_calls: Mutex)>>, replacement_targets_ready: Mutex, @@ -867,6 +868,9 @@ impl HealStorageAPI for MockStorage { async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option)> { *self.global_format_calls.lock().unwrap() += 1; + if let Some(error) = self.format_error.lock().unwrap().take() { + return Err(error); + } let no_heal_required = *self.format_no_heal_required.lock().unwrap(); if no_heal_required { Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::NoHealRequired)))) @@ -2052,6 +2056,30 @@ async fn test_erasure_set_heal_continues_after_format_no_heal_required() { ); } +#[tokio::test] +async fn erasure_set_format_slowdown_is_propagated() { + let storage = Arc::new(MockStorage { + format_error: Mutex::new(Some(Error::Storage(EcstoreError::SlowDown))), + ..Default::default() + }); + let request = HealRequest::new( + HealType::ErasureSet { + buckets: Vec::new(), + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions::default(), + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage); + + let error = task + .execute() + .await + .expect_err("format SlowDown must remain recoverable for the task manager"); + + assert!(matches!(error, Error::Storage(EcstoreError::SlowDown))); +} + #[tokio::test] async fn erasure_set_bucket_prepass_failure_stops_before_object_heal() { let temp = TempDir::new().expect("temporary directory should be created"); diff --git a/crates/test-utils/src/lib.rs b/crates/test-utils/src/lib.rs index 59b4f3e0c..0188c7064 100644 --- a/crates/test-utils/src/lib.rs +++ b/crates/test-utils/src/lib.rs @@ -245,6 +245,18 @@ impl TestECStoreEnvBuilder { .await .expect("build test ECStore"); + // The production bootstrap only persists pool.bin from the elected + // first cluster node. Test stores intentionally have no cluster + // election, but heal-format still requires that durable fence before + // it can write any disk format. Materialize the validated topology + // here so the shared fixture models a ready single-node store. + let mut pool_meta = ecstore.pool_meta.read().await.clone(); + pool_meta.dont_save = false; + pool_meta + .save(ecstore.pools.clone()) + .await + .expect("persist test pool metadata"); + if self.init_bucket_metadata { let buckets_list = ecstore .list_bucket(&BucketOptions {