feat(heal): pace running admin work at safe boundaries (#7255)

crates/heal/src/heal/erasure_healer.rs

@@ -108,6 +108,144 @@ pub struct ErasureSetHealer {
     target_endpoints: Arc<[String]>,
     replacement_task_id: Option<String>,
     replacement_target_identities: Option<Arc<[ReplacementTargetIdentity]>>,
+    mainline_pacer: Option<Arc<super::pacing::MainlinePacer>>,
+}
+
+async fn acquire_page_permit(
+    semaphore: Arc<Semaphore>,
+    pacer: Option<&super::pacing::MainlinePacer>,
+    cancel: &tokio_util::sync::CancellationToken,
+) -> Result<tokio::sync::OwnedSemaphorePermit> {
+    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<Arc<super::pacing::MainlinePacer>>) -> Self {
+        self.mainline_pacer = pacer;
+        self
+    }
+
     pub(crate) fn with_replacement_targets(
         mut self,
         mut target_endpoints: Vec<String>,
@@ -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,

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,
 }
 

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);

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]

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<PressureProbe>,
+    namespace: Mutex<()>,
+    io: Arc<Semaphore>,
+    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<Option<HealObjectInfo>> {
+        Ok(None)
+    }
+    async fn ec_decode_rebuild(&self, _: &str, _: &str) -> Result<Vec<u8>> {
+        Ok(Vec::new())
+    }
+    async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>> {
+        Ok(Some(BucketInfo {
+            name: bucket.into(),
+            ..Default::default()
+        }))
+    }
+    async fn list_buckets(&self) -> Result<Vec<BucketInfo>> {
+        Ok(Vec::new())
+    }
+    async fn object_exists(&self, _: &str, _: &str) -> Result<bool> {
+        Ok(true)
+    }
+    async fn heal_bucket(&self, _: &str, _: &HealOpts) -> Result<HealResultItem> {
+        Ok(HealResultItem::default())
+    }
+    async fn heal_format(&self, _: bool) -> Result<(HealResultItem, Option<Error>)> {
+        Ok((HealResultItem::default(), None))
+    }
+    async fn get_disk_for_resume(&self, _: &str) -> Result<DiskStore> {
+        Err(Error::other("no resume disk in bucket fixture"))
+    }
+    async fn list_objects_for_heal_page(
+        &self,
+        _: &str,
+        _: &str,
+        _: Option<&str>,
+        _: bool,
+    ) -> Result<(Vec<HealListItem>, Option<String>, 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<Error>)> {
+        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<RunningStorage>, Arc<PressureProbe>, Arc<HealTask>) {
+    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
+            }
+        );
+    }
+}

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;

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<Instant>,
+}
+
+pub(crate) enum PacingDecision {
+    Ready,
+    Wait(Option<ForegroundPressure>),
+}
+
+/// Cooperative pacing for one admin execution, not a storage admission permit.
+pub(crate) struct MainlinePacer {
+    provider: Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>,
+    read_high: usize,
+    write_high: usize,
+    pause: Duration,
+    state: Mutex<PacingState>,
+}
+
+impl MainlinePacer {
+    pub(crate) fn new(
+        provider: Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>,
+        read_high: usize,
+        write_high: usize,
+        pause: Duration,
+    ) -> Option<Self> {
+        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<ForegroundPressure>,
+    ) -> Result<bool> {
+        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");
+    }
+}

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<dyn HealStorageAPI>,
+    mainline_pacer: Option<Arc<super::pacing::MainlinePacer>>,
 }
 
 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<Arc<super::pacing::MainlinePacer>>) -> 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;

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 =

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;