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

fix: bound tier free-version recovery sweeps (#7125)

Browse Source
This commit is contained in:
cxymds
2026-09-04 16:12:01 +08:00
committed by GitHub
parent 5f739944a3
commit fdb28e611f
3 changed files with 369 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+282 -54
View File
@@ -2777,7 +2777,8 @@ async fn run_tier_free_version_recovery_loop<F, Fut>(
match recovery_result {
Ok(stats) => {
let elapsed = started_at.elapsed();
schedule.record_success(&stats, elapsed);
schedule.record_success(&stats);
rustfs_io_metrics::record_stage_duration("lifecycle_free_version_recovery", elapsed.as_secs_f64() * 1000.0);
let (pending_tasks, active_tasks) = {
let state = expiry_state.read().await;
(state.pending_tasks(), state.stats.active_tasks())
@@ -2806,7 +2807,7 @@ async fn run_tier_free_version_recovery_loop<F, Fut>(
}
Err(err) => {
let elapsed = started_at.elapsed();
schedule.record_failure(elapsed);
schedule.record_failure();
rustfs_io_metrics::record_stage_duration(
"lifecycle_free_version_recovery_failed",
elapsed.as_secs_f64() * 1000.0,
@@ -2838,10 +2839,10 @@ async fn wait_for_tier_free_version_recovery(
} else {
schedule.next_delay
};
let sleep_delay = next_delay.saturating_sub(schedule.previous_run_duration);
schedule.previous_run_duration = StdDuration::ZERO;
// Recovery delays are completion-relative. Discounting the previous run
// would let a page that took at least one interval restart with no cooldown.
schedule.jitter_next_delay = false;
let sleep = tokio::time::sleep(sleep_delay);
let sleep = tokio::time::sleep(next_delay);
tokio::pin!(sleep);
let mut recovery_request_consumed = false;
@@ -2891,7 +2892,6 @@ struct TierFreeVersionRecoverySchedule {
next_delay: StdDuration,
idle_interval: StdDuration,
failure_interval: StdDuration,
previous_run_duration: StdDuration,
jitter_next_delay: bool,
bucket_marker: Option<String>,
object_marker: Option<String>,
@@ -2904,7 +2904,6 @@ impl Default for TierFreeVersionRecoverySchedule {
next_delay: StdDuration::ZERO,
idle_interval: TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL,
failure_interval: TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL,
previous_run_duration: StdDuration::ZERO,
jitter_next_delay: false,
bucket_marker: None,
object_marker: None,
@@ -2917,7 +2916,6 @@ impl TierFreeVersionRecoverySchedule {
fn reset_idle_interval(&mut self) {
self.idle_interval = TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL;
self.next_delay = TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL;
self.previous_run_duration = StdDuration::ZERO;
self.jitter_next_delay = false;
}
@@ -2928,18 +2926,15 @@ impl TierFreeVersionRecoverySchedule {
self.reset_idle_interval();
}
fn record_failure(&mut self, _run_duration: StdDuration) {
fn record_failure(&mut self) {
self.idle_interval = TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL;
self.next_delay = self.failure_interval;
self.failure_interval =
std::cmp::min(self.failure_interval.saturating_mul(2), TIER_FREE_VERSION_RECOVERY_MAX_IDLE_INTERVAL);
// Keep the full backoff even after a long failed run, so a run whose
// duration exceeds the interval cannot restart immediately.
self.previous_run_duration = StdDuration::ZERO;
self.jitter_next_delay = false;
}
fn record_success(&mut self, stats: &FreeVersionRecoveryStats, run_duration: StdDuration) {
fn record_success(&mut self, stats: &FreeVersionRecoveryStats) {
self.failure_interval = TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL;
if stats.enqueued > 0 || stats.failed > 0 {
self.follow_up_sweep = true;
@@ -2949,7 +2944,6 @@ impl TierFreeVersionRecoverySchedule {
self.object_marker = stats.next_object_marker.clone();
if stats.truncated {
self.next_delay = TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL;
self.previous_run_duration = run_duration;
self.jitter_next_delay = false;
return;
}
@@ -2960,13 +2954,11 @@ impl TierFreeVersionRecoverySchedule {
self.follow_up_sweep = false;
self.idle_interval = TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL;
self.next_delay = TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL;
self.previous_run_duration = run_duration;
self.jitter_next_delay = false;
return;
}
self.next_delay = self.idle_interval;
self.previous_run_duration = run_duration;
self.jitter_next_delay = true;
self.idle_interval = std::cmp::min(self.idle_interval.saturating_mul(2), TIER_FREE_VERSION_RECOVERY_MAX_IDLE_INTERVAL);
}
@@ -5623,8 +5615,9 @@ mod tests {
use crate::bucket::lifecycle::replication_sink::{ReplicationStatusType, VersionPurgeStatusType};
use crate::bucket::lifecycle::runtime_boundary as runtime_sources;
use crate::bucket::lifecycle::tier_free_version_recovery::{
FreeVersionRecoveryStats, RecoveryWalkTestAction, list_tier_free_versions, recover_tier_free_versions_with_cancel,
set_recovery_bucket_list_wait_hook, set_recovery_walk_test_hook,
FreeVersionRecoveryStats, RecoveryWalkTestAction, RecoveryWorkBudget, list_tier_free_versions,
list_tier_free_versions_with_budget, recover_tier_free_versions_with_cancel, set_recovery_bucket_list_wait_hook,
set_recovery_walk_test_hook,
};
use crate::bucket::lifecycle::tier_last_day_stats::LastDayTierStats;
use crate::bucket::lifecycle::tier_sweeper::Jentry;
@@ -5712,7 +5705,7 @@ mod tests {
assert_eq!(schedule.next_delay, StdDuration::ZERO);
for expected in [60, 120, 240, 480, 600, 600, 600, 600] {
schedule.record_success(&idle, StdDuration::ZERO);
schedule.record_success(&idle);
assert_eq!(schedule.next_delay, StdDuration::from_secs(expected));
}
assert_eq!(schedule.next_delay.as_secs() / TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL.as_secs(), 10);
@@ -5723,15 +5716,13 @@ mod tests {
let mut schedule = TierFreeVersionRecoverySchedule::default();
for expected in [60, 120, 240, 480, 600, 600] {
schedule.record_failure(StdDuration::from_secs(75));
schedule.record_failure();
assert_eq!(schedule.next_delay, StdDuration::from_secs(expected));
assert_eq!(schedule.previous_run_duration, StdDuration::ZERO);
}
schedule.record_success(&free_version_recovery_stats(0, 0, false), StdDuration::ZERO);
schedule.record_failure(StdDuration::from_secs(75));
schedule.record_success(&free_version_recovery_stats(0, 0, false));
schedule.record_failure();
assert_eq!(schedule.next_delay, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL);
assert_eq!(schedule.previous_run_duration, StdDuration::ZERO);
}
#[test]
@@ -5746,22 +5737,22 @@ mod tests {
fn tier_free_version_recovery_pagination_preserves_full_sweep_backoff() {
let idle = free_version_recovery_stats(0, 0, false);
let mut schedule = TierFreeVersionRecoverySchedule::default();
schedule.record_success(&idle, StdDuration::ZERO);
schedule.record_success(&idle, StdDuration::ZERO);
schedule.record_success(&idle);
schedule.record_success(&idle);
assert_eq!(schedule.next_delay, StdDuration::from_secs(120));
assert_eq!(schedule.idle_interval, StdDuration::from_secs(240));
schedule.record_success(&free_version_recovery_stats(0, 0, true), StdDuration::ZERO);
schedule.record_success(&free_version_recovery_stats(0, 0, true));
assert_eq!(schedule.next_delay, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL);
assert_eq!(schedule.idle_interval, StdDuration::from_secs(240));
assert_eq!(schedule.bucket_marker.as_deref(), Some("bucket"));
assert_eq!(schedule.object_marker.as_deref(), Some("object"));
schedule.record_success(&free_version_recovery_stats(0, 0, true), StdDuration::ZERO);
schedule.record_success(&free_version_recovery_stats(0, 0, true));
assert_eq!(schedule.next_delay, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL);
assert_eq!(schedule.idle_interval, StdDuration::from_secs(240));
schedule.record_success(&idle, StdDuration::ZERO);
schedule.record_success(&idle);
assert_eq!(schedule.next_delay, StdDuration::from_secs(240));
assert_eq!(schedule.idle_interval, StdDuration::from_secs(480));
assert!(schedule.bucket_marker.is_none());
@@ -5771,7 +5762,7 @@ mod tests {
#[test]
fn tier_free_version_recovery_wake_during_pagination_keeps_one_full_follow_up() {
let mut schedule = TierFreeVersionRecoverySchedule::default();
schedule.record_success(&free_version_recovery_stats(0, 0, true), StdDuration::ZERO);
schedule.record_success(&free_version_recovery_stats(0, 0, true));
schedule.request_retry();
schedule.request_retry();
@@ -5779,7 +5770,7 @@ mod tests {
assert_eq!(schedule.bucket_marker.as_deref(), Some("bucket"));
assert_eq!(schedule.object_marker.as_deref(), Some("object"));
schedule.record_success(&free_version_recovery_stats(0, 0, false), StdDuration::ZERO);
schedule.record_success(&free_version_recovery_stats(0, 0, false));
assert_eq!(schedule.next_delay, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL);
assert!(!schedule.follow_up_sweep);
assert!(schedule.bucket_marker.is_none());
@@ -5794,22 +5785,22 @@ mod tests {
free_version_recovery_stats(0, 1, true),
] {
let mut schedule = TierFreeVersionRecoverySchedule::default();
schedule.record_success(&idle, StdDuration::ZERO);
schedule.record_success(&idle, StdDuration::ZERO);
schedule.record_success(&idle);
schedule.record_success(&idle);
assert_eq!(schedule.idle_interval, StdDuration::from_secs(240));
schedule.record_success(&work, StdDuration::ZERO);
schedule.record_success(&work);
assert!(schedule.follow_up_sweep);
assert_eq!(schedule.idle_interval, StdDuration::from_secs(240));
assert!(!schedule.jitter_next_delay);
schedule.record_success(&idle, StdDuration::ZERO);
schedule.record_success(&idle);
assert!(!schedule.follow_up_sweep);
assert_eq!(schedule.next_delay, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL);
assert_eq!(schedule.idle_interval, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL);
assert!(!schedule.jitter_next_delay);
schedule.record_success(&idle, StdDuration::ZERO);
schedule.record_success(&idle);
assert_eq!(schedule.next_delay, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL);
assert_eq!(schedule.idle_interval, StdDuration::from_secs(120));
}
@@ -5823,17 +5814,17 @@ mod tests {
free_version_recovery_stats(0, 1, false),
] {
let mut schedule = TierFreeVersionRecoverySchedule::default();
schedule.record_success(&idle, StdDuration::ZERO);
schedule.record_success(&idle, StdDuration::ZERO);
schedule.record_success(&idle);
schedule.record_success(&idle);
assert_eq!(schedule.idle_interval, StdDuration::from_secs(240));
schedule.record_success(&work, StdDuration::ZERO);
schedule.record_success(&work);
assert_eq!(schedule.next_delay, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL);
assert_eq!(schedule.idle_interval, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL);
assert!(!schedule.follow_up_sweep);
assert!(!schedule.jitter_next_delay);
schedule.record_success(&idle, StdDuration::ZERO);
schedule.record_success(&idle);
assert_eq!(schedule.next_delay, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL);
assert_eq!(schedule.idle_interval, StdDuration::from_secs(120));
}
@@ -5972,7 +5963,7 @@ mod tests {
);
}
async fn tier_free_version_recovery_page_call_times(run_duration: StdDuration) -> Vec<StdDuration> {
async fn tier_free_version_recovery_page_call_times(run_durations: &[StdDuration]) -> Vec<StdDuration> {
RECOVERY_JITTER_CALLS.store(0, Ordering::SeqCst);
let cancel = CancellationToken::new();
let state = ExpiryState::new();
@@ -5981,6 +5972,7 @@ mod tests {
let recorded_call_times = Arc::clone(&call_times);
let call_index = Arc::new(AtomicUsize::new(0));
let recorded_call_index = Arc::clone(&call_index);
let recovery_run_durations = run_durations.to_vec();
let loop_cancel = cancel.clone();
let recovery_cancel = cancel.clone();
let worker = tokio::spawn(async move {
@@ -5991,8 +5983,9 @@ mod tests {
.push(tokio::time::Instant::now().duration_since(started_at));
let index = recorded_call_index.fetch_add(1, Ordering::SeqCst);
let cancel = recovery_cancel.clone();
let run_duration = recovery_run_durations.get(index).copied();
async move {
if index == 0 {
if let Some(run_duration) = run_duration {
tokio::time::sleep(run_duration).await;
Ok(free_version_recovery_stats(0, 0, true))
} else {
@@ -6005,15 +5998,14 @@ mod tests {
});
tokio::task::yield_now().await;
tokio::time::advance(run_duration).await;
tokio::task::yield_now().await;
if run_duration < TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL {
let remaining = TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL - run_duration;
tokio::time::advance(remaining - StdDuration::from_secs(1)).await;
assert_eq!(call_index.load(Ordering::SeqCst), 1);
for (index, run_duration) in run_durations.iter().copied().enumerate() {
tokio::time::advance(run_duration).await;
tokio::task::yield_now().await;
tokio::time::advance(TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL - StdDuration::from_secs(1)).await;
assert_eq!(call_index.load(Ordering::SeqCst), index + 1);
tokio::time::advance(StdDuration::from_secs(1)).await;
tokio::task::yield_now().await;
}
tokio::task::yield_now().await;
worker.await.expect("recovery loop should stop after cancellation");
assert_eq!(
RECOVERY_JITTER_CALLS.load(Ordering::SeqCst),
@@ -6029,14 +6021,75 @@ mod tests {
#[tokio::test(start_paused = true)]
#[serial]
async fn tier_free_version_recovery_preserves_start_to_start_page_cadence() {
async fn tier_free_version_recovery_waits_after_each_page_completes() {
assert_eq!(
tier_free_version_recovery_page_call_times(StdDuration::from_secs(45)).await,
vec![StdDuration::ZERO, StdDuration::from_secs(60)]
tier_free_version_recovery_page_call_times(&[StdDuration::from_secs(45)]).await,
vec![StdDuration::ZERO, StdDuration::from_secs(105)]
);
assert_eq!(
tier_free_version_recovery_page_call_times(StdDuration::from_secs(75)).await,
vec![StdDuration::ZERO, StdDuration::from_secs(75)]
tier_free_version_recovery_page_call_times(&[StdDuration::from_secs(60)]).await,
vec![StdDuration::ZERO, StdDuration::from_secs(120)]
);
assert_eq!(
tier_free_version_recovery_page_call_times(&[StdDuration::from_secs(75)]).await,
vec![StdDuration::ZERO, StdDuration::from_secs(135)]
);
assert_eq!(
tier_free_version_recovery_page_call_times(&[StdDuration::from_secs(75), StdDuration::from_secs(45)]).await,
vec![StdDuration::ZERO, StdDuration::from_secs(135), StdDuration::from_secs(240)]
);
}
#[tokio::test(start_paused = true)]
async fn tier_free_version_recovery_notify_during_page_keeps_completion_cooldown() {
let cancel = CancellationToken::new();
let state = ExpiryState::new();
let recovery_notify = Arc::clone(&state.read().await.recovery_notify);
let started_at = tokio::time::Instant::now();
let call_times = Arc::new(StdMutex::new(Vec::new()));
let recorded_call_times = Arc::clone(&call_times);
let call_index = Arc::new(AtomicUsize::new(0));
let recorded_call_index = Arc::clone(&call_index);
let loop_cancel = cancel.clone();
let recovery_cancel = cancel.clone();
let worker = tokio::spawn(async move {
run_tier_free_version_recovery_loop(loop_cancel, state, std::convert::identity, move |_, _, _| {
recorded_call_times
.lock()
.expect("recovery call times lock should not be poisoned")
.push(tokio::time::Instant::now().duration_since(started_at));
let index = recorded_call_index.fetch_add(1, Ordering::SeqCst);
let cancel = recovery_cancel.clone();
async move {
if index == 0 {
tokio::time::sleep(StdDuration::from_secs(75)).await;
Ok(free_version_recovery_stats(0, 0, true))
} else {
cancel.cancel();
Ok(free_version_recovery_stats(0, 0, false))
}
}
})
.await;
});
tokio::task::yield_now().await;
tokio::time::advance(StdDuration::from_secs(30)).await;
recovery_notify.notify_one();
tokio::time::advance(StdDuration::from_secs(45)).await;
tokio::task::yield_now().await;
tokio::time::advance(StdDuration::from_secs(59)).await;
assert_eq!(call_index.load(Ordering::SeqCst), 1);
tokio::time::advance(StdDuration::from_secs(1)).await;
tokio::task::yield_now().await;
worker.await.expect("recovery loop should stop after cancellation");
assert_eq!(
call_times
.lock()
.expect("recovery call times lock should not be poisoned")
.as_slice(),
&[StdDuration::ZERO, StdDuration::from_secs(135)]
);
}
@@ -12557,6 +12610,181 @@ mod tests {
.expect("empty recovery test bucket should be removed");
}
#[tokio::test]
#[serial]
async fn tier_free_version_recovery_object_budget_is_global_across_buckets() {
let (_disk_paths, ecstore) = setup_test_env().await;
for budget in [
RecoveryWorkBudget {
max_objects: 0,
max_buckets: 1,
},
RecoveryWorkBudget {
max_objects: 1,
max_buckets: 0,
},
] {
let err = list_tier_free_versions_with_budget(Arc::clone(&ecstore), 1, None, None, CancellationToken::new(), budget)
.await
.expect_err("a zero recovery work budget must fail before starting an unbounded walk");
assert!(matches!(err, Error::Io(_)));
}
let suffix = Uuid::new_v4().simple();
let buckets = [
format!("zzzz-recovery-budget-{suffix}-a"),
format!("zzzz-recovery-budget-{suffix}-b"),
format!("zzzz-recovery-budget-{suffix}-c"),
];
for bucket in &buckets {
create_test_bucket(&ecstore, bucket).await;
}
let first_bucket = buckets[0].clone();
let first_bucket_for_hook = first_bucket.clone();
let _walk = set_recovery_walk_test_hook(move |bucket| {
(bucket == first_bucket_for_hook).then(|| {
RecoveryWalkTestAction::SendItems(
["ordinary-object-0", "ordinary-object-0", "ordinary-object-1"]
.into_iter()
.map(|name| ObjectInfo {
bucket: first_bucket_for_hook.clone(),
name: name.to_string(),
..Default::default()
})
.collect(),
)
})
});
let page = list_tier_free_versions_with_budget(
Arc::clone(&ecstore),
1,
None,
None,
CancellationToken::new(),
RecoveryWorkBudget {
max_objects: 1,
max_buckets: 10,
},
)
.await
.expect("the decoded-object boundary should enforce the recovery budget");
assert!(page.items.is_empty());
assert_eq!(page.scanned_entries, 3);
assert_eq!(page.buckets_scanned, 1);
assert!(page.truncated);
assert_eq!(page.next_bucket_marker.as_deref(), Some(first_bucket.as_str()));
assert_eq!(page.next_object_marker.as_deref(), Some("ordinary-object-0"));
drop(_walk);
let walked = Arc::new(StdMutex::new(Vec::new()));
let walked_by_hook = Arc::clone(&walked);
let test_buckets = buckets.clone();
let _walk = set_recovery_walk_test_hook(move |bucket| {
test_buckets
.iter()
.find(|candidate| candidate.as_str() == bucket)
.map(|bucket| {
walked_by_hook
.lock()
.expect("recovery walk log should not be poisoned")
.push(bucket.clone());
RecoveryWalkTestAction::SendItems(vec![ObjectInfo {
bucket: bucket.clone(),
name: "ordinary-object".to_string(),
..Default::default()
}])
})
});
let page = list_tier_free_versions_with_budget(
Arc::clone(&ecstore),
1,
None,
None,
CancellationToken::new(),
RecoveryWorkBudget {
max_objects: 2,
max_buckets: 10,
},
)
.await
.expect("the bounded recovery page should be listed");
assert!(page.items.is_empty());
assert_eq!(page.scanned_entries, 2);
assert_eq!(page.buckets_scanned, 2);
assert!(page.truncated);
assert_eq!(page.next_bucket_marker.as_deref(), Some(buckets[1].as_str()));
assert_eq!(page.next_object_marker.as_deref(), Some("ordinary-object"));
assert_eq!(walked.lock().expect("recovery walk log should not be poisoned").as_slice(), &buckets[..2]);
for bucket in &buckets {
ecstore
.delete_bucket(bucket, &DeleteBucketOptions::default())
.await
.expect("empty recovery test bucket should be removed");
}
}
#[tokio::test]
#[serial]
async fn tier_free_version_recovery_bucket_budget_resumes_at_unscanned_bucket() {
let (_disk_paths, ecstore) = setup_test_env().await;
let suffix = Uuid::new_v4().simple();
let buckets = [
format!("zzzz-recovery-buckets-{suffix}-a"),
format!("zzzz-recovery-buckets-{suffix}-b"),
];
for bucket in &buckets {
create_test_bucket(&ecstore, bucket).await;
}
let test_buckets = buckets.clone();
let _walk = set_recovery_walk_test_hook(move |bucket| {
test_buckets
.iter()
.any(|candidate| candidate.as_str() == bucket)
.then(|| RecoveryWalkTestAction::SendItems(Vec::new()))
});
let budget = RecoveryWorkBudget {
max_objects: 10,
max_buckets: 1,
};
let first = list_tier_free_versions_with_budget(Arc::clone(&ecstore), 1, None, None, CancellationToken::new(), budget)
.await
.expect("the first bucket-bounded page should be listed");
assert_eq!(first.buckets_scanned, 1);
assert!(first.truncated);
assert_eq!(first.next_bucket_marker.as_deref(), Some(buckets[1].as_str()));
assert!(first.next_object_marker.is_none());
let second = list_tier_free_versions_with_budget(
Arc::clone(&ecstore),
1,
first.next_bucket_marker,
first.next_object_marker,
CancellationToken::new(),
budget,
)
.await
.expect("the second bucket-bounded page should resume");
assert_eq!(second.buckets_scanned, 1);
assert!(!second.truncated);
assert!(second.next_bucket_marker.is_none());
assert!(second.next_object_marker.is_none());
for bucket in &buckets {
ecstore
.delete_bucket(bucket, &DeleteBucketOptions::default())
.await
.expect("empty recovery test bucket should be removed");
}
}
#[tokio::test]
#[serial]
async fn tier_free_version_recovery_propagates_walk_item_error() {
crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs
+77 -3
View File
@@ -32,7 +32,10 @@ use crate::store::ECStore;
use rustfs_filemeta::FileInfo;
pub const DEFAULT_FREE_VERSION_RECOVERY_LIMIT: usize = 1_000;
// These are page-wide repair budgets. Applying them per bucket would still let
// one recovery pass walk an unbounded namespace before the scheduler can cool down.
const DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT: usize = 10_000;
const DEFAULT_FREE_VERSION_RECOVERY_BUCKET_LIMIT: usize = 100;
#[cfg(not(test))]
const BACKGROUND_WALK_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
#[cfg(test)]
@@ -41,6 +44,12 @@ const BACKGROUND_WALK_SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(100);
type ObjectInfoOrErr = StorageObjectInfoOrErr<ObjectInfo, crate::error::Error>;
type WalkOptions = StorageWalkOptions<fn(&FileInfo) -> bool>;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct RecoveryWorkBudget {
pub(super) max_objects: usize,
pub(super) max_buckets: usize,
}
fn recovery_walk_options(limit: usize, marker: Option<String>) -> WalkOptions {
WalkOptions {
include_free_versions: true,
@@ -58,6 +67,7 @@ fn recovery_walk_options(limit: usize, marker: Option<String>) -> WalkOptions {
#[cfg(test)]
pub(super) enum RecoveryWalkTestAction {
SendItems(Vec<ObjectInfo>),
SendItemsThenError(Vec<ObjectInfo>, crate::error::Error),
SendItemsThenHang(Vec<ObjectInfo>, Arc<tokio::sync::Notify>),
SendItemsUntilReceiverCloses(Arc<tokio::sync::Notify>),
@@ -277,6 +287,17 @@ pub(super) async fn list_tier_free_versions(
bucket_marker: Option<String>,
object_marker: Option<String>,
cancel_token: CancellationToken,
) -> Result<FreeVersionRecoveryPage> {
list_tier_free_versions_with_budget(api, limit, bucket_marker, object_marker, cancel_token, recovery_work_budget(limit)).await
}
pub(super) async fn list_tier_free_versions_with_budget(
api: Arc<ECStore>,
limit: usize,
bucket_marker: Option<String>,
object_marker: Option<String>,
cancel_token: CancellationToken,
work_budget: RecoveryWorkBudget,
) -> Result<FreeVersionRecoveryPage> {
let mut page = FreeVersionRecoveryPage {
items: Vec::new(),
@@ -290,6 +311,9 @@ pub(super) async fn list_tier_free_versions(
if limit == 0 {
return Ok(page);
}
if work_budget.max_objects == 0 || work_budget.max_buckets == 0 {
return Err(std::io::Error::other("free-version recovery work budget must be greater than zero").into());
}
let bucket_options = BucketOptions::default();
let list_buckets = async {
@@ -313,7 +337,7 @@ pub(super) async fn list_tier_free_versions(
};
let mut bucket_seen = bucket_marker.is_none();
let mut truncated_after: Option<RecoveryCursor> = None;
let walk_scan_limit = recovery_walk_scan_limit(limit);
let mut remaining_scan_objects = work_budget.max_objects;
for bucket in buckets {
if cancel_token.is_cancelled() {
@@ -329,12 +353,20 @@ pub(super) async fn list_tier_free_versions(
bucket_seen = true;
}
if page.buckets_scanned >= work_budget.max_buckets {
page.truncated = true;
page.next_bucket_marker = Some(bucket.name);
page.next_object_marker = None;
break;
}
page.buckets_scanned += 1;
let bucket_object_marker = if bucket_marker.as_deref() == Some(bucket.name.as_str()) {
object_marker.clone()
} else {
None
};
let bucket_walk_limit = remaining_scan_objects;
let (tx, mut rx) = mpsc::channel::<ObjectInfoOrErr>(100);
let cancel = cancel_token.child_token();
@@ -358,6 +390,21 @@ pub(super) async fn list_tier_free_versions(
#[cfg(test)]
if let Some(action) = test_action {
match action {
RecoveryWalkTestAction::SendItems(items) => {
for item in items {
if tx
.send(ObjectInfoOrErr {
item: Some(item),
err: None,
})
.await
.is_err()
{
return Ok(());
}
}
return Ok(());
}
RecoveryWalkTestAction::SendItemsThenError(items, err) => {
for item in items {
if tx
@@ -425,7 +472,7 @@ pub(super) async fn list_tier_free_versions(
}
}
api.walk(cancel, &bucket_name, "", tx, recovery_walk_options(walk_scan_limit, object_marker))
api.walk(cancel, &bucket_name, "", tx, recovery_walk_options(bucket_walk_limit, object_marker))
.await
}
});
@@ -466,6 +513,18 @@ pub(super) async fn list_tier_free_versions(
let Some(oi) = item.item else {
continue;
};
if last_seen_object.as_deref() != Some(oi.name.as_str()) && scanned_objects >= bucket_walk_limit {
// The disk listing limit follows S3-visible counting rules and
// does not charge metadata containing only hidden/free versions.
// Enforce the repair budget again at the decoded-object boundary.
page.truncated = true;
page.next_bucket_marker = Some(bucket.name.clone());
page.next_object_marker = last_seen_object.clone();
cancel.cancel();
draining_after_truncation = true;
drain_deadline = Some(tokio::time::Instant::now() + BACKGROUND_WALK_SHUTDOWN_TIMEOUT);
continue;
}
record_scanned_object(&mut last_seen_object, &mut scanned_objects, &oi.name);
if let Some(cursor) = &truncated_after
&& (cursor.bucket.as_str() != bucket.name.as_str() || cursor.object.as_str() != oi.name.as_str())
@@ -509,7 +568,8 @@ pub(super) async fn list_tier_free_versions(
return Err(err);
}
walk_result?;
mark_scan_truncated_if_needed(&mut page, scanned_objects, walk_scan_limit, &bucket.name, last_seen_object.as_deref());
remaining_scan_objects = remaining_scan_objects.saturating_sub(scanned_objects);
mark_scan_truncated_if_needed(&mut page, scanned_objects, bucket_walk_limit, &bucket.name, last_seen_object.as_deref());
if page.truncated {
break;
@@ -528,6 +588,13 @@ fn recovery_walk_scan_limit(limit: usize) -> usize {
DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT.max(limit.saturating_add(1))
}
fn recovery_work_budget(limit: usize) -> RecoveryWorkBudget {
RecoveryWorkBudget {
max_objects: recovery_walk_scan_limit(limit),
max_buckets: DEFAULT_FREE_VERSION_RECOVERY_BUCKET_LIMIT,
}
}
fn record_scanned_object(last_seen_object: &mut Option<String>, scanned_objects: &mut usize, object: &str) {
if last_seen_object.as_deref() == Some(object) {
return;
@@ -700,6 +767,13 @@ mod tests {
recovery_walk_scan_limit(DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT),
DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT + 1
);
assert_eq!(
recovery_work_budget(DEFAULT_FREE_VERSION_RECOVERY_LIMIT),
RecoveryWorkBudget {
max_objects: DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT,
max_buckets: DEFAULT_FREE_VERSION_RECOVERY_BUCKET_LIMIT,
}
);
}
#[test]