heal: verify replacement pool metadata repair (#7471)

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/api/mod.rs

@@ -563,7 +563,7 @@ pub mod store_list {
 }
 
 pub mod storage {
-    pub use crate::core::pools::HealLifecycleExpiryContext;
+    pub use crate::core::pools::{HealLifecycleExpiryContext, POOL_META_NAME};
     pub use crate::store::HealWalkVersion;
     pub use crate::store::{
         BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk,

crates/heal/src/heal/erasure_healer.rs

@@ -34,7 +34,7 @@ use std::time::{Duration, UNIX_EPOCH};
 use tokio::sync::{RwLock, Semaphore};
 use tracing::{debug, error, warn};
 
-use super::{DiskStore, EcstoreError};
+use super::{DiskStore, EcstoreError, POOL_META_NAME, RUSTFS_META_BUCKET};
 
 /// Outcome of classifying an error returned by [`HealStorageAPI::heal_object`].
 enum HealObjectOutcome {
@@ -68,6 +68,13 @@ struct PageConcurrencyGuard {
     set_label: String,
 }
 
+struct ErasureSetPassCounters<'a> {
+    processed_objects: &'a mut u64,
+    successful_objects: &'a mut u64,
+    failed_objects: &'a mut u64,
+    skipped_objects: &'a mut u64,
+}
+
 impl PageConcurrencyGuard {
     fn new(in_flight: Arc<AtomicUsize>, set_label: String) -> Self {
         let current = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
@@ -834,6 +841,21 @@ impl ErasureSetHealer {
             current_object_index = 0;
         }
 
+        if failed_objects == 0 && skipped_objects == 0 && failed_buckets == 0 {
+            self.heal_replacement_pool_metadata(
+                set_disk_id,
+                &mut ErasureSetPassCounters {
+                    processed_objects: &mut processed_objects,
+                    successful_objects: &mut successful_objects,
+                    failed_objects: &mut failed_objects,
+                    skipped_objects: &mut skipped_objects,
+                },
+                resume_manager,
+                checkpoint_manager,
+            )
+            .await?;
+        }
+
         // 5. finalize. Only declare the set healed when nothing failed AND
         // nothing was transiently skipped — otherwise the resume/checkpoint
         // state must survive so the failed/skipped versions are retried instead
@@ -919,6 +941,207 @@ impl ErasureSetHealer {
         Ok(())
     }
 
+    async fn heal_replacement_pool_metadata(
+        &self,
+        set_disk_id: &str,
+        counters: &mut ErasureSetPassCounters<'_>,
+        resume_manager: &ResumeManager,
+        checkpoint_manager: &CheckpointManager,
+    ) -> Result<()> {
+        if self.replacement_task_id.is_none() {
+            return Ok(());
+        }
+        if self.target_endpoints.is_empty() {
+            return Err(Error::TaskExecutionFailed {
+                message: "Replacement pool metadata heal requires target endpoints".to_string(),
+            });
+        }
+
+        let object_key = format!("{RUSTFS_META_BUCKET}/{POOL_META_NAME}");
+        let checkpoint_key = compose_key(&object_key, None);
+        let checkpoint = checkpoint_manager.get_checkpoint().await;
+        if checkpoint.processed_objects.contains(&checkpoint_key)
+            || checkpoint.failed_objects.contains(&checkpoint_key)
+            || checkpoint.skipped_objects.contains(&checkpoint_key)
+        {
+            return Ok(());
+        }
+        drop(checkpoint);
+
+        self.verify_replacement_identity_fence("pool metadata").await?;
+        resume_manager
+            .set_current_item(Some(RUSTFS_META_BUCKET.to_string()), Some(POOL_META_NAME.to_string()))
+            .await?;
+
+        let result = match self
+            .storage
+            .heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &self.heal_opts)
+            .await
+        {
+            Ok((result, None)) if target_outcomes_complete(&result, &self.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,
+                    )
+                    .await
+                {
+                    Ok(true) => (object_size, Ok(())),
+                    Ok(false) => (
+                        object_size,
+                        Err(Error::transient_skip(
+                            "Skipped replacement pool metadata heal because target readback did not confirm the committed version",
+                        )),
+                    ),
+                    Err(err) => (
+                        object_size,
+                        Err(Error::transient_skip(format!(
+                            "Skipped replacement pool metadata heal because target readback failed: {err}"
+                        ))),
+                    ),
+                }
+            }
+            Ok((result, None)) => (
+                result_object_size_u64(&result),
+                Err(Error::transient_skip(
+                    "Skipped replacement pool metadata heal because a replacement target was not committed",
+                )),
+            ),
+            Ok((result, Some(err))) => {
+                let object_size = result_object_size_u64(&result);
+                match Self::classify_heal_object_error(&err) {
+                    HealObjectOutcome::Absent | HealObjectOutcome::Transient => (
+                        object_size,
+                        Err(Error::transient_skip(format!(
+                            "Skipped replacement pool metadata heal due to transient error: {err}"
+                        ))),
+                    ),
+                    HealObjectOutcome::Failed => (object_size, Err(err)),
+                }
+            }
+            Err(err @ Error::TaskCancelled) | Err(err @ Error::TaskTimeout) => return Err(err),
+            Err(err) => match Self::classify_heal_object_error(&err) {
+                HealObjectOutcome::Absent | HealObjectOutcome::Transient => (
+                    0,
+                    Err(Error::transient_skip(format!(
+                        "Skipped replacement pool metadata heal due to transient error: {err}"
+                    ))),
+                ),
+                HealObjectOutcome::Failed => (0, Err(err)),
+            },
+        };
+
+        let (object_size, result) = result;
+        let mut bytes_processed = self.progress.read().await.bytes_processed;
+        let mut telemetry_unknown = false;
+        let checkpoint_outcome = match result {
+            Ok(()) => {
+                telemetry_unknown |= !increment_counter(counters.successful_objects);
+                telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size);
+                debug!(
+                    target: "rustfs::heal::erasure_healer",
+                    event = EVENT_HEAL_ERASURE_OBJECT_STATE,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
+                    set_disk_id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    state = "healed",
+                    "Replacement pool metadata healed"
+                );
+                CheckpointObjectOutcome::Processed
+            }
+            Err(Error::TransientSkip { message }) => {
+                telemetry_unknown |= !increment_counter(counters.skipped_objects);
+                telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size);
+                warn!(
+                    target: "rustfs::heal::erasure_healer",
+                    event = EVENT_HEAL_ERASURE_OBJECT_STATE,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
+                    set_disk_id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    state = "transient_skip",
+                    error = %message,
+                    "Replacement pool metadata heal skipped due to transient error"
+                );
+                CheckpointObjectOutcome::Skipped
+            }
+            Err(err) => {
+                telemetry_unknown |= !increment_counter(counters.failed_objects);
+                telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size);
+                warn!(
+                    target: "rustfs::heal::erasure_healer",
+                    event = EVENT_HEAL_ERASURE_OBJECT_STATE,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
+                    set_disk_id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    state = "failed",
+                    error = %err,
+                    "Replacement pool metadata heal failed"
+                );
+                CheckpointObjectOutcome::Failed
+            }
+        };
+
+        telemetry_unknown |= !increment_counter(counters.processed_objects);
+        let (outcome_record, counter_unknown, skipped_new_versions, skipped_ilm_expired) = {
+            let mut progress = self.progress.write().await;
+            progress.set_current_object(Some(object_key.clone()));
+            progress.update_object_progress(
+                *counters.processed_objects,
+                *counters.successful_objects,
+                *counters.failed_objects,
+                *counters.skipped_objects,
+                bytes_processed,
+            );
+            if telemetry_unknown {
+                progress.mark_unknown();
+            }
+            (
+                CheckpointObjectOutcomeRecord {
+                    object: checkpoint_key,
+                    outcome: checkpoint_outcome,
+                    successful: progress.objects_healed,
+                    failed: progress.objects_failed,
+                    skipped: progress.skipped_objects,
+                    bytes: progress.bytes_processed,
+                    skipped_new_versions: progress.skipped_new_versions,
+                    skipped_ilm_expired: progress.skipped_ilm_expired,
+                    counter_unknown: progress.counter_unknown,
+                },
+                progress.counter_unknown,
+                progress.skipped_new_versions,
+                progress.skipped_ilm_expired,
+            )
+        };
+        checkpoint_manager.record_object_outcome(outcome_record).await?;
+        resume_manager
+            .update_progress_with_bytes(
+                *counters.processed_objects,
+                *counters.successful_objects,
+                *counters.failed_objects,
+                *counters.skipped_objects,
+                bytes_processed,
+            )
+            .await?;
+        resume_manager
+            .set_skipped_version_counts(skipped_new_versions, skipped_ilm_expired)
+            .await?;
+        if counter_unknown {
+            resume_manager.mark_counter_unknown().await?;
+        }
+        Ok(())
+    }
+
     /// heal single bucket with resume
     #[allow(clippy::too_many_arguments)]
     #[tracing::instrument(skip(self, current_object_index, processed_objects, successful_objects, failed_objects, skipped_objects, resume_manager, checkpoint_manager), fields(bucket = %bucket, bucket_index = bucket_index))]
