gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-24 21:26:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): harden resumable set repair failures (#5693)

Browse Source 
* fix(heal): enforce resumable task control

* fix(ecstore): surface bucket and metadata heal errors

* chore: refresh guardrail path references

---------

Signed-off-by: cxymds <cxymds@gmail.com>
This commit is contained in:
cxymds
2026-08-04 20:35:58 +08:00
committed by GitHub
parent 3c8bd5b929
commit 31959b90db
15 changed files with 1276 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/error.rs
+18 -3
View File
@@ -107,13 +107,21 @@ impl Error {
Error::TaskTimeout | Error::TransientSkip { .. } => true,
Error::Storage(err) => {
err.is_quorum_error()
|| matches!(err, EcstoreError::SlowDown | EcstoreError::OperationCanceled | EcstoreError::Lock(_))
|| matches!(
err,
EcstoreError::DiskNotFound
| EcstoreError::VolumeNotFound
| EcstoreError::SlowDown
| EcstoreError::OperationCanceled
| EcstoreError::Lock(_)
)
|| is_recoverable_heal_error_message(&err.to_string())
}
Error::Disk(err) => {
matches!(
err,
DiskError::ErasureReadQuorum
DiskError::DiskNotFound
| DiskError::ErasureReadQuorum
| DiskError::ErasureWriteQuorum
| DiskError::Timeout
| DiskError::SourceStalled
@@ -159,7 +167,7 @@ impl From<Error> for std::io::Error {
#[cfg(test)]
mod tests {
use super::Error;
use crate::heal::EcstoreError;
use crate::heal::{DiskError, EcstoreError};
#[test]
fn incomplete_target_rename_is_recoverable() {
@@ -173,4 +181,11 @@ mod tests {
assert!(task_error.is_recoverable_heal());
assert!(storage_error.is_recoverable_heal());
}
#[test]
fn offline_disk_errors_are_recoverable() {
assert!(Error::Disk(DiskError::DiskNotFound).is_recoverable_heal());
assert!(Error::Storage(EcstoreError::DiskNotFound).is_recoverable_heal());
assert!(Error::Storage(EcstoreError::VolumeNotFound).is_recoverable_heal());
}
}
crates/heal/src/heal/channel.rs
+16
View File
@@ -623,6 +623,7 @@ impl HealChannelProcessor {
update_parity: request.update_parity.unwrap_or(true),
recursive,
dry_run: request.dry_run.unwrap_or(false),
no_lock: request.no_lock.unwrap_or(false),
timeout: request.timeout_seconds.map(std::time::Duration::from_secs),
pool_index: request.pool_index,
set_index: request.set_index,
@@ -852,6 +853,7 @@ mod tests {
update_parity: None,
recursive: None,
dry_run: None,
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -883,6 +885,7 @@ mod tests {
update_parity: Some(true),
recursive: Some(true),
dry_run: Some(false),
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -914,6 +917,7 @@ mod tests {
update_parity: Some(true),
recursive: Some(false),
dry_run: Some(false),
no_lock: Some(true),
timeout_seconds: Some(300),
pool_index: Some(0),
set_index: Some(1),
@@ -928,6 +932,7 @@ mod tests {
assert_eq!(heal_request.options.scan_mode, HealScanMode::Deep);
assert!(heal_request.options.remove_corrupted);
assert!(heal_request.options.recreate_missing);
assert!(heal_request.options.no_lock);
}
#[tokio::test]
@@ -948,6 +953,7 @@ mod tests {
update_parity: None,
recursive: Some(false),
dry_run: None,
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -984,6 +990,7 @@ mod tests {
update_parity: None,
recursive: Some(false),
dry_run: None,
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -1022,6 +1029,7 @@ mod tests {
update_parity: None,
recursive: Some(false),
dry_run: None,
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -1053,6 +1061,7 @@ mod tests {
update_parity: Some(true),
recursive: Some(true),
dry_run: Some(false),
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -1087,6 +1096,7 @@ mod tests {
update_parity: None,
recursive: None,
dry_run: None,
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -1117,6 +1127,7 @@ mod tests {
update_parity: None,
recursive: None,
dry_run: None,
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -1154,6 +1165,7 @@ mod tests {
update_parity: None,
recursive: None,
dry_run: None,
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -1184,6 +1196,7 @@ mod tests {
update_parity: Some(false),
recursive: None,
dry_run: None,
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -1216,6 +1229,7 @@ mod tests {
update_parity: None,
recursive: None,
dry_run: None,
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -1252,6 +1266,7 @@ mod tests {
update_parity: None,
recursive: None,
dry_run: None,
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
@@ -1529,6 +1544,7 @@ mod tests {
update_parity: None,
recursive: None,
dry_run: None,
no_lock: None,
timeout_seconds: None,
pool_index: None,
set_index: None,
crates/heal/src/heal/erasure_healer.rs
+223 -63
View File
@@ -43,6 +43,34 @@ enum HealObjectOutcome {
Failed,
}
struct PageConcurrencyGuard {
in_flight: Arc<AtomicUsize>,
set_label: String,
}
impl PageConcurrencyGuard {
fn new(in_flight: Arc<AtomicUsize>, set_label: String) -> Self {
let current = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
gauge!(
"rustfs_heal_page_concurrency_current",
"set" => set_label.clone()
)
.set(current as f64);
Self { in_flight, set_label }
}
}
impl Drop for PageConcurrencyGuard {
fn drop(&mut self) {
let current = self.in_flight.fetch_sub(1, Ordering::SeqCst) - 1;
gauge!(
"rustfs_heal_page_concurrency_current",
"set" => self.set_label.clone()
)
.set(current as f64);
}
}
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_ERASURE_HEALER: &str = "erasure_healer";
const EVENT_HEAL_ERASURE_RESUME_STATE: &str = "heal_erasure_resume_state";
@@ -183,35 +211,13 @@ impl ErasureSetHealer {
.execute_heal_with_resume(buckets, set_disk_id, &resume_manager, &checkpoint_manager)
.await;
// 4. cleanup resume state
if result.is_ok() {
if let Err(e) = resume_manager.cleanup().await {
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
state = "resume_cleanup_failed",
error = %e,
"Erasure set resume cleanup failed"
);
}
if let Err(e) = checkpoint_manager.cleanup().await {
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
state = "checkpoint_cleanup_failed",
error = %e,
"Erasure set checkpoint cleanup failed"
);
}
}
result?;
result
// The healing marker is cleared by the caller only after both cleanup
// operations succeed. Cleanup is idempotent, so a retry is safe.
checkpoint_manager.cleanup().await?;
resume_manager.cleanup().await?;
Ok(())
}
/// get or create task id
@@ -223,7 +229,10 @@ impl ErasureSetHealer {
match ResumeManager::load_from_disk(self.disk.clone(), &task_id).await {
Ok(manager) => {
let state = manager.get_state().await;
if state.set_disk_id == set_disk_id && ResumeUtils::can_resume_task(&self.disk, &task_id).await {
if !state.completed
&& state.set_disk_id == set_disk_id
&& ResumeUtils::can_resume_task(&self.disk, &task_id).await
{
debug!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
@@ -295,6 +304,21 @@ impl ErasureSetHealer {
CheckpointManager::new(self.disk.clone(), task_id.to_string()).await?
};
let state = resume_manager.get_state().await;
if state.retry_count > 0
&& state.completed_buckets.is_empty()
&& state.resume_cursor.is_none()
&& state.processed_objects == 0
&& state.successful_objects == 0
&& state.failed_objects == 0
&& state.skipped_objects == 0
{
// schedule_retry persists the authoritative resume reset before
// resetting the checkpoint. Reapply the checkpoint reset after
// a crash in that window so stale positions cannot skip work.
checkpoint_manager.reset_for_retry().await?;
}
Ok((resume_manager, checkpoint_manager))
} else {
debug!(
@@ -358,6 +382,7 @@ impl ErasureSetHealer {
let mut successful_objects = state.successful_objects;
let mut failed_objects = state.failed_objects;
let mut skipped_objects = state.skipped_objects;
let mut failed_buckets = 0u64;
// 4. process remaining buckets
for (bucket_idx, bucket) in buckets.iter().enumerate().skip(current_bucket_index) {
@@ -385,6 +410,10 @@ impl ErasureSetHealer {
)
.await;
if matches!(bucket_result, Err(Error::TaskCancelled | Error::TaskTimeout)) {
return bucket_result;
}
// update checkpoint position
checkpoint_manager.update_position(bucket_idx, current_object_index).await?;
@@ -422,7 +451,9 @@ impl ErasureSetHealer {
"Erasure set bucket completed"
);
}
Err(err @ Error::TaskCancelled) | Err(err @ Error::TaskTimeout) => return Err(err),
Err(e) => {
failed_buckets = failed_buckets.saturating_add(1);
error!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
@@ -453,7 +484,7 @@ impl ErasureSetHealer {
// skip may be because the disk is still down, so these are deferred to a
// later heal cycle via the same bounded-retry mechanism as failures —
// never hot-retried in place here.
if failed_objects > 0 || skipped_objects > 0 {
if failed_objects > 0 || skipped_objects > 0 || failed_buckets > 0 {
if resume_manager.schedule_retry().await? {
// Both persistence layers must be reset together: schedule_retry
// rewinds the resume state (cursor + counters), and the
@@ -471,13 +502,14 @@ impl ErasureSetHealer {
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
failed_buckets,
failed_objects,
skipped_objects,
state = "retry_scheduled",
"Erasure set heal pass finished with unhealed versions; scheduled full re-heal retry"
);
return Err(Error::other(format!(
"Erasure set heal incomplete: {failed_objects} failed, {skipped_objects} skipped object(s); retry scheduled"
"Erasure set heal incomplete: {failed_buckets} bucket(s) failed, {failed_objects} object(s) failed, {skipped_objects} object(s) skipped; retry scheduled"
)));
}
@@ -491,15 +523,16 @@ impl ErasureSetHealer {
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
failed_buckets,
failed_objects,
skipped_objects,
state = "failed_after_retries",
"Erasure set heal exhausted retries with unrecovered versions"
);
let _ = resume_manager.cleanup().await;
let _ = checkpoint_manager.cleanup().await;
checkpoint_manager.cleanup().await?;
resume_manager.cleanup().await?;
return Err(Error::other(format!(
"Erasure set heal exhausted retries with {failed_objects} failed, {skipped_objects} skipped object(s)"
"Erasure set heal exhausted retries with {failed_buckets} bucket(s) failed, {failed_objects} object(s) failed, {skipped_objects} object(s) skipped"
)));
}
@@ -646,12 +679,7 @@ impl ErasureSetHealer {
Err(err) => return (dedup_key, object_name, version_id, Err(err)),
};
let current_in_flight = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
gauge!(
"rustfs_heal_page_concurrency_current",
"set" => set_label.clone()
)
.set(current_in_flight as f64);
let _in_flight_guard = PageConcurrencyGuard::new(in_flight, set_label);
// Always go through heal_object. Genuine absence flows through
// heal_object -> FileVersionNotFound/FileNotFound ->
@@ -677,13 +705,6 @@ impl ErasureSetHealer {
}
};
let current = in_flight.fetch_sub(1, Ordering::SeqCst) - 1;
gauge!(
"rustfs_heal_page_concurrency_current",
"set" => set_label.clone()
)
.set(current as f64);
(dedup_key, object_name, version_id, result)
});
}
@@ -723,14 +744,7 @@ impl ErasureSetHealer {
"Erasure set missing object treated as ok"
);
}
Err(Error::TaskCancelled) => {
gauge!(
"rustfs_heal_page_concurrency_current",
"set" => set_disk_id.to_string()
)
.set(0.0);
return Err(Error::TaskCancelled);
}
Err(err @ Error::TaskCancelled) | Err(err @ Error::TaskTimeout) => return Err(err),
Err(Error::TransientSkip { message }) => {
*skipped_objects += 1;
checkpoint_manager.add_skipped_object(key).await?;
@@ -783,12 +797,6 @@ impl ErasureSetHealer {
let next_cursor = if is_truncated { next_token.clone() } else { None };
resume_manager.set_resume_cursor(next_cursor.clone()).await?;
checkpoint_manager.complete_page(bucket_index, *current_object_index).await?;
gauge!(
"rustfs_heal_page_concurrency_current",
"set" => set_disk_id.to_string()
)
.set(0.0);
// Check if there are more pages
if !is_truncated {
break;
@@ -835,8 +843,30 @@ impl ErasureSetHealer {
#[cfg(test)]
mod tests {
use super::ErasureSetHealer;
use super::{ErasureSetHealer, PageConcurrencyGuard};
use rustfs_common::heal_channel::{HealRequestSource, HealScanMode};
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
};
#[tokio::test]
async fn dropping_pending_page_heal_releases_concurrency_slot() {
let in_flight = Arc::new(AtomicUsize::new(0));
let mut pending_heal = Box::pin({
let in_flight = in_flight.clone();
async move {
let _guard = PageConcurrencyGuard::new(in_flight, "pool_0_set_0".to_string());
std::future::pending::<()>().await;
}
});
assert!(futures::poll!(pending_heal.as_mut()).is_pending());
assert_eq!(in_flight.load(Ordering::SeqCst), 1);
drop(pending_heal);
assert_eq!(in_flight.load(Ordering::SeqCst), 0);
}
#[test]
fn heal_page_object_concurrency_uses_default_when_env_is_unset() {
@@ -979,7 +1009,7 @@ mod resume_loop_tests {
//! handling) — not merely a mock's own output.
use super::ErasureSetHealer;
use crate::heal::progress::HealProgress;
use crate::heal::resume::{CheckpointManager, ResumeManager, compose_key};
use crate::heal::resume::{CheckpointManager, RESUME_CHECKPOINT_FILE, ResumeDeleteFailure, ResumeManager, compose_key};
use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo, HealStorageAPI};
use crate::heal::storage_api::status::BucketInfo;
use crate::heal::{
@@ -989,6 +1019,7 @@ mod resume_loop_tests {
use rustfs_common::heal_channel::{HealOpts, HealRequestSource};
use rustfs_madmin::heal_commands::HealResultItem;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use tempfile::TempDir;
use tokio::sync::RwLock;
@@ -1017,6 +1048,7 @@ mod resume_loop_tests {
/// A transient infrastructure condition (offline disk / unmet quorum):
/// the version must be recorded as skipped and retried on a later pass.
Transient,
Timeout,
}
#[derive(Default)]
@@ -1027,6 +1059,7 @@ mod resume_loop_tests {
outcomes: Mutex<HashMap<String, HealOutcome>>,
/// every heal_object call recorded as (name, version_id)
heal_calls: Mutex<Vec<(String, Option<String>)>>,
fail_listing: AtomicBool,
}
impl FakeStorage {
@@ -1039,6 +1072,9 @@ mod resume_loop_tests {
fn calls(&self) -> Vec<(String, Option<String>)> {
self.heal_calls.lock().unwrap().clone()
}
fn fail_listing(&self) {
self.fail_listing.store(true, Ordering::SeqCst);
}
}
#[async_trait::async_trait]
@@ -1108,6 +1144,7 @@ mod resume_loop_tests {
Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileVersionNotFound))))
}
HealOutcome::Transient => Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::DiskNotFound)))),
HealOutcome::Timeout => Err(Error::TaskTimeout),
}
}
async fn heal_bucket(&self, _b: &str, _o: &HealOpts) -> Result<HealResultItem> {
@@ -1125,6 +1162,9 @@ mod resume_loop_tests {
_prefix: &str,
continuation_token: Option<&str>,
) -> Result<(Vec<HealListItem>, Option<String>, bool)> {
if self.fail_listing.load(Ordering::SeqCst) {
return Err(Error::other("injected listing failure"));
}
let key = continuation_token.map(str::to_string);
let page = self.pages.lock().unwrap().get(&key).cloned();
match page {
@@ -1233,6 +1273,126 @@ mod resume_loop_tests {
assert_eq!(env.resume.resume_cursor().await, None);
}
#[tokio::test]
async fn object_timeout_aborts_the_bucket_page_immediately() {
let env = make_env().await;
env.storage.set_page(
None,
Page {
items: vec![item("timed-out", None, false)],
next: None,
truncated: false,
},
);
env.storage.set_outcome("timed-out", None, HealOutcome::Timeout);
let (processed, successful, failed, skipped, result) = run(&env).await;
assert!(matches!(result, Err(Error::TaskTimeout)));
assert_eq!(processed, 0);
assert_eq!(successful, 0);
assert_eq!(failed, 0);
assert_eq!(skipped, 0);
}
#[tokio::test]
async fn bucket_listing_failure_does_not_mark_set_completed() {
let env = make_env().await;
env.storage.fail_listing();
let result = env
.healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint)
.await;
assert!(result.is_err(), "a bucket listing failure must fail the set heal pass");
let state = env.resume.get_state().await;
assert!(!state.completed, "a failed bucket must not mark the set completed");
assert_eq!(state.retry_count, 1, "the failed bucket must schedule a bounded retry");
assert!(state.completed_buckets.is_empty(), "the failed bucket must remain resumable");
}
#[tokio::test]
async fn completed_resume_state_is_not_selected_for_a_new_heal() {
let env = make_env().await;
env.resume
.mark_completed()
.await
.expect("completed resume state should persist");
let task_id = env
.healer
.get_or_create_task_id("pool_0_set_0")
.await
.expect("new heal should allocate a task id");
assert_ne!(task_id, "task", "a completed resume state must not suppress a new heal");
}
#[tokio::test]
async fn cleanup_failure_keeps_erasure_set_heal_incomplete() {
let env = make_env().await;
let checkpoint_path = format!("{BUCKET_META_PREFIX}/task_{RESUME_CHECKPOINT_FILE}");
let _failure = ResumeDeleteFailure::install(checkpoint_path, crate::heal::DiskError::DiskAccessDenied);
let error = env
.healer
.heal_erasure_set(&["b".to_string()], "pool_0_set_0")
.await
.expect_err("checkpoint cleanup failure must fail the erasure-set heal");
assert!(matches!(error, Error::Disk(crate::heal::DiskError::DiskAccessDenied)));
let state = ResumeManager::load_from_disk(env.healer.disk.clone(), "task")
.await
.expect("completed state must remain discoverable after cleanup failure")
.get_state()
.await;
assert!(state.completed, "successful data heal must be persisted before cleanup is attempted");
}
#[tokio::test]
async fn retry_resume_repairs_checkpoint_after_crash_between_resets() {
let env = make_env().await;
env.resume
.update_progress(3, 1, 1, 1)
.await
.expect("dirty resume progress should persist");
env.resume
.complete_bucket("b")
.await
.expect("dirty completed bucket should persist");
env.resume
.set_resume_cursor(Some("stale-cursor".to_string()))
.await
.expect("dirty resume cursor should persist");
env.checkpoint
.add_skipped_object(compose_key("stale-object", None))
.await
.expect("dirty checkpoint object should be recorded");
env.checkpoint
.update_position(4, 9)
.await
.expect("dirty checkpoint position should persist");
assert!(
env.resume.schedule_retry().await.expect("resume retry reset should persist"),
"retry budget should remain"
);
let (_, checkpoint) = env
.healer
.initialize_resume_state("task", "pool_0_set_0", &["b".to_string()])
.await
.expect("resume initialization should repair a stale checkpoint");
let checkpoint = checkpoint.get_checkpoint().await;
assert_eq!(checkpoint.current_bucket_index, 0);
assert_eq!(checkpoint.current_object_index, 0);
assert!(checkpoint.processed_objects.is_empty());
assert!(checkpoint.failed_objects.is_empty());
assert!(checkpoint.skipped_objects.is_empty());
}
#[tokio::test]
async fn test_resume_across_page_boundary_no_drop_no_double() {
let env = make_env().await;
crates/heal/src/heal/manager.rs
+71 -1
View File
@@ -2495,7 +2495,8 @@ impl HealManager {
queue.pop_next()
};
if let Some(request) = selected_request {
if let Some(mut request) = selected_request {
request.options.timeout.get_or_insert(config.task_timeout);
let task_priority = request.priority;
let task_type_label = heal_request_type_label(&request).to_string();
let task_set_label = heal_request_set_metric_label(&request);
@@ -4925,6 +4926,75 @@ mod tests {
assert_eq!(manager.get_queue_length().await, 0);
}
#[tokio::test]
async fn configured_task_timeout_applies_only_when_request_timeout_is_absent() {
let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
let manager = HealManager::new(
storage,
Some(HealConfig {
max_concurrent_heals: 1,
task_timeout: Duration::ZERO,
..HealConfig::default()
}),
);
let mut defaulted = bucket_request("defaulted-timeout", HealPriority::Normal, HealRequestSource::Admin);
defaulted.options.timeout = None;
let defaulted_id = defaulted.id.clone();
manager
.submit_heal_request(defaulted)
.await
.expect("request without timeout should be queued");
process_manager_queue_once(&manager).await;
let defaulted_status = tokio::time::timeout(Duration::from_secs(1), async {
loop {
if let Ok(status @ HealTaskStatus::Retrying { .. }) = manager.get_task_status(&defaulted_id).await {
break status;
}
tokio::task::yield_now().await;
}
})
.await
.expect("configured timeout should finish the task");
assert!(matches!(defaulted_status, HealTaskStatus::Retrying { .. }));
assert_eq!(
manager
.retrying_heals
.lock()
.await
.get(&defaulted_id)
.expect("timed out task should retain its retry request")
.request
.options
.timeout,
Some(Duration::ZERO)
);
manager
.cancel_task(&defaulted_id)
.await
.expect("retrying timeout task should be cancelled");
let mut explicit = bucket_request("explicit-timeout", HealPriority::Normal, HealRequestSource::Admin);
explicit.options.timeout = Some(Duration::from_secs(60));
let explicit_id = explicit.id.clone();
manager
.submit_heal_request(explicit)
.await
.expect("request with explicit timeout should be queued");
process_manager_queue_once(&manager).await;
let explicit_status = tokio::time::timeout(Duration::from_secs(1), async {
loop {
if let Ok(status @ HealTaskStatus::Failed { .. }) = manager.get_task_status(&explicit_id).await {
break status;
}
tokio::task::yield_now().await;
}
})
.await
.expect("explicit timeout request should finish without using the zero default");
assert!(matches!(explicit_status, HealTaskStatus::Failed { .. }));
}
#[tokio::test]
async fn test_force_start_bypasses_duplicate_and_full_admission() {
let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
crates/heal/src/heal/resume.rs
+145 -21
View File
@@ -14,6 +14,8 @@
use crate::{Error, Result};
use serde::{Deserialize, Serialize};
#[cfg(test)]
use std::collections::HashMap;
use std::collections::HashSet;
use std::path::Path;
use std::sync::{Arc, Mutex};
@@ -32,7 +34,7 @@ const EVENT_HEAL_CHECKPOINT_STATE: &str = "heal_checkpoint_state";
/// resume state file constants
const RESUME_STATE_FILE: &str = "ahm_resume_state.json";
const RESUME_PROGRESS_FILE: &str = "ahm_progress.json";
const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json";
pub(super) const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json";
/// Current on-disk schema version for `ResumeState`. Snapshots written by an
/// older schema (which tracked latest-only object names and a positional
@@ -92,6 +94,64 @@ fn path_to_str(path: &Path) -> Result<&str> {
.ok_or_else(|| Error::other(format!("Invalid UTF-8 path: {path:?}")))
}
#[cfg(test)]
pub(super) struct ResumeDeleteFailure {
path: String,
}
#[cfg(test)]
fn resume_delete_failures() -> &'static Mutex<HashMap<String, DiskError>> {
static FAILURES: std::sync::OnceLock<Mutex<HashMap<String, DiskError>>> = std::sync::OnceLock::new();
FAILURES.get_or_init(|| Mutex::new(HashMap::new()))
}
#[cfg(test)]
impl ResumeDeleteFailure {
pub(super) fn install(path: String, error: DiskError) -> Self {
let previous = resume_delete_failures()
.lock()
.expect("resume delete failure registry should not poison")
.insert(path.clone(), error);
assert!(previous.is_none(), "resume delete failure already installed");
Self { path }
}
}
#[cfg(test)]
impl Drop for ResumeDeleteFailure {
fn drop(&mut self) {
resume_delete_failures()
.lock()
.expect("resume delete failure registry should not poison")
.remove(&self.path);
}
}
#[cfg(test)]
fn injected_resume_delete_error(path: &str) -> Option<DiskError> {
resume_delete_failures()
.lock()
.expect("resume delete failure registry should not poison")
.get(path)
.cloned()
}
#[cfg(not(test))]
fn injected_resume_delete_error(_path: &str) -> Option<DiskError> {
None
}
async fn delete_resume_file(disk: &DiskStore, path: &Path) -> Result<()> {
let path_str = path_to_str(path)?;
if let Some(err) = injected_resume_delete_error(path_str) {
return Err(err.into());
}
match disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await {
Ok(()) | Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => Ok(()),
Err(err) => Err(err.into()),
}
}
/// resume state
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResumeState {
@@ -407,7 +467,7 @@ impl ResumeManager {
let mut state = self.state.write().await;
state.mark_completed();
drop(state);
self.save_state_throttled().await
self.save_state().await
}
/// set error message
@@ -444,24 +504,14 @@ impl ResumeManager {
/// cleanup resume state
pub async fn cleanup(&self) -> Result<()> {
let state = self.state.read().await;
let task_id = &state.task_id;
let task_id = self.state.read().await.task_id.clone();
// delete state files
let state_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_STATE_FILE}"));
let progress_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_PROGRESS_FILE}"));
let checkpoint_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"));
// ignore delete errors, files may not exist
if let Ok(path_str) = path_to_str(&state_file) {
let _ = self.disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await;
}
if let Ok(path_str) = path_to_str(&progress_file) {
let _ = self.disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await;
}
if let Ok(path_str) = path_to_str(&checkpoint_file) {
let _ = self.disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await;
}
delete_resume_file(&self.disk, &progress_file).await?;
// Delete the state file last so a partial cleanup remains discoverable.
delete_resume_file(&self.disk, &state_file).await?;
debug!(
target: "rustfs::heal::resume",
@@ -763,13 +813,10 @@ impl CheckpointManager {
/// cleanup checkpoint
pub async fn cleanup(&self) -> Result<()> {
let checkpoint = self.checkpoint.read().await;
let task_id = &checkpoint.task_id;
let task_id = self.checkpoint.read().await.task_id.clone();
let checkpoint_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"));
if let Ok(path_str) = path_to_str(&checkpoint_file) {
let _ = self.disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await;
}
delete_resume_file(&self.disk, &checkpoint_file).await?;
debug!(
target: "rustfs::heal::resume",
@@ -1211,6 +1258,83 @@ mod tests {
assert!(!throttle.record(), "counter must reset after a save");
}
#[tokio::test]
async fn completion_persists_immediately_and_cleanup_propagates_delete_errors() {
use super::super::{DiskOption, Endpoint, new_disk};
use tempfile::TempDir;
let temp_dir = TempDir::new().expect("create resume persistence test directory");
let endpoint = Endpoint::try_from(temp_dir.path().to_string_lossy().as_ref()).expect("create test disk endpoint");
let disk = new_disk(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("create resume persistence test disk");
match disk.make_volume(RUSTFS_META_BUCKET).await {
Ok(()) | Err(DiskError::VolumeExists) => {}
Err(err) => panic!("create metadata volume for resume persistence test: {err}"),
}
let task_id = "completion-persistence".to_string();
let manager = ResumeManager::new(
disk.clone(),
task_id.clone(),
"erasure_set".to_string(),
"pool_0_set_0".to_string(),
vec!["bucket".to_string()],
)
.await
.expect("create resume manager");
manager
.update_progress(1, 1, 0, 0)
.await
.expect("buffer progress below the persistence threshold");
manager.mark_completed().await.expect("persist completed resume state");
let persisted = ResumeManager::load_from_disk(disk.clone(), &task_id)
.await
.expect("reload completed resume state")
.get_state()
.await;
assert!(persisted.completed, "completion must be persisted without waiting for the throttle");
assert_eq!(persisted.processed_objects, 1, "the completion write must include buffered progress");
let state_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_STATE_FILE}");
let failure = ResumeDeleteFailure::install(state_path, DiskError::DiskAccessDenied);
let error = manager
.cleanup()
.await
.expect_err("resume cleanup must propagate a real delete failure");
assert!(matches!(error, Error::Disk(DiskError::DiskAccessDenied)));
drop(failure);
manager.cleanup().await.expect("resume cleanup must be retryable");
manager
.cleanup()
.await
.expect("missing resume files must be idempotent success");
let checkpoint = CheckpointManager::new(disk.clone(), task_id.clone())
.await
.expect("create checkpoint manager");
let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
let failure = ResumeDeleteFailure::install(checkpoint_path, DiskError::DiskAccessDenied);
let error = checkpoint
.cleanup()
.await
.expect_err("checkpoint cleanup must propagate a real delete failure");
assert!(matches!(error, Error::Disk(DiskError::DiskAccessDenied)));
drop(failure);
checkpoint.cleanup().await.expect("checkpoint cleanup must be retryable");
checkpoint
.cleanup()
.await
.expect("missing checkpoint must be idempotent success");
}
#[tokio::test]
async fn test_resume_utils() {
let task_id1 = ResumeUtils::generate_task_id();
crates/heal/src/heal/task.rs
+138 -17
View File
@@ -144,6 +144,9 @@ pub struct HealOptions {
pub recursive: bool,
/// Whether to dry run
pub dry_run: bool,
/// Whether to skip namespace locking
#[serde(default)]
pub no_lock: bool,
/// Timeout
pub timeout: Option<Duration>,
/// pool index
@@ -161,7 +164,8 @@ impl Default for HealOptions {
update_parity: true,
recursive: false,
dry_run: false,
timeout: Some(Duration::from_secs(300)), // 5 minutes default timeout
no_lock: false,
timeout: None,
pool_index: None,
set_index: None,
}
@@ -896,7 +900,7 @@ impl HealTask {
recreate: self.options.recreate_missing,
scan_mode: self.options.scan_mode,
update_parity: self.options.update_parity,
no_lock: false,
no_lock: self.options.no_lock,
pool: self.options.pool_index,
set: self.options.set_index,
};
@@ -1104,7 +1108,7 @@ impl HealTask {
recreate: true,
scan_mode: HealScanMode::Deep,
update_parity: true,
no_lock: false,
no_lock: self.options.no_lock,
pool: None,
set: None,
};
@@ -1260,7 +1264,7 @@ impl HealTask {
recreate: self.options.recreate_missing,
scan_mode: self.options.scan_mode,
update_parity: self.options.update_parity,
no_lock: false,
no_lock: self.options.no_lock,
pool: self.options.pool_index,
set: self.options.set_index,
};
@@ -1426,7 +1430,7 @@ impl HealTask {
recreate: self.options.recreate_missing,
scan_mode: self.options.scan_mode,
update_parity: self.options.update_parity,
no_lock: false,
no_lock: self.options.no_lock,
pool: self.options.pool_index,
set: self.options.set_index,
};
@@ -1680,7 +1684,7 @@ impl HealTask {
recreate: false,
scan_mode: HealScanMode::Deep,
update_parity: false,
no_lock: false,
no_lock: self.options.no_lock,
pool: self.options.pool_index,
set: self.options.set_index,
};
@@ -1809,7 +1813,7 @@ impl HealTask {
recreate: self.options.recreate_missing,
scan_mode: HealScanMode::Deep,
update_parity: true,
no_lock: false,
no_lock: self.options.no_lock,
pool: None,
set: None,
};
@@ -1973,7 +1977,7 @@ impl HealTask {
recreate: true,
scan_mode: HealScanMode::Deep,
update_parity: true,
no_lock: false,
no_lock: self.options.no_lock,
pool: None,
set: None,
};
@@ -2214,7 +2218,7 @@ impl HealTask {
recreate: self.options.recreate_missing,
scan_mode: self.options.scan_mode,
update_parity: self.options.update_parity,
no_lock: false,
no_lock: self.options.no_lock,
pool: self.options.pool_index,
set: self.options.set_index,
};
@@ -2230,10 +2234,6 @@ impl HealTask {
self.record_result_item(result).await;
}
Err(err) => {
// Check if error is due to cancellation or timeout
if matches!(err, Error::TaskCancelled | Error::TaskTimeout) {
return Err(err);
}
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_ERASURE_SET_RESULT,
@@ -2246,6 +2246,7 @@ impl HealTask {
error = %err,
"Heal erasure set bucket prepass failed"
);
return Err(err);
}
}
}
@@ -2268,7 +2269,7 @@ impl HealTask {
recreate: self.options.recreate_missing,
scan_mode: self.options.scan_mode,
update_parity: self.options.update_parity,
no_lock: false,
no_lock: self.options.no_lock,
pool: self.options.pool_index,
set: self.options.set_index,
};
@@ -2297,7 +2298,9 @@ impl HealTask {
stage = "execute_resumable_heal",
"Heal erasure set stage entered"
);
let result = erasure_healer.heal_erasure_set(&buckets, &set_disk_id).await;
let result = self
.await_with_control(erasure_healer.heal_erasure_set(&buckets, &set_disk_id))
.await;
// Keep the markers on failure: the resume state also persists, and the
// next run of this set heal re-marks and eventually clears them.
@@ -2360,12 +2363,13 @@ impl std::fmt::Debug for HealTask {
#[cfg(test)]
mod tests {
use super::super::{DiskStore, Endpoint};
use super::super::{DiskOption, DiskStore, Endpoint, HealDiskExt as _, new_disk};
use super::*;
use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo};
use rustfs_madmin::heal_commands::HealResultItem;
use std::collections::{HashMap, VecDeque};
use std::sync::Mutex;
use tempfile::TempDir;
use super::super::storage_api::status::BucketInfo;
#[derive(Default)]
@@ -2388,6 +2392,8 @@ mod tests {
listed_buckets: Mutex<Option<Vec<String>>>,
bucket_heal_errors: Mutex<HashMap<String, VecDeque<&'static str>>>,
bucket_heal_calls: Mutex<Vec<String>>,
block_heal_object: Mutex<bool>,
resume_disk: Mutex<Option<DiskStore>>,
}
/// Build a latest, non-delete-marker heal list item with no version id.
@@ -2520,6 +2526,10 @@ mod tests {
.unwrap()
.push(version_id.map(ToString::to_string));
self.object_heal_opts.lock().unwrap().push(*opts);
let block_heal_object = *self.block_heal_object.lock().unwrap();
if block_heal_object {
std::future::pending::<()>().await;
}
if let Some(outcome) = self
.heal_object_outcomes
.lock()
@@ -2635,10 +2645,35 @@ mod tests {
}
async fn get_disk_for_resume(&self, _set_disk_id: &str) -> Result<DiskStore> {
Err(Error::other("not implemented in tests"))
self.resume_disk
.lock()
.unwrap()
.clone()
.ok_or_else(|| Error::other("not implemented in tests"))
}
}
async fn make_resume_disk(temp: &TempDir) -> DiskStore {
let disk_path = temp.path().join("test_disk");
std::fs::create_dir_all(&disk_path).expect("test disk directory should be created");
let endpoint = Endpoint::try_from(disk_path.to_string_lossy().as_ref()).expect("test disk endpoint should be valid");
let disk = new_disk(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("test disk should initialize");
let metadata_volume = disk.make_volume(RUSTFS_META_BUCKET).await;
assert!(
matches!(metadata_volume, Ok(()) | Err(DiskError::VolumeExists)),
"metadata volume should exist: {metadata_volume:?}"
);
disk
}
#[tokio::test]
async fn test_recursive_bucket_heal_visits_objects() {
let storage = Arc::new(MockStorage::default());
@@ -2932,6 +2967,7 @@ mod tests {
remove_corrupted: true,
recreate_missing: true,
scan_mode: HealScanMode::Deep,
no_lock: true,
timeout: None,
..Default::default()
},
@@ -2948,12 +2984,14 @@ mod tests {
assert!(!bucket_opts[0].remove);
assert!(bucket_opts[0].recreate);
assert_eq!(bucket_opts[0].scan_mode, HealScanMode::Deep);
assert!(bucket_opts[0].no_lock);
let object_opts = storage.object_heal_opts.lock().unwrap();
assert_eq!(object_opts.len(), 2);
assert!(object_opts.iter().all(|opts| opts.remove));
assert!(object_opts.iter().all(|opts| opts.recreate));
assert!(object_opts.iter().all(|opts| opts.scan_mode == HealScanMode::Deep));
assert!(object_opts.iter().all(|opts| opts.no_lock));
}
#[tokio::test]
@@ -3584,4 +3622,87 @@ mod tests {
"erasure-set heal should continue past NoHealRequired format result, got: {err}"
);
}
#[tokio::test]
async fn erasure_set_bucket_prepass_failure_stops_before_object_heal() {
let temp = TempDir::new().expect("temporary directory should be created");
let disk = make_resume_disk(&temp).await;
let storage = Arc::new(MockStorage {
bucket_heal_errors: Mutex::new(HashMap::from([(
"bucket-a".to_string(),
VecDeque::from(["injected bucket prepass failure"]),
)])),
resume_disk: Mutex::new(Some(disk)),
..Default::default()
});
let request = HealRequest::new(
HealType::ErasureSet {
buckets: vec!["bucket-a".to_string()],
set_disk_id: "pool_0_set_0".to_string(),
},
HealOptions {
timeout: None,
..Default::default()
},
HealPriority::Normal,
);
let task = HealTask::from_request(request, storage.clone());
let error = task
.heal_erasure_set(vec!["bucket-a".to_string()], "pool_0_set_0".to_string())
.await
.expect_err("bucket prepass failure must stop the erasure-set heal");
assert!(error.to_string().contains("injected bucket prepass failure"));
assert_eq!(storage.bucket_heal_calls.lock().unwrap().as_slice(), ["bucket-a".to_string()]);
assert!(storage.object_heal_opts.lock().unwrap().is_empty());
}
#[tokio::test]
async fn resumable_erasure_set_execution_is_cancelled_while_object_heal_is_pending() {
let temp = TempDir::new().expect("temporary directory should be created");
let disk = make_resume_disk(&temp).await;
let storage = Arc::new(MockStorage {
block_heal_object: Mutex::new(true),
resume_disk: Mutex::new(Some(disk)),
..Default::default()
});
let request = HealRequest::new(
HealType::ErasureSet {
buckets: vec!["bucket-a".to_string()],
set_disk_id: "pool_0_set_0".to_string(),
},
HealOptions {
no_lock: true,
timeout: None,
..Default::default()
},
HealPriority::Normal,
);
let task = Arc::new(HealTask::from_request(request, storage.clone()));
let execution = tokio::spawn({
let task = task.clone();
async move { task.execute().await }
});
tokio::time::timeout(Duration::from_secs(5), async {
loop {
if !storage.object_heal_opts.lock().unwrap().is_empty() {
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("resumable object heal should start");
task.cancel().await.expect("task cancellation should succeed");
let result = tokio::time::timeout(Duration::from_secs(1), execution)
.await
.expect("cancellation should interrupt the pending resumable heal")
.expect("task execution should join");
assert!(matches!(result, Err(Error::TaskCancelled)));
assert!(storage.bucket_heal_opts.lock().unwrap().iter().all(|opts| opts.no_lock));
assert!(storage.object_heal_opts.lock().unwrap().iter().all(|opts| opts.no_lock));
}
}