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

fix(heal): reconcile dangling objects after node reconnect (#6567)

Browse Source
This commit is contained in:
Henry Guo
2026-08-25 21:20:47 +08:00
committed by GitHub
parent b4a78fc907
commit 9db29c8a6f
6 changed files with 458 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/storage.rs
+18
View File
@@ -428,6 +428,16 @@ pub trait HealStorageAPI: Send + Sync {
include_lifecycle_object_info: bool,
) -> Result<(Vec<HealListItem>, Option<String>, bool)>;
/// Return the live erasure sets selected by this heal request.
///
/// Recursive admin heals use these scopes with the cross-disk union walk so
/// objects surviving on only one returning disk are still discovered. The
/// `None` default preserves the read-quorum listing for alternate backends;
/// `Some(Vec::new())` means the selected topology currently has no live set.
async fn heal_erasure_set_scopes(&self, _opts: &HealOpts) -> Result<Option<Vec<(usize, usize)>>> {
Ok(None)
}
/// List versions for healing via a per-erasure-set DISK-WALK union enumerator
/// (backlog#920). Unlike `list_objects_for_heal_page` (which reflects only the
/// READ-QUORUM metadata view via `list_object_versions`), this surfaces every
@@ -1294,6 +1304,14 @@ impl HealStorageAPI for ECStoreHealStorage {
Ok((page_objects, next_token, list_info.is_truncated))
}
async fn heal_erasure_set_scopes(&self, opts: &HealOpts) -> Result<Option<Vec<(usize, usize)>>> {
self.ecstore
.heal_erasure_set_scopes(opts)
.await
.map(Some)
.map_err(Error::Storage)
}
async fn list_versions_for_heal_page_disk_walk(
&self,
set_disk_id: &str,
crates/heal/src/heal/task/heal_bucket.rs
+154 -122
View File
@@ -14,6 +14,7 @@
/// bucket/cluster/prefix heal: the recursive bucket-objects sweep and the erasure-set usage baseline
use super::*;
use crate::heal::progress::{add_bytes, increment_counter, stable_generation};
use crate::heal::utils::format_set_disk_id;
impl HealTask {
pub(super) async fn heal_bucket(&self, bucket: &str) -> Result<()> {
@@ -242,7 +243,6 @@ impl HealTask {
#[hotpath::measure]
async fn heal_bucket_objects(&self, bucket: &str, prefix: &str) -> Result<()> {
let mut continuation_token: Option<String> = None;
let mut scanned = 0u64;
let mut healed = 0u64;
let mut failed = 0u64;
@@ -266,108 +266,101 @@ impl HealTask {
set: self.options.set_index,
};
loop {
self.check_control_flags().await?;
let (objects, next_token, is_truncated) = self
.await_with_control(
self.storage
.list_objects_for_heal_page(bucket, prefix, continuation_token.as_deref(), false),
)
.await?;
let erasure_set_scopes = self
.await_with_control(self.storage.heal_erasure_set_scopes(&heal_opts))
.await?;
let listing_scopes = match erasure_set_scopes {
None => vec![(None, heal_opts)],
Some(erasure_set_scopes) => erasure_set_scopes
.into_iter()
.map(|(pool_idx, set_idx)| {
let mut scoped_opts = heal_opts;
scoped_opts.pool = Some(pool_idx);
scoped_opts.set = Some(set_idx);
(Some(format_set_disk_id(pool_idx, set_idx)), scoped_opts)
})
.collect(),
};
let mut pending = objects;
let mut retry_attempt = 0_u32;
while !pending.is_empty() {
if retry_attempt > 0 {
self.await_with_control(async {
tokio::time::sleep(self.bucket_object_retry_delay(retry_attempt)).await;
Ok(())
})
.await?;
}
let mut retry = Vec::with_capacity(pending.len());
for item in pending {
self.check_control_flags().await?;
let mut telemetry_unknown = false;
let object = item.name.as_str();
{
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("{bucket}/{object}")));
for (set_disk_id, heal_opts) in listing_scopes {
let mut continuation_token: Option<String> = None;
loop {
self.check_control_flags().await?;
let (objects, next_token, is_truncated) = if let Some(set_disk_id) = set_disk_id.as_deref() {
self.await_with_control(self.storage.list_versions_for_heal_page_disk_walk(
set_disk_id,
bucket,
prefix,
continuation_token.as_deref(),
false,
))
.await?
} else {
self.await_with_control(self.storage.list_objects_for_heal_page(
bucket,
prefix,
continuation_token.as_deref(),
false,
))
.await?
};
let mut pending = objects;
let mut retry_attempt = 0_u32;
while !pending.is_empty() {
if retry_attempt > 0 {
self.await_with_control(async {
tokio::time::sleep(self.bucket_object_retry_delay(retry_attempt)).await;
Ok(())
})
.await?;
}
let mut terminal_outcome = true;
let error = match self
.await_with_control(
self.storage
.heal_object(bucket, object, item.version_id.as_deref(), &heal_opts),
)
.await
{
Ok((result, None)) => {
telemetry_unknown |= !increment_counter(&mut healed);
telemetry_unknown |= !add_bytes(&mut bytes, u64::try_from(result.object_size).unwrap_or(u64::MAX));
self.record_result_item(result).await;
None
let mut retry = Vec::with_capacity(pending.len());
for item in pending {
self.check_control_flags().await?;
let mut telemetry_unknown = false;
let object = item.name.as_str();
{
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("{bucket}/{object}")));
}
Ok((_, Some(err))) if is_missing_object_dir_heal_result(object, &err) => {
telemetry_unknown |= !increment_counter(&mut healed);
debug!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object,
result = "object_dir_not_found_skipped",
"Heal bucket object-dir candidate skipped after not-found result"
);
None
}
Ok((_, Some(err))) | Err(err) => Some(err),
};
if let Some(err) = error {
if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) {
telemetry_unknown |= !increment_counter(&mut skipped);
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object,
result = "transient_skip",
error = %err,
"Heal bucket object repair skipped due to transient metadata error"
);
} else if err.is_recoverable_heal() && retry_attempt < MAX_BUCKET_OBJECT_HEAL_RETRIES {
terminal_outcome = false;
debug!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object,
retry_attempt = retry_attempt.saturating_add(1),
error = %err,
result = "object_retry_scheduled",
"Heal bucket object retry scheduled"
);
retry.push(item);
} else {
telemetry_unknown |= !increment_counter(&mut failed);
if err.is_recoverable_heal() {
retryable_failed = retryable_failed.saturating_add(1);
} else {
permanent_failed = permanent_failed.saturating_add(1);
let mut terminal_outcome = true;
let error = match self
.await_with_control(
self.storage
.heal_object(bucket, object, item.version_id.as_deref(), &heal_opts),
)
.await
{
Ok((result, None)) => {
telemetry_unknown |= !increment_counter(&mut healed);
telemetry_unknown |=
!add_bytes(&mut bytes, u64::try_from(result.object_size).unwrap_or(u64::MAX));
self.record_result_item(result).await;
None
}
first_failed_object.get_or_insert_with(|| object.to_string());
first_error.get_or_insert_with(|| err.to_string());
if take_failure_log_sample(&mut failure_samples_logged) {
Ok((_, Some(err))) if is_missing_object_dir_heal_result(object, &err) => {
telemetry_unknown |= !increment_counter(&mut healed);
debug!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object,
result = "object_dir_not_found_skipped",
"Heal bucket object-dir candidate skipped after not-found result"
);
None
}
Ok((_, Some(err))) | Err(err) => Some(err),
};
if let Some(err) = error {
if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) {
telemetry_unknown |= !increment_counter(&mut skipped);
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
@@ -376,41 +369,80 @@ impl HealTask {
task_id = %self.id,
bucket,
object,
retry_attempt,
result = "transient_skip",
error = %err,
result = "object_failed",
"Heal bucket object repair failed"
"Heal bucket object repair skipped due to transient metadata error"
);
} else if err.is_recoverable_heal() && retry_attempt < MAX_BUCKET_OBJECT_HEAL_RETRIES {
terminal_outcome = false;
debug!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object,
retry_attempt = retry_attempt.saturating_add(1),
error = %err,
result = "object_retry_scheduled",
"Heal bucket object retry scheduled"
);
retry.push(item);
} else {
telemetry_unknown |= !increment_counter(&mut failed);
if err.is_recoverable_heal() {
retryable_failed = retryable_failed.saturating_add(1);
} else {
permanent_failed = permanent_failed.saturating_add(1);
}
first_failed_object.get_or_insert_with(|| object.to_string());
first_error.get_or_insert_with(|| err.to_string());
if take_failure_log_sample(&mut failure_samples_logged) {
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object,
retry_attempt,
error = %err,
result = "object_failed",
"Heal bucket object repair failed"
);
}
}
}
}
if terminal_outcome {
telemetry_unknown |= !increment_counter(&mut scanned);
}
if terminal_outcome {
telemetry_unknown |= !increment_counter(&mut scanned);
}
if !terminal_outcome {
continue;
}
if !terminal_outcome {
continue;
}
let mut progress = self.progress.write().await;
progress.update_object_progress(scanned, healed, failed, skipped, bytes);
if telemetry_unknown {
progress.mark_unknown();
let mut progress = self.progress.write().await;
progress.update_object_progress(scanned, healed, failed, skipped, bytes);
if telemetry_unknown {
progress.mark_unknown();
}
}
pending = retry;
retry_attempt = retry_attempt.saturating_add(1);
}
pending = retry;
retry_attempt = retry_attempt.saturating_add(1);
}
if !is_truncated {
break;
}
if !is_truncated {
break;
}
continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
if continuation_token.is_none() {
// Truncated but no continuation token: end of listing.
break;
continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
if continuation_token.is_none() {
// Truncated but no continuation token: end of listing.
break;
}
}
}
crates/heal/src/heal/task/tests.rs
+63
View File
@@ -564,6 +564,8 @@ struct MockStorage {
replacement_resume_disk: Mutex<Option<DiskStore>>,
usage_baseline: Mutex<Option<HealBucketUsageBaseline>>,
usage_baseline_error: Mutex<bool>,
erasure_set_scopes: Mutex<Vec<(usize, usize)>>,
disk_walk_calls: Mutex<Vec<String>>,
}
#[test]
@@ -946,6 +948,26 @@ impl HealStorageAPI for MockStorage {
}
}
async fn heal_erasure_set_scopes(&self, _opts: &HealOpts) -> Result<Option<Vec<(usize, usize)>>> {
let scopes = self.erasure_set_scopes.lock().unwrap().clone();
Ok((!scopes.is_empty()).then_some(scopes))
}
async fn list_versions_for_heal_page_disk_walk(
&self,
set_disk_id: &str,
_bucket: &str,
_prefix: &str,
continuation_token: Option<&str>,
_include_lifecycle_object_info: bool,
) -> Result<(Vec<HealListItem>, Option<String>, bool)> {
self.disk_walk_calls.lock().unwrap().push(set_disk_id.to_string());
if continuation_token.is_some() {
return Ok((Vec::new(), None, false));
}
Ok((vec![heal_item(&format!("{set_disk_id}-object"))], None, false))
}
async fn get_disk_for_resume(&self, _set_disk_id: &str) -> Result<DiskStore> {
self.resume_disk
.lock()
@@ -1081,6 +1103,47 @@ async fn test_recursive_bucket_heal_visits_objects() {
assert_eq!(result_items.iter().filter(|item| item.object_size == 1).count(), 2);
}
#[tokio::test]
async fn recursive_bucket_heal_uses_each_erasure_set_union_scope() {
let storage = Arc::new(MockStorage {
erasure_set_scopes: Mutex::new(vec![(0, 0), (1, 2)]),
..Default::default()
});
let request = HealRequest::new(
HealType::Bucket {
bucket: "bucket-a".to_string(),
},
HealOptions {
recursive: true,
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 consume every selected set union");
assert_eq!(
storage.disk_walk_calls.lock().unwrap().as_slice(),
["pool_0_set_0".to_string(), "pool_1_set_2".to_string()]
);
assert_eq!(
storage.healed_objects.lock().unwrap().as_slice(),
["pool_0_set_0-object".to_string(), "pool_1_set_2-object".to_string()]
);
assert!(
!*storage.listed.lock().unwrap(),
"read-quorum listing must not hide returning-disk candidates"
);
let object_opts = storage.object_heal_opts.lock().unwrap();
assert_eq!(object_opts.len(), 2);
assert_eq!((object_opts[0].pool, object_opts[0].set), (Some(0), Some(0)));
assert_eq!((object_opts[1].pool, object_opts[1].set), (Some(1), Some(2)));
}
#[tokio::test]
async fn result_items_are_bounded_and_report_truncation() {
let storage = Arc::new(MockStorage::default());
crates/heal/tests/heal_b920_subquorum_union_test.rs
+3 -3
View File
@@ -354,9 +354,9 @@ mod serial_tests {
with_dangling_grace_disabled(heal_storage.heal_object(bucket, object, Some(&v1), &deep_heal_opts()))
.await
.expect("heal_object call must not itself error");
// A dangling delete reports the version as gone (FileVersionNotFound),
// proving the destructive path fired for a genuinely torn write.
assert!(error.is_some(), "a torn (< data_blocks) version must NOT be silently treated as healed");
// Successful cleanup is a successful heal outcome. The on-disk checks
// below prove that the torn version was deleted rather than healed.
assert!(error.is_none(), "successful torn-version cleanup must not report a heal error");
// Destructive-path PROOF: the stale minority copy on disk0 was purged by
// the dangling delete (the guard correctly did NOT rescue a torn write).
assert_eq!(