gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-06 12:09:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): bound cross-page object retry delays (#7273)

Browse Source 
Retain failed identities in an execution-local count and byte bounded window so healthy later pages can advance. Preserve retry jitter, deadlines, terminal accounting and pressure pacing, with deterministic head-of-line and capacity regressions.

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-06 16:44:59 +08:00
committed by GitHub
parent 71859ff83c
commit 3a4afe9b38
4 changed files with 979 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/task/heal_bucket.rs
+378 -204
View File
@@ -14,8 +14,107 @@
/// 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::storage::HealListItem;
use crate::heal::utils::format_set_disk_id;
const MAX_DEFERRED_OBJECTS: usize = 256;
const MAX_DEFERRED_BYTES: usize = 256 * 1024;
const MAX_DEFERRED_FORWARD_PAGES: u64 = 2;
const MAX_DEFERRED_AGE: Duration = Duration::from_secs(30);
struct DeferredObject {
name: String,
version_id: Option<String>,
attempt: u32,
page: u64,
first_failure: Option<tokio::time::Instant>,
due: tokio::time::Instant,
}
impl DeferredObject {
fn new(item: HealListItem, page: u64) -> Self {
Self {
name: item.name,
version_id: item.version_id,
attempt: 0,
page,
first_failure: None,
due: tokio::time::Instant::now(),
}
}
fn payload_bytes(&self) -> usize {
self.name
.capacity()
.saturating_add(self.version_id.as_ref().map_or(0, String::capacity))
}
fn expired(&self) -> bool {
self.first_failure.is_some_and(|first| first.elapsed() >= MAX_DEFERRED_AGE)
}
fn defer(&mut self, delay: Duration) {
let now = tokio::time::Instant::now();
let first = *self.first_failure.get_or_insert(now);
self.attempt += 1;
self.due = (now + delay).min(first + MAX_DEFERRED_AGE);
}
}
// Only failed identities are retained. The current listing page remains owned
// by the caller; capacity pressure stops fetching, never discards that page.
struct DeferredWindow {
objects: VecDeque<DeferredObject>,
bytes: usize,
}
impl Default for DeferredWindow {
fn default() -> Self {
Self {
objects: VecDeque::new(),
// Charge every possible slot up front, including spare capacity.
bytes: MAX_DEFERRED_OBJECTS * size_of::<DeferredObject>(),
}
}
}
impl DeferredWindow {
fn push(&mut self, item: DeferredObject) -> std::result::Result<(), DeferredObject> {
let bytes = item.payload_bytes();
if self.objects.len() >= MAX_DEFERRED_OBJECTS || bytes > MAX_DEFERRED_BYTES.saturating_sub(self.bytes) {
return Err(item);
}
self.bytes += bytes;
self.objects.push_back(item);
Ok(())
}
fn pop_due(&mut self) -> Option<DeferredObject> {
let now = tokio::time::Instant::now();
let index = self.objects.iter().position(|item| item.due <= now)?;
let item = self.objects.remove(index)?;
self.bytes -= item.payload_bytes();
Some(item)
}
fn next_due(&self) -> Option<tokio::time::Instant> {
self.objects.iter().map(|item| item.due).min()
}
fn can_advance(&self, page: u64) -> bool {
self.objects.len() < MAX_DEFERRED_OBJECTS
&& self.bytes < MAX_DEFERRED_BYTES
&& self
.objects
.iter()
.all(|item| page.saturating_sub(item.page) < MAX_DEFERRED_FORWARD_PAGES)
}
}
#[cfg(test)]
#[path = "tests/deferred_retry_window.rs"]
mod deferred_retry_window;
fn unavailable_recreate_error(result: &HealResultItem, opts: &HealOpts) -> Option<Error> {
if opts.dry_run || !opts.recreate {
return None;
@@ -305,83 +404,122 @@ impl HealTask {
for (set_disk_id, heal_opts) in listing_scopes {
let mut continuation_token: Option<String> = None;
loop {
self.check_control_flags().await?;
let mut listing_attempt = 0;
let (objects, next_token, is_truncated) = loop {
let mut deferred = DeferredWindow::default();
let mut inline_retry: Option<DeferredObject> = None;
let mut page_number = 0_u64;
let mut aborted_progress_unknown = false;
let mut pending = Vec::<HealListItem>::new().into_iter();
let mut listing_finished = false;
let mut listing_attempt = 0;
let mut listing_due = tokio::time::Instant::now();
let scope_result: Result<()> = async {
loop {
self.check_control_flags().await?;
if listing_finished && pending.as_slice().is_empty() && deferred.objects.is_empty() && inline_retry.is_none()
{
break;
}
self.pace_mainline().await?;
let page = 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
};
match page {
Ok(page) => break page,
Err(error @ (Error::TaskCancelled | Error::TaskTimeout)) => return Err(error),
Err(error) => {
self.outcome.write().await.attempt_failed();
if error.is_recoverable_heal() && listing_attempt < MAX_BUCKET_OBJECT_HEAL_RETRIES {
listing_attempt += 1;
// Listing and object retries share this safe boundary. A
// failed listing never hides an already-due object retry.
let item = deferred.pop_due().or_else(|| {
if inline_retry
.as_ref()
.is_some_and(|item| item.due <= tokio::time::Instant::now())
{
inline_retry.take()
} else if inline_retry.is_none() {
pending.next().map(|item| DeferredObject::new(item, page_number))
} else {
None
}
});
let Some(mut item) = item else {
let can_list = !listing_finished && inline_retry.is_none() && deferred.can_advance(page_number);
if can_list && listing_due <= tokio::time::Instant::now() {
let page = 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
};
match page {
Ok((objects, next_token, is_truncated)) => {
page_number = page_number.saturating_add(1);
continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
listing_finished = continuation_token.is_none();
listing_attempt = 0;
listing_due = tokio::time::Instant::now();
pending = objects.into_iter();
}
Err(error @ (Error::TaskCancelled | Error::TaskTimeout)) => return Err(error),
Err(error) => {
self.outcome.write().await.attempt_failed();
if error.is_recoverable_heal() && listing_attempt < MAX_BUCKET_OBJECT_HEAL_RETRIES {
listing_attempt += 1;
listing_due =
tokio::time::Instant::now() + self.bucket_object_retry_delay(listing_attempt);
continue;
}
self.outcome.write().await.mark_untraversable();
return Err(Error::HealListingFailed {
bucket: bucket.to_string(),
source: Box::new(error),
});
}
}
continue;
} else {
let due = deferred
.next_due()
.into_iter()
.chain(inline_retry.as_ref().map(|item| item.due))
.chain(can_list.then_some(listing_due))
.min();
if let Some(due) = due {
self.await_with_control(async {
tokio::time::sleep(self.bucket_object_retry_delay(listing_attempt)).await;
tokio::time::sleep_until(due).await;
Ok(())
})
.await?;
continue;
}
self.outcome.write().await.mark_untraversable();
return Err(Error::HealListingFailed {
bucket: bucket.to_string(),
source: Box::new(error),
});
}
continue;
};
let retry_attempt = item.attempt;
let mut telemetry_unknown = false;
let object = item.name.as_str();
let identity =
self.outcome_identity(bucket, object, item.version_id.as_deref(), heal_opts.pool, heal_opts.set);
let mut disposition = if heal_opts.dry_run {
HealObjectDisposition::DryRunObserved
} else {
HealObjectDisposition::Unknown
};
let mut detail = None;
{
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("{bucket}/{object}")));
}
};
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?;
self.pace_mainline().await?;
let mut telemetry_unknown = false;
let object = item.name.as_str();
let identity =
self.outcome_identity(bucket, object, item.version_id.as_deref(), heal_opts.pool, heal_opts.set);
let mut disposition = if heal_opts.dry_run {
HealObjectDisposition::DryRunObserved
} else {
HealObjectDisposition::Unknown
};
let mut detail = None;
{
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("{bucket}/{object}")));
}
let mut terminal_outcome = true;
let error = match self
let mut terminal_outcome = true;
let age_exhausted = item.expired();
let error = if age_exhausted {
Some(Error::other("heal object retry age exhausted"))
} else {
match self
.await_with_control(
self.storage
.heal_object(bucket, object, item.version_id.as_deref(), &heal_opts),
@@ -414,152 +552,188 @@ impl HealTask {
None
}
Ok((_, Some(err))) | Err(err) => Some(err),
};
}
};
if let Some(err) = error {
match err {
Error::TaskCancelled | Error::TaskTimeout => {
let disposition = if matches!(err, Error::TaskCancelled) {
HealObjectDisposition::Cancelled
} else {
HealObjectDisposition::Deferred {
reason: HealDeferredReason::Deadline,
retry_not_before: None,
}
};
self.outcome.write().await.record(HealObjectOutcome {
identity,
disposition,
detail: None,
});
return Err(err);
}
_ => self.outcome.write().await.attempt_failed(),
}
detail = Some(err.to_string());
if Self::is_dangling_delete_grace_error(&err) {
disposition = HealObjectDisposition::Deferred {
reason: HealDeferredReason::DanglingDeleteGrace,
retry_not_before: None,
};
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 = "dangling_delete_grace_skip",
error = %err,
"Heal bucket object dangling cleanup deferred by grace window"
);
} else if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) {
disposition = HealObjectDisposition::Deferred {
reason: HealDeferredReason::TransientUsageCache,
retry_not_before: None,
};
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 {
disposition = HealObjectDisposition::Failed(if err.is_recoverable_heal() {
HealFailureClass::RetryExhausted
if let Some(err) = error {
match err {
Error::TaskCancelled | Error::TaskTimeout => {
let disposition = if matches!(err, Error::TaskCancelled) {
HealObjectDisposition::Cancelled
} else {
HealFailureClass::Permanent
HealObjectDisposition::Deferred {
reason: HealDeferredReason::Deadline,
retry_not_before: None,
}
};
self.outcome.write().await.record(HealObjectOutcome {
identity,
disposition,
detail: None,
});
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"
);
}
aborted_progress_unknown |= !increment_counter(&mut scanned);
aborted_progress_unknown |= !increment_counter(&mut skipped);
return Err(err);
}
_ if !age_exhausted => self.outcome.write().await.attempt_failed(),
_ => {}
}
if terminal_outcome {
telemetry_unknown |= !increment_counter(&mut scanned);
}
if !terminal_outcome {
continue;
}
self.outcome.write().await.record(HealObjectOutcome {
identity,
disposition,
detail,
});
let mut progress = self.progress.write().await;
progress.update_object_progress(
previous_progress.objects_scanned.saturating_add(scanned),
previous_progress.objects_healed.saturating_add(healed),
previous_progress.objects_failed.saturating_add(failed),
previous_progress.skipped_objects.saturating_add(skipped),
previous_progress.bytes_processed.saturating_add(bytes),
);
if telemetry_unknown {
progress.mark_unknown();
detail = Some(err.to_string());
if Self::is_dangling_delete_grace_error(&err) {
disposition = HealObjectDisposition::Deferred {
reason: HealDeferredReason::DanglingDeleteGrace,
retry_not_before: None,
};
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 = "dangling_delete_grace_skip",
error = %err,
"Heal bucket object dangling cleanup deferred by grace window"
);
} else if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) {
disposition = HealObjectDisposition::Deferred {
reason: HealDeferredReason::TransientUsageCache,
retry_not_before: None,
};
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 !age_exhausted && 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"
);
item.defer(self.bucket_object_retry_delay(retry_attempt + 1));
if let Err(item) = deferred.push(item) {
inline_retry = Some(item);
}
} else {
disposition = HealObjectDisposition::Failed(if age_exhausted || err.is_recoverable_heal() {
HealFailureClass::RetryExhausted
} else {
HealFailureClass::Permanent
});
telemetry_unknown |= !increment_counter(&mut failed);
if age_exhausted || 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"
);
}
}
}
pending = retry;
retry_attempt = retry_attempt.saturating_add(1);
}
if !is_truncated {
break;
}
if terminal_outcome {
telemetry_unknown |= !increment_counter(&mut scanned);
}
continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
if continuation_token.is_none() {
// Truncated without a continuation token is a compatibility EOF.
break;
if !terminal_outcome {
continue;
}
self.outcome.write().await.record(HealObjectOutcome {
identity,
disposition,
detail,
});
let mut progress = self.progress.write().await;
progress.update_object_progress(
previous_progress.objects_scanned.saturating_add(scanned),
previous_progress.objects_healed.saturating_add(healed),
previous_progress.objects_failed.saturating_add(failed),
previous_progress.skipped_objects.saturating_add(skipped),
previous_progress.bytes_processed.saturating_add(bytes),
);
if telemetry_unknown {
progress.mark_unknown();
}
}
Ok(())
}
.await;
if let Err(error) = scope_result {
let disposition = match error {
Error::TaskCancelled => HealObjectDisposition::Cancelled,
Error::TaskTimeout => HealObjectDisposition::Deferred {
reason: HealDeferredReason::Deadline,
retry_not_before: None,
},
_ => HealObjectDisposition::Unknown,
};
// Only attempted identities have terminal outcomes. Unstarted
// page tails remain unprocessed under the task's partial coverage.
// No detached sleepers survive abort.
for item in deferred.objects.into_iter().chain(inline_retry) {
self.outcome.write().await.record(HealObjectOutcome {
identity: self.outcome_identity(
bucket,
&item.name,
item.version_id.as_deref(),
heal_opts.pool,
heal_opts.set,
),
disposition: disposition.clone(),
detail: None,
});
aborted_progress_unknown |= !increment_counter(&mut scanned);
aborted_progress_unknown |= !increment_counter(&mut skipped);
}
let mut progress = self.progress.write().await;
progress.update_object_progress(
previous_progress.objects_scanned.saturating_add(scanned),
previous_progress.objects_healed.saturating_add(healed),
previous_progress.objects_failed.saturating_add(failed),
previous_progress.skipped_objects.saturating_add(skipped),
previous_progress.bytes_processed.saturating_add(bytes),
);
if aborted_progress_unknown {
progress.mark_unknown();
}
return Err(error);
}
}
crates/heal/src/heal/task/tests.rs
+38
View File
@@ -15,6 +15,8 @@
use super::super::{DiskOption, DiskStore, Endpoint, new_disk};
use super::*;
mod deferred_retry;
mod canonical_outcome {
use super::*;
use crate::heal::outcome::{HealExecutionOutcome, HealTraversalCoverage};
@@ -939,6 +941,10 @@ async fn verified_recovery_keeps_state_when_marker_clear_fails() {
#[derive(Default)]
struct MockStorage {
retry_test_pages: Option<Vec<Vec<HealListItem>>>,
retry_test_delays: HashMap<String, Duration>,
retry_test_listing_delays: Mutex<VecDeque<Duration>>,
retry_test_events: Mutex<Vec<String>>,
listed: Mutex<bool>,
list_each_bucket: bool,
fail_second_listing_page: bool,
@@ -1109,6 +1115,7 @@ fn replacement_identity(
}
enum MockHealObjectOutcome {
RetryableLock,
OkWithOtherError(&'static str),
ErrOther(&'static str),
DanglingGraceDeferred,
@@ -1213,6 +1220,10 @@ impl HealStorageAPI for MockStorage {
opts: &HealOpts,
) -> Result<(HealResultItem, Option<Error>)> {
self.heal_object_calls.lock().unwrap().push(object.to_string());
self.retry_test_events.lock().expect("events").push(format!("heal:{object}"));
if let Some(delay) = self.retry_test_delays.get(object) {
tokio::time::sleep(*delay).await;
}
self.heal_object_version_ids
.lock()
.unwrap()
@@ -1241,6 +1252,13 @@ impl HealStorageAPI for MockStorage {
bucket.to_string(),
object.to_string(),
))),
MockHealObjectOutcome::RetryableLock => Ok((
HealResultItem::default(),
Some(Error::Storage(EcstoreError::Lock(rustfs_lock::LockError::AlreadyLocked {
resource: object.to_string(),
owner: "competing-writer".to_string(),
}))),
)),
MockHealObjectOutcome::RetryableSlowDown => {
Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::SlowDown))))
}
@@ -1266,6 +1284,13 @@ impl HealStorageAPI for MockStorage {
bucket.to_string(),
object.to_string(),
))),
MockHealObjectOutcome::RetryableLock => Ok((
HealResultItem::default(),
Some(Error::Storage(EcstoreError::Lock(rustfs_lock::LockError::AlreadyLocked {
resource: object.to_string(),
owner: "competing-writer".to_string(),
}))),
)),
MockHealObjectOutcome::RetryableSlowDown => {
Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::SlowDown))))
}
@@ -1361,6 +1386,19 @@ impl HealStorageAPI for MockStorage {
.lock()
.expect("listing tokens")
.push(continuation_token.map(ToOwned::to_owned));
self.retry_test_events
.lock()
.expect("events")
.push(format!("list:{}", continuation_token.unwrap_or("first")));
let delay = self.retry_test_listing_delays.lock().expect("listing delays").pop_front();
if let Some(delay) = delay {
tokio::time::sleep(delay).await;
}
if let Some(pages) = &self.retry_test_pages {
let page = continuation_token.map_or(0, |token| token.parse::<usize>().expect("test page token"));
let next = (page + 1 < pages.len()).then(|| (page + 1).to_string());
return Ok((pages[page].clone(), next.clone(), next.is_some()));
}
if let Some(remaining) = self
.recoverable_second_page_failures
.lock()
crates/heal/src/heal/task/tests/deferred_retry.rs
+487
View File
@@ -0,0 +1,487 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use super::*;
fn bucket_task(storage: Arc<MockStorage>) -> HealTask {
HealTask::from_request(
HealRequest::new(
HealType::Bucket {
bucket: "bucket-a".to_string(),
},
HealOptions {
recursive: true,
timeout: None,
..Default::default()
},
HealPriority::Normal,
),
storage,
)
}
fn pages_storage(pages: &[&[&str]]) -> MockStorage {
MockStorage {
retry_test_pages: Some(
pages
.iter()
.map(|page| page.iter().map(|name| heal_item(name)).collect())
.collect(),
),
..Default::default()
}
}
fn fail_once(storage: &MockStorage, name: &str) {
storage
.heal_object_outcomes
.lock()
.expect("outcomes")
.insert(name.to_string(), VecDeque::from([MockHealObjectOutcome::RetryableLock]));
}
#[tokio::test(start_paused = true)]
async fn slow_listing_retry_services_due_object_then_age_before_next_listing() {
let storage = Arc::new(MockStorage {
recoverable_second_page_failures: Mutex::new(Some(1)),
retry_test_listing_delays: Mutex::new(VecDeque::from([Duration::ZERO, Duration::from_secs(29)])),
..Default::default()
});
storage.heal_object_outcomes.lock().expect("outcomes").insert(
"object-a".to_string(),
VecDeque::from([
MockHealObjectOutcome::RetryableSlowDown,
MockHealObjectOutcome::RetryableSlowDown,
]),
);
let task = bucket_task(storage.clone());
let execution = task.execute();
tokio::pin!(execution);
assert!(
tokio::time::timeout(Duration::from_millis(30_500), &mut execution)
.await
.is_err()
);
assert_eq!(
storage.retry_test_events.lock().expect("events").as_slice(),
["list:first", "heal:object-a", "list:second", "heal:object-a"]
);
let outcome = task.get_outcome().await;
assert_eq!(outcome.counters.processed, 1);
assert_eq!(
outcome.objects[0].disposition,
HealObjectDisposition::Failed(HealFailureClass::RetryExhausted)
);
execution.await.expect_err("age exhausted object must remain a batch failure");
assert_eq!(
storage.retry_test_events.lock().expect("events").as_slice(),
[
"list:first",
"heal:object-a",
"list:second",
"heal:object-a",
"list:second",
"heal:object-b"
]
);
let outcome = task.get_outcome().await;
assert_eq!((outcome.counters.processed, outcome.counters.attempt_failures), (2, 3));
}
#[tokio::test(start_paused = true)]
async fn listing_return_after_age_expires_does_not_start_another_heal_attempt() {
let storage = Arc::new(MockStorage {
recoverable_second_page_failures: Mutex::new(Some(1)),
retry_test_listing_delays: Mutex::new(VecDeque::from([Duration::ZERO, Duration::from_secs(31)])),
..Default::default()
});
fail_once(&storage, "object-a");
let task = bucket_task(storage.clone());
let execution = task.execute();
tokio::pin!(execution);
assert!(
tokio::time::timeout(Duration::from_millis(31_500), &mut execution)
.await
.is_err()
);
assert_eq!(
storage.retry_test_events.lock().expect("events").as_slice(),
["list:first", "heal:object-a", "list:second"]
);
assert_eq!(task.get_outcome().await.counters.failed, 1);
execution.await.expect_err("age exhaustion remains a failure");
assert_eq!(
storage.retry_test_events.lock().expect("events").as_slice(),
["list:first", "heal:object-a", "list:second", "list:second", "heal:object-b"]
);
}
#[tokio::test(start_paused = true)]
async fn full_window_abort_accounts_inline_once_and_leaves_unstarted_tail_unprocessed() {
for cancel in [true, false] {
let names: Vec<String> = (0..258).map(|index| format!("blocked-{index}")).collect();
let mut page: Vec<HealListItem> = names.iter().map(|name| heal_item(name)).collect();
page[256].version_id = Some("inline-version".to_string());
let storage = Arc::new(MockStorage {
retry_test_pages: Some(vec![page, vec![heal_item("healthy")]]),
..Default::default()
});
for name in &names {
fail_once(&storage, name);
}
let mut task = bucket_task(storage.clone());
if !cancel {
task.options.timeout = Some(Duration::from_secs(1));
}
let execution = task.execute();
tokio::pin!(execution);
assert!(
tokio::time::timeout(Duration::from_millis(500), &mut execution)
.await
.is_err()
);
assert_eq!(storage.heal_object_calls.lock().expect("calls").len(), 257);
assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 1);
if cancel {
task.cancel().await.expect("cancel");
}
let result = execution.await;
assert!(matches!(
(&result, cancel),
(Err(Error::TaskCancelled), true) | (Err(Error::TaskTimeout), false)
));
let outcome = task.get_outcome().await;
assert_eq!(
(
outcome.counters.processed,
outcome.counters.skipped,
outcome.counters.failed,
outcome.counters.healed
),
(257, 257, 0, 0)
);
assert_eq!(outcome.coverage, crate::heal::outcome::HealTraversalCoverage::Partial);
assert_eq!(
outcome.execution,
crate::heal::outcome::HealExecutionOutcome::Aborted(if cancel {
HealAbortReason::Cancelled
} else {
HealAbortReason::Deadline
})
);
let inline: Vec<_> = outcome
.objects
.iter()
.filter(|item| item.identity.object == "blocked-256")
.collect();
assert_eq!(inline.len(), 1);
assert_eq!(inline[0].identity.version_id.as_deref(), Some("inline-version"));
assert_eq!(
inline[0].disposition,
if cancel {
HealObjectDisposition::Cancelled
} else {
HealObjectDisposition::Deferred {
reason: HealDeferredReason::Deadline,
retry_not_before: None,
}
}
);
let progress = task.get_progress().await;
assert_eq!(
(
progress.objects_scanned,
progress.skipped_objects,
progress.objects_failed,
progress.objects_healed
),
(257, 257, 0, 0)
);
assert!(
!outcome
.objects
.iter()
.any(|item| item.identity.object == "blocked-257" || item.identity.object == "healthy")
);
tokio::time::advance(Duration::from_secs(60)).await;
assert_eq!(storage.heal_object_calls.lock().expect("calls").len(), 257);
assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 1);
}
}
#[tokio::test(start_paused = true)]
async fn repeated_slowdown_keeps_attempts_and_forward_pages_bounded() {
let storage = Arc::new(pages_storage(&[&["a"], &["b"], &["c"], &["d"]]));
for name in ["a", "b", "c", "d"] {
storage
.heal_object_outcomes
.lock()
.expect("outcomes")
.insert(name.to_string(), (0..4).map(|_| MockHealObjectOutcome::RetryableSlowDown).collect());
}
let task = bucket_task(storage.clone());
let execution = task.execute();
tokio::pin!(execution);
assert!(tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err());
assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 3);
execution.await.expect_err("all four objects exhaust retries");
let outcome = task.get_outcome().await;
assert_eq!(
(outcome.counters.processed, outcome.counters.failed, outcome.counters.attempt_failures),
(4, 4, 16)
);
assert_eq!(storage.heal_object_calls.lock().expect("calls").len(), 16);
for name in ["a", "b", "c", "d"] {
assert_eq!(outcome.objects.iter().filter(|item| item.identity.object == name).count(), 1);
}
}
#[tokio::test(start_paused = true)]
async fn listing_retry_keeps_cursor_and_does_not_replay_successful_objects() {
let storage = Arc::new(MockStorage {
recoverable_second_page_failures: Mutex::new(Some(1)),
..Default::default()
});
fail_once(&storage, "object-a");
let task = bucket_task(storage.clone());
task.execute().await.expect("both retries complete");
assert_eq!(
storage.listing_tokens.lock().expect("tokens").as_slice(),
[None, Some("second".to_string()), Some("second".to_string())]
);
assert_eq!(
storage.heal_object_calls.lock().expect("calls").as_slice(),
["object-a", "object-a", "object-b"]
);
let outcome = task.get_outcome().await;
assert_eq!((outcome.counters.processed, outcome.counters.attempt_failures), (2, 2));
}
#[tokio::test(start_paused = true)]
async fn typed_lock_contention_allows_only_two_forward_pages() {
let storage = Arc::new(pages_storage(&[&["a"], &["b"], &["c"], &["d"]]));
fail_once(&storage, "a");
let task = bucket_task(storage.clone());
let execution = task.execute();
tokio::pin!(execution);
assert!(tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err());
assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b", "c"]);
assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 3);
execution.await.expect("all objects complete");
assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b", "c", "a", "d"]);
assert_eq!(task.get_outcome().await.counters.processed, 4);
}
#[tokio::test(start_paused = true)]
async fn due_retry_runs_before_next_object_in_a_slow_healthy_page() {
let mut storage = pages_storage(&[&["a"], &["b", "c"]]);
storage.retry_test_delays.insert("b".to_string(), Duration::from_secs(3));
fail_once(&storage, "a");
let storage = Arc::new(storage);
bucket_task(storage.clone()).execute().await.expect("all objects complete");
assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b", "a", "c"]);
}
#[tokio::test(start_paused = true)]
async fn expired_retry_is_terminal_without_an_extra_storage_attempt() {
let mut storage = pages_storage(&[&["a"], &["b"]]);
storage.retry_test_delays.insert("b".to_string(), Duration::from_secs(31));
fail_once(&storage, "a");
let storage = Arc::new(storage);
let task = bucket_task(storage.clone());
task.execute().await.expect_err("aged pending responsibility is not success");
assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b"]);
let outcome = task.get_outcome().await;
assert_eq!(outcome.counters.processed, 2);
assert_eq!(outcome.counters.attempt_failures, 1);
assert_eq!(outcome.counters.failed, 1);
assert_eq!(
outcome
.objects
.iter()
.find(|item| item.identity.object == "a")
.expect("a outcome")
.disposition,
HealObjectDisposition::Failed(HealFailureClass::RetryExhausted)
);
}
#[tokio::test(start_paused = true)]
async fn cancellation_drains_owned_retries_once() {
let storage = Arc::new(pages_storage(&[&["a"], &["b"]]));
fail_once(&storage, "a");
let task = bucket_task(storage.clone());
let execution = task.execute();
tokio::pin!(execution);
assert!(tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err());
task.cancel().await.expect("cancel");
assert!(matches!(execution.await, Err(Error::TaskCancelled)));
tokio::time::advance(Duration::from_secs(60)).await;
assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b"]);
let outcome = task.get_outcome().await;
assert_eq!(outcome.counters.processed, 2);
assert_eq!(outcome.objects.iter().filter(|item| item.identity.object == "a").count(), 1);
assert_eq!(
outcome
.objects
.iter()
.find(|item| item.identity.object == "a")
.expect("a")
.disposition,
HealObjectDisposition::Cancelled
);
}
#[tokio::test(start_paused = true)]
async fn deadline_drains_owned_retries_without_false_completion() {
let storage = Arc::new(pages_storage(&[&["a"], &["b"]]));
fail_once(&storage, "a");
let mut task = bucket_task(storage.clone());
task.options.timeout = Some(Duration::from_secs(1));
assert!(matches!(task.execute().await, Err(Error::TaskTimeout)));
let outcome = task.get_outcome().await;
assert_eq!(outcome.counters.processed, 2);
assert_eq!(
outcome
.objects
.iter()
.find(|item| item.identity.object == "a")
.expect("a")
.disposition,
HealObjectDisposition::Deferred {
reason: HealDeferredReason::Deadline,
retry_not_before: None
}
);
tokio::time::advance(Duration::from_secs(60)).await;
assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b"]);
}
#[tokio::test(start_paused = true)]
async fn full_window_backpressures_without_losing_the_current_page_tail() {
let names: Vec<String> = (0..258).map(|index| format!("blocked-{index}")).collect();
let mut storage = MockStorage {
retry_test_pages: Some(vec![names.iter().map(|name| heal_item(name)).collect(), vec![heal_item("healthy")]]),
..Default::default()
};
for name in &names {
fail_once(&storage, name);
}
// The last item has a version, proving the current-page tail is not rebuilt
// from names alone when the window fills.
storage.retry_test_pages.as_mut().expect("pages")[0][257].version_id = Some("version-tail".to_string());
let storage = Arc::new(storage);
let task = bucket_task(storage.clone());
let execution = task.execute();
tokio::pin!(execution);
assert!(tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err());
assert_eq!(storage.heal_object_calls.lock().expect("calls").len(), 257);
assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 1);
execution.await.expect("every owned item eventually completes");
assert_eq!(task.get_outcome().await.counters.processed, 259);
let calls = storage.heal_object_calls.lock().expect("calls");
for name in &names {
assert_eq!(calls.iter().filter(|called| *called == name).count(), 2);
}
let versions = storage.heal_object_version_ids.lock().expect("versions");
for (name, version) in calls.iter().zip(versions.iter()) {
if name == "blocked-257" {
assert_eq!(version.as_deref(), Some("version-tail"));
}
}
}
#[tokio::test(start_paused = true)]
async fn oversized_identity_stays_inline_without_losing_version() {
let name = "k".repeat(256 * 1024);
let mut item = heal_item(&name);
item.version_id = Some("v".repeat(1024));
let storage = Arc::new(MockStorage {
retry_test_pages: Some(vec![vec![item], vec![heal_item("healthy")]]),
..Default::default()
});
fail_once(&storage, &name);
let task = bucket_task(storage.clone());
let execution = task.execute();
tokio::pin!(execution);
assert!(tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err());
assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 1);
execution.await.expect("oversized identity retries inline");
assert_eq!(task.get_outcome().await.counters.processed, 2);
let versions = storage.heal_object_version_ids.lock().expect("versions");
assert_eq!(versions[0], versions[1]);
assert_eq!(versions[0].as_ref().expect("version").len(), 1024);
}
#[tokio::test(start_paused = true)]
async fn terminal_listing_failure_keeps_deferred_identity_unknown() {
let storage = Arc::new(MockStorage {
fail_second_listing_page: true,
..Default::default()
});
fail_once(&storage, "object-a");
let task = bucket_task(storage.clone());
task.execute().await.expect_err("listing cannot continue");
let outcome = task.get_outcome().await;
assert_eq!(outcome.counters.processed, 1);
assert_eq!(outcome.objects[0].disposition, HealObjectDisposition::Unknown);
assert_eq!(outcome.coverage, crate::heal::outcome::HealTraversalCoverage::Partial);
assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["object-a"]);
}
#[tokio::test(start_paused = true)]
async fn healthy_second_page_advances_before_first_retry_is_due() {
let storage = Arc::new(MockStorage {
recoverable_second_page_failures: Mutex::new(Some(0)),
..Default::default()
});
storage
.heal_object_outcomes
.lock()
.expect("outcomes")
.insert("object-a".to_string(), VecDeque::from([MockHealObjectOutcome::RetryableSlowDown]));
let task = HealTask::from_request(
HealRequest::new(
HealType::Bucket {
bucket: "bucket-a".to_string(),
},
HealOptions {
recursive: true,
timeout: None,
..Default::default()
},
HealPriority::Normal,
),
storage.clone(),
);
let execution = task.execute();
tokio::pin!(execution);
assert!(
tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err(),
"the deferred first object must remain pending before its retry is due"
);
assert_eq!(
storage.heal_object_calls.lock().expect("calls").as_slice(),
["object-a", "object-b"],
"a retryable page head must not hold the healthy second page behind its backoff"
);
execution.await.expect("retry eventually succeeds");
let outcome = task.get_outcome().await;
assert_eq!(outcome.counters.processed, 2);
assert_eq!(outcome.counters.attempt_failures, 1);
assert_eq!(
storage.heal_object_calls.lock().expect("calls").as_slice(),
["object-a", "object-b", "object-a"]
);
}
crates/heal/src/heal/task/tests/deferred_retry_window.rs
+76
View File
@@ -0,0 +1,76 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use super::*;
fn item(name: String, version_id: Option<String>) -> DeferredObject {
DeferredObject::new(
HealListItem {
name,
version_id,
mod_time_unix_nanos: None,
lifecycle_object_info: None,
is_delete_marker: false,
},
1,
)
}
#[tokio::test(start_paused = true)]
async fn count_cap_and_next_item_preserve_ownership() {
let mut window = DeferredWindow::default();
for _ in 0..MAX_DEFERRED_OBJECTS {
assert!(window.push(item("key".to_string(), None)).is_ok());
}
let rejected = window
.push(item("next".to_string(), Some("version".to_string())))
.expect_err("count cap");
assert_eq!(rejected.name, "next");
assert_eq!(rejected.version_id.as_deref(), Some("version"));
assert_eq!(window.objects.len(), MAX_DEFERRED_OBJECTS);
assert!(window.bytes <= MAX_DEFERRED_BYTES);
assert!(!window.can_advance(1));
assert!(window.pop_due().is_some());
assert!(window.push(rejected).is_ok());
}
#[tokio::test(start_paused = true)]
async fn byte_cap_counts_key_version_and_reserved_slots() {
let mut window = DeferredWindow::default();
let available = MAX_DEFERRED_BYTES - window.bytes;
let key = "k".repeat(available / 2);
let version = "v".repeat(available - key.capacity());
assert_eq!(key.capacity() + version.capacity(), available);
assert!(window.push(item(key, Some(version))).is_ok());
assert_eq!(window.bytes, MAX_DEFERRED_BYTES);
assert!(!window.can_advance(1));
assert!(window.push(item("x".to_string(), None)).is_err());
assert!(window.pop_due().is_some());
assert_eq!(window.bytes, MAX_DEFERRED_OBJECTS * size_of::<DeferredObject>());
assert!(window.push(item("x".to_string(), None)).is_ok());
}
#[tokio::test(start_paused = true)]
async fn retry_age_caps_due_time_and_is_not_reset_by_rescheduling() {
let mut entry = item("a".to_string(), None);
entry.defer(Duration::from_secs(2));
let first = entry.first_failure.expect("first failure");
tokio::time::advance(Duration::from_secs(29)).await;
entry.defer(Duration::from_secs(8));
assert_eq!(entry.first_failure, Some(first));
assert_eq!(entry.due, first + MAX_DEFERRED_AGE);
assert!(!entry.expired());
tokio::time::advance(Duration::from_secs(1)).await;
assert!(entry.expired());
}