diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index a6c288dfd..e3d4e9b53 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -108,6 +108,144 @@ pub struct ErasureSetHealer { target_endpoints: Arc<[String]>, replacement_task_id: Option, replacement_target_identities: Option>, + mainline_pacer: Option>, +} + +async fn acquire_page_permit( + semaphore: Arc, + pacer: Option<&super::pacing::MainlinePacer>, + cancel: &tokio_util::sync::CancellationToken, +) -> Result { + let acquire = || async { + tokio::select! { + biased; + _ = cancel.cancelled() => Err(Error::TaskCancelled), + permit = semaphore.clone().acquire_owned() => permit.map_err(|err| Error::other(format!("Failed to acquire page concurrency permit: {err}"))), + } + }; + let mut paid_pause = false; + loop { + let permit = acquire().await?; + if let Some(pacer) = pacer { + // Keep the real permit on the low-pressure path. Every acquisition + // gets a fresh decision, including a waiter that queued a second + // time. One completed pause is a bounded minimum-progress grant. + match pacer.admission_decision() { + super::pacing::PacingDecision::Wait(pressure) if !paid_pause => { + drop(permit); + paid_pause = pacer.wait_after_admission(cancel, pressure).await?; + continue; + } + _ => {} + } + } + return Ok(permit); + } +} + +#[cfg(test)] +mod mainline_pacing_tests { + use super::*; + use crate::heal::pacing::{MainlinePacer, TestPressure}; + use rustfs_concurrency::WorkloadClass; + use std::sync::atomic::Ordering; + use tokio_util::sync::CancellationToken; + + #[tokio::test(start_paused = true)] + async fn running_mainline_page_waiters_resample_after_capacity_and_release_permits() { + let semaphore = Arc::new(Semaphore::new(1)); + let occupied = semaphore + .clone() + .acquire_owned() + .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 cancel = CancellationToken::new(); + let waiting = tokio::spawn({ + let semaphore = semaphore.clone(); + let pacer = pacer.clone(); + let cancel = cancel.clone(); + async move { acquire_page_permit(semaphore, Some(&pacer), &cancel).await } + }); + tokio::task::yield_now().await; + provider.active.store(100, Ordering::SeqCst); + drop(occupied); + provider.sampled.notified().await; + assert_eq!(semaphore.available_permits(), 1, "running pressure wait cannot retain the page permit"); + cancel.cancel(); + assert!(matches!(waiting.await.expect("page waiter"), Err(Error::TaskCancelled))); + assert_eq!(semaphore.available_permits(), 1); + let deadline = tokio::time::timeout( + Duration::from_millis(10), + acquire_page_permit(semaphore.clone(), Some(&pacer), &CancellationToken::new()), + ) + .await; + assert!(deadline.is_err()); + assert_eq!(semaphore.available_permits(), 1, "deadline must release all permits"); + let permit = acquire_page_permit(semaphore.clone(), None, &CancellationToken::new()) + .await + .expect("unpaced admission"); + assert_eq!(semaphore.available_permits(), 0, "disabling pacing cannot disable the hard cap"); + drop(permit); + assert_eq!(semaphore.available_permits(), 1); + } + + #[tokio::test(start_paused = true)] + async fn running_mainline_two_page_waiters_check_pressure_at_final_admission() { + use std::task::Poll; + for raise_pressure in [false, true] { + let semaphore = Arc::new(Semaphore::new(1)); + 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 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)); + assert!(futures::poll!(first.as_mut()).is_pending()); + assert!(futures::poll!(second.as_mut()).is_pending()); + drop(occupied); + let first_ready = match futures::poll!(first.as_mut()) { + Poll::Ready(result) => Some(result.expect("first admission")), + Poll::Pending => None, + }; + assert!(futures::poll!(second.as_mut()).is_pending()); + let first_permit = match first_ready { + Some(permit) => permit, + None => tokio::time::timeout(Duration::from_millis(1), first) + .await + .expect("low-pressure waiters must not bounce capacity forever") + .expect("first permit"), + }; + // The first object owns real page capacity while its commit runs. + tokio::time::advance(Duration::from_millis(100)).await; + if raise_pressure { + provider.active.store(100, Ordering::SeqCst); + } + drop(first_permit); + let admitted = if raise_pressure { + assert!( + futures::poll!(second.as_mut()).is_pending(), + "a second acquisition cannot reuse the earlier low-pressure sample" + ); + assert_eq!(semaphore.available_permits(), 1, "pressure wait must release page capacity"); + tokio::time::advance(pause).await; + tokio::time::timeout(Duration::from_millis(1), second) + .await + .expect("sustained pressure must allow one unit after its bounded pause") + .expect("second permit") + } else { + tokio::time::timeout(Duration::from_millis(1), second) + .await + .expect("low pressure must make progress") + .expect("second permit") + }; + assert_eq!(semaphore.available_permits(), 0); + drop(admitted); + assert_eq!(semaphore.available_permits(), 1); + } + } } pub(crate) fn target_outcomes_complete(result: &HealResultItem, target_endpoints: &[String]) -> bool { @@ -219,9 +357,15 @@ impl ErasureSetHealer { target_endpoints: Vec::new().into(), replacement_task_id: None, replacement_target_identities: None, + mainline_pacer: None, } } + pub(crate) fn with_mainline_pacer(mut self, pacer: Option>) -> Self { + self.mainline_pacer = pacer; + self + } + pub(crate) fn with_replacement_targets( mut self, mut target_endpoints: Vec, @@ -856,6 +1000,9 @@ impl ErasureSetHealer { let include_lifecycle_object_info = lifecycle_expiry_context.is_some(); loop { + if let Some(pacer) = &self.mainline_pacer { + pacer.wait(&self.cancel_token).await?; + } self.verify_replacement_identity_fence("page scan").await?; // Get one page of object versions let (objects, next_token, is_truncated) = if use_disk_walk { @@ -1034,13 +1181,10 @@ 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 mainline_pacer = self.mainline_pacer.clone(); page_tasks.push(async move { - let permit = semaphore - .clone() - .acquire_owned() - .await - .map_err(|e| Error::other(format!("Failed to acquire page concurrency permit: {e}"))); + let permit = acquire_page_permit(semaphore, mainline_pacer.as_deref(), &cancel_token).await; let _permit = match permit { Ok(permit) => permit, diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index c90a18729..74ec6d546 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -602,13 +602,13 @@ pub struct HealConfig { pub set_bulkhead_enable: bool, /// Whether erasure-set page parallelism is enabled. pub page_parallel_enable: bool, - /// Whether foreground read pressure can delay best-effort heal task starts. + /// Whether foreground pressure delays best-effort starts and paces running admin work. pub mainline_throttle_enable: bool, - /// Foreground read permit utilization percentage that delays best-effort heal starts. + /// Foreground read utilization high watermark for start admission and admin pacing. pub mainline_read_utilization_high_percent: usize, - /// Foreground write utilization percentage that delays best-effort heal starts. + /// Foreground write utilization high watermark for start admission and admin pacing. pub mainline_write_utilization_high_percent: usize, - /// Delay before rechecking foreground pressure after delaying heal starts. + /// Start recheck interval; running admin pacing caps each holder's pause at one second. pub mainline_max_sleep: Duration, } diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index 704a7cbeb..5247c9ea7 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/crates/heal/src/heal/manager/scheduler.rs @@ -175,11 +175,23 @@ impl HealManager { .unwrap_or_else(|poisoned| poisoned.into_inner()) .get(&request.id) .cloned(); - let task = Arc::new(HealTask::from_replacement_recovery_request( - request, - storage.clone(), - replacement_resume_endpoint, - )); + 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 task = Arc::new( + HealTask::from_replacement_recovery_request(request, storage.clone(), replacement_resume_endpoint) + .with_mainline_pacer(mainline_pacer), + ); let task_id = task.id.clone(); active_heals_guard.insert(task_id.clone(), task.clone()); publish_active_heal_count(&active_heals_guard); diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 17feee1b5..364f594f0 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -26,6 +26,8 @@ use rustfs_madmin::heal_commands::HealResultItem; use std::sync::Mutex as StdMutex; use tempfile::TempDir; +mod running_mainline; + use super::super::{DiskOption, DiskStore, Endpoint, new_disk, storage_api::status::BucketInfo}; #[tokio::test] diff --git a/crates/heal/src/heal/manager/tests/running_mainline.rs b/crates/heal/src/heal/manager/tests/running_mainline.rs new file mode 100644 index 000000000..4f21d7238 --- /dev/null +++ b/crates/heal/src/heal/manager/tests/running_mainline.rs @@ -0,0 +1,249 @@ +// Copyright 2026 RustFS Team +// Licensed under the Apache License, Version 2.0. + +use super::*; +use crate::heal::storage::HealListItem; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use tokio::sync::Semaphore; + +#[derive(Default)] +struct PressureProbe { + active: AtomicUsize, + commit_open: AtomicBool, + high_sampled: Notify, +} + +impl WorkloadAdmissionSnapshotProvider for PressureProbe { + fn workload_admission_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot { + assert!( + !self.commit_open.load(Ordering::SeqCst), + "pressure must not be sampled inside an object commit" + ); + let active = self.active.load(Ordering::SeqCst); + if active >= 80 { + self.high_sampled.notify_one(); + } + WorkloadAdmissionRegistrySnapshot::new(vec![ + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, AdmissionState::Open).with_counts( + Some(active), + None, + Some(100), + ), + ]) + } +} + +struct RunningStorage { + provider: Arc, + namespace: Mutex<()>, + io: Arc, + first_started: Notify, + release_first: Notify, + first_finished: Notify, + second_finished: Notify, + started: AtomicUsize, + committed: AtomicUsize, +} + +#[async_trait::async_trait] +impl HealStorageAPI for RunningStorage { + async fn get_object_meta(&self, _: &str, _: &str) -> Result> { + Ok(None) + } + async fn ec_decode_rebuild(&self, _: &str, _: &str) -> Result> { + Ok(Vec::new()) + } + async fn get_bucket_info(&self, bucket: &str) -> Result> { + Ok(Some(BucketInfo { + name: bucket.into(), + ..Default::default() + })) + } + async fn list_buckets(&self) -> Result> { + Ok(Vec::new()) + } + async fn object_exists(&self, _: &str, _: &str) -> Result { + Ok(true) + } + async fn heal_bucket(&self, _: &str, _: &HealOpts) -> Result { + Ok(HealResultItem::default()) + } + async fn heal_format(&self, _: bool) -> Result<(HealResultItem, Option)> { + Ok((HealResultItem::default(), None)) + } + async fn get_disk_for_resume(&self, _: &str) -> Result { + Err(Error::other("no resume disk in bucket fixture")) + } + async fn list_objects_for_heal_page( + &self, + _: &str, + _: &str, + _: Option<&str>, + _: bool, + ) -> Result<(Vec, Option, bool)> { + Ok(( + ["a", "b"] + .into_iter() + .map(|name| HealListItem { + name: name.into(), + version_id: None, + mod_time_unix_nanos: None, + lifecycle_object_info: None, + is_delete_marker: false, + }) + .collect(), + None, + false, + )) + } + async fn heal_object(&self, _: &str, _: &str, _: Option<&str>, _: &HealOpts) -> Result<(HealResultItem, Option)> { + let permit = self.io.clone().acquire_owned().await.expect("fixture I/O permit"); + let namespace = self.namespace.lock().await; + self.provider.commit_open.store(true, Ordering::SeqCst); + let index = self.started.fetch_add(1, Ordering::SeqCst); + if index == 0 { + self.first_started.notify_one(); + self.release_first.notified().await; + } + self.committed.fetch_add(1, Ordering::SeqCst); + self.provider.commit_open.store(false, Ordering::SeqCst); + drop(namespace); + drop(permit); + if index == 0 { + self.first_finished.notify_one(); + } else { + self.second_finished.notify_one(); + } + Ok(( + HealResultItem { + object_size: 1, + ..Default::default() + }, + None, + )) + } +} + +async fn start_fixture( + provider_enabled: bool, + pacing_enabled: bool, + timeout: Duration, +) -> (HealManager, Arc, Arc, Arc) { + let provider = Arc::new(PressureProbe::default()); + let storage = Arc::new(RunningStorage { + provider: provider.clone(), + namespace: Mutex::new(()), + io: Arc::new(Semaphore::new(1)), + first_started: Notify::new(), + release_first: Notify::new(), + first_finished: Notify::new(), + second_finished: Notify::new(), + started: AtomicUsize::new(0), + committed: AtomicUsize::new(0), + }); + let manager = HealManager::new_with_workload_provider( + storage.clone(), + Some(HealConfig { + mainline_throttle_enable: pacing_enabled, + mainline_read_utilization_high_percent: 80, + mainline_write_utilization_high_percent: 80, + mainline_max_sleep: Duration::from_millis(250), + max_concurrent_heals: 1, + ..HealConfig::default() + }), + provider_enabled.then(|| provider.clone() as WorkloadSnapshotProviderRef), + ); + let mut request = bucket_request("running-mainline", HealPriority::High, HealRequestSource::Admin); + request.options.recursive = true; + request.options.timeout = Some(timeout); + let task_id = request.id.clone(); + manager.submit_heal_request(request).await.expect("queue admin heal"); + process_manager_queue_once(&manager).await; + storage.first_started.notified().await; + let task = manager + .active_heals + .lock() + .await + .get(&task_id) + .cloned() + .expect("running task"); + (manager, storage, provider, task) +} + +#[tokio::test(start_paused = true)] +async fn running_mainline_admin_resamples_after_commit_and_yields_without_io_guards() { + let (_manager, storage, provider, _task) = start_fixture(true, true, Duration::from_secs(60)).await; + provider.active.store(100, Ordering::SeqCst); + assert!(storage.provider.commit_open.load(Ordering::SeqCst)); + assert_eq!(storage.committed.load(Ordering::SeqCst), 0); + storage.release_first.notify_one(); + storage.first_finished.notified().await; + tokio::time::timeout(Duration::from_millis(1), provider.high_sampled.notified()) + .await + .expect("running admin heal must re-sample rising pressure before its next object"); + assert_eq!(storage.started.load(Ordering::SeqCst), 1); + assert_eq!( + storage.committed.load(Ordering::SeqCst), + 1, + "in-flight commit must finish despite pressure" + ); + assert_eq!(storage.io.available_permits(), 1, "pacing must release I/O permits"); + assert!(storage.namespace.try_lock().is_ok(), "pacing must not hold the namespace lock"); + tokio::time::advance(Duration::from_millis(250)).await; + storage.second_finished.notified().await; + assert_eq!( + storage.committed.load(Ordering::SeqCst), + 2, + "sustained pressure must still allow bounded maintenance progress" + ); +} + +#[tokio::test(start_paused = true)] +async fn running_mainline_missing_provider_or_disabled_pacing_preserves_progress() { + for (provider_enabled, pacing_enabled) in [(false, true), (true, false)] { + let (_manager, storage, provider, _task) = start_fixture(provider_enabled, pacing_enabled, Duration::from_secs(60)).await; + provider.active.store(100, Ordering::SeqCst); + let before = tokio::time::Instant::now(); + storage.release_first.notify_one(); + storage.second_finished.notified().await; + assert_eq!(storage.committed.load(Ordering::SeqCst), 2); + assert_eq!(tokio::time::Instant::now(), before); + assert_eq!(storage.io.available_permits(), 1); + } +} + +#[tokio::test(start_paused = true)] +async fn running_mainline_cancellation_and_deadline_leave_next_object_unstarted() { + for cancelled in [true, false] { + let (_manager, storage, provider, task) = start_fixture(true, true, Duration::from_millis(100)).await; + provider.active.store(100, Ordering::SeqCst); + storage.release_first.notify_one(); + provider.high_sampled.notified().await; + if cancelled { + task.cancel_token.cancel(); + } else { + tokio::time::advance(Duration::from_millis(100)).await; + } + tokio::time::timeout(Duration::from_secs(1), async { + while matches!(task.get_status().await, HealTaskStatus::Running) { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await + .expect("pacing must not mask cancellation or timeout"); + assert_eq!(storage.started.load(Ordering::SeqCst), 1); + assert_eq!(storage.committed.load(Ordering::SeqCst), 1); + assert_eq!(storage.io.available_permits(), 1); + assert!(storage.namespace.try_lock().is_ok()); + let outcome = task.get_outcome().await; + assert_eq!(outcome.counters.processed, 1); + assert_eq!( + task.get_status().await, + if cancelled { + HealTaskStatus::Cancelled + } else { + HealTaskStatus::Timeout + } + ); + } +} diff --git a/crates/heal/src/heal/mod.rs b/crates/heal/src/heal/mod.rs index 5f17c8cd8..ce2058086 100644 --- a/crates/heal/src/heal/mod.rs +++ b/crates/heal/src/heal/mod.rs @@ -17,6 +17,7 @@ pub mod erasure_healer; pub mod manager; pub mod mrf_queue; pub mod outcome; +pub(crate) mod pacing; pub mod progress; pub(crate) mod replacement_readiness; pub mod resume; diff --git a/crates/heal/src/heal/pacing.rs b/crates/heal/src/heal/pacing.rs new file mode 100644 index 000000000..ff8e7bea7 --- /dev/null +++ b/crates/heal/src/heal/pacing.rs @@ -0,0 +1,238 @@ +// Copyright 2026 RustFS Team +// Licensed under the Apache License, Version 2.0. + +use crate::{Error, Result}; +use rustfs_concurrency::{ + WorkloadAdmissionSnapshotProvider, + workload::{ForegroundPressure, foreground_pressure}, +}; +use std::{sync::Arc, time::Duration}; +use tokio::{sync::Mutex, time::Instant}; +use tokio_util::sync::CancellationToken; + +#[derive(Default)] +struct PacingState { + throttled: bool, + low_since: Option, +} + +pub(crate) enum PacingDecision { + Ready, + Wait(Option), +} + +/// Cooperative pacing for one admin execution, not a storage admission permit. +pub(crate) struct MainlinePacer { + provider: Arc, + read_high: usize, + write_high: usize, + pause: Duration, + state: Mutex, +} + +impl MainlinePacer { + pub(crate) fn new( + provider: Arc, + read_high: usize, + write_high: usize, + pause: Duration, + ) -> Option { + if (read_high == 0 && write_high == 0) || pause.is_zero() { + return None; + } + Some(Self { + provider, + read_high: read_high.min(100), + write_high: write_high.min(100), + pause: pause.min(Duration::from_secs(1)), + state: Mutex::new(PacingState::default()), + }) + } + + /// Fresh, nonblocking decision while the caller owns actual page capacity. + /// A contended pacing latch is conservative, but never awaited here. + pub(crate) fn admission_decision(&self) -> PacingDecision { + let snapshot = self.provider.workload_admission_snapshot(); + let pressure = foreground_pressure(&snapshot, self.read_high, self.write_high); + if pressure.is_none() && self.state.try_lock().is_ok_and(|state| !state.throttled) { + PacingDecision::Ready + } else { + PacingDecision::Wait(pressure) + } + } + + /// Call only between storage operations, with no namespace lock or I/O + /// permit held. The pacing-only mutex serializes starts within this task; + /// each holder waits at most one pause so persistent pressure cannot stop + /// all maintenance progress. Cancellation also interrupts queued waiters. + pub(crate) async fn wait(&self, cancel: &CancellationToken) -> Result<()> { + self.wait_after_admission(cancel, None).await.map(|_| ()) + } + + /// Returns whether this unit paid a bounded pause. That grant permits one + /// unit even if pressure persists when page capacity becomes available. + pub(crate) async fn wait_after_admission( + &self, + cancel: &CancellationToken, + observed: Option, + ) -> Result { + let mut state = tokio::select! { + biased; + _ = cancel.cancelled() => return Err(Error::TaskCancelled), + state = self.state.lock() => state, + }; + if observed.is_some() { + state.throttled = true; + state.low_since = None; + } + let snapshot = self.provider.workload_admission_snapshot(); + let pressure = foreground_pressure(&snapshot, self.read_high, self.write_high); + if pressure.is_some() { + state.throttled = true; + state.low_since = None; + } else if state.throttled { + let low = |high: usize| if high == 0 { 0 } else { (high * 3 / 4).max(1) }; + if foreground_pressure(&snapshot, low(self.read_high), low(self.write_high)).is_none() { + let now = Instant::now(); + let since = state.low_since.get_or_insert(now); + if now.duration_since(*since) >= self.pause.saturating_mul(4) { + state.throttled = false; + state.low_since = None; + } + } else { + state.low_since = None; + } + } + if !state.throttled { + return Ok(false); + } + metrics::counter!( + "rustfs_heal_mainline_throttle_total", + "source" => "admin", + "result" => "delayed", + "reason" => pressure.or(observed).map_or("recovery_window", |pressure| pressure.reason()) + ) + .increment(1); + tokio::select! { + biased; + _ = cancel.cancelled() => Err(Error::TaskCancelled), + _ = tokio::time::sleep(self.pause) => Ok(true), + } + } +} + +#[cfg(test)] +pub(crate) struct TestPressure { + pub(crate) active: std::sync::atomic::AtomicUsize, + pub(crate) sampled: tokio::sync::Notify, + class: rustfs_concurrency::WorkloadClass, +} + +#[cfg(test)] +impl TestPressure { + pub(crate) fn new(class: rustfs_concurrency::WorkloadClass, active: usize) -> Self { + Self { + active: std::sync::atomic::AtomicUsize::new(active), + sampled: tokio::sync::Notify::new(), + class, + } + } +} + +#[cfg(test)] +impl WorkloadAdmissionSnapshotProvider for TestPressure { + fn workload_admission_snapshot(&self) -> rustfs_concurrency::WorkloadAdmissionRegistrySnapshot { + let active = self.active.load(std::sync::atomic::Ordering::SeqCst); + self.sampled.notify_one(); + rustfs_concurrency::WorkloadAdmissionRegistrySnapshot::new(vec![ + rustfs_concurrency::WorkloadAdmissionSnapshot::new(self.class, rustfs_concurrency::AdmissionState::Open).with_counts( + Some(active), + None, + Some(100), + ), + ]) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_concurrency::WorkloadClass; + use std::sync::atomic::Ordering; + + #[tokio::test(start_paused = true)] + async fn running_mainline_hysteresis_uses_configured_watermarks_and_stable_low_window() { + for class in [WorkloadClass::ForegroundRead, WorkloadClass::ForegroundWrite] { + let provider = Arc::new(TestPressure::new(class, 0)); + let pause = Duration::from_millis(250); + let pacer = MainlinePacer::new( + provider.clone(), + if class == WorkloadClass::ForegroundRead { 40 } else { 0 }, + if class == WorkloadClass::ForegroundWrite { 40 } else { 0 }, + pause, + ) + .expect("enabled pacer"); + let cancel = CancellationToken::new(); + let now = Instant::now(); + pacer.wait(&cancel).await.expect("quiet work"); + assert_eq!(Instant::now(), now); + // Low watermark is 30 for the configured high watermark 40. + for utilization in [40, 29, 35, 29, 29, 29, 29] { + provider.active.store(utilization, Ordering::SeqCst); + let before = Instant::now(); + pacer.wait(&cancel).await.expect("bounded maintenance progress"); + assert_eq!(Instant::now() - before, pause); + } + let before = Instant::now(); + pacer.wait(&cancel).await.expect("stable low pressure restores unpaced work"); + assert_eq!(Instant::now(), before); + } + } + + #[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"); + let before = Instant::now(); + pacer.wait(&CancellationToken::new()).await.expect("hard-capped pause"); + assert_eq!(Instant::now() - before, Duration::from_secs(1)); + } + + #[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 cancel_first = CancellationToken::new(); + let first = tokio::spawn({ + let paced = paced.clone(); + let cancel = cancel_first.clone(); + async move { paced.wait(&cancel).await } + }); + provider.sampled.notified().await; + let cancel_second = CancellationToken::new(); + let second = tokio::spawn({ + let paced = paced.clone(); + let cancel = cancel_second.clone(); + async move { paced.wait(&cancel).await } + }); + tokio::task::yield_now().await; + 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 before = Instant::now(); + other_task + .wait(&CancellationToken::new()) + .await + .expect("another task has no inherited latch"); + assert_eq!(Instant::now(), before, "task/set pacing state must not be global"); + cancel_first.cancel(); + assert!(matches!(first.await.expect("sleeping waiter"), Err(Error::TaskCancelled))); + tokio::time::timeout(Duration::from_secs(2), paced.wait(&CancellationToken::new())) + .await + .expect("pacing lock released") + .expect("bounded work after cancellation"); + } +} diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index a1b09c3b7..f1707be24 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -446,6 +446,7 @@ pub struct HealTask { pub cancel_token: tokio_util::sync::CancellationToken, /// Storage layer interface pub storage: Arc, + mainline_pacer: Option>, } impl HealTask { @@ -493,6 +494,7 @@ impl HealTask { task_start_instant: Arc::new(RwLock::new(None)), cancel_token: tokio_util::sync::CancellationToken::new(), storage, + mainline_pacer: None, } } @@ -529,6 +531,18 @@ impl HealTask { task } + pub(crate) fn with_mainline_pacer(mut self, pacer: Option>) -> Self { + self.mainline_pacer = pacer; + self + } + + async fn pace_mainline(&self) -> Result<()> { + if let Some(pacer) = &self.mainline_pacer { + self.await_with_control(pacer.wait(&self.cancel_token)).await?; + } + Ok(()) + } + pub fn metric_type_label(&self) -> &'static str { self.heal_type.kind_label() } @@ -933,24 +947,32 @@ impl HealTask { }); self.emit_trace_task_state("started", Duration::ZERO, None); - let result = match &self.heal_type { - HealType::Cluster => self.heal_cluster().await, - HealType::Object { - bucket, - object, - version_id, - } => self.heal_object(bucket, object, version_id.as_deref()).await, - HealType::Bucket { bucket } => self.heal_bucket(bucket).await, - HealType::Prefix { bucket, prefix } => self.heal_prefix(bucket, prefix).await, + let result = async { + if self.heal_type.is_per_object() { + self.pace_mainline().await?; + } + match &self.heal_type { + HealType::Cluster => self.heal_cluster().await, + HealType::Object { + bucket, + object, + version_id, + } => self.heal_object(bucket, object, version_id.as_deref()).await, + HealType::Bucket { bucket } => self.heal_bucket(bucket).await, + HealType::Prefix { bucket, prefix } => self.heal_prefix(bucket, prefix).await, - HealType::Metadata { bucket, object } => self.heal_metadata(bucket, object).await, - HealType::ECDecode { - bucket, - object, - version_id, - } => self.heal_ec_decode(bucket, object, version_id.as_deref()).await, - HealType::ErasureSet { buckets, set_disk_id } => self.heal_erasure_set(buckets.clone(), set_disk_id.clone()).await, - }; + HealType::Metadata { bucket, object } => self.heal_metadata(bucket, object).await, + HealType::ECDecode { + bucket, + object, + version_id, + } => self.heal_ec_decode(bucket, object, version_id.as_deref()).await, + HealType::ErasureSet { buckets, set_disk_id } => { + self.heal_erasure_set(buckets.clone(), set_disk_id.clone()).await + } + } + } + .await; #[cfg(test)] pause_outcome_finish(&self.id).await; diff --git a/crates/heal/src/heal/task/heal_bucket.rs b/crates/heal/src/heal/task/heal_bucket.rs index c6225c844..cb8d230b4 100644 --- a/crates/heal/src/heal/task/heal_bucket.rs +++ b/crates/heal/src/heal/task/heal_bucket.rs @@ -34,6 +34,7 @@ fn unavailable_recreate_error(result: &HealResultItem, opts: &HealOpts) -> Optio impl HealTask { pub(super) async fn heal_bucket(&self, bucket: &str) -> Result<()> { + self.pace_mainline().await?; debug!( target: "rustfs::heal::task", event = EVENT_HEAL_BUCKET_STAGE, @@ -308,6 +309,7 @@ impl HealTask { self.check_control_flags().await?; let mut listing_attempt = 0; let (objects, next_token, is_truncated) = loop { + self.pace_mainline().await?; let page = if let Some(set_disk_id) = set_disk_id.as_deref() { self.await_with_control(self.storage.list_versions_for_heal_page_disk_walk( set_disk_id, @@ -362,6 +364,7 @@ impl HealTask { let mut retry = Vec::with_capacity(pending.len()); for item in pending { self.check_control_flags().await?; + self.pace_mainline().await?; let mut telemetry_unknown = false; let object = item.name.as_str(); let identity = diff --git a/crates/heal/src/heal/task/heal_erasure_set.rs b/crates/heal/src/heal/task/heal_erasure_set.rs index cc1d638ba..446cb07d7 100644 --- a/crates/heal/src/heal/task/heal_erasure_set.rs +++ b/crates/heal/src/heal/task/heal_erasure_set.rs @@ -422,7 +422,8 @@ impl HealTask { self.source, ) .with_replacement_targets(self.heal_endpoints.clone(), is_auto_replacement.then(|| self.id.clone())) - .with_replacement_identity_fence(replacement_target_identities.clone()); + .with_replacement_identity_fence(replacement_target_identities.clone()) + .with_mainline_pacer(self.mainline_pacer.clone()); { let mut progress = self.progress.write().await; diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index 31fe6124c..55e40cc31 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -280,10 +280,10 @@ Heal knobs are environment-only and read by `HealConfig::default` (`crates/heal/ | `RUSTFS_HEAL_SET_BULKHEAD_ENABLE` | `true` (`DEFAULT_HEAL_SET_BULKHEAD_ENABLE`) | Per-set bulkhead scheduling. | | `RUSTFS_HEAL_PAGE_PARALLEL_ENABLE` | `true` (`DEFAULT_HEAL_PAGE_PARALLEL_ENABLE`) | Page-level parallel object healing during erasure-set repair. | | `RUSTFS_HEAL_PAGE_OBJECT_CONCURRENCY` | `8` (`DEFAULT_HEAL_PAGE_OBJECT_CONCURRENCY`) | Concurrent object heals within one erasure-set page. Forced to `1` when page parallelism is off, for `Deep` scan mode, and for `AutoHeal`-sourced requests (`ErasureSetHealer::effective_heal_page_object_concurrency_for_source`). | -| `RUSTFS_HEAL_MAINLINE_THROTTLE_ENABLE` | `true` (`DEFAULT_HEAL_MAINLINE_THROTTLE_ENABLE`) | Pause best-effort heal task starts while foreground I/O is saturated. | -| `RUSTFS_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT` | `80` (`DEFAULT_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT`, capped at 100) | Foreground read-permit utilization at which heal starts pause. | -| `RUSTFS_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT` | `80` (`DEFAULT_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT`, capped at 100) | Foreground write utilization at which heal starts pause. | -| `RUSTFS_HEAL_MAINLINE_MAX_SLEEP_MS` | `250` (`DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS`) | Recheck delay after deferring heal starts for foreground pressure. | +| `RUSTFS_HEAL_MAINLINE_THROTTLE_ENABLE` | `true` (`DEFAULT_HEAL_MAINLINE_THROTTLE_ENABLE`) | Defer best-effort starts and cooperatively pace running admin heal at safe work boundaries. | +| `RUSTFS_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT` | `80` (`DEFAULT_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT`, capped at 100) | Read-utilization high watermark for start admission and running admin pacing; zero disables this class. | +| `RUSTFS_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT` | `80` (`DEFAULT_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT`, capped at 100) | Write-utilization high watermark for start admission and running admin pacing; zero disables this class. | +| `RUSTFS_HEAL_MAINLINE_MAX_SLEEP_MS` | `250` (`DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS`) | Start recheck interval; running admin waits cap each pacing-gate holder at 1000 ms. Zero disables running pacing. | | `RUSTFS_HEAL_OVERLAP_POLICY` | `merge` (`DEFAULT_HEAL_OVERLAP_POLICY`) | `merge` dedups an admin heal start that overlaps a running or queued heal; `minio_error` returns a typed already-running / overlapping-paths rejection like madmin. | | `RUSTFS_HEAL_MRF_ENABLE` | `true` (`DEFAULT_HEAL_MRF_ENABLE`) | MRF intent pipeline: error paths deliver repair intents to the heal runtime and unconsumed intents replay from the durable journal after restart. | | `RUSTFS_HEAL_MRF_QUEUE_SIZE` | `100000` (`DEFAULT_HEAL_MRF_QUEUE_SIZE`) | MRF in-memory queue capacity. | @@ -291,6 +291,16 @@ Heal knobs are environment-only and read by `HealConfig::default` (`crates/heal/ | `RUSTFS_HEAL_MRF_REPLAY_BATCH` | `256` (`DEFAULT_HEAL_MRF_REPLAY_BATCH`) | Intents per replay push round. | | `RUSTFS_HEAL_DANGLING_DELETE_GRACE_SECS` | `3600` (`DEFAULT_HEAL_DANGLING_DELETE_GRACE_SECS`, `crates/ecstore/src/set_disk/core/io_primitives.rs`) | A recently modified object is never deleted as dangling inside this window; `0` disables the grace window. | +### Running admin heal pacing + +The manager passes its existing workload provider and a configuration snapshot into each admin execution. Bucket/prefix listing and object boundaries resample foreground pressure; erasure-set page workers also resample after earlier work releases page capacity. `High`, `Urgent`, and `force_start` do not exempt ordinary admin execution from this runtime pacing. The existing start-time bypass and overlap-control meanings are unchanged. + +Each execution has its own pacing latch, with no new global manager or cross-set pacing lock. The low watermark for each enabled class is `max(1, floor(high * 3 / 4))`: the default high watermark 80 therefore recovers below 60. High pressure latches pacing, and intermediate pressure resets the recovery window. Unpaced starts resume after sampled pressure remains below the low watermarks for four pause intervals, normally one second. While pressure persists, a pacing-gate holder waits only one interval, at most one second, then permits maintenance to continue. Concurrent page waiters serialize through this task-local gate; queue waiting still counts against the existing task execution timeout. + +The pacing gate holds neither namespace locks nor I/O/page permits while sleeping. At final page admission, each real permit acquisition gets a fresh, nonblocking pressure decision. Low-pressure work keeps that permit; only a unit that needs a pause releases capacity to wait. A unit that has completed one bounded pause may proceed despite persistent pressure, which supplies minimum maintenance progress without an endless acquire/pause loop. Existing object operations and commit tails are not interrupted because pressure rose. Cancellation and deadlines remain interruptible, and disabling pacing cannot bypass the global, per-set or page-concurrency hard caps. The existing `RUSTFS_HEAL_MAINLINE_THROTTLE_ENABLE=false` setting is the operational opt-out for newly created executions; no additional request override is introduced. + +A missing provider, zero pause, or both class thresholds set to zero preserves unpaced execution. Missing counts follow the existing shared pressure interpreter; they are observations, not health, quorum or resource-ownership proof. The current provider exposes node-level workload classes, so this does not claim independent per-set foreground measurements or a hard global resource budget. Runtime waits increment `rustfs_heal_mainline_throttle_total` with `source=admin`, `result=delayed`, and a foreground-pressure or `recovery_window` reason. Real p99/throughput protection requires the separate W20 fixed-load ABBA measurements. + ## Deliberate non-parity with MinIO These differences from MinIO are design decisions, recorded so they are not re-filed as gaps.