@@ -1679,7 +1902,8 @@ mod resume_loop_tests {
     use crate::heal::storage::{HealLifecycleExpiryContext, HealListItem, HealObjectInfo, HealStorageAPI};
     use crate::heal::storage_api::status::BucketInfo;
     use crate::heal::{
-        BUCKET_META_PREFIX, DiskOption, DiskStore, EcstoreError, Endpoint, HealDiskExt as _, RUSTFS_META_BUCKET, new_disk,
+        BUCKET_META_PREFIX, DiskOption, DiskStore, EcstoreError, Endpoint, HealDiskExt as _, POOL_META_NAME, RUSTFS_META_BUCKET,
+        new_disk,
     };
     use crate::{Error, Result};
     use rustfs_heal_contracts::heal_channel::{HealOpts, HealRequestSource};
@@ -1773,6 +1997,28 @@ mod resume_loop_tests {
         assert!(!target_outcomes_complete(&duplicate, &["replacement-a".to_string()]));
     }
 
+    fn replacement_target_ok_result(endpoint: &str, object: &str) -> HealResultItem {
+        HealResultItem {
+            object: object.to_string(),
+            object_size: 1024,
+            before: Infos {
+                drives: vec![HealDriveInfo {
+                    endpoint: endpoint.to_string(),
+                    state: "missing".to_string(),
+                    ..Default::default()
+                }],
+            },
+            after: Infos {
+                drives: vec![HealDriveInfo {
+                    endpoint: endpoint.to_string(),
+                    state: "ok".to_string(),
+                    ..Default::default()
+                }],
+            },
+            ..Default::default()
+        }
+    }
+
     #[derive(Clone)]
     struct Page {
         items: Vec<HealListItem>,
@@ -2444,6 +2690,8 @@ mod resume_loop_tests {
             HealRequestSource::AutoHeal,
         )
         .with_replacement_targets(vec!["replacement-a".to_string()], Some(replacement_task_id.clone()));
+        env.storage
+            .set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME));
 
         healer
             .heal_erasure_set(&["b".to_string()], "pool_0_set_0")
