mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 22:37:13 +00:00
chore: merge main into scanner feedback
This commit is contained in:
overtrue
@@ -19,6 +19,7 @@ use thiserror::Error;
|
||||
use super::heal::{DiskError, EcstoreError};
|
||||
|
||||
const HEAL_DANGLING_DELETE_GRACE_MESSAGE: &str = "dangling object deletion deferred by heal grace window";
|
||||
pub(crate) const MRF_MISSING_PROOF_PREFIX: &str = "Missing exact storage proof for durable MRF repair ";
|
||||
|
||||
/// Custom error type for heal operations
|
||||
/// This enum defines various error variants that can occur during
|
||||
@@ -221,6 +222,7 @@ pub(crate) fn classify_heal_error(error: &Error) -> HealErrorClass {
|
||||
Error::Storage(error) => classify_storage_error(error),
|
||||
Error::Disk(error) => classify_disk_error(error),
|
||||
Error::Io(error) => classify_io_error(error),
|
||||
Error::TaskExecutionFailed { message } if message.starts_with(MRF_MISSING_PROOF_PREFIX) => HealErrorClass::Permanent,
|
||||
Error::TaskExecutionFailed { message } | Error::Other(message) => {
|
||||
classify_error_display(message).unwrap_or(HealErrorClass::Permanent)
|
||||
}
|
||||
|
||||
@@ -58,6 +58,8 @@ fn result_object_size_u64(result: &HealResultItem) -> u64 {
|
||||
|
||||
const NEW_VERSION_SKIP_GRACE_SECS: u64 = 60;
|
||||
const NANOS_PER_SECOND: i128 = 1_000_000_000;
|
||||
const DEFAULT_REPLACEMENT_PAGE_OBJECT_CONCURRENCY: usize = 2;
|
||||
const REPLACEMENT_OBJECT_MAX_RETRIES: u32 = 2;
|
||||
|
||||
fn should_skip_new_version(mod_time_unix_nanos: Option<i128>, started_at_secs: u64) -> bool {
|
||||
let Some(mod_time_unix_nanos) = mod_time_unix_nanos else {
|
||||
@@ -177,7 +179,9 @@ mod mainline_pacing_tests {
|
||||
.await
|
||||
.expect("existing object owns capacity");
|
||||
let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 0));
|
||||
let pacer = Arc::new(MainlinePacer::new(provider.clone(), 80, 80, Duration::from_millis(250)).expect("pacer"));
|
||||
let pacer = Arc::new(
|
||||
MainlinePacer::new(provider.clone(), 80, 80, Duration::from_millis(250), HealRequestSource::Admin).expect("pacer"),
|
||||
);
|
||||
let cancel = CancellationToken::new();
|
||||
let waiting = tokio::spawn({
|
||||
let semaphore = semaphore.clone();
|
||||
@@ -216,7 +220,7 @@ mod mainline_pacing_tests {
|
||||
let occupied = semaphore.clone().acquire_owned().await.expect("queue both waiters");
|
||||
let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 0));
|
||||
let pause = Duration::from_millis(250);
|
||||
let pacer = MainlinePacer::new(provider.clone(), 80, 80, pause).expect("pacer");
|
||||
let pacer = MainlinePacer::new(provider.clone(), 80, 80, pause, HealRequestSource::Admin).expect("pacer");
|
||||
let cancel = CancellationToken::new();
|
||||
let mut first = Box::pin(acquire_page_permit(semaphore.clone(), Some(&pacer), &cancel));
|
||||
let mut second = Box::pin(acquire_page_permit(semaphore.clone(), Some(&pacer), &cancel));
|
||||
@@ -304,9 +308,21 @@ impl ErasureSetHealer {
|
||||
}
|
||||
}
|
||||
|
||||
fn effective_heal_page_object_concurrency_for_source(source: HealRequestSource, scan_mode: HealScanMode) -> usize {
|
||||
fn effective_heal_page_object_concurrency_for_source(
|
||||
source: HealRequestSource,
|
||||
scan_mode: HealScanMode,
|
||||
replacement: bool,
|
||||
) -> usize {
|
||||
if matches!(source, HealRequestSource::AutoHeal) {
|
||||
1
|
||||
if !replacement || matches!(scan_mode, HealScanMode::Deep) || !Self::page_parallel_enabled() {
|
||||
1
|
||||
} else {
|
||||
rustfs_utils::get_env_usize(
|
||||
rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY,
|
||||
DEFAULT_REPLACEMENT_PAGE_OBJECT_CONCURRENCY,
|
||||
)
|
||||
.clamp(1, Semaphore::MAX_PERMITS)
|
||||
}
|
||||
} else {
|
||||
Self::effective_heal_page_object_concurrency_for_scan_mode(scan_mode)
|
||||
}
|
||||
@@ -1467,7 +1483,16 @@ impl ErasureSetHealer {
|
||||
// transient cache id that changes between identical pages.
|
||||
let mut previous_page_last: Option<(String, Option<String>)> = None;
|
||||
let page_concurrency_limit =
|
||||
Self::effective_heal_page_object_concurrency_for_source(self.source, self.heal_opts.scan_mode);
|
||||
if self.source == HealRequestSource::AutoHeal && self.replacement_task_id.is_some() && self.heal_opts.no_lock {
|
||||
// Versions of the same object must not race without its namespace lock.
|
||||
1
|
||||
} else {
|
||||
Self::effective_heal_page_object_concurrency_for_source(
|
||||
self.source,
|
||||
self.heal_opts.scan_mode,
|
||||
self.replacement_task_id.is_some(),
|
||||
)
|
||||
};
|
||||
let in_flight = Arc::new(AtomicUsize::new(0));
|
||||
// Per-bucket sample caps for per-object warn! lines: a flapping rebuild
|
||||
// disk can fail/skip hundreds of thousands of versions in one sweep, so
|
||||
@@ -1689,6 +1714,7 @@ impl ErasureSetHealer {
|
||||
let semaphore = semaphore.clone();
|
||||
let target_endpoints = self.target_endpoints.clone();
|
||||
let replacement_commit_evidence_required = self.replacement_task_id.is_some();
|
||||
let retry_replacement = replacement_commit_evidence_required && self.source == HealRequestSource::AutoHeal;
|
||||
let mainline_pacer = self.mainline_pacer.clone();
|
||||
|
||||
let admin_identity = self.admin_identity(bucket, &object_name, version_id.as_deref(), incarnation);
|
||||
@@ -1698,107 +1724,125 @@ impl ErasureSetHealer {
|
||||
.as_ref()
|
||||
.map(|_| (dedup_key.clone(), object_name.clone(), version_id.clone()));
|
||||
let work = async move {
|
||||
let permit = acquire_page_permit(semaphore, mainline_pacer.as_deref(), &cancel_token).await;
|
||||
let mut retries = 0;
|
||||
let mut admin_identity = admin_identity;
|
||||
let mut admin_task_id = admin_task_id;
|
||||
loop {
|
||||
let permit = acquire_page_permit(semaphore.clone(), mainline_pacer.as_deref(), &cancel_token).await;
|
||||
|
||||
let _permit = match permit {
|
||||
Ok(permit) => permit,
|
||||
Err(err) => return (dedup_key, object_name, version_id, (0, Err(err)), None),
|
||||
};
|
||||
let permit = match permit {
|
||||
Ok(permit) => permit,
|
||||
Err(err) => return (dedup_key, object_name, version_id, (0, Err(err)), None),
|
||||
};
|
||||
|
||||
let _in_flight_guard = PageConcurrencyGuard::new(in_flight, set_label);
|
||||
let in_flight_guard = PageConcurrencyGuard::new(in_flight.clone(), set_label.clone());
|
||||
|
||||
if let (Some(identity), Some(task_id)) = (admin_identity, admin_task_id) {
|
||||
let (result, canonical) =
|
||||
admin::heal_object(storage.as_ref(), &heal_opts, identity, &task_id, &cancel_token).await;
|
||||
return (dedup_key, object_name, version_id, result, canonical);
|
||||
}
|
||||
if let (Some(identity), Some(task_id)) = (admin_identity.take(), admin_task_id.take()) {
|
||||
let (result, canonical) =
|
||||
admin::heal_object(storage.as_ref(), &heal_opts, identity, &task_id, &cancel_token).await;
|
||||
return (dedup_key, object_name, version_id, result, canonical);
|
||||
}
|
||||
|
||||
// Always go through heal_object. Genuine absence flows through
|
||||
// heal_object -> FileVersionNotFound/FileNotFound ->
|
||||
// classify_heal_object_error -> Absent, so gone versions are
|
||||
// recorded as skipped-ok rather than failed. The delete-marker
|
||||
// vs data path is chosen internally in ops/heal.rs.
|
||||
let result = if cancel_token.is_cancelled() {
|
||||
(0, Err(Error::TaskCancelled))
|
||||
} else {
|
||||
match storage
|
||||
.heal_object(&bucket_name, &object_name, version_id.as_deref(), &heal_opts)
|
||||
.await
|
||||
{
|
||||
Ok((result, None)) if target_outcomes_complete(&result, &target_endpoints) => {
|
||||
let object_size = result_object_size_u64(&result);
|
||||
if !replacement_commit_evidence_required {
|
||||
(object_size, Ok(true))
|
||||
} else {
|
||||
match storage
|
||||
.replacement_targets_have_version(
|
||||
&bucket_name,
|
||||
&object_name,
|
||||
version_id.as_deref(),
|
||||
&heal_opts,
|
||||
&target_endpoints,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(true) => (object_size, Ok(true)),
|
||||
Ok(false) => (
|
||||
object_size,
|
||||
Err(Error::transient_skip(format!(
|
||||
"Skipped heal for {bucket_name}/{object_name} because replacement target readback did not confirm the committed version"
|
||||
))),
|
||||
),
|
||||
Err(err) => (
|
||||
object_size,
|
||||
Err(Error::transient_skip(format!(
|
||||
"Skipped heal for {bucket_name}/{object_name} because replacement target readback failed: {err}"
|
||||
))),
|
||||
),
|
||||
// Always go through heal_object. Genuine absence flows through
|
||||
// heal_object -> FileVersionNotFound/FileNotFound ->
|
||||
// classify_heal_object_error -> Absent, so gone versions are
|
||||
// recorded as skipped-ok rather than failed. The delete-marker
|
||||
// vs data path is chosen internally in ops/heal.rs.
|
||||
let result = if cancel_token.is_cancelled() {
|
||||
(0, Err(Error::TaskCancelled))
|
||||
} else {
|
||||
match storage
|
||||
.heal_object(&bucket_name, &object_name, version_id.as_deref(), &heal_opts)
|
||||
.await
|
||||
{
|
||||
Ok((result, None)) if target_outcomes_complete(&result, &target_endpoints) => {
|
||||
let object_size = result_object_size_u64(&result);
|
||||
if !replacement_commit_evidence_required {
|
||||
(object_size, Ok(true))
|
||||
} else {
|
||||
match storage
|
||||
.replacement_targets_have_version(
|
||||
&bucket_name,
|
||||
&object_name,
|
||||
version_id.as_deref(),
|
||||
&heal_opts,
|
||||
&target_endpoints,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(true) => (object_size, Ok(true)),
|
||||
Ok(false) => (
|
||||
object_size,
|
||||
Err(Error::transient_skip(format!(
|
||||
"Skipped heal for {bucket_name}/{object_name} because replacement target readback did not confirm the committed version"
|
||||
))),
|
||||
),
|
||||
Err(err) => (
|
||||
object_size,
|
||||
Err(Error::transient_skip(format!(
|
||||
"Skipped heal for {bucket_name}/{object_name} because replacement target readback failed: {err}"
|
||||
))),
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok((result, None)) if !target_endpoints.is_empty() => (
|
||||
result_object_size_u64(&result),
|
||||
Err(Error::transient_skip(format!(
|
||||
"Skipped heal for {bucket_name}/{object_name} because a replacement target was not committed"
|
||||
))),
|
||||
),
|
||||
Ok((result, None)) => (result_object_size_u64(&result), Ok(true)),
|
||||
Ok((result, Some(err))) if is_missing_object_dir_heal_result(&object_name, &err) => {
|
||||
(result_object_size_u64(&result), Ok(false))
|
||||
}
|
||||
Ok((result, Some(err))) => {
|
||||
let object_size = result_object_size_u64(&result);
|
||||
match Self::classify_heal_object_error(&err) {
|
||||
HealObjectOutcome::Absent => (object_size, Ok(false)),
|
||||
Ok((result, None)) if !target_endpoints.is_empty() => (
|
||||
result_object_size_u64(&result),
|
||||
Err(Error::transient_skip(format!(
|
||||
"Skipped heal for {bucket_name}/{object_name} because a replacement target was not committed"
|
||||
))),
|
||||
),
|
||||
Ok((result, None)) => (result_object_size_u64(&result), Ok(true)),
|
||||
Ok((result, Some(err))) if is_missing_object_dir_heal_result(&object_name, &err) => {
|
||||
(result_object_size_u64(&result), Ok(false))
|
||||
}
|
||||
Ok((result, Some(err))) => {
|
||||
let object_size = result_object_size_u64(&result);
|
||||
match Self::classify_heal_object_error(&err) {
|
||||
HealObjectOutcome::Absent => (object_size, Ok(false)),
|
||||
HealObjectOutcome::DanglingDeleteDeferred { retry_not_before } => {
|
||||
(object_size, Err(Error::DanglingDeleteDeferred { retry_not_before }))
|
||||
}
|
||||
HealObjectOutcome::Transient => (
|
||||
object_size,
|
||||
Err(Error::transient_skip(format!(
|
||||
"Skipped heal for {bucket_name}/{object_name} due to transient error: {err}"
|
||||
))),
|
||||
),
|
||||
HealObjectOutcome::Failed => (object_size, Err(err)),
|
||||
}
|
||||
}
|
||||
Err(err) => match Self::classify_heal_object_error(&err) {
|
||||
HealObjectOutcome::Absent => (0, Ok(false)),
|
||||
HealObjectOutcome::DanglingDeleteDeferred { retry_not_before } => {
|
||||
(object_size, Err(Error::DanglingDeleteDeferred { retry_not_before }))
|
||||
(0, Err(Error::DanglingDeleteDeferred { retry_not_before }))
|
||||
}
|
||||
HealObjectOutcome::Transient => (
|
||||
object_size,
|
||||
0,
|
||||
Err(Error::transient_skip(format!(
|
||||
"Skipped heal for {bucket_name}/{object_name} due to transient error: {err}"
|
||||
))),
|
||||
),
|
||||
HealObjectOutcome::Failed => (object_size, Err(err)),
|
||||
}
|
||||
HealObjectOutcome::Failed => (0, Err(err)),
|
||||
},
|
||||
}
|
||||
Err(err) => match Self::classify_heal_object_error(&err) {
|
||||
HealObjectOutcome::Absent => (0, Ok(false)),
|
||||
HealObjectOutcome::DanglingDeleteDeferred { retry_not_before } => {
|
||||
(0, Err(Error::DanglingDeleteDeferred { retry_not_before }))
|
||||
}
|
||||
HealObjectOutcome::Transient => (
|
||||
0,
|
||||
Err(Error::transient_skip(format!(
|
||||
"Skipped heal for {bucket_name}/{object_name} due to transient error: {err}"
|
||||
))),
|
||||
),
|
||||
HealObjectOutcome::Failed => (0, Err(err)),
|
||||
},
|
||||
}
|
||||
};
|
||||
};
|
||||
|
||||
(dedup_key, object_name, version_id, result, None)
|
||||
drop(in_flight_guard);
|
||||
drop(permit);
|
||||
if retry_replacement
|
||||
&& retries < REPLACEMENT_OBJECT_MAX_RETRIES
|
||||
&& matches!(&result.1, Err(Error::TransientSkip { .. }))
|
||||
{
|
||||
retries += 1;
|
||||
tokio::select! {
|
||||
_ = cancel_token.cancelled() => return (dedup_key, object_name, version_id, (0, Err(Error::TaskCancelled)), None),
|
||||
_ = tokio::time::sleep(Duration::from_millis(250 * u64::from(retries))) => {}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
return (dedup_key, object_name, version_id, result, None);
|
||||
}
|
||||
};
|
||||
page_tasks.push(async move {
|
||||
if let (Some(execution), Some((key, object, version))) = (execution, failure_identity) {
|
||||
@@ -2129,6 +2173,7 @@ mod tests {
|
||||
ErasureSetHealer::effective_heal_page_object_concurrency_for_source(
|
||||
HealRequestSource::AutoHeal,
|
||||
HealScanMode::Normal,
|
||||
false,
|
||||
),
|
||||
1
|
||||
);
|
||||
@@ -2136,6 +2181,73 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn automatic_replacement_uses_a_conservative_default_and_respects_serial_controls() {
|
||||
temp_env::with_vars(
|
||||
[
|
||||
(rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")),
|
||||
(rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, None),
|
||||
],
|
||||
|| {
|
||||
assert_eq!(
|
||||
ErasureSetHealer::effective_heal_page_object_concurrency_for_source(
|
||||
HealRequestSource::AutoHeal,
|
||||
HealScanMode::Normal,
|
||||
true,
|
||||
),
|
||||
2
|
||||
);
|
||||
assert_eq!(
|
||||
ErasureSetHealer::effective_heal_page_object_concurrency_for_source(
|
||||
HealRequestSource::AutoHeal,
|
||||
HealScanMode::Deep,
|
||||
true,
|
||||
),
|
||||
1
|
||||
);
|
||||
},
|
||||
);
|
||||
temp_env::with_vars(
|
||||
[
|
||||
(rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("false")),
|
||||
(rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("8")),
|
||||
],
|
||||
|| {
|
||||
assert_eq!(
|
||||
ErasureSetHealer::effective_heal_page_object_concurrency_for_source(
|
||||
HealRequestSource::AutoHeal,
|
||||
HealScanMode::Normal,
|
||||
true,
|
||||
),
|
||||
1
|
||||
);
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn automatic_replacement_concurrency_fits_the_semaphore() {
|
||||
let maximum = usize::MAX.to_string();
|
||||
for (value, expected) in [("0", 1), (maximum.as_str(), tokio::sync::Semaphore::MAX_PERMITS)] {
|
||||
temp_env::with_vars(
|
||||
[
|
||||
(rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")),
|
||||
(rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some(value)),
|
||||
],
|
||||
|| {
|
||||
assert_eq!(
|
||||
ErasureSetHealer::effective_heal_page_object_concurrency_for_source(
|
||||
HealRequestSource::AutoHeal,
|
||||
HealScanMode::Normal,
|
||||
true,
|
||||
),
|
||||
expected
|
||||
);
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn non_auto_normal_scan_page_object_concurrency_uses_effective_limit() {
|
||||
temp_env::with_vars(
|
||||
@@ -2148,6 +2260,7 @@ mod tests {
|
||||
ErasureSetHealer::effective_heal_page_object_concurrency_for_source(
|
||||
HealRequestSource::Admin,
|
||||
HealScanMode::Normal,
|
||||
false,
|
||||
),
|
||||
11
|
||||
);
|
||||
@@ -2520,6 +2633,8 @@ mod resume_loop_tests {
|
||||
lifecycle_expired: Mutex<HashSet<String>>,
|
||||
/// every heal_object call recorded as (name, version_id)
|
||||
heal_calls: Mutex<Vec<(String, Option<String>)>>,
|
||||
heal_gate: Mutex<Option<Arc<tokio::sync::Semaphore>>>,
|
||||
heal_started: tokio::sync::Notify,
|
||||
list_include_lifecycle_object_info: Mutex<Vec<bool>>,
|
||||
replacement_target_identity_sequences: Mutex<VecDeque<Vec<ReplacementTargetIdentity>>>,
|
||||
replacement_targets_unready: AtomicBool,
|
||||
@@ -2704,6 +2819,11 @@ mod resume_loop_tests {
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push((object.to_string(), version_id.map(str::to_string)));
|
||||
self.heal_started.notify_one();
|
||||
let heal_gate = self.heal_gate.lock().unwrap().clone();
|
||||
if let Some(gate) = heal_gate {
|
||||
gate.acquire().await.unwrap().forget();
|
||||
}
|
||||
let key = compose_key(object, version_id);
|
||||
let outcome = {
|
||||
let mut outcomes = self.outcomes.lock().unwrap();
|
||||
@@ -3898,6 +4018,287 @@ mod resume_loop_tests {
|
||||
assert_eq!(unique.len(), 50, "no version healed twice");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_replacement_runs_two_object_repairs_within_the_configured_limit() {
|
||||
temp_env::async_with_vars(
|
||||
[
|
||||
(rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")),
|
||||
(rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("2")),
|
||||
],
|
||||
async {
|
||||
let mut env = make_env().await;
|
||||
env.healer.source = HealRequestSource::AutoHeal;
|
||||
env.healer = env
|
||||
.healer
|
||||
.with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone()));
|
||||
let gate = Arc::new(tokio::sync::Semaphore::new(0));
|
||||
*env.storage.heal_gate.lock().unwrap() = Some(gate.clone());
|
||||
let objects = ["a", "b", "c"];
|
||||
for name in objects {
|
||||
env.storage
|
||||
.set_result(name, None, replacement_target_ok_result("replacement-a", name));
|
||||
}
|
||||
env.storage.set_page(
|
||||
None,
|
||||
Page {
|
||||
items: objects.into_iter().map(|name| item(name, None, false)).collect(),
|
||||
next: None,
|
||||
truncated: false,
|
||||
},
|
||||
);
|
||||
|
||||
let observe = async {
|
||||
let two_started = tokio::time::timeout(std::time::Duration::from_secs(2), async {
|
||||
loop {
|
||||
let notified = env.storage.heal_started.notified();
|
||||
if env.storage.calls().len() >= 2 {
|
||||
break;
|
||||
}
|
||||
notified.await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.is_ok();
|
||||
let started = env.storage.calls().len();
|
||||
gate.add_permits(objects.len());
|
||||
assert!(two_started, "automatic replacement should start two repairs before either finishes");
|
||||
assert_eq!(started, 2, "configured concurrency must bound in-flight repairs");
|
||||
};
|
||||
let ((processed, successful, failed, skipped, result), ()) = tokio::join!(run(&env), observe);
|
||||
result.expect("all replacement objects must be verified");
|
||||
assert_eq!((processed, successful, failed, skipped), (3, 3, 0, 0));
|
||||
assert_eq!(env.storage.calls().len(), 3);
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_replacement_without_object_locks_stays_serial() {
|
||||
temp_env::async_with_vars(
|
||||
[
|
||||
(rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")),
|
||||
(rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("2")),
|
||||
],
|
||||
async {
|
||||
let mut env = make_env().await;
|
||||
env.healer.source = HealRequestSource::AutoHeal;
|
||||
env.healer.heal_opts.no_lock = true;
|
||||
env.healer = env
|
||||
.healer
|
||||
.with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone()));
|
||||
let gate = Arc::new(tokio::sync::Semaphore::new(0));
|
||||
*env.storage.heal_gate.lock().unwrap() = Some(gate.clone());
|
||||
for version in ["v1", "v2"] {
|
||||
env.storage
|
||||
.set_result("object", Some(version), replacement_target_ok_result("replacement-a", "object"));
|
||||
}
|
||||
env.storage.set_page(
|
||||
None,
|
||||
Page {
|
||||
items: vec![item("object", Some("v1"), false), item("object", Some("v2"), false)],
|
||||
next: None,
|
||||
truncated: false,
|
||||
},
|
||||
);
|
||||
let observe = async {
|
||||
let first_started = tokio::time::timeout(std::time::Duration::from_secs(5), async {
|
||||
loop {
|
||||
let notified = env.storage.heal_started.notified();
|
||||
if !env.storage.calls().is_empty() {
|
||||
break;
|
||||
}
|
||||
notified.await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.is_ok();
|
||||
let two_started = tokio::time::timeout(std::time::Duration::from_secs(2), async {
|
||||
loop {
|
||||
let notified = env.storage.heal_started.notified();
|
||||
if env.storage.calls().len() >= 2 {
|
||||
break;
|
||||
}
|
||||
notified.await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.is_ok();
|
||||
gate.add_permits(2);
|
||||
assert!(first_started, "the first object repair must start");
|
||||
assert!(!two_started, "unlocked object repairs must not overlap");
|
||||
};
|
||||
let ((processed, successful, failed, skipped, result), ()) = tokio::join!(run(&env), observe);
|
||||
result.expect("the serial replacement must finish");
|
||||
assert_eq!((processed, successful, failed, skipped), (2, 2, 0, 0));
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_replacement_retries_a_transient_object_before_rewinding_the_scan() {
|
||||
let mut env = make_env().await;
|
||||
env.healer.source = HealRequestSource::AutoHeal;
|
||||
env.healer = env
|
||||
.healer
|
||||
.with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone()));
|
||||
env.storage.set_page(
|
||||
None,
|
||||
Page {
|
||||
items: vec![item("object", Some("version"), false)],
|
||||
next: None,
|
||||
truncated: false,
|
||||
},
|
||||
);
|
||||
env.storage
|
||||
.set_outcome("object", Some("version"), HealOutcome::RpcCancelled(1));
|
||||
env.storage
|
||||
.set_result("object", Some("version"), replacement_target_ok_result("replacement-a", "object"));
|
||||
|
||||
let (processed, successful, failed, skipped, result) = run(&env).await;
|
||||
result.expect("the transient repair should succeed during the current pass");
|
||||
assert_eq!((processed, successful, failed, skipped), (1, 1, 0, 0));
|
||||
assert_eq!(env.storage.calls().len(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_replacement_releases_capacity_during_a_bounded_retry() {
|
||||
temp_env::async_with_vars(
|
||||
[
|
||||
(rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")),
|
||||
(rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("1")),
|
||||
],
|
||||
async {
|
||||
let mut env = make_env().await;
|
||||
env.healer.source = HealRequestSource::AutoHeal;
|
||||
env.healer = env
|
||||
.healer
|
||||
.with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone()));
|
||||
env.storage.set_page(
|
||||
None,
|
||||
Page {
|
||||
items: vec![item("retry", None, false), item("ready", None, false)],
|
||||
next: None,
|
||||
truncated: false,
|
||||
},
|
||||
);
|
||||
env.storage.set_outcome("retry", None, HealOutcome::Transient);
|
||||
env.storage
|
||||
.set_result("ready", None, replacement_target_ok_result("replacement-a", "ready"));
|
||||
|
||||
let (processed, successful, failed, skipped, result) = run(&env).await;
|
||||
result.expect("a failed object is retained as a transient skip");
|
||||
assert_eq!((processed, successful, failed, skipped), (2, 1, 0, 1));
|
||||
let calls = env.storage.calls();
|
||||
let retries: Vec<_> = calls.iter().enumerate().filter(|(_, call)| call.0 == "retry").collect();
|
||||
let ready = calls.iter().position(|call| call.0 == "ready").expect("ready object ran");
|
||||
assert_eq!(retries.len(), 3, "only two retries are allowed");
|
||||
assert!(ready < retries[1].0, "the backoff must release the only page permit");
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_replacement_cancellation_interrupts_object_retry() {
|
||||
let mut env = make_env().await;
|
||||
env.healer.source = HealRequestSource::AutoHeal;
|
||||
env.healer = env
|
||||
.healer
|
||||
.with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone()));
|
||||
env.storage.set_page(
|
||||
None,
|
||||
Page {
|
||||
items: vec![item("retry", None, false)],
|
||||
next: None,
|
||||
truncated: false,
|
||||
},
|
||||
);
|
||||
env.storage.set_outcome("retry", None, HealOutcome::Transient);
|
||||
let cancel = env.healer.cancel_token.clone();
|
||||
let observe = async {
|
||||
loop {
|
||||
let notified = env.storage.heal_started.notified();
|
||||
if !env.storage.calls().is_empty() {
|
||||
break;
|
||||
}
|
||||
notified.await;
|
||||
}
|
||||
cancel.cancel();
|
||||
};
|
||||
|
||||
let ((_, successful, _, _, result), ()) = tokio::join!(run(&env), observe);
|
||||
assert!(matches!(result, Err(Error::TaskCancelled)));
|
||||
assert_eq!(successful, 0);
|
||||
assert_eq!(env.storage.calls().len(), 1);
|
||||
assert!(env.checkpoint.get_checkpoint().await.processed_objects.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_replacement_cancel_retains_workers_until_inflight_repairs_settle() {
|
||||
temp_env::async_with_vars(
|
||||
[
|
||||
(rustfs_config::ENV_HEAL_PAGE_PARALLEL_ENABLE, Some("true")),
|
||||
(rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("2")),
|
||||
],
|
||||
async {
|
||||
let mut env = make_env().await;
|
||||
env.healer.source = HealRequestSource::AutoHeal;
|
||||
let execution = crate::heal::storage::ReplacementExecution::for_test(Vec::new(), Vec::new());
|
||||
env.healer = env
|
||||
.healer
|
||||
.with_replacement_targets(vec!["replacement-a".to_string()], Some(env.task_id.clone()))
|
||||
.with_replacement_execution(Some(execution.clone()));
|
||||
let gate = Arc::new(tokio::sync::Semaphore::new(0));
|
||||
*env.storage.heal_gate.lock().unwrap() = Some(gate.clone());
|
||||
for name in ["a", "b", "c"] {
|
||||
env.storage
|
||||
.set_result(name, None, replacement_target_ok_result("replacement-a", name));
|
||||
}
|
||||
env.storage.set_page(
|
||||
None,
|
||||
Page {
|
||||
items: ["a", "b", "c"].into_iter().map(|name| item(name, None, false)).collect(),
|
||||
next: None,
|
||||
truncated: false,
|
||||
},
|
||||
);
|
||||
let cancel = env.healer.cancel_token.clone();
|
||||
let observe = async {
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), async {
|
||||
loop {
|
||||
let notified = env.storage.heal_started.notified();
|
||||
if env.storage.calls().len() >= 2 {
|
||||
break;
|
||||
}
|
||||
notified.await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("two repairs must start");
|
||||
cancel.cancel();
|
||||
};
|
||||
|
||||
let ((_, successful, _, _, result), ()) = tokio::join!(run(&env), observe);
|
||||
assert!(matches!(result, Err(Error::TaskCancelled)));
|
||||
assert_eq!(successful, 0);
|
||||
assert_eq!(env.storage.calls().len(), 2, "queued work must not start after cancellation");
|
||||
assert!(env.checkpoint.get_checkpoint().await.processed_objects.is_empty());
|
||||
assert!(Arc::strong_count(&execution) >= 4, "in-flight workers must retain execution ownership");
|
||||
gate.add_permits(2);
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), async {
|
||||
while Arc::strong_count(&execution) > 2 {
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("workers must release execution ownership after their storage work settles");
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_resume_after_version_deleted_midheal_no_skip() {
|
||||
let env = make_env().await;
|
||||
@@ -4312,7 +4713,11 @@ mod resume_loop_tests {
|
||||
assert!(!state.completed, "readback errors must leave the replacement task incomplete");
|
||||
assert_eq!(state.retry_count, 1, "readback errors must arm the bounded retry path");
|
||||
assert!(env.checkpoint.get_checkpoint().await.processed_objects.is_empty());
|
||||
assert_eq!(env.storage.calls(), vec![("object".to_string(), Some("dm-v1".to_string()))]);
|
||||
assert_eq!(
|
||||
env.storage.calls(),
|
||||
vec![("object".to_string(), Some("dm-v1".to_string())); 3],
|
||||
"unverified readback must exhaust only the bounded object retries"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -30,6 +30,30 @@ pub(super) fn retry_admission_delay(
|
||||
Some(delay)
|
||||
}
|
||||
|
||||
pub(super) fn running_mainline_pacer(
|
||||
request: &HealRequest,
|
||||
config: &HealConfig,
|
||||
provider: &Option<WorkloadSnapshotProviderRef>,
|
||||
) -> Option<Arc<crate::heal::pacing::MainlinePacer>> {
|
||||
let pace_running = request.source == HealRequestSource::Admin
|
||||
|| (request.source == HealRequestSource::AutoHeal
|
||||
&& matches!(&request.heal_type, HealType::ErasureSet { .. })
|
||||
&& !request.heal_endpoints.is_empty());
|
||||
if !pace_running || !config.mainline_throttle_enable {
|
||||
return None;
|
||||
}
|
||||
provider.as_ref().and_then(|provider| {
|
||||
crate::heal::pacing::MainlinePacer::new(
|
||||
provider.clone(),
|
||||
config.mainline_read_utilization_high_percent,
|
||||
config.mainline_write_utilization_high_percent,
|
||||
config.mainline_max_sleep,
|
||||
request.source,
|
||||
)
|
||||
.map(Arc::new)
|
||||
})
|
||||
}
|
||||
|
||||
impl HealManager {
|
||||
/// Start scheduler
|
||||
pub(super) async fn start_scheduler(&self) -> Result<()> {
|
||||
@@ -202,19 +226,7 @@ impl HealManager {
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
.get(&request.id)
|
||||
.cloned();
|
||||
let mainline_pacer = if request.source == HealRequestSource::Admin && config.mainline_throttle_enable {
|
||||
workload_provider.as_ref().and_then(|provider| {
|
||||
crate::heal::pacing::MainlinePacer::new(
|
||||
provider.clone(),
|
||||
config.mainline_read_utilization_high_percent,
|
||||
config.mainline_write_utilization_high_percent,
|
||||
config.mainline_max_sleep,
|
||||
)
|
||||
.map(Arc::new)
|
||||
})
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let mainline_pacer = running_mainline_pacer(&request, &config, workload_provider);
|
||||
let task = Arc::new(
|
||||
HealTask::from_replacement_recovery_request(request, storage.clone(), replacement_resume_endpoint)
|
||||
.with_mainline_pacer(mainline_pacer)
|
||||
|
||||
@@ -6,6 +6,41 @@ use crate::heal::storage::HealListItem;
|
||||
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
||||
use tokio::sync::Semaphore;
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn automatic_replacement_uses_running_foreground_pacing() {
|
||||
let provider = Arc::new(super::super::super::pacing::TestPressure::new(WorkloadClass::ForegroundRead, 100));
|
||||
let provider: Option<WorkloadSnapshotProviderRef> = Some(provider);
|
||||
let mut config = HealConfig {
|
||||
mainline_throttle_enable: true,
|
||||
mainline_read_utilization_high_percent: 80,
|
||||
mainline_max_sleep: Duration::from_millis(250),
|
||||
..HealConfig::default()
|
||||
};
|
||||
let mut request = HealRequest::new(
|
||||
HealType::ErasureSet {
|
||||
buckets: Vec::new(),
|
||||
set_disk_id: "pool_0_set_0".to_owned(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Low,
|
||||
);
|
||||
request.source = HealRequestSource::AutoHeal;
|
||||
request.heal_endpoints = vec!["replacement-a".to_owned()];
|
||||
|
||||
let pacer = super::super::scheduler::running_mainline_pacer(&request, &config, &provider)
|
||||
.expect("automatic replacement must pace under foreground pressure");
|
||||
let started = tokio::time::Instant::now();
|
||||
pacer.wait(&CancellationToken::new()).await.expect("bounded progress");
|
||||
assert_eq!(started.elapsed(), Duration::from_millis(250));
|
||||
|
||||
assert!(super::super::scheduler::running_mainline_pacer(&request, &config, &None).is_none());
|
||||
config.mainline_throttle_enable = false;
|
||||
assert!(super::super::scheduler::running_mainline_pacer(&request, &config, &provider).is_none());
|
||||
config.mainline_throttle_enable = true;
|
||||
request.heal_endpoints.clear();
|
||||
assert!(super::super::scheduler::running_mainline_pacer(&request, &config, &provider).is_none());
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct PressureProbe {
|
||||
active: AtomicUsize,
|
||||
|
||||
@@ -6,6 +6,7 @@ use rustfs_concurrency::{
|
||||
WorkloadAdmissionSnapshotProvider,
|
||||
workload::{ForegroundPressure, foreground_pressure},
|
||||
};
|
||||
use rustfs_heal_contracts::heal_channel::HealRequestSource;
|
||||
use std::{sync::Arc, time::Duration};
|
||||
use tokio::{sync::Mutex, time::Instant};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
@@ -21,12 +22,13 @@ pub(crate) enum PacingDecision {
|
||||
Wait(Option<ForegroundPressure>),
|
||||
}
|
||||
|
||||
/// Cooperative pacing for one admin execution, not a storage admission permit.
|
||||
/// Cooperative pacing for one heal execution, not a storage admission permit.
|
||||
pub(crate) struct MainlinePacer {
|
||||
provider: Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>,
|
||||
read_high: usize,
|
||||
write_high: usize,
|
||||
pause: Duration,
|
||||
source: HealRequestSource,
|
||||
state: Mutex<PacingState>,
|
||||
}
|
||||
|
||||
@@ -36,6 +38,7 @@ impl MainlinePacer {
|
||||
read_high: usize,
|
||||
write_high: usize,
|
||||
pause: Duration,
|
||||
source: HealRequestSource,
|
||||
) -> Option<Self> {
|
||||
if (read_high == 0 && write_high == 0) || pause.is_zero() {
|
||||
return None;
|
||||
@@ -45,6 +48,7 @@ impl MainlinePacer {
|
||||
read_high: read_high.min(100),
|
||||
write_high: write_high.min(100),
|
||||
pause: pause.min(Duration::from_secs(1)),
|
||||
source,
|
||||
state: Mutex::new(PacingState::default()),
|
||||
})
|
||||
}
|
||||
@@ -108,7 +112,7 @@ impl MainlinePacer {
|
||||
}
|
||||
metrics::counter!(
|
||||
"rustfs_heal_mainline_throttle_total",
|
||||
"source" => "admin",
|
||||
"source" => self.source.as_str(),
|
||||
"result" => "delayed",
|
||||
"reason" => pressure.or(observed).map_or("recovery_window", |pressure| pressure.reason())
|
||||
)
|
||||
@@ -170,6 +174,7 @@ mod tests {
|
||||
if class == WorkloadClass::ForegroundRead { 40 } else { 0 },
|
||||
if class == WorkloadClass::ForegroundWrite { 40 } else { 0 },
|
||||
pause,
|
||||
HealRequestSource::Admin,
|
||||
)
|
||||
.expect("enabled pacer");
|
||||
let cancel = CancellationToken::new();
|
||||
@@ -192,9 +197,9 @@ mod tests {
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn running_mainline_huge_pause_is_capped_and_disabled_classes_do_not_sleep() {
|
||||
let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 100));
|
||||
assert!(MainlinePacer::new(provider.clone(), 0, 0, Duration::from_secs(1)).is_none());
|
||||
assert!(MainlinePacer::new(provider.clone(), 80, 80, Duration::ZERO).is_none());
|
||||
let pacer = MainlinePacer::new(provider, 80, 80, Duration::from_secs(3600)).expect("pacer");
|
||||
assert!(MainlinePacer::new(provider.clone(), 0, 0, Duration::from_secs(1), HealRequestSource::Admin).is_none());
|
||||
assert!(MainlinePacer::new(provider.clone(), 80, 80, Duration::ZERO, HealRequestSource::Admin).is_none());
|
||||
let pacer = MainlinePacer::new(provider, 80, 80, Duration::from_secs(3600), HealRequestSource::Admin).expect("pacer");
|
||||
let before = Instant::now();
|
||||
pacer.wait(&CancellationToken::new()).await.expect("hard-capped pause");
|
||||
assert_eq!(Instant::now() - before, Duration::from_secs(1));
|
||||
@@ -203,7 +208,9 @@ mod tests {
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn running_mainline_waiters_cancel_and_task_latches_are_isolated() {
|
||||
let provider = Arc::new(TestPressure::new(WorkloadClass::ForegroundRead, 100));
|
||||
let paced = Arc::new(MainlinePacer::new(provider.clone(), 80, 80, Duration::from_secs(1)).expect("pacer"));
|
||||
let paced = Arc::new(
|
||||
MainlinePacer::new(provider.clone(), 80, 80, Duration::from_secs(1), HealRequestSource::Admin).expect("pacer"),
|
||||
);
|
||||
let cancel_first = CancellationToken::new();
|
||||
let first = tokio::spawn({
|
||||
let paced = paced.clone();
|
||||
@@ -221,7 +228,8 @@ mod tests {
|
||||
cancel_second.cancel();
|
||||
assert!(matches!(second.await.expect("queued waiter"), Err(Error::TaskCancelled)));
|
||||
provider.active.store(0, Ordering::SeqCst);
|
||||
let other_task = MainlinePacer::new(provider.clone(), 80, 80, Duration::from_secs(1)).expect("independent task");
|
||||
let other_task = MainlinePacer::new(provider.clone(), 80, 80, Duration::from_secs(1), HealRequestSource::Admin)
|
||||
.expect("independent task");
|
||||
let before = Instant::now();
|
||||
other_task
|
||||
.wait(&CancellationToken::new())
|
||||
|
||||
@@ -489,9 +489,13 @@ impl HealTask {
|
||||
))
|
||||
.await?;
|
||||
if let Some(error) = storage_result.error {
|
||||
return Err(Error::TaskExecutionFailed {
|
||||
message: format!("Failed to recreate missing object {bucket}/{object}: {error}"),
|
||||
});
|
||||
if error.is_dangling_delete_grace() {
|
||||
// Preserve the existing MRF grace retry ownership.
|
||||
return Err(Error::TaskExecutionFailed {
|
||||
message: format!("Failed to recreate missing object {bucket}/{object}: {error}"),
|
||||
});
|
||||
}
|
||||
return Err(error);
|
||||
}
|
||||
|
||||
let object_size = storage_result.item.object_size as u64;
|
||||
@@ -526,7 +530,7 @@ impl HealTask {
|
||||
}
|
||||
} else {
|
||||
Error::TaskExecutionFailed {
|
||||
message: format!("Missing exact storage proof for durable MRF repair {bucket}/{object}"),
|
||||
message: format!("{}{bucket}/{object}", crate::error::MRF_MISSING_PROOF_PREFIX),
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -2912,6 +2912,44 @@ fn mrf_task_for_item(object: &str, item: HealResultItem) -> (HealTask, Arc<MockS
|
||||
(HealTask::from_request(request, storage.clone()), storage)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn durable_mrf_slowdown_remains_retryable_without_a_storage_receipt() {
|
||||
let (task, storage) = mrf_task_for_item("object", HealResultItem::default());
|
||||
*storage.heal_object_outcome.lock().unwrap() = Some(MockHealObjectOutcome::RetryableSlowDown);
|
||||
|
||||
let error = task
|
||||
.execute()
|
||||
.await
|
||||
.expect_err("a failed repair cannot discharge the responsibility");
|
||||
assert!(matches!(error, Error::Storage(EcstoreError::SlowDown)), "{error:?}");
|
||||
assert!(error.is_recoverable_heal());
|
||||
assert_eq!(task.get_outcome().await.counters.healed, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn durable_mrf_grace_remains_owned_by_the_responsibility() {
|
||||
let (task, storage) = mrf_task_for_item("object", HealResultItem::default());
|
||||
*storage.heal_object_outcome.lock().unwrap() = Some(MockHealObjectOutcome::DanglingGraceDeferred);
|
||||
|
||||
let error = task.execute().await.expect_err("grace must retain the responsibility");
|
||||
assert!(error.is_dangling_delete_grace());
|
||||
assert!(!error.is_recoverable_heal());
|
||||
assert_eq!(task.get_outcome().await.counters.healed, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn durable_mrf_unverified_object_names_cannot_change_retry_classification() {
|
||||
for object in ["ordinary", "timed out", "network error"] {
|
||||
let (task, _) = mrf_task_for_item(object, HealResultItem::default());
|
||||
let error = task
|
||||
.execute()
|
||||
.await
|
||||
.expect_err("missing proof must retain the responsibility");
|
||||
assert!(!error.is_recoverable_heal(), "object name must not change retry classification: {object}");
|
||||
assert_eq!(task.get_outcome().await.counters.healed, 0);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn mrf_legacy_object_names_cannot_change_retry_classification() {
|
||||
for object in [
|
||||
|
||||
@@ -565,11 +565,11 @@ Heal knobs are environment-only and read by `HealConfig::default` (`crates/heal/
|
||||
| `RUSTFS_HEAL_EVENT_DRIVEN_SCHEDULER_ENABLE` | `true` (`DEFAULT_HEAL_EVENT_DRIVEN_SCHEDULER_ENABLE`) | Notify-driven scheduler wakeups. |
|
||||
| `RUSTFS_HEAL_SET_BULKHEAD_ENABLE` | `true` (`DEFAULT_HEAL_SET_BULKHEAD_ENABLE`) | Per-set bulkhead scheduling. |
|
||||
| `RUSTFS_HEAL_PAGE_PARALLEL_ENABLE` | `true` (`DEFAULT_HEAL_PAGE_PARALLEL_ENABLE`) | Page-level parallel object healing during erasure-set repair. |
|
||||
| `RUSTFS_HEAL_PAGE_OBJECT_CONCURRENCY` | `8` (`DEFAULT_HEAL_PAGE_OBJECT_CONCURRENCY`) | Concurrent object heals within one erasure-set page. Forced to `1` when page parallelism is off, for `Deep` scan mode, and for `AutoHeal`-sourced requests (`ErasureSetHealer::effective_heal_page_object_concurrency_for_source`). |
|
||||
| `RUSTFS_HEAL_MAINLINE_THROTTLE_ENABLE` | `true` (`DEFAULT_HEAL_MAINLINE_THROTTLE_ENABLE`) | Defer best-effort starts and cooperatively pace running admin heal at safe work boundaries. |
|
||||
| `RUSTFS_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT` | `80` (`DEFAULT_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT`, capped at 100) | Read-utilization high watermark for start admission and running admin pacing; zero disables this class. |
|
||||
| `RUSTFS_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT` | `80` (`DEFAULT_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT`, capped at 100) | Write-utilization high watermark for start admission and running admin pacing; zero disables this class. |
|
||||
| `RUSTFS_HEAL_MAINLINE_MAX_SLEEP_MS` | `250` (`DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS`) | Start recheck interval; running admin waits cap each pacing-gate holder at 1000 ms. Zero disables running pacing. |
|
||||
| `RUSTFS_HEAL_PAGE_OBJECT_CONCURRENCY` | `8` for ordinary heals; `2` for automatic replacement | Concurrent object heals within one erasure-set page. An explicit value applies to both; automatic replacement clamps it to the semaphore's supported range. Forced to `1` when page parallelism is off, for `Deep` scans, and for automatic heals without a replacement generation or object locks. |
|
||||
| `RUSTFS_HEAL_MAINLINE_THROTTLE_ENABLE` | `true` (`DEFAULT_HEAL_MAINLINE_THROTTLE_ENABLE`) | Defer best-effort starts and cooperatively pace running admin and automatic replacement heals at safe work boundaries. |
|
||||
| `RUSTFS_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT` | `80` (`DEFAULT_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT`, capped at 100) | Read-utilization high watermark for start admission and running heal pacing; zero disables this class. |
|
||||
| `RUSTFS_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT` | `80` (`DEFAULT_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT`, capped at 100) | Write-utilization high watermark for start admission and running heal pacing; zero disables this class. |
|
||||
| `RUSTFS_HEAL_MAINLINE_MAX_SLEEP_MS` | `250` (`DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS`) | Start recheck interval; running waits cap each pacing-gate holder at 1000 ms. Zero disables running pacing. |
|
||||
| `RUSTFS_HEAL_OVERLAP_POLICY` | `merge` (`DEFAULT_HEAL_OVERLAP_POLICY`) | `merge` reuses the token of an equivalent active, queued, or retrying admin heal; incompatible same-target starts and intersecting admin scopes return `already_running` / `overlapping_paths`. Durable-only owners reject until recovered or replaced with `forceStart`. `minio_error` also rejects equivalent starts and preserves overlap rejection against background tasks. |
|
||||
| `RUSTFS_HEAL_MRF_ENABLE` | `true` (`DEFAULT_HEAL_MRF_ENABLE`) | MRF intent pipeline: error paths deliver repair intents to the heal runtime and unconsumed intents replay from the durable journal after restart. |
|
||||
| `RUSTFS_HEAL_MRF_QUEUE_SIZE` | `100000` (`DEFAULT_HEAL_MRF_QUEUE_SIZE`) | MRF in-memory queue capacity. |
|
||||
@@ -599,6 +599,8 @@ Admin heal has three separate budgets. Increasing one does not extend the others
|
||||
| Task execution | `RUSTFS_HEAL_TASK_TIMEOUT_SECS` supplies the default of 300 seconds when the execution has no explicit timeout. Elapsed execution time is deducted before a recoverable scheduler retry, which keeps the request identity and remaining budget. Object/listing backoff and pressure pacing inside an execution consume that budget; time queued or in the scheduler's between-attempt backoff is not a new absolute wall-clock deadline. Zero is not an unlimited execution budget. |
|
||||
| Object/listing retry | A recursive bucket/prefix traversal retries recoverable failures at most three times after the initial attempt, with 2/4/8-second delays plus the existing task-derived jitter. Object retries also obey the bounded delayed window and its 30-second age, checked at safe boundaries; listing retries keep their cursor. These limits do not extend the task execution budget or force an in-flight storage operation to finish within the retry age. |
|
||||
|
||||
Automatic replacement retries a transient object repair twice within its current page, waiting 250 ms and then 500 ms without holding page capacity. An exhausted retry remains unverified and still triggers the existing generation retry; target readback and checkpoint requirements are unchanged.
|
||||
|
||||
For a large admin heal, select a finite configured execution budget appropriate to the expected work and contention, and inspect progress while it runs. Investigate stalled work and object failures before increasing this budget; a larger timeout must not hide lock or quorum problems. A longer HTTP or internode timeout does not keep a repair task alive past its execution budget. A successful start response returns the canonical token for accepted or merged work; it does not prove that repair has completed. Query that token without `forceStart`; terminal timeout/cancellation reports retain completed progress while the task report remains available.
|
||||
|
||||
Capability preflight runs before constructing and admitting a start request. However, the public `cluster heal coordination unavailable` error can also follow a transport failure or an invalid coordinator response after a request was sent. An unknown or lost HTTP response is therefore not proof that no task was admitted.
|
||||
@@ -609,13 +611,13 @@ Implementation references: `rustfs/src/admin/handlers/heal.rs` (`submit_cluster_
|
||||
|
||||
### Running admin heal pacing
|
||||
|
||||
The manager passes its existing workload provider and a configuration snapshot into each admin execution. Bucket/prefix listing and object boundaries resample foreground pressure; erasure-set page workers also resample after earlier work releases page capacity. `High`, `Urgent`, and `force_start` do not exempt ordinary admin execution from this runtime pacing. The existing start-time bypass and overlap-control meanings are unchanged.
|
||||
The manager passes its existing workload provider and a configuration snapshot into each admin execution and automatic replacement generation. Bucket/prefix listing and object boundaries resample foreground pressure; erasure-set page workers also resample after earlier work releases page capacity. `High`, `Urgent`, and `force_start` do not exempt ordinary admin execution from this runtime pacing. The existing start-time bypass and overlap-control meanings are unchanged.
|
||||
|
||||
Each execution has its own pacing latch, with no new global manager or cross-set pacing lock. The low watermark for each enabled class is `max(1, floor(high * 3 / 4))`: the default high watermark 80 therefore recovers below 60. High pressure latches pacing, and intermediate pressure resets the recovery window. Unpaced starts resume after sampled pressure remains below the low watermarks for four pause intervals, normally one second. While pressure persists, a pacing-gate holder waits only one interval, at most one second, then permits maintenance to continue. Concurrent page waiters serialize through this task-local gate; queue waiting still counts against the existing task execution timeout.
|
||||
|
||||
The pacing gate holds neither namespace locks nor I/O/page permits while sleeping. At final page admission, each real permit acquisition gets a fresh, nonblocking pressure decision. Low-pressure work keeps that permit; only a unit that needs a pause releases capacity to wait. A unit that has completed one bounded pause may proceed despite persistent pressure, which supplies minimum maintenance progress without an endless acquire/pause loop. Existing object operations and commit tails are not interrupted because pressure rose. Cancellation and deadlines remain interruptible, and disabling pacing cannot bypass the global, per-set or page-concurrency hard caps. The existing `RUSTFS_HEAL_MAINLINE_THROTTLE_ENABLE=false` setting is the operational opt-out for newly created executions; no additional request override is introduced.
|
||||
The pacing gate holds neither namespace locks nor I/O/page permits while sleeping. At final page admission, each real permit acquisition gets a fresh, nonblocking pressure decision. Low-pressure work keeps that permit; only a unit that needs a pause releases capacity to wait. A unit that has completed one bounded pause may proceed despite persistent pressure, which supplies minimum maintenance progress without an endless acquire/pause loop. Existing object operations and commit tails are not interrupted because pressure rose. Cancellation interrupts the pacing wait; replacement operations retain their execution lease until storage work settles. Disabling pacing cannot bypass the global, per-set or page-concurrency hard caps. The existing `RUSTFS_HEAL_MAINLINE_THROTTLE_ENABLE=false` setting is the operational opt-out for newly created executions; no additional request override is introduced.
|
||||
|
||||
A missing provider, zero pause, or both class thresholds set to zero preserves unpaced execution. Missing counts follow the existing shared pressure interpreter; they are observations, not health, quorum or resource-ownership proof. The current provider exposes node-level workload classes, so this does not claim independent per-set foreground measurements or a hard global resource budget. Runtime waits increment `rustfs_heal_mainline_throttle_total` with `source=admin`, `result=delayed`, and a foreground-pressure or `recovery_window` reason. Real p99/throughput protection requires the separate W20 fixed-load ABBA measurements.
|
||||
A missing provider, zero pause, or both class thresholds set to zero preserves unpaced execution. Missing counts follow the existing shared pressure interpreter; they are observations, not health, quorum or resource-ownership proof. The current provider exposes node-level workload classes, so this does not claim independent per-set foreground measurements or a hard global resource budget. Runtime waits increment `rustfs_heal_mainline_throttle_total` with `source=admin` or `source=auto_heal`, `result=delayed`, and a foreground-pressure or `recovery_window` reason. Real p99/throughput protection requires the separate W20 fixed-load ABBA measurements.
|
||||
|
||||
## Live Scanner Heal Completions
|
||||
|
||||
|
||||
Reference in New Issue
Block a user