diff --git a/Cargo.lock b/Cargo.lock index 1914a8505..7c9c75fd4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9449,6 +9449,7 @@ dependencies = [ "http 1.4.2", "metrics", "rustfs-common", + "rustfs-concurrency", "rustfs-config", "rustfs-ecstore", "rustfs-madmin", diff --git a/crates/common/src/heal_channel.rs b/crates/common/src/heal_channel.rs index c60fe8088..2446db0d7 100644 --- a/crates/common/src/heal_channel.rs +++ b/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); diff --git a/crates/config/src/constants/heal.rs b/crates/config/src/constants/heal.rs index 743ea88cc..8f067f509 100644 --- a/crates/config/src/constants/heal.rs +++ b/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; diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index fd8ec62c5..a3a4e57b6 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/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, diff --git a/crates/ecstore/src/set_disk/heal.rs b/crates/ecstore/src/set_disk/heal.rs index 046923687..462de2085 100644 --- a/crates/ecstore/src/set_disk/heal.rs +++ b/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) - .await - .map_err(DiskError::other)?; + warn!( + event = EVENT_HEAL_OBJECT_RENAME, + 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) - .await - .map_err(DiskError::other)?; - + rename_successes += 1; 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); diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index a5e35fdf9..2c1ba494b 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/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>> = OnceLock::new(); + +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct ReadRepairHealCacheKey { + bucket: String, + object: String, + version_id: Option, + 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 { + 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 { + 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 + 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, + 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, + 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( tasks: Vec, 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 _ = - rustfs_common::heal_channel::send_heal_request(rustfs_common::heal_channel::create_heal_request_with_options( - fi.volume.to_string(), // bucket - Some(fi.name.to_string()), // object_prefix - false, // force_start - Some(HealChannelPriority::Normal), // priority - Some(self.pool_index), // pool_index - Some(self.set_index), // set_index - )) - .await; + let version_id = resolved_read_repair_version_id(&fi, opts.version_id.as_deref()); + submit_read_repair_heal( + &fi.volume, + &fi.name, + version_id.as_deref(), + self.pool_index, + self.set_index, + None, + "metadata_read_error", + ) + .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) = - rustfs_common::heal_channel::send_heal_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), - )) - .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"); - } + let version_id = fi.version_id.as_ref().map(ToString::to_string); + submit_read_repair_heal( + bucket, + object, + version_id.as_deref(), + pool_index, + set_index, + Some(part_number), + "missing_shards", + ) + .await; } // 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); - if let Err(e) = rustfs_common::heal_channel::send_heal_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), - ), + debug!( + bucket, + object, + part_number, + error = ?de_err, + "Recoverable decode error triggered read repair" + ); + let version_id = fi.version_id.as_ref().map(ToString::to_string); + submit_read_repair_heal( + bucket, + object, + version_id.as_deref(), + pool_index, + set_index, + Some(part_number), + "decode_error", ) - .await - { - warn!( - bucket, - object, - part_number, - error = %e, - "Failed to enqueue heal request after decode error" - ); - } + .await; 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 - && let Err(e) = - rustfs_common::heal_channel::send_heal_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), - )) - .await - { - warn!( + if missing_shards > 0 { + let version_id = fi.version_id.as_ref().map(ToString::to_string); + submit_read_repair_heal( bucket, object, - part_number, - error = %e, - "Failed to enqueue heal request for missing shards" - ); + version_id.as_deref(), + pool_index, + 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::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; diff --git a/crates/ecstore/src/set_disk/write.rs b/crates/ecstore/src/set_disk/write.rs index ca27521ee..2ff88fd4f 100644 --- a/crates/ecstore/src/set_disk/write.rs +++ b/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, "".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) } diff --git a/crates/heal/Cargo.toml b/crates/heal/Cargo.toml index ecc798cae..e08517bc0 100644 --- a/crates/heal/Cargo.toml +++ b/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 } diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 5d7b709ee..03e085ac0 100644 --- a/crates/heal/src/heal/channel.rs +++ b/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}"), diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 86eace9d5..198eafa6e 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -20,6 +20,7 @@ use crate::heal::{ use crate::{Error, Result}; use metrics::{counter, gauge}; use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult, HealRequestSource}; +use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass}; use rustfs_madmin::heal_commands::HealResultItem; use std::{ collections::{BinaryHeap, HashMap, HashSet}, @@ -44,11 +45,14 @@ const EVENT_HEAL_AUTO_SCAN_DISK: &str = "heal_auto_scan_disk"; const EVENT_HEAL_AUTO_SCAN_ENQUEUE: &str = "heal_auto_scan_enqueue"; const EVENT_HEAL_MANAGER_STATE: &str = "heal_manager_state"; const EVENT_HEAL_QUEUE_ADMISSION: &str = "heal_queue_admission"; +const EVENT_HEAL_MAINLINE_THROTTLE: &str = "heal_mainline_throttle"; const EVENT_HEAL_SCHEDULER_STATE: &str = "heal_scheduler_state"; const EVENT_HEAL_QUEUE_STATE: &str = "heal_queue_state"; const MAX_RECOVERABLE_HEAL_RETRIES: u32 = 3; const MAX_RECOVERABLE_HEAL_RETRY_DELAY: Duration = Duration::from_secs(30); +type WorkloadSnapshotProviderRef = Arc; + /// Priority queue wrapper for heal requests /// Uses BinaryHeap for priority-based ordering while maintaining FIFO for same-priority items #[derive(Debug)] @@ -156,6 +160,7 @@ pub struct HealSourceCounts { pub admin: u64, pub auto_heal: u64, pub internal: u64, + pub read_repair: u64, } impl HealSourceCounts { @@ -165,6 +170,7 @@ impl HealSourceCounts { HealRequestSource::Admin => self.admin += 1, HealRequestSource::AutoHeal => self.auto_heal += 1, HealRequestSource::Internal => self.internal += 1, + HealRequestSource::ReadRepair => self.read_repair += 1, } } } @@ -626,6 +632,12 @@ 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. + pub mainline_throttle_enable: bool, + /// Foreground read permit utilization percentage that delays best-effort heal starts. + pub mainline_read_utilization_high_percent: usize, + /// Delay before rechecking foreground read pressure after delaying heal starts. + pub mainline_max_sleep: Duration, } impl Default for HealConfig { @@ -670,6 +682,19 @@ impl Default for HealConfig { rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, rustfs_config::DEFAULT_HEAL_PAGE_PARALLEL_ENABLE, ); + let mainline_throttle_enable = rustfs_utils::get_env_bool( + rustfs_config::ENV_HEAL_MAINLINE_THROTTLE_ENABLE, + rustfs_config::DEFAULT_HEAL_MAINLINE_THROTTLE_ENABLE, + ); + let mainline_read_utilization_high_percent = rustfs_utils::get_env_usize( + rustfs_config::ENV_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT, + rustfs_config::DEFAULT_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT, + ) + .min(100); + let mainline_max_sleep = Duration::from_millis(rustfs_utils::get_env_u64( + rustfs_config::ENV_HEAL_MAINLINE_MAX_SLEEP_MS, + rustfs_config::DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS, + )); Self { enable_auto_heal, heal_interval, // 10 seconds @@ -682,6 +707,9 @@ impl Default for HealConfig { event_driven_scheduler_enable, set_bulkhead_enable, page_parallel_enable, + mainline_throttle_enable, + mainline_read_utilization_high_percent, + mainline_max_sleep, } } } @@ -727,6 +755,8 @@ pub struct HealManager { statistics: Arc>, /// Scheduler wake-up notifier for event-driven dispatch notify: Arc, + /// Optional runtime workload snapshot provider used to protect foreground data-plane work. + workload_provider: Option, } struct HealQueueContext<'a> { @@ -739,17 +769,52 @@ struct HealQueueContext<'a> { storage: &'a Arc, notify: &'a Arc, cancel_token: &'a CancellationToken, + workload_provider: &'a Option, } impl HealManager { fn classify_full_admission(request: &HealRequest, config: &HealConfig) -> HealAdmissionResult { - if request.priority == HealPriority::Low && config.low_priority_drop_when_full { + let best_effort_source = matches!( + request.source, + HealRequestSource::Scanner | HealRequestSource::AutoHeal | HealRequestSource::ReadRepair + ); + if best_effort_source || (request.priority == HealPriority::Low && config.low_priority_drop_when_full) { HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull) } else { HealAdmissionResult::Full } } + fn queue_usage_pct(queue_len: usize, queue_capacity: usize) -> usize { + queue_len.saturating_mul(100).checked_div(queue_capacity).unwrap_or(0) + } + + fn classify_pressure_admission( + request: &HealRequest, + queue_len: usize, + queue_capacity: usize, + ) -> Option { + if request.force_start || queue_capacity == 0 { + return None; + } + + let queue_usage_pct = Self::queue_usage_pct(queue_len, queue_capacity); + if queue_usage_pct < 80 { + return None; + } + + match request.source { + HealRequestSource::ReadRepair => Some(HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped)), + HealRequestSource::Scanner if request.priority == HealPriority::Low => { + Some(HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped)) + } + HealRequestSource::AutoHeal if queue_usage_pct >= 95 => { + Some(HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped)) + } + _ => None, + } + } + fn duplicate_admission_for_request(request: &HealRequest, config: &HealConfig) -> HealAdmissionResult { if request.priority == HealPriority::Low && !config.low_priority_merge_enable { HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped) @@ -758,6 +823,285 @@ impl HealManager { } } + fn can_displace_queued_work(request: &HealRequest) -> bool { + matches!(request.source, HealRequestSource::Admin | HealRequestSource::Internal) + } + + fn request_bypasses_mainline_throttle(request: &HealRequest) -> bool { + request.force_start + || matches!(request.source, HealRequestSource::Admin | HealRequestSource::Internal) + || matches!(request.priority, HealPriority::High | HealPriority::Urgent) + } + + fn mainline_throttle_active(config: &HealConfig, provider: &Option) -> Option { + if !config.mainline_throttle_enable || config.mainline_read_utilization_high_percent == 0 { + return None; + } + + let provider = provider.as_ref()?; + let snapshot = provider.workload_admission_snapshot(); + let foreground_read = snapshot.get(WorkloadClass::ForegroundRead)?; + if matches!(foreground_read.state, AdmissionState::Saturated) { + return Some(100); + } + + let limit = foreground_read.limit?; + if limit == 0 { + return None; + } + let usage_pct = foreground_read + .active + .unwrap_or(0) + .saturating_mul(100) + .checked_div(limit) + .unwrap_or(100); + (usage_pct >= config.mainline_read_utilization_high_percent).then_some(usage_pct) + } + + fn schedule_mainline_throttle_recheck(notify: Arc, delay: Duration) { + if delay.is_zero() { + notify.notify_one(); + return; + } + + tokio::spawn(async move { + sleep(delay).await; + notify.notify_one(); + }); + } + + fn record_mainline_throttle_delay(usage_pct: usize, config: &HealConfig) { + counter!( + "rustfs_heal_mainline_throttle_total", + "source" => "background", + "result" => "delayed", + "reason" => "foreground_read_pressure" + ) + .increment(1); + debug!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_MAINLINE_THROTTLE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + state = "delayed", + foreground_read_usage_pct = usage_pct, + threshold_pct = config.mainline_read_utilization_high_percent, + recheck_delay_ms = config.mainline_max_sleep.as_millis(), + "Heal scheduler delayed background work under foreground read pressure" + ); + } + + fn record_admission_metric(source: HealRequestSource, admission: HealAdmissionResult, context: &'static str) { + counter!( + "rustfs_heal_admission_total", + "source" => source.as_str().to_string(), + "result" => admission.result_label().to_string(), + "reason" => admission.reason_label().to_string(), + "context" => context.to_string() + ) + .increment(1); + } + + fn admit_request_to_queue( + queue: &mut PriorityHealQueue, + request: HealRequest, + config: &HealConfig, + context: &'static str, + ) -> HealAdmissionResult { + let queue_len = queue.len(); + publish_heal_queue_length(queue); + let queue_capacity = config.queue_size; + + if queue_len >= queue_capacity && !request.force_start { + if Self::can_displace_queued_work(&request) && queue.can_displace_lower_priority(request.priority) { + let request_id = request.id.clone(); + let priority = request.priority; + let source = request.source; + if let Some(displaced) = queue.push_displacing_lower_priority(request) { + publish_heal_queue_length(queue); + Self::record_admission_metric(source, HealAdmissionResult::Accepted, context); + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %request_id, + priority = ?priority, + source = source.as_str(), + context, + displaced_request_id = %displaced.id, + displaced_priority = ?displaced.priority, + queue_len, + queue_capacity, + result = "accepted_by_displacement", + "Heal queue request accepted by displacement" + ); + return HealAdmissionResult::Accepted; + } + + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %request_id, + priority = ?priority, + source = source.as_str(), + context, + queue_len, + queue_capacity, + result = "full_no_displacement_candidate", + "Heal queue request rejected without displacement" + ); + Self::record_admission_metric(source, HealAdmissionResult::Full, context); + return HealAdmissionResult::Full; + } + + let admission = Self::classify_full_admission(&request, config); + Self::record_admission_metric(request.source, admission, context); + match admission { + HealAdmissionResult::Dropped(reason) => { + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %request.id, + priority = ?request.priority, + source = request.source.as_str(), + context, + queue_len, + queue_capacity, + reason = reason.as_str(), + result = "dropped_full", + "Heal queue request dropped" + ); + } + HealAdmissionResult::Full => { + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %request.id, + priority = ?request.priority, + source = request.source.as_str(), + context, + queue_len, + queue_capacity, + result = "rejected_full", + "Heal queue request rejected" + ); + } + HealAdmissionResult::Accepted | HealAdmissionResult::Merged => {} + } + return admission; + } + + if let Some(admission) = Self::classify_pressure_admission(&request, queue_len, queue_capacity) { + Self::record_admission_metric(request.source, admission, context); + if let HealAdmissionResult::Dropped(reason) = admission { + debug!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %request.id, + priority = ?request.priority, + source = request.source.as_str(), + context, + queue_len, + queue_capacity, + queue_usage_pct = Self::queue_usage_pct(queue_len, queue_capacity), + reason = reason.as_str(), + result = "dropped_pressure", + "Heal queue request dropped under pressure" + ); + } + return admission; + } + + if queue_capacity > 0 { + let capacity_threshold = queue_capacity.saturating_mul(80) / 100; + if queue_len >= capacity_threshold { + debug!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + queue_len, + queue_capacity, + queue_usage_pct = Self::queue_usage_pct(queue_len, queue_capacity), + context, + result = "queue_pressure_high", + "Heal queue pressure high" + ); + } + } + + let request_id = request.id.clone(); + let priority = request.priority; + let source = request.source; + + match queue.push(request) { + QueuePushOutcome::Accepted => { + publish_heal_queue_length(queue); + Self::record_admission_metric(source, HealAdmissionResult::Accepted, context); + if matches!(priority, HealPriority::High | HealPriority::Urgent) { + let stats = queue.get_priority_stats(); + debug!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %request_id, + priority = ?priority, + source = source.as_str(), + context, + queue_len = queue_len + 1, + urgent = *stats.get(&HealPriority::Urgent).unwrap_or(&0), + high = *stats.get(&HealPriority::High).unwrap_or(&0), + normal = *stats.get(&HealPriority::Normal).unwrap_or(&0), + low = *stats.get(&HealPriority::Low).unwrap_or(&0), + result = "accepted", + "Heal queue snapshot recorded" + ); + } + debug!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %request_id, + priority = ?priority, + source = source.as_str(), + context, + queue_len = queue_len + 1, + result = "accepted", + "Heal queue request accepted" + ); + HealAdmissionResult::Accepted + } + QueuePushOutcome::Merged => { + Self::record_admission_metric(source, HealAdmissionResult::Merged, context); + debug!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %request_id, + priority = ?priority, + source = source.as_str(), + context, + queue_len, + result = "merged_duplicate", + "Heal queue request merged" + ); + HealAdmissionResult::Merged + } + } + } + async fn insert_task_alias(&self, alias_id: &str, task_id: &str) { if alias_id == task_id { return; @@ -789,6 +1133,15 @@ impl HealManager { /// Create new HealManager pub fn new(storage: Arc, config: Option) -> Self { + Self::new_with_workload_provider(storage, config, None) + } + + /// Create new HealManager with an optional workload admission snapshot provider. + pub fn new_with_workload_provider( + storage: Arc, + config: Option, + workload_provider: Option, + ) -> Self { let config = config.unwrap_or_default(); Self { config: Arc::new(RwLock::new(config)), @@ -802,6 +1155,7 @@ impl HealManager { cancel_token: CancellationToken::new(), statistics: Arc::new(RwLock::new(HealStatistics::new())), notify: Arc::new(Notify::new()), + workload_provider, } } @@ -835,7 +1189,18 @@ impl HealManager { self.start_scheduler().await?; // start auto disk scanner to heal unformatted disks - self.start_auto_disk_scanner().await?; + if self.config.read().await.enable_auto_heal { + self.start_auto_disk_scanner().await?; + } else { + info!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_AUTO_SCAN_STATE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_DISK_SCANNER, + state = "disabled", + "Heal auto disk scanner disabled" + ); + } info!( target: "rustfs::heal::manager", @@ -995,10 +1360,6 @@ impl HealManager { let mut queue = self.heal_queue.lock().await; - let queue_len = queue.len(); - publish_heal_queue_length(&queue); - let queue_capacity = config.queue_size; - let queued_duplicate = (!request.force_start) .then(|| queue.request_for_dedup_key(&dedup_key).map(|queued| queued.id.clone())) .flatten(); @@ -1040,145 +1401,15 @@ impl HealManager { return Ok(admission); } - if queue_len >= queue_capacity && !request.force_start { - if queue.can_displace_lower_priority(request.priority) { - let request_id = request.id.clone(); - let priority = request.priority; - if let Some(displaced) = queue.push_displacing_lower_priority(request) { - publish_heal_queue_length(&queue); - warn!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_QUEUE_ADMISSION, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - request_id = %request_id, - priority = ?priority, - displaced_request_id = %displaced.id, - displaced_priority = ?displaced.priority, - queue_len, - queue_capacity, - result = "accepted_by_displacement", - "Heal queue request accepted by displacement" - ); - drop(queue); - if config.event_driven_scheduler_enable { - self.notify.notify_one(); - } - return Ok(HealAdmissionResult::Accepted); - } - - warn!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_QUEUE_ADMISSION, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - request_id = %request_id, - priority = ?priority, - queue_len, - queue_capacity, - result = "full_no_displacement_candidate", - "Heal queue request rejected without displacement" - ); - return Ok(HealAdmissionResult::Full); - } - - let admission = Self::classify_full_admission(&request, &config); - match admission { - HealAdmissionResult::Dropped(reason) => { - warn!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_QUEUE_ADMISSION, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - request_id = %request.id, - priority = ?request.priority, - queue_len, - queue_capacity, - reason = reason.as_str(), - result = "dropped_full", - "Heal queue request dropped" - ); - } - HealAdmissionResult::Full => { - warn!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_QUEUE_ADMISSION, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - request_id = %request.id, - priority = ?request.priority, - queue_len, - queue_capacity, - result = "rejected_full", - "Heal queue request rejected" - ); - } - HealAdmissionResult::Accepted | HealAdmissionResult::Merged => {} - } - return Ok(admission); - } - - // Warn when queue is getting full (>80% capacity) - let capacity_threshold = (queue_capacity as f64 * 0.8) as usize; - if queue_len >= capacity_threshold { - warn!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_QUEUE_ADMISSION, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - queue_len, - queue_capacity, - queue_usage_pct = (queue_len * 100) / queue_capacity, - result = "queue_pressure_high", - "Heal queue pressure high" - ); - } - - let request_id = request.id.clone(); - let priority = request.priority; - - let push_outcome = queue.push(request); - debug_assert_eq!(push_outcome, QueuePushOutcome::Accepted); - publish_heal_queue_length(&queue); - - // Log queue statistics periodically (when adding high/urgent priority items) - if matches!(priority, HealPriority::High | HealPriority::Urgent) { - let stats = queue.get_priority_stats(); - debug!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_QUEUE_ADMISSION, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - request_id = %request_id, - priority = ?priority, - queue_len = queue_len + 1, - urgent = *stats.get(&HealPriority::Urgent).unwrap_or(&0), - high = *stats.get(&HealPriority::High).unwrap_or(&0), - normal = *stats.get(&HealPriority::Normal).unwrap_or(&0), - low = *stats.get(&HealPriority::Low).unwrap_or(&0), - result = "accepted", - "Heal queue snapshot recorded" - ); - } - + let admission = Self::admit_request_to_queue(&mut queue, request, &config, "submit"); + let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable; drop(queue); - debug!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_QUEUE_ADMISSION, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - request_id = %request_id, - priority = ?priority, - queue_len = queue_len + 1, - result = "accepted", - "Heal queue request accepted" - ); - if config.event_driven_scheduler_enable { + if should_notify { self.notify.notify_one(); } - Ok(HealAdmissionResult::Accepted) + Ok(admission) } /// Get task status @@ -1696,6 +1927,7 @@ impl HealManager { let statistics = self.statistics.clone(); let storage = self.storage.clone(); let notify = self.notify.clone(); + let workload_provider = self.workload_provider.clone(); tokio::spawn(async move { let mut interval = interval(config.read().await.heal_interval); @@ -1725,6 +1957,7 @@ impl HealManager { storage: &storage, notify: ¬ify, cancel_token: &cancel_token, + workload_provider: &workload_provider, }) .await; } @@ -1739,6 +1972,7 @@ impl HealManager { storage: &storage, notify: ¬ify, cancel_token: &cancel_token, + workload_provider: &workload_provider, }) .await; } @@ -1782,6 +2016,9 @@ impl HealManager { let mut skipped_duplicate_count = 0usize; let mut skipped_invalid_count = 0usize; let mut enqueued_count = 0usize; + let mut not_enqueued_count = 0usize; + let mut dropped_count = 0usize; + let mut full_count = 0usize; tokio::select! { _ = cancel_token.cancelled() => { info!( @@ -1951,11 +2188,15 @@ impl HealManager { HealPriority::Normal, ); req.source = HealRequestSource::AutoHeal; + let config = config.read().await; let mut queue = heal_queue.lock().await; - if matches!(queue.push(req), QueuePushOutcome::Accepted) { - publish_heal_queue_length(&queue); - let config = config.read().await; - if config.event_driven_scheduler_enable { + let admission = Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan"); + let should_notify = + matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable; + drop(queue); + drop(config); + if matches!(admission, HealAdmissionResult::Accepted) { + if should_notify { notify.notify_one(); } enqueued_count += 1; @@ -1970,6 +2211,31 @@ impl HealManager { result = "enqueued", "Heal auto-scan task enqueued" ); + } else { + if matches!(admission, HealAdmissionResult::Merged) { + skipped_duplicate_count += 1; + } else { + not_enqueued_count += 1; + } + if matches!(admission, HealAdmissionResult::Full) { + full_count += 1; + } + if matches!(admission, HealAdmissionResult::Dropped(_)) { + dropped_count += 1; + } + debug!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_AUTO_SCAN_ENQUEUE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_DISK_SCANNER, + endpoint = %ep, + set_disk_id, + bucket_count = buckets.len(), + admission = admission.result_label(), + reason = admission.reason_label(), + result = "not_enqueued", + "Heal auto-scan task not enqueued" + ); } } info!( @@ -1980,6 +2246,9 @@ impl HealManager { state = "cycle_completed", candidate_count, enqueued_count, + not_enqueued_count, + dropped_count, + full_count, skipped_duplicate_count, skipped_invalid_count, "Heal auto-scan cycle completed" @@ -2004,9 +2273,11 @@ impl HealManager { storage, notify, cancel_token, + workload_provider, } = context; let config = config.read().await; + let mainline_usage_pct = Self::mainline_throttle_active(&config, workload_provider); let mut active_heals_guard = active_heals.lock().await; publish_active_heal_count(&active_heals_guard); @@ -2029,12 +2300,18 @@ impl HealManager { let mut running_per_set = running_erasure_set_counts(&active_heals_guard); let mut tasks_started = 0usize; + let mut delayed_by_mainline_throttle = false; for _ in 0..available_slots { - let selected_request = if config.set_bulkhead_enable { + let selected_request = if config.set_bulkhead_enable || mainline_usage_pct.is_some() { let max_concurrent_per_set = config.max_concurrent_per_set; let (selected_request, skipped_sets) = queue.pop_runnable_with_skips( - |request| can_schedule_request(request, &running_per_set, max_concurrent_per_set), + |request| { + let set_allowed = !config.set_bulkhead_enable + || can_schedule_request(request, &running_per_set, max_concurrent_per_set); + let mainline_allowed = mainline_usage_pct.is_none() || Self::request_bypasses_mainline_throttle(request); + set_allowed && mainline_allowed + }, |request| heal_request_set_key(request).map(|_| heal_request_set_metric_label(request)), ); for skipped_set in skipped_sets { @@ -2068,6 +2345,7 @@ impl HealManager { let manager_cancel_token = cancel_token.clone(); let task_type_label_for_spawn = task_type_label.clone(); let task_set_label_for_spawn = task_set_label.clone(); + let config_for_spawn = config.clone(); // start heal task tokio::spawn(async move { @@ -2176,9 +2454,11 @@ impl HealManager { let retry_active_heals = active_heals_clone.clone(); let retry_heal_queue = heal_queue_clone.clone(); let retrying_heals_for_spawn = retrying_heals_clone.clone(); + let retry_completed_heals = completed_heals_clone.clone(); let retry_notify = notify_clone.clone(); let retry_cancel_token = CancellationToken::new(); let retry_manager_cancel_token = manager_cancel_token.clone(); + let retry_config = config_for_spawn.clone(); retrying_heals_clone.lock().await.insert( retry_request_id.clone(), RetryingHeal { @@ -2188,38 +2468,42 @@ impl HealManager { }, ); tokio::spawn(async move { - tokio::select! { - _ = retry_cancel_token.cancelled() => { - info!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_QUEUE_ADMISSION, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - request_id = %retry_request_id, - priority = ?retry_priority, - retry_attempt, - result = "retry_cancelled", - "Heal retry admission decided" - ); - return; + loop { + tokio::select! { + _ = retry_cancel_token.cancelled() => { + info!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %retry_request_id, + priority = ?retry_priority, + retry_attempt, + result = "retry_cancelled", + "Heal retry admission decided" + ); + return; + } + _ = retry_manager_cancel_token.cancelled() => { + retrying_heals_for_spawn.lock().await.remove(&retry_request_id); + return; + } + _ = sleep(retry_delay) => {} } - _ = retry_manager_cancel_token.cancelled() => { + + { + let retrying_heals_guard = retrying_heals_for_spawn.lock().await; + if !retrying_heals_guard.contains_key(&retry_request_id) { + return; + } + } + + let active_duplicate = { + let active_heals_guard = retry_active_heals.lock().await; + active_heals_contains_dedup_key(&active_heals_guard, &retry_key) + }; + if active_duplicate { retrying_heals_for_spawn.lock().await.remove(&retry_request_id); - return; - } - _ = sleep(retry_delay) => {} - } - - { - let mut retrying_heals_guard = retrying_heals_for_spawn.lock().await; - if retrying_heals_guard.remove(&retry_request_id).is_none() { - return; - } - } - - { - let active_heals_guard = retry_active_heals.lock().await; - if active_heals_contains_dedup_key(&active_heals_guard, &retry_key) { info!( target: "rustfs::heal::manager", event = EVENT_HEAL_QUEUE_ADMISSION, @@ -2233,41 +2517,78 @@ impl HealManager { ); return; } - } - let mut queue = retry_heal_queue.lock().await; - let outcome = queue.push(retry_request); - publish_heal_queue_length(&queue); - match outcome { - QueuePushOutcome::Accepted => { - info!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_QUEUE_ADMISSION, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - request_id = %retry_request_id, - priority = ?retry_priority, - retry_attempt, - retry_delay_ms = retry_delay.as_millis(), - error = %retry_error, - result = "retry_enqueued", - "Heal retry admission decided" - ); - drop(queue); - retry_notify.notify_one(); - } - QueuePushOutcome::Merged => { - info!( - target: "rustfs::heal::manager", - event = EVENT_HEAL_QUEUE_ADMISSION, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_MANAGER, - request_id = %retry_request_id, - priority = ?retry_priority, - retry_attempt, - result = "retry_merged_duplicate", - "Heal retry admission decided" - ); + let mut queue = retry_heal_queue.lock().await; + let admission = + Self::admit_request_to_queue(&mut queue, retry_request.clone(), &retry_config, "retry"); + let should_notify = matches!(admission, HealAdmissionResult::Accepted) + && retry_config.event_driven_scheduler_enable; + match admission { + HealAdmissionResult::Accepted => { + drop(queue); + retrying_heals_for_spawn.lock().await.remove(&retry_request_id); + retry_completed_heals.lock().await.remove(&retry_request_id); + info!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %retry_request_id, + priority = ?retry_priority, + retry_attempt, + retry_delay_ms = retry_delay.as_millis(), + error = %retry_error, + result = "retry_enqueued", + "Heal retry admission decided" + ); + if should_notify { + retry_notify.notify_one(); + } + return; + } + HealAdmissionResult::Merged => { + drop(queue); + retrying_heals_for_spawn.lock().await.remove(&retry_request_id); + info!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %retry_request_id, + priority = ?retry_priority, + retry_attempt, + result = "retry_merged_duplicate", + "Heal retry admission decided" + ); + return; + } + HealAdmissionResult::Full => { + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %retry_request_id, + priority = ?retry_priority, + retry_attempt, + result = "retry_rejected_full", + "Heal retry admission decided" + ); + } + HealAdmissionResult::Dropped(reason) => { + debug!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_QUEUE_ADMISSION, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + request_id = %retry_request_id, + priority = ?retry_priority, + retry_attempt, + reason = reason.as_str(), + result = "retry_dropped", + "Heal retry admission decided" + ); + } } } }); @@ -2276,6 +2597,7 @@ impl HealManager { }); tasks_started += 1; } else { + delayed_by_mainline_throttle = mainline_usage_pct.is_some(); break; } } @@ -2287,6 +2609,11 @@ impl HealManager { publish_active_heal_count(&active_heals_guard); publish_heal_queue_length(&queue); + if delayed_by_mainline_throttle && let Some(usage_pct) = mainline_usage_pct { + Self::record_mainline_throttle_delay(usage_pct, &config); + Self::schedule_mainline_throttle_recheck(notify.clone(), config.mainline_max_sleep); + } + // Log queue status if items remain if !queue.is_empty() { let remaining = queue.len(); @@ -2406,10 +2733,46 @@ mod tests { use crate::heal::storage::{HealObjectInfo, HealStorageAPI}; use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealTask, HealType}; use rustfs_common::heal_channel::{HealOpts, HealRequestSource}; + use rustfs_concurrency::{WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot}; use rustfs_madmin::heal_commands::HealResultItem; use super::super::{DiskStore, Endpoint, storage_api::status::BucketInfo}; + #[derive(Debug)] + struct FixedWorkloadProvider { + active: usize, + limit: usize, + state: AdmissionState, + } + + impl WorkloadAdmissionSnapshotProvider for FixedWorkloadProvider { + fn workload_admission_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot { + WorkloadAdmissionRegistrySnapshot::new(vec![ + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, self.state).with_counts( + Some(self.active), + None, + Some(self.limit), + ), + ]) + } + } + + async fn process_manager_queue_once(manager: &HealManager) { + HealManager::process_heal_queue(HealQueueContext { + heal_queue: &manager.heal_queue, + active_heals: &manager.active_heals, + completed_heals: &manager.completed_heals, + retrying_heals: &manager.retrying_heals, + config: &manager.config, + statistics: &manager.statistics, + storage: &manager.storage, + notify: &manager.notify, + cancel_token: &manager.cancel_token, + workload_provider: &manager.workload_provider, + }) + .await; + } + struct MockStorage; #[async_trait::async_trait] @@ -2506,6 +2869,18 @@ mod tests { } } + fn bucket_request(bucket: &str, priority: HealPriority, source: HealRequestSource) -> HealRequest { + let mut request = HealRequest::new( + HealType::Bucket { + bucket: bucket.to_string(), + }, + HealOptions::default(), + priority, + ); + request.source = source; + request + } + #[test] fn test_priority_queue_ordering() { let mut queue = PriorityHealQueue::new(); @@ -3811,6 +4186,169 @@ mod tests { ); } + #[tokio::test] + async fn test_submit_heal_request_drops_read_repair_under_pressure() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new( + storage, + Some(HealConfig { + queue_size: 10, + ..HealConfig::default() + }), + ); + + for index in 0..8 { + assert_eq!( + manager + .submit_heal_request(bucket_request( + &format!("queued-{index}"), + HealPriority::Normal, + HealRequestSource::Internal, + )) + .await + .expect("seed request should be accepted"), + HealAdmissionResult::Accepted + ); + } + + let admission = manager + .submit_heal_request(bucket_request("read-repair", HealPriority::Normal, HealRequestSource::ReadRepair)) + .await + .expect("read repair admission should return a result"); + + assert_eq!(admission, HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped)); + assert_eq!(manager.get_queue_length().await, 8); + } + + #[tokio::test] + async fn test_submit_heal_request_drops_low_scanner_under_pressure() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new( + storage, + Some(HealConfig { + queue_size: 10, + ..HealConfig::default() + }), + ); + + for index in 0..8 { + assert_eq!( + manager + .submit_heal_request(bucket_request( + &format!("queued-{index}"), + HealPriority::Normal, + HealRequestSource::Internal, + )) + .await + .expect("seed request should be accepted"), + HealAdmissionResult::Accepted + ); + } + + let admission = manager + .submit_heal_request(bucket_request("scanner", HealPriority::Low, HealRequestSource::Scanner)) + .await + .expect("scanner admission should return a result"); + + assert_eq!(admission, HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped)); + assert_eq!(manager.get_queue_length().await, 8); + } + + #[tokio::test] + async fn test_submit_heal_request_accepts_admin_high_under_pressure() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new( + storage, + Some(HealConfig { + queue_size: 10, + ..HealConfig::default() + }), + ); + + for index in 0..8 { + assert_eq!( + manager + .submit_heal_request(bucket_request( + &format!("queued-{index}"), + HealPriority::Normal, + HealRequestSource::Internal, + )) + .await + .expect("seed request should be accepted"), + HealAdmissionResult::Accepted + ); + } + + let admission = manager + .submit_heal_request(bucket_request("admin", HealPriority::High, HealRequestSource::Admin)) + .await + .expect("admin admission should return a result"); + + assert_eq!(admission, HealAdmissionResult::Accepted); + assert_eq!(manager.get_queue_length().await, 9); + } + + #[tokio::test] + async fn test_mainline_throttle_delays_background_heal_start() { + let storage: Arc = Arc::new(MockStorage); + let provider: WorkloadSnapshotProviderRef = Arc::new(FixedWorkloadProvider { + active: 8, + limit: 10, + state: AdmissionState::Open, + }); + let manager = HealManager::new_with_workload_provider( + storage, + Some(HealConfig { + max_concurrent_heals: 1, + mainline_throttle_enable: true, + mainline_read_utilization_high_percent: 80, + mainline_max_sleep: Duration::from_millis(1), + ..HealConfig::default() + }), + Some(provider), + ); + + manager + .submit_heal_request(bucket_request("read-repair", HealPriority::Normal, HealRequestSource::ReadRepair)) + .await + .expect("read repair request should be queued"); + + process_manager_queue_once(&manager).await; + + assert_eq!(manager.get_queue_length().await, 1); + assert_eq!(manager.get_active_task_count().await, 0); + } + + #[tokio::test] + async fn test_mainline_throttle_allows_admin_high_start() { + let storage: Arc = Arc::new(MockStorage); + let provider: WorkloadSnapshotProviderRef = Arc::new(FixedWorkloadProvider { + active: 10, + limit: 10, + state: AdmissionState::Saturated, + }); + let manager = HealManager::new_with_workload_provider( + storage, + Some(HealConfig { + max_concurrent_heals: 1, + mainline_throttle_enable: true, + mainline_read_utilization_high_percent: 80, + mainline_max_sleep: Duration::from_millis(1), + ..HealConfig::default() + }), + Some(provider), + ); + + manager + .submit_heal_request(bucket_request("admin", HealPriority::High, HealRequestSource::Admin)) + .await + .expect("admin request should be queued"); + + process_manager_queue_once(&manager).await; + + assert_eq!(manager.get_queue_length().await, 0); + } + #[tokio::test] async fn test_force_start_bypasses_duplicate_and_full_admission() { let storage: Arc = Arc::new(MockStorage); diff --git a/crates/heal/src/lib.rs b/crates/heal/src/lib.rs index 9297f3fca..65b3056b2 100644 --- a/crates/heal/src/lib.rs +++ b/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, config: Option, +) -> Result> { + 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, + config: Option, + workload_provider: Option>, ) -> Result> { // 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?; diff --git a/rustfs/src/startup_background.rs b/rustfs/src/startup_background.rs index 8f855b117..d3ce62b11 100644 --- a/rustfs/src/startup_background.rs +++ b/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) -> 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 {