diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 435b29225..08432c7e4 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/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, diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index e3d4e9b53..dbd62e422 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/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, 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, @@ -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; diff --git a/crates/heal/src/heal/mod.rs b/crates/heal/src/heal/mod.rs index e7885fb91..92a806f19 100644 --- a/crates/heal/src/heal/mod.rs +++ b/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 diff --git a/crates/heal/src/heal/storage_api.rs b/crates/heal/src/heal/storage_api.rs index 597cb675b..8138185be 100644 --- a/crates/heal/src/heal/storage_api.rs +++ b/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};