@@ -2461,9 +2709,59 @@ mod resume_loop_tests {
             CheckpointManager::has_checkpoint(&env.healer.disk, &replacement_task_id).await,
             "the checkpoint must survive until the caller clears the healing marker"
         );
+        assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
         drop(checkpoint);
     }
 
+    #[tokio::test]
+    async fn replacement_pool_metadata_readback_failure_schedules_retry() {
+        let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
+        let replacement_task_id = ResumeUtils::generate_task_id();
+        ResumeManager::new_replacement_intent(
+            env.healer.disk.clone(),
+            replacement_task_id.clone(),
+            "pool_0_set_0".to_string(),
+            vec!["b".to_string()],
+            vec!["replacement-a".to_string()],
+            vec![crate::heal::resume::ReplacementTargetIdentity {
+                endpoint: "replacement-a".to_string(),
+                canonical_path: "/mnt/replacement-a".to_string(),
+                physical_device_ids: vec!["device-a".to_string()],
+                filesystem_identity: "1:2:3".to_string(),
+            }],
+        )
+        .await
+        .expect("replacement intent should persist");
+        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 healer = ErasureSetHealer::new(
+            env.storage.clone(),
+            Arc::new(RwLock::new(HealProgress::new())),
+            CancellationToken::new(),
+            env.healer.disk.clone(),
+            HealOpts::default(),
+            HealRequestSource::AutoHeal,
+        )
+        .with_replacement_targets(vec!["replacement-a".to_string()], Some(replacement_task_id.clone()));
+
+        let error = healer
+            .heal_erasure_set(&["b".to_string()], "pool_0_set_0")
+            .await
+            .expect_err("unconfirmed pool metadata readback must keep the replacement incomplete");
+
+        assert!(error.to_string().contains("Replacement erasure set heal incomplete"));
+        let state = ResumeManager::load_replacement_intent(env.healer.disk.clone(), &replacement_task_id)
+            .await
+            .expect("replacement retry state must remain")
+            .get_state()
+            .await;
+        assert!(!state.completed);
+        assert_eq!(state.replacement_phase, crate::heal::resume::ReplacementPhase::Intent);
+        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;

crates/heal/src/heal/mod.rs

@@ -27,10 +27,10 @@ pub mod task;
 pub mod utils;
 
 use storage_api::owner::{
-    ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET,
-    EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskOption,
-    EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ObjectIO,
-    ObjectOperations, ecstore_local_disk_map_read, ecstore_new_disk,
+    ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_POOL_META_NAME,
+    ECSTORE_RUSTFS_META_BUCKET, EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes,
+    EcstoreDiskError, EcstoreDiskOption, EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType,
+    EcstoreStorageError, EcstoreStore, ObjectIO, ObjectOperations, ecstore_local_disk_map_read, ecstore_new_disk,
 };
 
 pub use erasure_healer::ErasureSetHealer;
@@ -41,6 +41,7 @@ pub use task::{HealOptions, HealPriority, HealRequest, HealTask, HealType};
 pub(crate) const DATA_USAGE_CACHE_NAME: &str = ECSTORE_DATA_USAGE_CACHE_NAME;
 pub(crate) const BUCKET_META_PREFIX: &str = ECSTORE_BUCKET_META_PREFIX;
 pub(crate) const RUSTFS_META_BUCKET: &str = ECSTORE_RUSTFS_META_BUCKET;
+pub(crate) const POOL_META_NAME: &str = ECSTORE_POOL_META_NAME;
 
 /// Marker written to every local disk while the process runs; removed by
 /// [`clear_unclean_shutdown_markers`] on graceful shutdown. Finding it at

crates/heal/src/heal/storage_api.rs

@@ -29,6 +29,7 @@ pub(crate) use rustfs_ecstore::api::error::{Error as EcstoreErrorType, StorageEr
 pub(crate) use rustfs_ecstore::api::runtime::local_disk_map_read as ecstore_local_disk_map_read;
 pub(crate) use rustfs_ecstore::api::storage::{
     ECStore as EcstoreStore, HealLifecycleExpiryContext as EcstoreHealLifecycleExpiryContext,
+    POOL_META_NAME as ECSTORE_POOL_META_NAME,
 };
 use rustfs_storage_api as storage_contracts;
 
@@ -36,10 +37,11 @@ pub(crate) mod owner {
     pub(crate) use super::storage_contracts::{ObjectIO, ObjectOperations};
 
     pub(crate) use super::{
-        ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET,
-        EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError,
-        EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, EcstoreHealLifecycleExpiryContext,
-        EcstoreStorageError, EcstoreStore, ecstore_load_admin_data_usage_from_backend_cached, ecstore_local_disk_map_read,
+        ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_POOL_META_NAME,
+        ECSTORE_RUSTFS_META_BUCKET, EcstoreConditionalFileUpdate, EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes,
+        EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType,
+        EcstoreHealLifecycleExpiryContext, EcstoreStorageError, EcstoreStore, ecstore_load_admin_data_usage_from_backend_cached,
+        ecstore_local_disk_map_read,
     };
 
     pub(crate) use super::{EcstoreDiskOption, ecstore_new_disk};