From cc5487e7de309e58b7108f839f8b0ac7f9b80072 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 8 Sep 2026 16:57:13 +0800 Subject: [PATCH] heal: verify admin recreate pool metadata (#7474) Co-authored-by: zhi22915 --- crates/heal/src/heal/erasure_healer.rs | 91 +++++++++++++++++-- crates/heal/src/heal/task.rs | 2 +- crates/heal/src/heal/task/heal_bucket.rs | 79 ++++++++++++++++ crates/heal/src/heal/task/heal_erasure_set.rs | 5 + crates/heal/src/heal/task/heal_object.rs | 5 - crates/heal/src/heal/task/tests.rs | 88 +++++++++++++++++- 6 files changed, 252 insertions(+), 18 deletions(-) diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index dbd62e422..9cb5071bc 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -113,6 +113,7 @@ pub struct ErasureSetHealer { heal_opts: HealOpts, source: HealRequestSource, target_endpoints: Arc<[String]>, + pool_metadata_target_endpoints: Arc<[String]>, replacement_task_id: Option, replacement_target_identities: Option>, mainline_pacer: Option>, @@ -362,6 +363,7 @@ impl ErasureSetHealer { heal_opts, source, target_endpoints: Vec::new().into(), + pool_metadata_target_endpoints: Vec::new().into(), replacement_task_id: None, replacement_target_identities: None, mainline_pacer: None, @@ -385,6 +387,13 @@ impl ErasureSetHealer { self } + pub(crate) fn with_pool_metadata_targets(mut self, mut target_endpoints: Vec) -> Self { + target_endpoints.sort_unstable(); + target_endpoints.dedup(); + self.pool_metadata_target_endpoints = target_endpoints.into(); + self + } + pub(crate) fn with_replacement_identity_fence( mut self, replacement_target_identities: Option>, @@ -948,10 +957,16 @@ impl ErasureSetHealer { resume_manager: &ResumeManager, checkpoint_manager: &CheckpointManager, ) -> Result<()> { - if self.replacement_task_id.is_none() { + let target_endpoints = if self.pool_metadata_target_endpoints.is_empty() { + self.target_endpoints.as_ref() + } else { + self.pool_metadata_target_endpoints.as_ref() + }; + let target_scoped_recreate = !self.heal_opts.dry_run && self.heal_opts.recreate && !target_endpoints.is_empty(); + if self.replacement_task_id.is_none() && !target_scoped_recreate { return Ok(()); } - if self.target_endpoints.is_empty() { + if target_endpoints.is_empty() { return Err(Error::TaskExecutionFailed { message: "Replacement pool metadata heal requires target endpoints".to_string(), }); @@ -978,17 +993,11 @@ impl ErasureSetHealer { .heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &self.heal_opts) .await { - Ok((result, None)) if target_outcomes_complete(&result, &self.target_endpoints) => { + Ok((result, None)) if target_outcomes_complete(&result, target_endpoints) => { let object_size = result_object_size_u64(&result); match self .storage - .replacement_targets_have_version( - RUSTFS_META_BUCKET, - POOL_META_NAME, - None, - &self.heal_opts, - &self.target_endpoints, - ) + .replacement_targets_have_version(RUSTFS_META_BUCKET, POOL_META_NAME, None, &self.heal_opts, target_endpoints) .await { Ok(true) => (object_size, Ok(())), @@ -2762,6 +2771,68 @@ mod resume_loop_tests { assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]); } + #[tokio::test] + async fn admin_recreate_target_heals_pool_metadata_before_completion() { + let env = make_env_with_targets(vec!["replacement-a".to_string()]).await; + let healer = ErasureSetHealer::new( + env.storage.clone(), + Arc::new(RwLock::new(HealProgress::new())), + CancellationToken::new(), + env.healer.disk.clone(), + HealOpts { + recreate: true, + pool: Some(0), + set: Some(0), + ..Default::default() + }, + HealRequestSource::Admin, + ) + .with_pool_metadata_targets(vec!["replacement-a".to_string()]); + env.storage + .set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME)); + + healer + .execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint) + .await + .expect("admin recreate should heal and verify pool metadata"); + + assert!(env.resume.get_state().await.completed); + assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]); + } + + #[tokio::test] + async fn admin_recreate_pool_metadata_readback_failure_keeps_resume_state() { + let env = make_env_with_targets(vec!["replacement-a".to_string()]).await; + let healer = ErasureSetHealer::new( + env.storage.clone(), + Arc::new(RwLock::new(HealProgress::new())), + CancellationToken::new(), + env.healer.disk.clone(), + HealOpts { + recreate: true, + pool: Some(0), + set: Some(0), + ..Default::default() + }, + HealRequestSource::Admin, + ) + .with_pool_metadata_targets(vec!["replacement-a".to_string()]); + env.storage + .set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME)); + env.storage.set_replacement_commit_evidence(POOL_META_NAME, None, false); + + let error = healer + .execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint) + .await + .expect_err("unconfirmed admin recreate pool metadata must keep the set incomplete"); + + assert!(error.to_string().contains("Erasure set heal incomplete")); + let state = env.resume.get_state().await; + assert!(!state.completed); + assert_eq!(state.retry_count, 1); + assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]); + } + #[tokio::test] async fn retry_exhaustion_keeps_resume_artifacts_for_recovery() { let env = make_env().await; diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 2c171e76e..816f22afd 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -45,7 +45,7 @@ use tokio::sync::RwLock; use tracing::{debug, error, info, warn}; use uuid::Uuid; -use super::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, RUSTFS_META_BUCKET}; +use super::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, POOL_META_NAME, RUSTFS_META_BUCKET}; #[cfg(test)] pub(crate) struct OutcomeFinishTestHook { diff --git a/crates/heal/src/heal/task/heal_bucket.rs b/crates/heal/src/heal/task/heal_bucket.rs index d7465fdb5..cec17886a 100644 --- a/crates/heal/src/heal/task/heal_bucket.rs +++ b/crates/heal/src/heal/task/heal_bucket.rs @@ -340,9 +340,88 @@ impl HealTask { return Err(self.record_batch_failure(failure).await); } + if self.options.recreate_missing && !self.options.dry_run { + self.heal_cluster_pool_metadata().await?; + } + Ok(()) } + async fn heal_cluster_pool_metadata(&self) -> Result<()> { + let heal_opts = HealOpts { + recursive: false, + dry_run: self.options.dry_run, + remove: false, + recreate: self.options.recreate_missing, + scan_mode: self.options.scan_mode, + update_parity: self.options.update_parity, + no_lock: self.options.no_lock, + read_repair: false, + pool: self.options.pool_index, + set: self.options.set_index, + }; + + let heal_result = self + .await_with_control(self.storage.heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &heal_opts)) + .await; + match heal_result { + Ok((result, None)) => { + debug!( + target: "rustfs::heal::task", + event = EVENT_HEAL_BUCKET_RESULT, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_TASK, + task_id = %self.id, + bucket = RUSTFS_META_BUCKET, + object = POOL_META_NAME, + drives_healed = result.drives_healed(), + drives_total = result.drives_reported(), + result = "pool_metadata_ok", + "Heal cluster pool metadata repaired" + ); + self.record_result_item(result).await; + Ok(()) + } + Ok((result, Some(err))) => { + self.record_result_item(result).await; + warn!( + target: "rustfs::heal::task", + event = EVENT_HEAL_BUCKET_RESULT, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_TASK, + task_id = %self.id, + bucket = RUSTFS_META_BUCKET, + object = POOL_META_NAME, + result = "pool_metadata_failed", + error = %err, + "Heal cluster pool metadata failed" + ); + Err(Error::TaskExecutionFailed { + message: format!("Failed to heal cluster pool metadata: {err}"), + }) + } + Err(Error::TaskCancelled) => Err(Error::TaskCancelled), + Err(Error::TaskTimeout) => Err(Error::TaskTimeout), + Err(err) => { + warn!( + target: "rustfs::heal::task", + event = EVENT_HEAL_BUCKET_RESULT, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_TASK, + task_id = %self.id, + bucket = RUSTFS_META_BUCKET, + object = POOL_META_NAME, + result = "pool_metadata_failed", + error = %err, + "Heal cluster pool metadata failed" + ); + Err(Error::TaskExecutionFailed { + message: format!("Failed to heal cluster pool metadata: {err}"), + }) + } + } + } + pub(super) async fn heal_prefix(&self, bucket: &str, prefix: &str) -> Result<()> { debug!( target: "rustfs::heal::task", diff --git a/crates/heal/src/heal/task/heal_erasure_set.rs b/crates/heal/src/heal/task/heal_erasure_set.rs index 76b65a41a..2a6ccc494 100644 --- a/crates/heal/src/heal/task/heal_erasure_set.rs +++ b/crates/heal/src/heal/task/heal_erasure_set.rs @@ -451,6 +451,11 @@ impl HealTask { self.source, ) .with_replacement_targets(replacement_targets, is_auto_replacement.then(|| self.id.clone())) + .with_pool_metadata_targets(if self.options.recreate_missing && !self.options.dry_run { + self.heal_endpoints.clone() + } else { + Vec::new() + }) .with_replacement_identity_fence(replacement_target_identities.clone()) .with_mainline_pacer(self.mainline_pacer.clone()); diff --git a/crates/heal/src/heal/task/heal_object.rs b/crates/heal/src/heal/task/heal_object.rs index 2265d1919..c7635a79f 100644 --- a/crates/heal/src/heal/task/heal_object.rs +++ b/crates/heal/src/heal/task/heal_object.rs @@ -162,11 +162,6 @@ impl HealTask { pool: self.options.pool_index, set: self.options.set_index, }; - let expected_bucket_incarnation_id = self.storage.bucket_incarnation_id(bucket).await?; - let mut expected_identity = - self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index); - expected_identity.bucket_incarnation_id = expected_bucket_incarnation_id; - let mut expected_identity = self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index); expected_identity.bucket_incarnation_id = self.outcome_bucket_incarnation_id(bucket, self.options.dry_run).await?; diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 9981d73d3..63376d08e 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -66,7 +66,7 @@ mod canonical_outcome { assert_eq!(task.get_progress().await.objects_scanned, 2); assert_eq!( storage.heal_object_calls.lock().expect("object calls").as_slice(), - ["object-a", "object-b"] + ["object-a", "object-b", POOL_META_NAME] ); assert_eq!( storage.listing_tokens.lock().expect("listing tokens").as_slice(), @@ -2582,11 +2582,95 @@ async fn test_cluster_heal_visits_bucket_objects() { assert_eq!( storage.healed_objects.lock().unwrap().as_slice(), - ["object-a".to_string(), "object-b".to_string()] + ["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()] ); assert!(matches!(task.get_status().await, HealTaskStatus::Completed)); } +#[tokio::test] +async fn cluster_recreate_heals_pool_metadata_after_user_buckets() { + let storage = Arc::new(MockStorage::default()); + let request = HealRequest::new( + HealType::Cluster, + HealOptions { + recursive: true, + recreate_missing: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage.clone()); + + task.execute() + .await + .expect("cluster recreate heal should include pool metadata"); + + assert_eq!( + storage.heal_object_calls.lock().expect("object calls").as_slice(), + ["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()] + ); + let opts = storage.object_heal_opts.lock().expect("object opts"); + assert!(opts.last().expect("pool metadata opts").recreate); +} + +#[tokio::test] +async fn cluster_recreate_fails_when_pool_metadata_heal_fails() { + let storage = Arc::new(MockStorage::default()); + storage.heal_object_outcomes.lock().expect("object outcomes").insert( + POOL_META_NAME.to_string(), + VecDeque::from([MockHealObjectOutcome::ErrOther("pool metadata missing")]), + ); + let request = HealRequest::new( + HealType::Cluster, + HealOptions { + recursive: true, + recreate_missing: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage.clone()); + + let err = task + .execute() + .await + .expect_err("cluster recreate heal must not hide pool metadata failure"); + + assert!(matches!(err, Error::TaskExecutionFailed { .. })); + assert_eq!( + storage.heal_object_calls.lock().expect("object calls").as_slice(), + ["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()] + ); +} + +#[tokio::test] +async fn cluster_dry_run_does_not_heal_pool_metadata() { + let storage = Arc::new(MockStorage::default()); + let request = HealRequest::new( + HealType::Cluster, + HealOptions { + recursive: true, + dry_run: true, + recreate_missing: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage.clone()); + + task.execute() + .await + .expect("dry-run cluster heal should preserve existing coverage"); + + assert_eq!( + storage.heal_object_calls.lock().expect("object calls").as_slice(), + ["object-a".to_string(), "object-b".to_string()] + ); +} + #[tokio::test] async fn object_heal_skips_dangling_delete_grace_without_failing_task() { let storage = Arc::new(MockStorage {