fix(heal): add backpressure to repair admission (#3900)

Cargo.lock

@@ -9449,6 +9449,7 @@ dependencies = [
  "http 1.4.2",
  "metrics",
  "rustfs-common",
+ "rustfs-concurrency",
  "rustfs-config",
  "rustfs-ecstore",
  "rustfs-madmin",

crates/common/src/heal_channel.rs

@@ -273,6 +273,7 @@ pub enum HealRequestSource {
     Admin,
     Scanner,
     AutoHeal,
+    ReadRepair,
 }
 
 impl HealRequestSource {
@@ -282,6 +283,7 @@ impl HealRequestSource {
             Self::Admin => "admin",
             Self::Scanner => "scanner",
             Self::AutoHeal => "auto_heal",
+            Self::ReadRepair => "read_repair",
         }
     }
 }
@@ -680,6 +682,7 @@ mod tests {
         assert_eq!(HealRequestSource::Admin.as_str(), "admin");
         assert_eq!(HealRequestSource::AutoHeal.as_str(), "auto_heal");
         assert_eq!(HealRequestSource::Internal.as_str(), "internal");
+        assert_eq!(HealRequestSource::ReadRepair.as_str(), "read_repair");
 
         let request = HealChannelRequest::default();
         assert_eq!(request.source, HealRequestSource::Internal);

crates/config/src/constants/heal.rs

@@ -133,6 +133,17 @@ pub const ENV_HEAL_SET_BULKHEAD_ENABLE: &str = "RUSTFS_HEAL_SET_BULKHEAD_ENABLE"
 /// Environment variable that toggles page-level parallel object healing for erasure-set repair.
 pub const ENV_HEAL_PAGE_PARALLEL_ENABLE: &str = "RUSTFS_HEAL_PAGE_PARALLEL_ENABLE";
 
+/// Environment variable that toggles foreground read pressure gating for background heal work.
+pub const ENV_HEAL_MAINLINE_THROTTLE_ENABLE: &str = "RUSTFS_HEAL_MAINLINE_THROTTLE_ENABLE";
+
+/// Environment variable that controls the foreground read permit utilization percentage
+/// at which background heal work pauses starting new tasks.
+pub const ENV_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT: &str = "RUSTFS_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT";
+
+/// Environment variable that controls how soon the heal scheduler rechecks foreground
+/// pressure after delaying background work.
+pub const ENV_HEAL_MAINLINE_MAX_SLEEP_MS: &str = "RUSTFS_HEAL_MAINLINE_MAX_SLEEP_MS";
+
 /// Default behavior is to merge duplicate low-priority requests.
 pub const DEFAULT_HEAL_LOW_PRIORITY_MERGE_ENABLE: bool = true;
 
@@ -150,3 +161,12 @@ pub const DEFAULT_HEAL_SET_BULKHEAD_ENABLE: bool = true;
 
 /// Default behavior is to keep erasure-set page parallelism enabled.
 pub const DEFAULT_HEAL_PAGE_PARALLEL_ENABLE: bool = true;
+
+/// Default behavior is to pause best-effort heal task starts when foreground reads are saturated.
+pub const DEFAULT_HEAL_MAINLINE_THROTTLE_ENABLE: bool = true;
+
+/// Default foreground read permit utilization threshold for pausing best-effort heal task starts.
+pub const DEFAULT_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT: usize = 80;
+
+/// Default foreground pressure recheck delay for heal scheduler, in milliseconds.
+pub const DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS: u64 = 250;

crates/ecstore/src/set_disk.rs

@@ -77,7 +77,10 @@ use http::HeaderMap;
 use md5::{Digest as Md5Digest, Md5};
 use rand::{Rng, seq::SliceRandom};
 use regex::Regex;
-use rustfs_common::heal_channel::{DriveState, HealChannelPriority, HealItemType, HealOpts, HealScanMode, send_heal_disk};
+use rustfs_common::heal_channel::{
+    DriveState, HealAdmissionResult, HealChannelPriority, HealItemType, HealOpts, HealRequestSource, HealScanMode,
+    send_heal_disk, send_heal_request_with_admission,
+};
 use rustfs_config::MI_B;
 use rustfs_filemeta::{
     FileInfo, FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo,

crates/ecstore/src/set_disk/heal.rs

@@ -16,6 +16,10 @@ use super::*;
 use crate::storage_api_contracts::NamespaceLocking as _;
 use rustfs_config::{DEFAULT_OBJECT_ZERO_COPY_ENABLE, ENV_OBJECT_ZERO_COPY_ENABLE};
 
+const LOG_COMPONENT_ECSTORE: &str = "ecstore";
+const LOG_SUBSYSTEM_HEAL: &str = "heal";
+const EVENT_HEAL_OBJECT_RENAME: &str = "heal_object_rename";
+
 impl SetDisks {
     #[tracing::instrument(skip(self, opts), fields(bucket = %bucket, object = %object, version_id = %version_id))]
     pub(super) async fn heal_object(
@@ -513,8 +517,11 @@ impl SetDisks {
                             }
                         }
                         // Rename from tmp location to the actual location.
+                        let mut rename_attempts = 0usize;
+                        let mut rename_successes = 0usize;
                         for (index, outdated_disk) in out_dated_disks.iter().enumerate() {
                             if let Some(disk) = outdated_disk {
+                                rename_attempts += 1;
                                 // record the index of the updated disks
                                 parts_metadata[index].erasure.index = index + 1;
                                 // Attempt a rename now from healed data to final location.
@@ -525,14 +532,22 @@ impl SetDisks {
                                     .await;
 
                                 if let Err(err) = &rename_result {
-                                    self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id)
+                                    warn!(
-                                        .await
+                                        event = EVENT_HEAL_OBJECT_RENAME,
-                                        .map_err(DiskError::other)?;
+                                        component = LOG_COMPONENT_ECSTORE,
+                                        subsystem = LOG_SUBSYSTEM_HEAL,
+                                        bucket,
+                                        object,
+                                        version_id,
+                                        disk_index = index,
+                                        endpoint = %disk.endpoint(),
+                                        tmp_id,
+                                        result = "failed",
+                                        error = %err,
+                                        "Heal object rename failed"
+                                    );
                                 } else {
-                                    self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id)
+                                    rename_successes += 1;
-                                        .await
-                                        .map_err(DiskError::other)?;
-
                                     if parts_metadata[index].is_remote() {
                                         let rm_data_dir = parts_metadata[index].data_dir.unwrap().to_string();
 
@@ -560,6 +575,16 @@ impl SetDisks {
                                 }
                             }
                         }
+                        self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id)
+                            .await
+                            .map_err(DiskError::other)?;
+
+                        if rename_attempts > 0 && rename_successes == 0 {
+                            return Ok((
+                                result,
+                                Some(DiskError::other(format!("all healed data rename attempts failed for {bucket}/{object}"))),
+                            ));
+                        }
 
                         record_capacity_scope_if_needed(None, &out_dated_disks);
 

crates/ecstore/src/set_disk/read.rs

@@ -17,11 +17,224 @@ use crate::get_diagnostics::{
     GET_OBJECT_PATH_CODEC_STREAMING, GET_STAGE_DECODE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP, GetObjectFailureReason,
     classify_disk_error, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path,
 };
+use metrics::counter;
 use rustfs_config::{DEFAULT_OBJECT_ZERO_COPY_ENABLE, ENV_OBJECT_ZERO_COPY_ENABLE};
-use std::future::Future;
+use std::{
+    collections::HashMap,
+    future::Future,
+    pin::Pin,
+    sync::OnceLock,
+    time::{Duration, Instant},
+};
 use tokio::io::AsyncRead;
+use tokio::sync::RwLock;
 use tokio::task::JoinSet;
 
+const READ_REPAIR_HEAL_DEDUP_TTL: Duration = Duration::from_secs(60);
+const READ_REPAIR_HEAL_DEDUP_MAX_ENTRIES: usize = 4096;
+
+static READ_REPAIR_HEAL_CACHE: OnceLock<RwLock<HashMap<ReadRepairHealCacheKey, Instant>>> = OnceLock::new();
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+struct ReadRepairHealCacheKey {
+    bucket: String,
+    object: String,
+    version_id: Option<String>,
+    pool_index: usize,
+    set_index: usize,
+}
+
+impl ReadRepairHealCacheKey {
+    fn new(bucket: &str, object: &str, version_id: Option<&str>, pool_index: usize, set_index: usize) -> Self {
+        Self {
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            version_id: version_id.filter(|value| !value.is_empty()).map(str::to_string),
+            pool_index,
+            set_index,
+        }
+    }
+}
+
+fn resolved_read_repair_version_id(fi: &FileInfo, requested_version_id: Option<&str>) -> Option<String> {
+    fi.version_id
+        .as_ref()
+        .map(ToString::to_string)
+        .or_else(|| requested_version_id.filter(|value| !value.is_empty()).map(str::to_string))
+}
+
+async fn reserve_read_repair_heal(
+    bucket: &str,
+    object: &str,
+    version_id: Option<&str>,
+    pool_index: usize,
+    set_index: usize,
+) -> Option<ReadRepairHealCacheKey> {
+    let key = ReadRepairHealCacheKey::new(bucket, object, version_id, pool_index, set_index);
+    let now = Instant::now();
+    let cache = READ_REPAIR_HEAL_CACHE.get_or_init(|| RwLock::new(HashMap::new()));
+
+    {
+        let cache = cache.read().await;
+        if cache
+            .get(&key)
+            .is_some_and(|seen_at| now.saturating_duration_since(*seen_at) <= READ_REPAIR_HEAL_DEDUP_TTL)
+        {
+            return None;
+        }
+    }
+
+    let mut cache = cache.write().await;
+    if cache
+        .get(&key)
+        .is_some_and(|seen_at| now.saturating_duration_since(*seen_at) <= READ_REPAIR_HEAL_DEDUP_TTL)
+    {
+        return None;
+    }
+
+    if cache.len() >= READ_REPAIR_HEAL_DEDUP_MAX_ENTRIES {
+        cache.retain(|_, seen_at| now.saturating_duration_since(*seen_at) <= READ_REPAIR_HEAL_DEDUP_TTL);
+    }
+    if cache.len() >= READ_REPAIR_HEAL_DEDUP_MAX_ENTRIES
+        && let Some(oldest_key) = cache.iter().min_by_key(|(_, seen_at)| **seen_at).map(|(key, _)| key.clone())
+    {
+        cache.remove(&oldest_key);
+    }
+    cache.insert(key.clone(), now);
+    Some(key)
+}
+
+async fn release_read_repair_heal_reservation(key: &ReadRepairHealCacheKey) {
+    if let Some(cache) = READ_REPAIR_HEAL_CACHE.get() {
+        cache.write().await.remove(key);
+    }
+}
+
+fn record_read_repair_dedup(reason: &'static str) {
+    counter!("rustfs_heal_read_repair_dedup_total", "reason" => reason.to_string()).increment(1);
+}
+
+enum ReadRepairAdmissionOutcome {
+    Response(HealAdmissionResult),
+    Failed(String),
+}
+
+type ReadRepairAdmissionFuture = Pin<Box<dyn Future<Output = ReadRepairAdmissionOutcome> + Send>>;
+type ReadRepairAdmissionSubmitter = fn(rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture;
+
+fn send_read_repair_heal_request(request: rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture {
+    Box::pin(async {
+        match send_heal_request_with_admission(request).await {
+            Ok(result) => ReadRepairAdmissionOutcome::Response(result),
+            Err(err) => ReadRepairAdmissionOutcome::Failed(err),
+        }
+    })
+}
+
+async fn submit_read_repair_heal(
+    bucket: &str,
+    object: &str,
+    version_id: Option<&str>,
+    pool_index: usize,
+    set_index: usize,
+    part_number: Option<usize>,
+    reason: &'static str,
+) {
+    submit_read_repair_heal_with_submitter(
+        bucket,
+        object,
+        version_id,
+        pool_index,
+        set_index,
+        part_number,
+        reason,
+        send_read_repair_heal_request,
+    )
+    .await;
+}
+
+async fn submit_read_repair_heal_with_submitter(
+    bucket: &str,
+    object: &str,
+    version_id: Option<&str>,
+    pool_index: usize,
+    set_index: usize,
+    part_number: Option<usize>,
+    reason: &'static str,
+    submitter: ReadRepairAdmissionSubmitter,
+) {
+    let Some(dedup_key) = reserve_read_repair_heal(bucket, object, version_id, pool_index, set_index).await else {
+        record_read_repair_dedup("duplicate");
+        debug!(
+            bucket,
+            object, part_number, pool_index, set_index, reason, "Skipped duplicate read-repair heal request"
+        );
+        return;
+    };
+
+    let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
+        bucket.to_string(),
+        Some(object.to_string()),
+        false,
+        Some(HealChannelPriority::Normal),
+        Some(pool_index),
+        Some(set_index),
+    );
+    request.source = HealRequestSource::ReadRepair;
+    request.object_version_id = version_id.filter(|value| !value.is_empty()).map(str::to_string);
+    request.recreate_missing = Some(true);
+
+    let request_id = request.id.clone();
+    let bucket = bucket.to_string();
+    let object = object.to_string();
+    tokio::spawn(async move {
+        match submitter(request).await {
+            ReadRepairAdmissionOutcome::Response(result) if result.is_admitted() => {
+                debug!(
+                    bucket,
+                    object,
+                    part_number,
+                    pool_index,
+                    set_index,
+                    request_id,
+                    reason,
+                    admission = result.result_label(),
+                    "Read-repair heal request admitted"
+                );
+            }
+            ReadRepairAdmissionOutcome::Response(result) => {
+                release_read_repair_heal_reservation(&dedup_key).await;
+                debug!(
+                    bucket,
+                    object,
+                    part_number,
+                    pool_index,
+                    set_index,
+                    request_id,
+                    reason,
+                    admission = result.result_label(),
+                    drop_reason = result.reason_label(),
+                    "Read-repair heal request not admitted"
+                );
+            }
+            ReadRepairAdmissionOutcome::Failed(err) => {
+                release_read_repair_heal_reservation(&dedup_key).await;
+                debug!(
+                    bucket,
+                    object,
+                    part_number,
+                    pool_index,
+                    set_index,
+                    request_id,
+                    reason,
+                    error = %err,
+                    "Read-repair heal request could not be submitted"
+                );
+            }
+        }
+    });
+}
+
 async fn collect_read_multiple_results<F>(
     tasks: Vec<F>,
     read_quorum: usize,
@@ -709,16 +922,17 @@ impl SetDisks {
 
         let fi = Self::pick_valid_fileinfo(&parts_metadata, mot_time, etag, read_quorum as usize)?;
         if errs.iter().any(|err| err.is_some()) {
-            let _ =
+            let version_id = resolved_read_repair_version_id(&fi, opts.version_id.as_deref());
-                rustfs_common::heal_channel::send_heal_request(rustfs_common::heal_channel::create_heal_request_with_options(
+            submit_read_repair_heal(
-                    fi.volume.to_string(),             // bucket
+                &fi.volume,
-                    Some(fi.name.to_string()),         // object_prefix
+                &fi.name,
-                    false,                             // force_start
+                version_id.as_deref(),
-                    Some(HealChannelPriority::Normal), // priority
+                self.pool_index,
-                    Some(self.pool_index),             // pool_index
+                self.set_index,
-                    Some(self.set_index),              // set_index
+                None,
-                ))
+                "metadata_read_error",
-                .await;
+            )
+            .await;
         } else if use_metadata_cache {
             self.cache_get_object_fileinfo(bucket, object, &fi, &parts_metadata, &op_online_disks, read_quorum as usize)
                 .await;
@@ -1010,7 +1224,7 @@ impl SetDisks {
             let available_shards = nil_count;
             let missing_shards = total_shards - available_shards;
 
-            info!(
+            debug!(
                 bucket,
                 object,
                 part_number,
@@ -1024,7 +1238,7 @@ impl SetDisks {
 
             if missing_shards > 0 && available_shards >= erasure.data_shards {
                 // We have missing shards but enough to read - trigger background heal
-                info!(
+                debug!(
                     bucket,
                     object,
                     part_number,
@@ -1034,27 +1248,17 @@ impl SetDisks {
                     set_index,
                     "Detected missing shards during read, triggering background heal"
                 );
-                if let Err(e) =
+                let version_id = fi.version_id.as_ref().map(ToString::to_string);
-                    rustfs_common::heal_channel::send_heal_request(rustfs_common::heal_channel::create_heal_request_with_options(
+                submit_read_repair_heal(
-                        bucket.to_string(),
+                    bucket,
-                        Some(object.to_string()),
+                    object,
-                        false,
+                    version_id.as_deref(),
-                        Some(HealChannelPriority::Normal),
+                    pool_index,
-                        Some(pool_index),
+                    set_index,
-                        Some(set_index),
+                    Some(part_number),
-                    ))
+                    "missing_shards",
-                    .await
+                )
-                {
+                .await;
-                    warn!(
-                        bucket,
-                        object,
-                        part_number,
-                        error = %e,
-                        "Failed to enqueue heal request for missing shards"
-                    );
-                } else {
-                    warn!(bucket, object, part_number, "Successfully enqueued heal request for missing shards");
-                }
             }
 
             // debug!(
@@ -1079,27 +1283,24 @@ impl SetDisks {
                 if written == part_length {
                     match de_err {
                         DiskError::FileNotFound | DiskError::FileCorrupt => {
-                            error!("erasure.decode err 111 {:?}", &de_err);
+                            debug!(
-                            if let Err(e) = rustfs_common::heal_channel::send_heal_request(
+                                bucket,
-                                rustfs_common::heal_channel::create_heal_request_with_options(
+                                object,
-                                    bucket.to_string(),
+                                part_number,
-                                    Some(object.to_string()),
+                                error = ?de_err,
-                                    false,
+                                "Recoverable decode error triggered read repair"
-                                    Some(HealChannelPriority::Normal),
+                            );
-                                    Some(pool_index),
+                            let version_id = fi.version_id.as_ref().map(ToString::to_string);
-                                    Some(set_index),
+                            submit_read_repair_heal(
-                                ),
+                                bucket,
+                                object,
+                                version_id.as_deref(),
+                                pool_index,
+                                set_index,
+                                Some(part_number),
+                                "decode_error",
                             )
-                            .await
+                            .await;
-                            {
-                                warn!(
-                                    bucket,
-                                    object,
-                                    part_number,
-                                    error = %e,
-                                    "Failed to enqueue heal request after decode error"
-                                );
-                            }
                             has_err = false;
                         }
                         _ => {}
@@ -1234,25 +1435,18 @@ impl SetDisks {
 
         let total_shards = erasure.data_shards + erasure.parity_shards;
         let missing_shards = total_shards.saturating_sub(available_shards);
-        if missing_shards > 0
+        if missing_shards > 0 {
-            && let Err(e) =
+            let version_id = fi.version_id.as_ref().map(ToString::to_string);
-                rustfs_common::heal_channel::send_heal_request(rustfs_common::heal_channel::create_heal_request_with_options(
+            submit_read_repair_heal(
-                    bucket.to_string(),
-                    Some(object.to_string()),
-                    false,
-                    Some(HealChannelPriority::Normal),
-                    Some(pool_index),
-                    Some(set_index),
-                ))
-                .await
-        {
-            warn!(
                 bucket,
                 object,
-                part_number,
+                version_id.as_deref(),
-                error = %e,
+                pool_index,
-                "Failed to enqueue heal request for missing shards"
+                set_index,
-            );
+                Some(part_number),
+                "missing_shards_streaming",
+            )
+            .await;
         }
 
         let source = erasure_coding::decode::ParallelReader::new_with_metrics_path(
@@ -1285,6 +1479,27 @@ fn is_get_object_metadata_cache_request_eligible(bucket: &str, opts: &ObjectOpti
 #[cfg(test)]
 mod metadata_cache_tests {
     use super::*;
+    use rustfs_common::heal_channel::HealAdmissionDropReason;
+    use serial_test::serial;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    static SLOW_READ_REPAIR_SUBMITTER_CALLS: AtomicUsize = AtomicUsize::new(0);
+    static DROPPED_READ_REPAIR_SUBMITTER_CALLS: AtomicUsize = AtomicUsize::new(0);
+
+    fn slow_read_repair_submitter(_request: rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture {
+        SLOW_READ_REPAIR_SUBMITTER_CALLS.fetch_add(1, Ordering::Relaxed);
+        Box::pin(async {
+            tokio::time::sleep(Duration::from_millis(250)).await;
+            ReadRepairAdmissionOutcome::Response(HealAdmissionResult::Accepted)
+        })
+    }
+
+    fn dropped_read_repair_submitter(_request: rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture {
+        DROPPED_READ_REPAIR_SUBMITTER_CALLS.fetch_add(1, Ordering::Relaxed);
+        Box::pin(async {
+            ReadRepairAdmissionOutcome::Response(HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped))
+        })
+    }
 
     async fn new_metadata_cache_test_set() -> Arc<SetDisks> {
         SetDisks::new(
@@ -1373,6 +1588,115 @@ mod metadata_cache_tests {
         assert!(!is_get_object_metadata_cache_request_eligible("bucket", &opts, true));
     }
 
+    #[tokio::test]
+    async fn read_repair_heal_dedupes_same_object_version() {
+        let bucket = format!("bucket-{}", Uuid::new_v4());
+
+        assert!(reserve_read_repair_heal(&bucket, "object", None, 0, 0).await.is_some());
+        assert!(reserve_read_repair_heal(&bucket, "object", None, 0, 0).await.is_none());
+        assert!(
+            reserve_read_repair_heal(&bucket, "object", Some("version-2"), 0, 0)
+                .await
+                .is_some()
+        );
+        assert!(reserve_read_repair_heal(&bucket, "object", None, 0, 1).await.is_some());
+    }
+
+    #[tokio::test]
+    async fn read_repair_heal_reservation_can_be_released_after_rejection() {
+        let bucket = format!("bucket-{}", Uuid::new_v4());
+        let key = reserve_read_repair_heal(&bucket, "object", None, 0, 0)
+            .await
+            .expect("first reservation should be accepted");
+
+        assert!(reserve_read_repair_heal(&bucket, "object", None, 0, 0).await.is_none());
+        release_read_repair_heal_reservation(&key).await;
+        assert!(reserve_read_repair_heal(&bucket, "object", None, 0, 0).await.is_some());
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn submit_read_repair_heal_does_not_wait_for_slow_admission() {
+        SLOW_READ_REPAIR_SUBMITTER_CALLS.store(0, Ordering::Relaxed);
+        let bucket = format!("bucket-{}", Uuid::new_v4());
+        let started = Instant::now();
+
+        submit_read_repair_heal_with_submitter(
+            &bucket,
+            "object",
+            None,
+            0,
+            0,
+            Some(1),
+            "missing_shards",
+            slow_read_repair_submitter,
+        )
+        .await;
+
+        assert!(
+            started.elapsed() < Duration::from_millis(50),
+            "read-repair submission should not wait for admission response"
+        );
+
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while SLOW_READ_REPAIR_SUBMITTER_CALLS.load(Ordering::Relaxed) == 0 {
+                tokio::time::sleep(Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("background read-repair submitter should be called");
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn submit_read_repair_heal_releases_reservation_after_policy_drop() {
+        DROPPED_READ_REPAIR_SUBMITTER_CALLS.store(0, Ordering::Relaxed);
+        let bucket = format!("bucket-{}", Uuid::new_v4());
+
+        submit_read_repair_heal_with_submitter(
+            &bucket,
+            "object",
+            None,
+            0,
+            0,
+            Some(1),
+            "missing_shards",
+            dropped_read_repair_submitter,
+        )
+        .await;
+
+        let released_key = tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                if let Some(key) = reserve_read_repair_heal(&bucket, "object", None, 0, 0).await {
+                    break key;
+                }
+                tokio::time::sleep(Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("read-repair reservation should be released after policy drop");
+
+        assert_eq!(DROPPED_READ_REPAIR_SUBMITTER_CALLS.load(Ordering::Relaxed), 1);
+        release_read_repair_heal_reservation(&released_key).await;
+    }
+
+    #[test]
+    fn resolved_read_repair_version_prefers_selected_fileinfo_version() {
+        let mut fi = valid_test_fileinfo("object");
+        fi.version_id = Some(Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap());
+
+        assert_eq!(
+            resolved_read_repair_version_id(&fi, Some("00000000-0000-0000-0000-000000000002")).as_deref(),
+            Some("00000000-0000-0000-0000-000000000001")
+        );
+
+        fi.version_id = None;
+        assert_eq!(
+            resolved_read_repair_version_id(&fi, Some("00000000-0000-0000-0000-000000000002")).as_deref(),
+            Some("00000000-0000-0000-0000-000000000002")
+        );
+    }
+
     #[tokio::test]
     async fn get_object_metadata_cache_hit_returns_stored_metadata() {
         let set = new_metadata_cache_test_set().await;

crates/ecstore/src/set_disk/write.rs

@@ -625,19 +625,37 @@ impl SetDisks {
         }
 
         let results = join_all(futures).await;
+        let mut delete_errs = Vec::with_capacity(results.len());
         for (index, result) in results.into_iter().enumerate() {
             let key = format!("ddisk-{index}");
+            let already_absent = matches!(
+                errs.get(index).and_then(Option::as_ref),
+                Some(DiskError::FileNotFound | DiskError::FileVersionNotFound)
+            );
             match result {
                 Ok(_) => {
                     tags.insert(key, "<nil>".to_string());
+                    delete_errs.push(None);
                 }
                 Err(e) => {
                     tags.insert(key, e.to_string());
+                    if already_absent || matches!(&e, DiskError::FileNotFound | DiskError::FileVersionNotFound) {
+                        delete_errs.push(None);
+                    } else {
+                        delete_errs.push(Some(e));
+                    }
                 }
             }
         }
 
-        // TODO: audit
+        let write_quorum = if m.is_valid() {
+            m.write_quorum(self.default_write_quorum())
+        } else {
+            self.default_write_quorum()
+        };
+        if let Some(err) = reduce_write_quorum_errs(&delete_errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
+            return Err(err);
+        }
 
         Ok(m)
     }

crates/heal/Cargo.toml

@@ -28,6 +28,7 @@ categories = ["web-programming", "development-tools", "filesystem"]
 
 [dependencies]
 rustfs-config = { workspace = true }
+rustfs-concurrency = { workspace = true }
 rustfs-ecstore = { workspace = true }
 rustfs-storage-api = { workspace = true }
 rustfs-common = { workspace = true }

crates/heal/src/heal/channel.rs

@@ -462,7 +462,10 @@ impl HealChannelProcessor {
 
         let recreate_missing = request.recreate_missing.unwrap_or(match request.source {
             HealRequestSource::Scanner => false,
-            HealRequestSource::Admin | HealRequestSource::AutoHeal | HealRequestSource::Internal => true,
+            HealRequestSource::Admin
+            | HealRequestSource::AutoHeal
+            | HealRequestSource::Internal
+            | HealRequestSource::ReadRepair => true,
         });
 
         // Build HealOptions with all available fields
@@ -769,6 +772,7 @@ mod tests {
             HealRequestSource::Admin,
             HealRequestSource::AutoHeal,
             HealRequestSource::Internal,
+            HealRequestSource::ReadRepair,
         ] {
             let channel_request = HealChannelRequest {
                 id: format!("test-id-{source:?}"),
@@ -806,6 +810,7 @@ mod tests {
             (HealRequestSource::Admin, false),
             (HealRequestSource::AutoHeal, false),
             (HealRequestSource::Internal, false),
+            (HealRequestSource::ReadRepair, false),
         ] {
             let channel_request = HealChannelRequest {
                 id: format!("test-id-{source:?}-{recreate_missing}"),

crates/heal/src/lib.rs

@@ -20,6 +20,7 @@ pub use heal::{
     HealManager, HealOperationsSnapshot, HealOptions, HealPriority, HealPriorityCounts, HealRequest, HealSourceCounts, HealType,
     channel::HealChannelProcessor,
 };
+use rustfs_concurrency::WorkloadAdmissionSnapshotProvider;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, OnceLock};
 use tokio_util::sync::CancellationToken;
@@ -70,9 +71,18 @@ static GLOBAL_HEAL_QUEUE_LENGTH: AtomicU64 = AtomicU64::new(0);
 pub async fn init_heal_manager(
     storage: Arc<dyn heal::storage::HealStorageAPI>,
     config: Option<heal::manager::HealConfig>,
+) -> Result<Arc<HealManager>> {
+    init_heal_manager_with_workload_provider(storage, config, None).await
+}
+
+/// Initialize and start heal manager with channel processor and workload snapshots.
+pub async fn init_heal_manager_with_workload_provider(
+    storage: Arc<dyn heal::storage::HealStorageAPI>,
+    config: Option<heal::manager::HealConfig>,
+    workload_provider: Option<Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>>,
 ) -> Result<Arc<HealManager>> {
     // Create heal manager
-    let heal_manager = Arc::new(HealManager::new(storage, config));
+    let heal_manager = Arc::new(HealManager::new_with_workload_provider(storage, config, workload_provider));
 
     // Start heal manager
     heal_manager.start().await?;

rustfs/src/startup_background.rs

@@ -13,7 +13,10 @@
 // limitations under the License.
 
 use crate::storage_api::startup::ECStore;
-use rustfs_heal::{create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager};
+use crate::workload_admission::RustFsWorkloadAdmissionSnapshotProvider;
+use rustfs_heal::{
+    create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager_with_workload_provider,
+};
 use rustfs_utils::get_env_bool_with_aliases;
 use std::{io::Result, sync::Arc};
 use tracing::{debug, info};
@@ -44,7 +47,8 @@ pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Resu
 
     if enable_heal || enable_scanner {
         let heal_storage = Arc::new(ECStoreHealStorage::new(store));
-        init_heal_manager(heal_storage, None).await?;
+        let workload_provider = Arc::new(RustFsWorkloadAdmissionSnapshotProvider);
+        init_heal_manager_with_workload_provider(heal_storage, None, Some(workload_provider)).await?;
     }
 
     if !enable_heal && !enable_scanner {