From 5ab1d5191ee6fb7866113ecd43c29b0bb6c4c1b5 Mon Sep 17 00:00:00 2001 From: Chris An Date: Wed, 28 Oct 2026 01:10:44 +0800 Subject: [PATCH] fix(heal): pace and parallelize replacement recovery (#9005) --- crates/heal/src/error.rs | 2 + crates/heal/src/heal/erasure_healer.rs | 589 +++++++++++++++--- crates/heal/src/heal/manager/scheduler.rs | 38 +- .../heal/manager/tests/running_mainline.rs | 35 ++ crates/heal/src/heal/pacing.rs | 22 +- crates/heal/src/heal/task/heal_object.rs | 12 +- crates/heal/src/heal/task/tests.rs | 38 ++ docs/operations/scanner-runtime-controls.md | 18 +- 8 files changed, 630 insertions(+), 124 deletions(-) diff --git a/crates/heal/src/error.rs b/crates/heal/src/error.rs index be6d91406..b3d25f44e 100644 --- a/crates/heal/src/error.rs +++ b/crates/heal/src/error.rs @@ -19,6 +19,7 @@ use thiserror::Error; use super::heal::{DiskError, EcstoreError}; const HEAL_DANGLING_DELETE_GRACE_MESSAGE: &str = "dangling object deletion deferred by heal grace window"; +pub(crate) const MRF_MISSING_PROOF_PREFIX: &str = "Missing exact storage proof for durable MRF repair "; /// Custom error type for heal operations /// This enum defines various error variants that can occur during @@ -221,6 +222,7 @@ pub(crate) fn classify_heal_error(error: &Error) -> HealErrorClass { Error::Storage(error) => classify_storage_error(error), Error::Disk(error) => classify_disk_error(error), Error::Io(error) => classify_io_error(error), + Error::TaskExecutionFailed { message } if message.starts_with(MRF_MISSING_PROOF_PREFIX) => HealErrorClass::Permanent, Error::TaskExecutionFailed { message } | Error::Other(message) => { classify_error_display(message).unwrap_or(HealErrorClass::Permanent) } diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 999afd282..95aabffad 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -58,6 +58,8 @@ fn result_object_size_u64(result: &HealResultItem) -> u64 { const NEW_VERSION_SKIP_GRACE_SECS: u64 = 60; const NANOS_PER_SECOND: i128 = 1_000_000_000; +const DEFAULT_REPLACEMENT_PAGE_OBJECT_CONCURRENCY: usize = 2; +const REPLACEMENT_OBJECT_MAX_RETRIES: u32 = 2; fn should_skip_new_version(mod_time_unix_nanos: Option, started_at_secs: u64) -> bool { let Some(mod_time_unix_nanos) = mod_time_unix_nanos else { @@ -177,7 +179,9 @@ mod mainline_pacing_tests { .await .expect("existing object owns capacity"); let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 0)); - let pacer = Arc::new(MainlinePacer::new(provider.clone(), 80, 80, Duration::from_millis(250)).expect("pacer")); + let pacer = Arc::new( + MainlinePacer::new(provider.clone(), 80, 80, Duration::from_millis(250), HealRequestSource::Admin).expect("pacer"), + ); let cancel = CancellationToken::new(); let waiting = tokio::spawn({ let semaphore = semaphore.clone(); @@ -216,7 +220,7 @@ mod mainline_pacing_tests { let occupied = semaphore.clone().acquire_owned().await.expect("queue both waiters"); let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 0)); let pause = Duration::from_millis(250); - let pacer = MainlinePacer::new(provider.clone(), 80, 80, pause).expect("pacer"); + let pacer = MainlinePacer::new(provider.clone(), 80, 80, pause, HealRequestSource::Admin).expect("pacer"); let cancel = CancellationToken::new(); let mut first = Box::pin(acquire_page_permit(semaphore.clone(), Some(&pacer), &cancel)); let mut second = Box::pin(acquire_page_permit(semaphore.clone(), Some(&pacer), &cancel)); @@ -304,9 +308,21 @@ impl ErasureSetHealer { } } - fn effective_heal_page_object_concurrency_for_source(source: HealRequestSource, scan_mode: HealScanMode) -> usize { + fn effective_heal_page_object_concurrency_for_source( + source: HealRequestSource, + scan_mode: HealScanMode, + replacement: bool, + ) -> usize { if matches!(source, HealRequestSource::AutoHeal) { - 1 + if !replacement || matches!(scan_mode, HealScanMode::Deep) || !Self::page_parallel_enabled() { + 1 + } else { + rustfs_utils::get_env_usize( + rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, + DEFAULT_REPLACEMENT_PAGE_OBJECT_CONCURRENCY, + ) + .clamp(1, Semaphore::MAX_PERMITS) + } } else { Self::effective_heal_page_object_concurrency_for_scan_mode(scan_mode) } @@ -1467,7 +1483,16 @@ impl ErasureSetHealer { // transient cache id that changes between identical pages. let mut previous_page_last: Option<(String, Option)> = None; let page_concurrency_limit = - Self::effective_heal_page_object_concurrency_for_source(self.source, self.heal_opts.scan_mode); + if self.source == HealRequestSource::AutoHeal && self.replacement_task_id.is_some() && self.heal_opts.no_lock { + // Versions of the same object must not race without its namespace lock. + 1 + } else { + Self::effective_heal_page_object_concurrency_for_source( + self.source, + self.heal_opts.scan_mode, + self.replacement_task_id.is_some(), + ) + }; let in_flight = Arc::new(AtomicUsize::new(0)); // Per-bucket sample caps for per-object warn! lines: a flapping rebuild // disk can fail/skip hundreds of thousands of versions in one sweep, so @@ -1689,6 +1714,7 @@ impl ErasureSetHealer { let semaphore = semaphore.clone(); let target_endpoints = self.target_endpoints.clone(); let replacement_commit_evidence_required = self.replacement_task_id.is_some(); + let retry_replacement = replacement_commit_evidence_required && self.source == HealRequestSource::AutoHeal; let mainline_pacer = self.mainline_pacer.clone(); let admin_identity = self.admin_identity(bucket, &object_name, version_id.as_deref(), incarnation); @@ -1698,107 +1724,125 @@ impl ErasureSetHealer { .as_ref() .map(|_| (dedup_key.clone(), object_name.clone(), version_id.clone())); let work = async move { - let permit = acquire_page_permit(semaphore, mainline_pacer.as_deref(), &cancel_token).await; + let mut retries = 0; + let mut admin_identity = admin_identity; + let mut admin_task_id = admin_task_id; + loop { + let permit = acquire_page_permit(semaphore.clone(), mainline_pacer.as_deref(), &cancel_token).await; - let _permit = match permit { - Ok(permit) => permit, - Err(err) => return (dedup_key, object_name, version_id, (0, Err(err)), None), - }; + let permit = match permit { + Ok(permit) => permit, + Err(err) => return (dedup_key, object_name, version_id, (0, Err(err)), None), + }; - let _in_flight_guard = PageConcurrencyGuard::new(in_flight, set_label); + let in_flight_guard = PageConcurrencyGuard::new(in_flight.clone(), set_label.clone()); - if let (Some(identity), Some(task_id)) = (admin_identity, admin_task_id) { - let (result, canonical) = - admin::heal_object(storage.as_ref(), &heal_opts, identity, &task_id, &cancel_token).await; - return (dedup_key, object_name, version_id, result, canonical); - } + if let (Some(identity), Some(task_id)) = (admin_identity.take(), admin_task_id.take()) { + let (result, canonical) = + admin::heal_object(storage.as_ref(), &heal_opts, identity, &task_id, &cancel_token).await; + return (dedup_key, object_name, version_id, result, canonical); + } - // Always go through heal_object. Genuine absence flows through - // heal_object -> FileVersionNotFound/FileNotFound -> - // classify_heal_object_error -> Absent, so gone versions are - // recorded as skipped-ok rather than failed. The delete-marker - // vs data path is chosen internally in ops/heal.rs. - let result = if cancel_token.is_cancelled() { - (0, Err(Error::TaskCancelled)) - } else { - match storage - .heal_object(&bucket_name, &object_name, version_id.as_deref(), &heal_opts) - .await - { - Ok((result, None)) if target_outcomes_complete(&result, &target_endpoints) => { - let object_size = result_object_size_u64(&result); - if !replacement_commit_evidence_required { - (object_size, Ok(true)) - } else { - match storage - .replacement_targets_have_version( - &bucket_name, - &object_name, - version_id.as_deref(), - &heal_opts, - &target_endpoints, - ) - .await - { - Ok(true) => (object_size, Ok(true)), - Ok(false) => ( - object_size, - Err(Error::transient_skip(format!( - "Skipped heal for {bucket_name}/{object_name} because replacement target readback did not confirm the committed version" - ))), - ), - Err(err) => ( - object_size, - Err(Error::transient_skip(format!( - "Skipped heal for {bucket_name}/{object_name} because replacement target readback failed: {err}" - ))), - ), + // Always go through heal_object. Genuine absence flows through + // heal_object -> FileVersionNotFound/FileNotFound -> + // classify_heal_object_error -> Absent, so gone versions are + // recorded as skipped-ok rather than failed. The delete-marker + // vs data path is chosen internally in ops/heal.rs. + let result = if cancel_token.is_cancelled() { + (0, Err(Error::TaskCancelled)) + } else { + match storage + .heal_object(&bucket_name, &object_name, version_id.as_deref(), &heal_opts) + .await + { + Ok((result, None)) if target_outcomes_complete(&result, &target_endpoints) => { + let object_size = result_object_size_u64(&result); + if !replacement_commit_evidence_required { + (object_size, Ok(true)) + } else { + match storage + .replacement_targets_have_version( + &bucket_name, + &object_name, + version_id.as_deref(), + &heal_opts, + &target_endpoints, + ) + .await + { + Ok(true) => (object_size, Ok(true)), + Ok(false) => ( + object_size, + Err(Error::transient_skip(format!( + "Skipped heal for {bucket_name}/{object_name} because replacement target readback did not confirm the committed version" + ))), + ), + Err(err) => ( + object_size, + Err(Error::transient_skip(format!( + "Skipped heal for {bucket_name}/{object_name} because replacement target readback failed: {err}" + ))), + ), + } } } - } - Ok((result, None)) if !target_endpoints.is_empty() => ( - result_object_size_u64(&result), - Err(Error::transient_skip(format!( - "Skipped heal for {bucket_name}/{object_name} because a replacement target was not committed" - ))), - ), - Ok((result, None)) => (result_object_size_u64(&result), Ok(true)), - Ok((result, Some(err))) if is_missing_object_dir_heal_result(&object_name, &err) => { - (result_object_size_u64(&result), Ok(false)) - } - Ok((result, Some(err))) => { - let object_size = result_object_size_u64(&result); - match Self::classify_heal_object_error(&err) { - HealObjectOutcome::Absent => (object_size, Ok(false)), + Ok((result, None)) if !target_endpoints.is_empty() => ( + result_object_size_u64(&result), + Err(Error::transient_skip(format!( + "Skipped heal for {bucket_name}/{object_name} because a replacement target was not committed" + ))), + ), + Ok((result, None)) => (result_object_size_u64(&result), Ok(true)), + Ok((result, Some(err))) if is_missing_object_dir_heal_result(&object_name, &err) => { + (result_object_size_u64(&result), Ok(false)) + } + Ok((result, Some(err))) => { + let object_size = result_object_size_u64(&result); + match Self::classify_heal_object_error(&err) { + HealObjectOutcome::Absent => (object_size, Ok(false)), + HealObjectOutcome::DanglingDeleteDeferred { retry_not_before } => { + (object_size, Err(Error::DanglingDeleteDeferred { retry_not_before })) + } + HealObjectOutcome::Transient => ( + object_size, + Err(Error::transient_skip(format!( + "Skipped heal for {bucket_name}/{object_name} due to transient error: {err}" + ))), + ), + HealObjectOutcome::Failed => (object_size, Err(err)), + } + } + Err(err) => match Self::classify_heal_object_error(&err) { + HealObjectOutcome::Absent => (0, Ok(false)), HealObjectOutcome::DanglingDeleteDeferred { retry_not_before } => { - (object_size, Err(Error::DanglingDeleteDeferred { retry_not_before })) + (0, Err(Error::DanglingDeleteDeferred { retry_not_before })) } HealObjectOutcome::Transient => ( - object_size, + 0, Err(Error::transient_skip(format!( "Skipped heal for {bucket_name}/{object_name} due to transient error: {err}" ))), ), - HealObjectOutcome::Failed => (object_size, Err(err)), - } + HealObjectOutcome::Failed => (0, Err(err)), + }, } - Err(err) => match Self::classify_heal_object_error(&err) { - HealObjectOutcome::Absent => (0, Ok(false)), - HealObjectOutcome::DanglingDeleteDeferred { retry_not_before } => { - (0, Err(Error::DanglingDeleteDeferred { retry_not_before })) - } - HealObjectOutcome::Transient => ( - 0, - Err(Error::transient_skip(format!( - "Skipped heal for {bucket_name}/{object_name} due to transient error: {err}" - ))), - ), - HealObjectOutcome::Failed => (0, Err(err)), - }, - } - }; + }; - (dedup_key, object_name, version_id, result, None) + drop(in_flight_guard); + drop(permit); + if retry_replacement + && retries < REPLACEMENT_OBJECT_MAX_RETRIES + && matches!(&result.1, Err(Error::TransientSkip { .. })) + { + retries += 1; + tokio::select! { + _ = cancel_token.cancelled() => return (dedup_key, object_name, version_id, (0, Err(Error::TaskCancelled)), None), + _ = tokio::time::sleep(Duration::from_millis(250 * u64::from(retries))) => {} + } + continue; + } + return (dedup_key, object_name, version_id, result, None); + } }; page_tasks.push(async move { if let (Some(execution), Some((key, object, version))) = (execution, failure_identity) { @@ -2129,6 +2173,7 @@ mod tests { ErasureSetHealer::effective_heal_page_object_concurrency_for_source( HealRequestSource::AutoHeal, HealScanMode::Normal, + false, ), 1 ); @@ -2136,6 +2181,73 @@ mod tests { ); } + #[test] + fn automatic_replacement_uses_a_conservative_default_and_respects_serial_controls() { + temp_env::with_vars( + [ + (rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")), + (rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, None), + ], + || { + assert_eq!( + ErasureSetHealer::effective_heal_page_object_concurrency_for_source( + HealRequestSource::AutoHeal, + HealScanMode::Normal, + true, + ), + 2 + ); + assert_eq!( + ErasureSetHealer::effective_heal_page_object_concurrency_for_source( + HealRequestSource::AutoHeal, + HealScanMode::Deep, + true, + ), + 1 + ); + }, + ); + temp_env::with_vars( + [ + (rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("false")), + (rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("8")), + ], + || { + assert_eq!( + ErasureSetHealer::effective_heal_page_object_concurrency_for_source( + HealRequestSource::AutoHeal, + HealScanMode::Normal, + true, + ), + 1 + ); + }, + ); + } + + #[test] + fn automatic_replacement_concurrency_fits_the_semaphore() { + let maximum = usize::MAX.to_string(); + for (value, expected) in [("0", 1), (maximum.as_str(), tokio::sync::Semaphore::MAX_PERMITS)] { + temp_env::with_vars( + [ + (rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")), + (rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some(value)), + ], + || { + assert_eq!( + ErasureSetHealer::effective_heal_page_object_concurrency_for_source( + HealRequestSource::AutoHeal, + HealScanMode::Normal, + true, + ), + expected + ); + }, + ); + } + } + #[test] fn non_auto_normal_scan_page_object_concurrency_uses_effective_limit() { temp_env::with_vars( @@ -2148,6 +2260,7 @@ mod tests { ErasureSetHealer::effective_heal_page_object_concurrency_for_source( HealRequestSource::Admin, HealScanMode::Normal, + false, ), 11 ); @@ -2520,6 +2633,8 @@ mod resume_loop_tests { lifecycle_expired: Mutex>, /// every heal_object call recorded as (name, version_id) heal_calls: Mutex)>>, + heal_gate: Mutex>>, + heal_started: tokio::sync::Notify, list_include_lifecycle_object_info: Mutex>, replacement_target_identity_sequences: Mutex>>, replacement_targets_unready: AtomicBool, @@ -2704,6 +2819,11 @@ mod resume_loop_tests { .lock() .unwrap() .push((object.to_string(), version_id.map(str::to_string))); + self.heal_started.notify_one(); + let heal_gate = self.heal_gate.lock().unwrap().clone(); + if let Some(gate) = heal_gate { + gate.acquire().await.unwrap().forget(); + } let key = compose_key(object, version_id); let outcome = { let mut outcomes = self.outcomes.lock().unwrap(); @@ -3898,6 +4018,287 @@ mod resume_loop_tests { assert_eq!(unique.len(), 50, "no version healed twice"); } + #[tokio::test] + async fn auto_replacement_runs_two_object_repairs_within_the_configured_limit() { + temp_env::async_with_vars( + [ + (rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")), + (rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("2")), + ], + async { + let mut env = make_env().await; + env.healer.source = HealRequestSource::AutoHeal; + env.healer = env + .healer + .with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone())); + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + *env.storage.heal_gate.lock().unwrap() = Some(gate.clone()); + let objects = ["a", "b", "c"]; + for name in objects { + env.storage + .set_result(name, None, replacement_target_ok_result("replacement-a", name)); + } + env.storage.set_page( + None, + Page { + items: objects.into_iter().map(|name| item(name, None, false)).collect(), + next: None, + truncated: false, + }, + ); + + let observe = async { + let two_started = tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + let notified = env.storage.heal_started.notified(); + if env.storage.calls().len() >= 2 { + break; + } + notified.await; + } + }) + .await + .is_ok(); + let started = env.storage.calls().len(); + gate.add_permits(objects.len()); + assert!(two_started, "automatic replacement should start two repairs before either finishes"); + assert_eq!(started, 2, "configured concurrency must bound in-flight repairs"); + }; + let ((processed, successful, failed, skipped, result), ()) = tokio::join!(run(&env), observe); + result.expect("all replacement objects must be verified"); + assert_eq!((processed, successful, failed, skipped), (3, 3, 0, 0)); + assert_eq!(env.storage.calls().len(), 3); + }, + ) + .await; + } + + #[tokio::test] + async fn auto_replacement_without_object_locks_stays_serial() { + temp_env::async_with_vars( + [ + (rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")), + (rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("2")), + ], + async { + let mut env = make_env().await; + env.healer.source = HealRequestSource::AutoHeal; + env.healer.heal_opts.no_lock = true; + env.healer = env + .healer + .with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone())); + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + *env.storage.heal_gate.lock().unwrap() = Some(gate.clone()); + for version in ["v1", "v2"] { + env.storage + .set_result("object", Some(version), replacement_target_ok_result("replacement-a", "object")); + } + env.storage.set_page( + None, + Page { + items: vec![item("object", Some("v1"), false), item("object", Some("v2"), false)], + next: None, + truncated: false, + }, + ); + let observe = async { + let first_started = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let notified = env.storage.heal_started.notified(); + if !env.storage.calls().is_empty() { + break; + } + notified.await; + } + }) + .await + .is_ok(); + let two_started = tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + let notified = env.storage.heal_started.notified(); + if env.storage.calls().len() >= 2 { + break; + } + notified.await; + } + }) + .await + .is_ok(); + gate.add_permits(2); + assert!(first_started, "the first object repair must start"); + assert!(!two_started, "unlocked object repairs must not overlap"); + }; + let ((processed, successful, failed, skipped, result), ()) = tokio::join!(run(&env), observe); + result.expect("the serial replacement must finish"); + assert_eq!((processed, successful, failed, skipped), (2, 2, 0, 0)); + }, + ) + .await; + } + + #[tokio::test] + async fn auto_replacement_retries_a_transient_object_before_rewinding_the_scan() { + let mut env = make_env().await; + env.healer.source = HealRequestSource::AutoHeal; + env.healer = env + .healer + .with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone())); + env.storage.set_page( + None, + Page { + items: vec![item("object", Some("version"), false)], + next: None, + truncated: false, + }, + ); + env.storage + .set_outcome("object", Some("version"), HealOutcome::RpcCancelled(1)); + env.storage + .set_result("object", Some("version"), replacement_target_ok_result("replacement-a", "object")); + + let (processed, successful, failed, skipped, result) = run(&env).await; + result.expect("the transient repair should succeed during the current pass"); + assert_eq!((processed, successful, failed, skipped), (1, 1, 0, 0)); + assert_eq!(env.storage.calls().len(), 2); + } + + #[tokio::test] + async fn auto_replacement_releases_capacity_during_a_bounded_retry() { + temp_env::async_with_vars( + [ + (rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")), + (rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("1")), + ], + async { + let mut env = make_env().await; + env.healer.source = HealRequestSource::AutoHeal; + env.healer = env + .healer + .with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone())); + env.storage.set_page( + None, + Page { + items: vec![item("retry", None, false), item("ready", None, false)], + next: None, + truncated: false, + }, + ); + env.storage.set_outcome("retry", None, HealOutcome::Transient); + env.storage + .set_result("ready", None, replacement_target_ok_result("replacement-a", "ready")); + + let (processed, successful, failed, skipped, result) = run(&env).await; + result.expect("a failed object is retained as a transient skip"); + assert_eq!((processed, successful, failed, skipped), (2, 1, 0, 1)); + let calls = env.storage.calls(); + let retries: Vec<_> = calls.iter().enumerate().filter(|(_, call)| call.0 == "retry").collect(); + let ready = calls.iter().position(|call| call.0 == "ready").expect("ready object ran"); + assert_eq!(retries.len(), 3, "only two retries are allowed"); + assert!(ready < retries[1].0, "the backoff must release the only page permit"); + }, + ) + .await; + } + + #[tokio::test] + async fn auto_replacement_cancellation_interrupts_object_retry() { + let mut env = make_env().await; + env.healer.source = HealRequestSource::AutoHeal; + env.healer = env + .healer + .with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone())); + env.storage.set_page( + None, + Page { + items: vec![item("retry", None, false)], + next: None, + truncated: false, + }, + ); + env.storage.set_outcome("retry", None, HealOutcome::Transient); + let cancel = env.healer.cancel_token.clone(); + let observe = async { + loop { + let notified = env.storage.heal_started.notified(); + if !env.storage.calls().is_empty() { + break; + } + notified.await; + } + cancel.cancel(); + }; + + let ((_, successful, _, _, result), ()) = tokio::join!(run(&env), observe); + assert!(matches!(result, Err(Error::TaskCancelled))); + assert_eq!(successful, 0); + assert_eq!(env.storage.calls().len(), 1); + assert!(env.checkpoint.get_checkpoint().await.processed_objects.is_empty()); + } + + #[tokio::test] + async fn auto_replacement_cancel_retains_workers_until_inflight_repairs_settle() { + temp_env::async_with_vars( + [ + (rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")), + (rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("2")), + ], + async { + let mut env = make_env().await; + env.healer.source = HealRequestSource::AutoHeal; + let execution = crate::heal::storage::ReplacementExecution::for_test(Vec::new(), Vec::new()); + env.healer = env + .healer + .with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone())) + .with_replacement_execution(Some(execution.clone())); + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + *env.storage.heal_gate.lock().unwrap() = Some(gate.clone()); + for name in ["a", "b", "c"] { + env.storage + .set_result(name, None, replacement_target_ok_result("replacement-a", name)); + } + env.storage.set_page( + None, + Page { + items: ["a", "b", "c"].into_iter().map(|name| item(name, None, false)).collect(), + next: None, + truncated: false, + }, + ); + let cancel = env.healer.cancel_token.clone(); + let observe = async { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let notified = env.storage.heal_started.notified(); + if env.storage.calls().len() >= 2 { + break; + } + notified.await; + } + }) + .await + .expect("two repairs must start"); + cancel.cancel(); + }; + + let ((_, successful, _, _, result), ()) = tokio::join!(run(&env), observe); + assert!(matches!(result, Err(Error::TaskCancelled))); + assert_eq!(successful, 0); + assert_eq!(env.storage.calls().len(), 2, "queued work must not start after cancellation"); + assert!(env.checkpoint.get_checkpoint().await.processed_objects.is_empty()); + assert!(Arc::strong_count(&execution) >= 4, "in-flight workers must retain execution ownership"); + gate.add_permits(2); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while Arc::strong_count(&execution) > 2 { + tokio::task::yield_now().await; + } + }) + .await + .expect("workers must release execution ownership after their storage work settles"); + }, + ) + .await; + } + #[tokio::test] async fn test_resume_after_version_deleted_midheal_no_skip() { let env = make_env().await; @@ -4312,7 +4713,11 @@ mod resume_loop_tests { assert!(!state.completed, "readback errors must leave the replacement task incomplete"); assert_eq!(state.retry_count, 1, "readback errors must arm the bounded retry path"); assert!(env.checkpoint.get_checkpoint().await.processed_objects.is_empty()); - assert_eq!(env.storage.calls(), vec![("object".to_string(), Some("dm-v1".to_string()))]); + assert_eq!( + env.storage.calls(), + vec![("object".to_string(), Some("dm-v1".to_string())); 3], + "unverified readback must exhaust only the bounded object retries" + ); } #[tokio::test] diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index 720acb575..24b5e063c 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/crates/heal/src/heal/manager/scheduler.rs @@ -30,6 +30,30 @@ pub(super) fn retry_admission_delay( Some(delay) } +pub(super) fn running_mainline_pacer( + request: &HealRequest, + config: &HealConfig, + provider: &Option, +) -> Option> { + let pace_running = request.source == HealRequestSource::Admin + || (request.source == HealRequestSource::AutoHeal + && matches!(&request.heal_type, HealType::ErasureSet { .. }) + && !request.heal_endpoints.is_empty()); + if !pace_running || !config.mainline_throttle_enable { + return None; + } + provider.as_ref().and_then(|provider| { + crate::heal::pacing::MainlinePacer::new( + provider.clone(), + config.mainline_read_utilization_high_percent, + config.mainline_write_utilization_high_percent, + config.mainline_max_sleep, + request.source, + ) + .map(Arc::new) + }) +} + impl HealManager { /// Start scheduler pub(super) async fn start_scheduler(&self) -> Result<()> { @@ -202,19 +226,7 @@ impl HealManager { .unwrap_or_else(|poisoned| poisoned.into_inner()) .get(&request.id) .cloned(); - let mainline_pacer = if request.source == HealRequestSource::Admin && config.mainline_throttle_enable { - workload_provider.as_ref().and_then(|provider| { - crate::heal::pacing::MainlinePacer::new( - provider.clone(), - config.mainline_read_utilization_high_percent, - config.mainline_write_utilization_high_percent, - config.mainline_max_sleep, - ) - .map(Arc::new) - }) - } else { - None - }; + let mainline_pacer = running_mainline_pacer(&request, &config, workload_provider); let task = Arc::new( HealTask::from_replacement_recovery_request(request, storage.clone(), replacement_resume_endpoint) .with_mainline_pacer(mainline_pacer) diff --git a/crates/heal/src/heal/manager/tests/running_mainline.rs b/crates/heal/src/heal/manager/tests/running_mainline.rs index 4cc153d28..ce078bf50 100644 --- a/crates/heal/src/heal/manager/tests/running_mainline.rs +++ b/crates/heal/src/heal/manager/tests/running_mainline.rs @@ -6,6 +6,41 @@ use crate::heal::storage::HealListItem; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use tokio::sync::Semaphore; +#[tokio::test(start_paused = true)] +async fn automatic_replacement_uses_running_foreground_pacing() { + let provider = Arc::new(super::super::super::pacing::TestPressure::new(WorkloadClass::ForegroundRead, 100)); + let provider: Option = Some(provider); + let mut config = HealConfig { + mainline_throttle_enable: true, + mainline_read_utilization_high_percent: 80, + mainline_max_sleep: Duration::from_millis(250), + ..HealConfig::default() + }; + let mut request = HealRequest::new( + HealType::ErasureSet { + buckets: Vec::new(), + set_disk_id: "pool_0_set_0".to_owned(), + }, + HealOptions::default(), + HealPriority::Low, + ); + request.source = HealRequestSource::AutoHeal; + request.heal_endpoints = vec!["replacement-a".to_owned()]; + + let pacer = super::super::scheduler::running_mainline_pacer(&request, &config, &provider) + .expect("automatic replacement must pace under foreground pressure"); + let started = tokio::time::Instant::now(); + pacer.wait(&CancellationToken::new()).await.expect("bounded progress"); + assert_eq!(started.elapsed(), Duration::from_millis(250)); + + assert!(super::super::scheduler::running_mainline_pacer(&request, &config, &None).is_none()); + config.mainline_throttle_enable = false; + assert!(super::super::scheduler::running_mainline_pacer(&request, &config, &provider).is_none()); + config.mainline_throttle_enable = true; + request.heal_endpoints.clear(); + assert!(super::super::scheduler::running_mainline_pacer(&request, &config, &provider).is_none()); +} + #[derive(Default)] struct PressureProbe { active: AtomicUsize, diff --git a/crates/heal/src/heal/pacing.rs b/crates/heal/src/heal/pacing.rs index ff8e7bea7..c7c354483 100644 --- a/crates/heal/src/heal/pacing.rs +++ b/crates/heal/src/heal/pacing.rs @@ -6,6 +6,7 @@ use rustfs_concurrency::{ WorkloadAdmissionSnapshotProvider, workload::{ForegroundPressure, foreground_pressure}, }; +use rustfs_heal_contracts::heal_channel::HealRequestSource; use std::{sync::Arc, time::Duration}; use tokio::{sync::Mutex, time::Instant}; use tokio_util::sync::CancellationToken; @@ -21,12 +22,13 @@ pub(crate) enum PacingDecision { Wait(Option), } -/// Cooperative pacing for one admin execution, not a storage admission permit. +/// Cooperative pacing for one heal execution, not a storage admission permit. pub(crate) struct MainlinePacer { provider: Arc, read_high: usize, write_high: usize, pause: Duration, + source: HealRequestSource, state: Mutex, } @@ -36,6 +38,7 @@ impl MainlinePacer { read_high: usize, write_high: usize, pause: Duration, + source: HealRequestSource, ) -> Option { if (read_high == 0 && write_high == 0) || pause.is_zero() { return None; @@ -45,6 +48,7 @@ impl MainlinePacer { read_high: read_high.min(100), write_high: write_high.min(100), pause: pause.min(Duration::from_secs(1)), + source, state: Mutex::new(PacingState::default()), }) } @@ -108,7 +112,7 @@ impl MainlinePacer { } metrics::counter!( "rustfs_heal_mainline_throttle_total", - "source" => "admin", + "source" => self.source.as_str(), "result" => "delayed", "reason" => pressure.or(observed).map_or("recovery_window", |pressure| pressure.reason()) ) @@ -170,6 +174,7 @@ mod tests { if class == WorkloadClass::ForegroundRead { 40 } else { 0 }, if class == WorkloadClass::ForegroundWrite { 40 } else { 0 }, pause, + HealRequestSource::Admin, ) .expect("enabled pacer"); let cancel = CancellationToken::new(); @@ -192,9 +197,9 @@ mod tests { #[tokio::test(start_paused = true)] async fn running_mainline_huge_pause_is_capped_and_disabled_classes_do_not_sleep() { let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 100)); - assert!(MainlinePacer::new(provider.clone(), 0, 0, Duration::from_secs(1)).is_none()); - assert!(MainlinePacer::new(provider.clone(), 80, 80, Duration::ZERO).is_none()); - let pacer = MainlinePacer::new(provider, 80, 80, Duration::from_secs(3600)).expect("pacer"); + assert!(MainlinePacer::new(provider.clone(), 0, 0, Duration::from_secs(1), HealRequestSource::Admin).is_none()); + assert!(MainlinePacer::new(provider.clone(), 80, 80, Duration::ZERO, HealRequestSource::Admin).is_none()); + let pacer = MainlinePacer::new(provider, 80, 80, Duration::from_secs(3600), HealRequestSource::Admin).expect("pacer"); let before = Instant::now(); pacer.wait(&CancellationToken::new()).await.expect("hard-capped pause"); assert_eq!(Instant::now() - before, Duration::from_secs(1)); @@ -203,7 +208,9 @@ mod tests { #[tokio::test(start_paused = true)] async fn running_mainline_waiters_cancel_and_task_latches_are_isolated() { let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 100)); - let paced = Arc::new(MainlinePacer::new(provider.clone(), 80, 80, Duration::from_secs(1)).expect("pacer")); + let paced = Arc::new( + MainlinePacer::new(provider.clone(), 80, 80, Duration::from_secs(1), HealRequestSource::Admin).expect("pacer"), + ); let cancel_first = CancellationToken::new(); let first = tokio::spawn({ let paced = paced.clone(); @@ -221,7 +228,8 @@ mod tests { cancel_second.cancel(); assert!(matches!(second.await.expect("queued waiter"), Err(Error::TaskCancelled))); provider.active.store(0, Ordering::SeqCst); - let other_task = MainlinePacer::new(provider.clone(), 80, 80, Duration::from_secs(1)).expect("independent task"); + let other_task = MainlinePacer::new(provider.clone(), 80, 80, Duration::from_secs(1), HealRequestSource::Admin) + .expect("independent task"); let before = Instant::now(); other_task .wait(&CancellationToken::new()) diff --git a/crates/heal/src/heal/task/heal_object.rs b/crates/heal/src/heal/task/heal_object.rs index f83e775a0..b03026cf4 100644 --- a/crates/heal/src/heal/task/heal_object.rs +++ b/crates/heal/src/heal/task/heal_object.rs @@ -489,9 +489,13 @@ impl HealTask { )) .await?; if let Some(error) = storage_result.error { - return Err(Error::TaskExecutionFailed { - message: format!("Failed to recreate missing object {bucket}/{object}: {error}"), - }); + if error.is_dangling_delete_grace() { + // Preserve the existing MRF grace retry ownership. + return Err(Error::TaskExecutionFailed { + message: format!("Failed to recreate missing object {bucket}/{object}: {error}"), + }); + } + return Err(error); } let object_size = storage_result.item.object_size as u64; @@ -526,7 +530,7 @@ impl HealTask { } } else { Error::TaskExecutionFailed { - message: format!("Missing exact storage proof for durable MRF repair {bucket}/{object}"), + message: format!("{}{bucket}/{object}", crate::error::MRF_MISSING_PROOF_PREFIX), } }); } diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 885ead396..a153bd739 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -2912,6 +2912,44 @@ fn mrf_task_for_item(object: &str, item: HealResultItem) -> (HealTask, Arc