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

fix(heal): restore single disk data during deep heal (#3085)

Browse Source 
* fix(heal): restore single disk data during deep heal

* fix(heal): clarify erasure heal step logging

* fix(heal): avoid duplicate deep object heals

* fix(ecstore): preserve volume-not-found walk errors

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
Henry Guo
2026-05-27 01:36:17 +08:00
committed by GitHub
parent 0875f09a39
commit 0478505839
2 changed files with 148 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/erasure_healer.rs
+74 -14
View File
@@ -62,6 +62,18 @@ impl ErasureSetHealer {
}
}
fn effective_heal_page_object_concurrency_for_scan_mode(scan_mode: HealScanMode) -> usize {
if matches!(scan_mode, HealScanMode::Deep) {
1
} else {
Self::effective_heal_page_object_concurrency()
}
}
fn is_object_not_found_message(message: &str) -> bool {
message.contains("File not found") || message.contains("not found")
}
pub fn new(
storage: Arc<dyn HealStorageAPI>,
progress: Arc<RwLock<HealProgress>>,
@@ -292,7 +304,7 @@ impl ErasureSetHealer {
// 2. process objects with pagination to avoid loading all objects into memory
let mut continuation_token: Option<String> = None;
let mut global_obj_idx = 0usize;
let page_concurrency_limit = Self::effective_heal_page_object_concurrency();
let page_concurrency_limit = Self::effective_heal_page_object_concurrency_for_scan_mode(self.heal_opts.scan_mode);
let in_flight = Arc::new(AtomicUsize::new(0));
loop {
@@ -329,23 +341,50 @@ impl ErasureSetHealer {
let in_flight = in_flight.clone();
let set_label = set_disk_id.to_string();
let heal_opts = self.heal_opts;
let permit = semaphore
.clone()
.acquire_owned()
.await
.map_err(|e| Error::other(format!("Failed to acquire page concurrency permit: {e}")))?;
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 deep_scan = matches!(heal_opts.scan_mode, HealScanMode::Deep);
let semaphore = semaphore.clone();
page_tasks.push(async move {
let _permit = permit;
let permit = semaphore
.clone()
.acquire_owned()
.await
.map_err(|e| Error::other(format!("Failed to acquire page concurrency permit: {e}")));
let _permit = match permit {
Ok(permit) => permit,
Err(err) => return (object_name, 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 result = if cancel_token.is_cancelled() {
Err(Error::TaskCancelled)
} else if deep_scan {
match storage.heal_object(&bucket_name, &object_name, None, &heal_opts).await {
Ok((_result, None)) => Ok(true),
Ok((_, Some(err))) => {
let err_msg = err.to_string();
if Self::is_object_not_found_message(&err_msg) {
Ok(false)
} else {
Err(Error::other(err))
}
}
Err(err) => {
let err_msg = err.to_string();
if Self::is_object_not_found_message(&err_msg) {
Ok(false)
} else {
Err(err)
}
}
}
} else {
let object_exists = match storage.object_exists(&bucket_name, &object_name).await {
Ok(exists) => exists,
@@ -676,6 +715,7 @@ impl ErasureSetHealer {
#[cfg(test)]
mod tests {
use super::ErasureSetHealer;
use rustfs_common::heal_channel::HealScanMode;
#[test]
fn heal_page_object_concurrency_uses_default_when_env_is_unset() {
@@ -702,4 +742,24 @@ mod tests {
});
});
}
#[test]
fn deep_scan_heal_page_object_concurrency_is_serial() {
temp_env::with_var(rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("11"), || {
assert_eq!(
ErasureSetHealer::effective_heal_page_object_concurrency_for_scan_mode(HealScanMode::Deep),
1
);
});
}
#[test]
fn normal_scan_heal_page_object_concurrency_uses_effective_limit() {
temp_env::with_var(rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("11"), || {
assert_eq!(
ErasureSetHealer::effective_heal_page_object_concurrency_for_scan_mode(HealScanMode::Normal),
11
);
});
}
}
crates/heal/src/heal/task.rs
+74 -12
View File
@@ -662,7 +662,11 @@ impl HealTask {
let heal_opts = HealOpts {
recursive: self.options.recursive,
dry_run: self.options.dry_run,
remove: self.options.remove_corrupted,
remove: if self.options.recursive {
false
} else {
self.options.remove_corrupted
},
recreate: self.options.recreate_missing,
scan_mode: self.options.scan_mode,
update_parity: self.options.update_parity,
@@ -1130,22 +1134,40 @@ impl HealTask {
// Step 3: Heal bucket structure
// Check control flags before each iteration to ensure timely cancellation.
// Each heal_bucket call may handle timeout/cancellation internally, see its implementation for details.
let bucket_heal_opts = HealOpts {
recursive: false,
dry_run: self.options.dry_run,
remove: false,
recreate: self.options.recreate_missing,
scan_mode: self.options.scan_mode,
update_parity: self.options.update_parity,
no_lock: false,
pool: self.options.pool_index,
set: self.options.set_index,
};
for bucket in buckets.iter() {
// Check control flags before starting each bucket heal
self.check_control_flags().await?;
// heal_bucket internally uses await_with_control for timeout/cancellation handling
if let Err(err) = self.heal_bucket(bucket).await {
// Check if error is due to cancellation or timeout
if matches!(err, Error::TaskCancelled | Error::TaskTimeout) {
return Err(err);
let heal_result = self
.await_with_control(self.storage.heal_bucket(bucket, &bucket_heal_opts))
.await;
match heal_result {
Ok(result) => {
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);
}
info!("Bucket heal failed: {}", err.to_string());
}
info!("Bucket heal failed: {}", err.to_string());
}
}
// Step 3: Create erasure set healer with resume support
info!("Step 3: Creating erasure set healer with resume support");
// Create erasure set healer with resume support
info!("Creating erasure set healer with resume support");
let heal_opts = HealOpts {
recursive: self.options.recursive,
dry_run: self.options.dry_run,
@@ -1217,6 +1239,8 @@ mod tests {
struct MockStorage {
listed: Mutex<bool>,
healed_objects: Mutex<Vec<String>>,
bucket_heal_opts: Mutex<Vec<HealOpts>>,
object_heal_opts: Mutex<Vec<HealOpts>>,
}
#[async_trait::async_trait]
@@ -1285,9 +1309,10 @@ mod tests {
_bucket: &str,
object: &str,
_version_id: Option<&str>,
_opts: &HealOpts,
opts: &HealOpts,
) -> Result<(HealResultItem, Option<Error>)> {
self.healed_objects.lock().unwrap().push(object.to_string());
self.object_heal_opts.lock().unwrap().push(*opts);
Ok((
HealResultItem {
object_size: 1,
@@ -1297,7 +1322,8 @@ mod tests {
))
}
async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result<HealResultItem> {
async fn heal_bucket(&self, _bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
self.bucket_heal_opts.lock().unwrap().push(*opts);
Ok(HealResultItem::default())
}
@@ -1360,4 +1386,40 @@ mod tests {
assert_eq!(result_items.len(), 3);
assert_eq!(result_items.iter().filter(|item| item.object_size == 1).count(), 2);
}
#[tokio::test]
async fn test_recursive_bucket_heal_does_not_remove_bucket_metadata() {
let storage = Arc::new(MockStorage::default());
let request = HealRequest::new(
HealType::Bucket {
bucket: "bucket-a".to_string(),
},
HealOptions {
recursive: true,
remove_corrupted: true,
recreate_missing: true,
scan_mode: HealScanMode::Deep,
timeout: None,
..Default::default()
},
HealPriority::Normal,
);
let task = HealTask::from_request(request, storage.clone());
task.heal_bucket("bucket-a")
.await
.expect("recursive bucket heal should succeed");
let bucket_opts = storage.bucket_heal_opts.lock().unwrap();
assert_eq!(bucket_opts.len(), 1);
assert!(!bucket_opts[0].remove);
assert!(bucket_opts[0].recreate);
assert_eq!(bucket_opts[0].scan_mode, HealScanMode::Deep);
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));
}
}