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

fix(scanner): classify erasure candidates in one budgeted walk (#8869)

Browse Source 
* test(scanner): isolate async checkpoint log capture

Scope captured tracing subscribers to each future poll and cover the
competing-dispatcher failure without weakening checkpoint assertions.
Verify exact bounded erasure-probe accounting around the 64-entry limit.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(scanner): classify erasure candidates in one budgeted walk

Continue directory classification in cooperative 64-entry batches until
part evidence or EOF, retaining normal checkpoint and namespace work.
Defer metadata accounting and namespace reuse until classification ends.
Distinguish vanished opens from incomplete iteration and preserve quota,
sampling, failed-object and non-destructive heal behavior.

Pin test callsite interest alongside poll-scoped capture to prevent an
unsubscribed sibling from caching the checkpoint callsite as disabled.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser
2026-10-24 15:49:00 +08:00
committed by GitHub
parent 911ea265c5
commit 0c8b05585e
8 changed files with 1299 additions and 368 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/scanner_folder.rs
+341 -185
View File
@@ -96,8 +96,8 @@ const SCANNER_CHECKPOINT_OBJECT_INTERVAL: u64 = 1024;
const SCANNER_CHECKPOINT_MIN_INTERVAL: Duration = Duration::from_secs(5);
const SCANNER_CHECKPOINT_INTERVAL: Duration = Duration::from_secs(60);
const SCANNER_RAW_ENUMERATION_PAGE_ENTRY_LIMIT: usize = 128;
// Erasure data directories contain direct part.N files; keep namespace probes bounded.
const ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT: usize = 64;
// Bound cooperative directory work without treating a batch boundary as EOF.
const SCANNER_DIRECTORY_BATCH_ENTRIES: usize = 64;
const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024;
const ENV_DATA_USAGE_UPDATE_DIR_CYCLES: &str = "RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES";
const ENV_HEAL_OBJECT_SELECT_PROB: &str = "RUSTFS_HEAL_OBJECT_SELECT_PROB";
@@ -419,6 +419,25 @@ enum FolderScanSource {
Existing,
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum FolderScanMode {
Namespace,
ErasureCandidate { reuse_namespace: bool },
}
#[derive(Debug, PartialEq, Eq)]
enum FolderScanOutcome {
Namespace,
ErasureData,
Vanished,
}
fn is_erasure_candidate(name: &str) -> bool {
name.rsplit('/')
.next()
.is_some_and(|name| uuid::Uuid::parse_str(name).is_ok_and(|id| !id.is_nil()))
}
#[derive(Clone, Debug)]
struct QueuedFolder {
folder: CachedFolder,
@@ -920,16 +939,54 @@ async fn wait_for_scanner_io<T>(ctx: &CancellationToken, wait: impl std::future:
}
}
struct ScannerDirectory {
reader: tokio::fs::ReadDir,
batch_entries: usize,
#[cfg(test)]
ordered: Option<tests::erasure_probe::OrderedEntries>,
}
impl ScannerDirectory {
async fn open(path: &str) -> std::io::Result<Self> {
let reader = tokio::fs::read_dir(path).await?;
#[cfg(test)]
let (reader, ordered) = {
let mut reader = reader;
let ordered = tests::erasure_probe::ordered_entries(path, &mut reader).await?;
(reader, ordered)
};
Ok(Self {
reader,
batch_entries: 0,
#[cfg(test)]
ordered,
})
}
async fn next_entry(&mut self) -> std::io::Result<Option<tokio::fs::DirEntry>> {
#[cfg(test)]
if let Some(entries) = &mut self.ordered {
return entries.pop_front().transpose();
}
self.reader.next_entry().await
}
}
async fn next_scanner_dir_entry(
ctx: &CancellationToken,
sleeper: &DynamicSleeper,
entries: &mut tokio::fs::ReadDir,
entries: &mut ScannerDirectory,
) -> Result<Option<tokio::fs::DirEntry>, ScannerError> {
if entries.batch_entries == SCANNER_DIRECTORY_BATCH_ENTRIES {
wait_for_scanner_io(ctx, tokio::task::yield_now()).await?;
entries.batch_entries = 0;
}
let entry = wait_for_scanner_io(ctx, entries.next_entry()).await??;
if entry.is_some() {
// Tokio prefetches bounded chunks. Pacing consumption also paces the
// next refill, without charging an extra entry for EOF.
wait_for_scanner_io(ctx, sleeper.admit_disk_io(1, 0)).await?;
entries.batch_entries += 1;
}
Ok(entry)
}
@@ -1166,6 +1223,17 @@ impl FolderScanner {
self.record_scan_resume_hint(folder);
}
fn can_reuse_covered_namespace(&self, name: &str, compacted: bool) -> bool {
self.old_cache.info.scan_progress.is_some()
&& !compacted
&& self
.resume_frontier
.as_deref()
.is_some_and(|frontier| scan_frontier_covers_path(name, frontier))
&& self.old_cache.find(name).is_some()
&& !self.resume_subtree_has_failures(name)
}
fn resume_subtree_has_failures(&self, folder: &str) -> bool {
if self.resume_failed_paths.is_empty() {
return false;
@@ -1294,9 +1362,10 @@ impl FolderScanner {
parent_hash: &mut DataUsageHash,
parent_entry: &mut DataUsageEntry,
child_entry: &mut DataUsageEntry,
) -> Result<(), ScannerError> {
mode: FolderScanMode,
) -> Result<FolderScanOutcome, ScannerError> {
if self.checkpoint_tx.is_none() || self.old_cache.info.scan_progress.is_none() {
return Box::pin(self.scan_folder(ctx, folder, child_entry)).await;
return Box::pin(self.scan_folder_inner(ctx, folder, child_entry, mode)).await;
}
let depth = self.checkpoint_depth;
if depth >= MAX_DATA_USAGE_CACHE_DEPTH {
@@ -1310,7 +1379,7 @@ impl FolderScanner {
std::mem::swap(parent_hash, &mut self.checkpoint_ancestors[depth].hash);
std::mem::swap(parent_entry, &mut self.checkpoint_ancestors[depth].entry);
self.checkpoint_depth += 1;
let result = Box::pin(self.scan_folder(ctx, folder, child_entry)).await;
let result = Box::pin(self.scan_folder_inner(ctx, folder, child_entry, mode)).await;
self.checkpoint_depth -= 1;
// Restore the parent before cancellation/error handling stitches partial work.
std::mem::swap(parent_hash, &mut self.checkpoint_ancestors[depth].hash);
@@ -1632,23 +1701,79 @@ impl FolderScanner {
self.send_update().await;
}
async fn record_missing_erasure_metadata(
&mut self,
folder: &CachedFolder,
into: &mut DataUsageEntry,
) -> Result<(), ScannerError> {
let metadata_path = path_join_buf(&[&self.root, &folder.name, STORAGE_FORMAT_FILE]);
self.mark_failed_path_seen(&metadata_path);
let retry_suppressed = self.failed_retry_suppressed(&metadata_path);
into.failed_objects = into.failed_objects.saturating_add(1);
if !retry_suppressed {
self.record_failed(&metadata_path);
let failed_cache_entries = self.new_cache.info.failed_objects.len();
if failed_cache_entries > 0 && should_log_failed_object(failed_cache_entries) {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
path = %metadata_path,
failed_objects = failed_cache_entries,
state = "object_metadata_missing",
"Scanner found erasure object data without metadata"
);
}
}
if !retry_suppressed {
let (bucket, object) = path2_bucket_object_with_base_path(&self.root, &folder.name);
if !bucket.is_empty() && !object.is_empty() {
self.send_required_scanner_heal_request(
PendingScannerHealKind::Object,
bucket.clone(),
Some(object.clone()),
None,
build_non_destructive_object_heal_request(bucket, object, self.scan_mode, HealChannelPriority::High),
)
.await?;
}
}
Ok(())
}
/// Scan a folder recursively
/// Files found in the folders will be added to new_cache.
#[allow(clippy::never_loop)]
#[allow(unused_assignments)]
pub async fn scan_folder(
&mut self,
ctx: CancellationToken,
folder: CachedFolder,
into: &mut DataUsageEntry,
) -> Result<(), ScannerError> {
self.scan_folder_inner(ctx, folder, into, FolderScanMode::Namespace)
.await
.map(|_| ())
}
#[allow(clippy::never_loop)]
#[allow(unused_assignments)]
async fn scan_folder_inner(
&mut self,
ctx: CancellationToken,
folder: CachedFolder,
into: &mut DataUsageEntry,
mode: FolderScanMode,
) -> Result<FolderScanOutcome, ScannerError> {
let done_folder = Metrics::time(Metric::ScanFolder);
let trace_started_at = trace_start_instant();
if ctx.is_cancelled() {
return Err(ScannerError::Other("Operation cancelled".to_string()));
}
if !self.budget.try_start_directory() {
if mode == FolderScanMode::Namespace && !self.budget.try_start_directory() {
return Err(ScannerError::Other("Operation cancelled".to_string()));
}
@@ -1656,7 +1781,7 @@ impl FolderScanner {
// Store initial compaction state.
let was_compacted = into.compacted;
loop {
'scan: loop {
if ctx.is_cancelled() {
return Err(ScannerError::Other("Operation cancelled".to_string()));
}
@@ -1705,12 +1830,16 @@ impl FolderScanner {
};
let active_object_lock = self.old_cache.info.object_lock.clone();
wait_for_scanner_io(&ctx, self.sleeper.sleep_folder()).await?;
if mode != FolderScanMode::Namespace {
// A candidate may be object payload rather than a namespace.
wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?;
} else {
wait_for_scanner_io(&ctx, self.sleeper.sleep_folder()).await?;
}
let mut existing_folders: Vec<CachedFolder> = Vec::new();
let mut new_folders: Vec<CachedFolder> = Vec::new();
let mut found_object_metadata = false;
let mut erasure_data_directory_candidates: Vec<(CachedFolder, bool, String)> = Vec::new();
let mut object_count: u64 = 0;
let yield_every_objects = scanner_yield_every_n_objects();
@@ -1731,9 +1860,13 @@ impl FolderScanner {
// directory that vanished during iteration instead of failing the
// whole parent subtree. The bucket root (no parent) still propagates
// NotFound so namespace removal is detected by the caller.
let mut dir_reader = match wait_for_scanner_io(&ctx, tokio::fs::read_dir(&dir_path)).await? {
let mut dir_reader = match wait_for_scanner_io(&ctx, ScannerDirectory::open(&dir_path)).await? {
Ok(dir_reader) => Some(dir_reader),
Err(e) if e.kind() == ErrorKind::NotFound && folder.parent.is_some() => {
Err(e)
if folder.parent.is_some()
&& (e.kind() == ErrorKind::NotFound
|| (mode != FolderScanMode::Namespace && e.kind() == ErrorKind::NotADirectory)) =>
{
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
@@ -1744,6 +1877,11 @@ impl FolderScanner {
error = %e,
"Scanner folder state updated"
);
if mode != FolderScanMode::Namespace {
self.coverage_gap |= e.kind() == ErrorKind::NotADirectory && self.old_cache.info.scan_progress.is_some();
done_folder();
return Ok(FolderScanOutcome::Vanished);
}
None
}
Err(e) => return Err(ScannerError::Io(e)),
@@ -1753,67 +1891,111 @@ impl FolderScanner {
let mut raw_enumeration_complete = false;
let raw_enumeration_committed_entries = self.raw_enumeration_committed_entry_oracle(&folder.name);
let mut candidate_unclassified = mode != FolderScanMode::Namespace;
let mut deferred_metadata = None;
loop {
let Some(reader) = dir_reader.as_mut() else {
// Hold at most one metadata entry until a candidate reaches
// EOF: a later part must not leave object accounting or ILM work.
let (entry, replayed_metadata) = if let Some(reader) = dir_reader.as_mut() {
let entry = match next_scanner_dir_entry(&ctx, &self.sleeper, reader).await {
Ok(Some(entry)) => entry,
Ok(None) => {
raw_enumeration_complete = true;
dir_reader = None;
continue;
}
Err(ScannerError::Io(e))
if candidate_unclassified && matches!(e.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) =>
{
return Err(ScannerError::Io(e));
}
Err(ScannerError::Io(e)) if e.kind() == ErrorKind::NotFound => {
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
dir_path = %dir_path,
state = "dir_missing_during_iteration",
error = %e,
"Scanner folder state updated"
);
raw_enumeration_complete = true;
dir_reader = None;
continue;
}
Err(ScannerError::Io(e)) if e.kind() == ErrorKind::NotADirectory => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
dir_path = %dir_path,
state = "dir_became_non_directory",
error = %e,
"Scanner folder state updated"
);
raw_enumeration_complete = true;
dir_reader = None;
continue;
}
Err(e) => return Err(e),
};
(entry, false)
} else {
raw_enumeration_complete = true;
break;
};
let entry = match next_scanner_dir_entry(&ctx, &self.sleeper, reader).await {
Ok(Some(entry)) => entry,
Ok(None) => {
raw_enumeration_complete = true;
break;
if candidate_unclassified {
// Apply namespace reuse only after the current listing
// proves there is no direct erasure part.
if let FolderScanMode::ErasureCandidate { reuse_namespace: true } = mode
&& !into.compacted
&& self.old_cache.find(&folder.name).is_some()
{
// The caller stitches this cached child into its
// parent; do not clone a growing child set into the
// unused provisional accumulator as well.
self.new_cache.copy_with_children(&self.old_cache, &this_hash, &folder.parent);
self.budget.record_entries_visited(pending_entry_progress);
self.finish_raw_enumeration_parent(&folder.name);
done_folder();
return Ok(FolderScanOutcome::Namespace);
}
candidate_unclassified = false;
if !self.budget.try_start_directory() {
return Err(ScannerError::Other("Operation cancelled".to_string()));
}
wait_for_scanner_io(&ctx, self.sleeper.sleep_folder_preset()).await?;
}
Err(ScannerError::Io(e)) if e.kind() == ErrorKind::NotFound => {
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
dir_path = %dir_path,
state = "dir_missing_during_iteration",
error = %e,
"Scanner folder state updated"
);
raw_enumeration_complete = true;
break;
}
Err(ScannerError::Io(e)) if e.kind() == ErrorKind::NotADirectory => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
dir_path = %dir_path,
state = "dir_became_non_directory",
error = %e,
"Scanner folder state updated"
);
raw_enumeration_complete = true;
break;
}
Err(e) => return Err(e),
let Some(entry) = deferred_metadata.take() else { break };
(entry, true)
};
let file_name = entry.file_name().to_string_lossy().to_string();
if file_name.is_empty() || file_name == "." || file_name == ".." {
continue;
}
let raw_entry_consumed_by_owner_index = raw_enumeration_committed_entries.contains(&file_name);
if !raw_entry_consumed_by_owner_index {
#[cfg(test)]
tests::enumeration_restart::observe_raw_entry(&dir_path, &entry.file_name(), &self.budget);
pending_entry_progress = pending_entry_progress.saturating_add(1);
if pending_entry_progress >= SCANNER_ENTRY_PROGRESS_BATCH
|| last_entry_progress.elapsed() >= SCANNER_ENTRY_PROGRESS_INTERVAL
{
self.budget.record_entries_visited(pending_entry_progress);
pending_entry_progress = 0;
last_entry_progress = Instant::now();
if !replayed_metadata {
let raw_entry_consumed_by_owner_index = raw_enumeration_committed_entries.contains(&file_name);
if !raw_entry_consumed_by_owner_index {
#[cfg(test)]
tests::enumeration_restart::observe_raw_entry(&dir_path, &entry.file_name(), &self.budget);
pending_entry_progress = pending_entry_progress.saturating_add(1);
if pending_entry_progress >= SCANNER_ENTRY_PROGRESS_BATCH
|| last_entry_progress.elapsed() >= SCANNER_ENTRY_PROGRESS_INTERVAL
{
self.budget.record_entries_visited(pending_entry_progress);
pending_entry_progress = 0;
last_entry_progress = Instant::now();
}
}
self.record_raw_enumeration_entry(&folder.name, &file_name);
self.maybe_send_checkpoint(&folder, &this_hash, into);
}
self.record_raw_enumeration_entry(&folder.name, &file_name);
self.maybe_send_checkpoint(&folder, &this_hash, into);
let is_storage_format_entry = file_name == STORAGE_FORMAT_FILE;
if candidate_unclassified && is_storage_format_entry {
deferred_metadata = Some(entry);
continue;
}
let file_path = entry.path().to_string_lossy().to_string();
@@ -1858,6 +2040,15 @@ impl FolderScanner {
Err(e) => return Err(ScannerError::Io(e)),
};
if candidate_unclassified && entry_type.is_file() && is_erasure_part_name(&file_name) {
self.budget.record_entries_visited(pending_entry_progress);
// Positive data evidence completes classification, not raw
// namespace coverage. Discard its provisional resume owner.
self.finish_raw_enumeration_parent(&folder.name);
done_folder();
return Ok(FolderScanOutcome::ErasureData);
}
// Metadata presence establishes an erasure object boundary;
// parsing failures still belong to accounting and healing. A
// directory named `xl.meta` remains a valid namespace prefix,
@@ -1950,11 +2141,6 @@ impl FolderScanner {
object_heal_prob_div: folder.object_heal_prob_div,
};
if self.is_erasure_mode && uuid::Uuid::parse_str(&file_name).is_ok_and(|data_dir_id| !data_dir_id.is_nil()) {
erasure_data_directory_candidates.push((this, exists, file_path));
continue;
}
if exists && self.should_reuse_clean_root_child(&folder, into, &this, &h, &abandoned_children) {
abandoned_children.remove(&h.key());
self.reuse_clean_root_child(&this, &h, into);
@@ -2164,73 +2350,6 @@ impl FolderScanner {
self.finish_raw_enumeration_parent(&folder.name);
}
let mut found_erasure_data_directory = false;
if self.is_erasure_mode && !found_object_metadata {
for (_, _, path) in &erasure_data_directory_candidates {
if contains_erasure_part_file(path, &ctx, &self.sleeper).await? {
found_erasure_data_directory = true;
break;
}
}
}
if !found_object_metadata && !found_erasure_data_directory {
for (candidate, exists, _) in erasure_data_directory_candidates {
let h = hash_path(&candidate.name);
if exists && self.should_reuse_clean_root_child(&folder, into, &candidate, &h, &abandoned_children) {
abandoned_children.remove(&h.key());
self.reuse_clean_root_child(&candidate, &h, into);
} else if exists {
abandoned_children.remove(&h.key());
self.update_cache.copy_with_children(&self.old_cache, &h, &candidate.parent);
existing_folders.push(candidate);
} else {
abandoned_children.remove(&h.key());
new_folders.push(candidate);
}
}
}
if self.is_erasure_mode && found_erasure_data_directory && !found_object_metadata {
found_object_metadata = true;
let metadata_path = path_join_buf(&[&dir_path, STORAGE_FORMAT_FILE]);
self.mark_failed_path_seen(&metadata_path);
let retry_suppressed = self.failed_retry_suppressed(&metadata_path);
into.failed_objects = into.failed_objects.saturating_add(1);
if !retry_suppressed {
self.record_failed(&metadata_path);
let failed_cache_entries = self.new_cache.info.failed_objects.len();
if failed_cache_entries > 0 && should_log_failed_object(failed_cache_entries) {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
path = %metadata_path,
failed_objects = failed_cache_entries,
state = "object_metadata_missing",
"Scanner found erasure object data without metadata"
);
}
}
if !retry_suppressed {
let (bucket, object) = path2_bucket_object_with_base_path(&self.root, &folder.name);
if !bucket.is_empty() && !object.is_empty() {
self.send_required_scanner_heal_request(
PendingScannerHealKind::Object,
bucket.clone(),
Some(object.clone()),
None,
build_non_destructive_object_heal_request(bucket, object, self.scan_mode, HealChannelPriority::High),
)
.await?;
}
}
}
if ctx.is_cancelled() {
return Err(ScannerError::Other("Operation cancelled".to_string()));
}
@@ -2320,7 +2439,6 @@ impl FolderScanner {
}));
let has_queued_folders = !queued_folders.is_empty();
let forward_sweep = self.old_cache.info.scan_progress.is_some();
let forward_resume_after = self.resume_frontier.clone();
let resume_order = if forward_sweep {
order_queued_folders_for_resume(&mut queued_folders, None)
} else {
@@ -2342,14 +2460,15 @@ impl FolderScanner {
let mut folder_item = queued_folder.folder;
let h = hash_path(&folder_item.name);
if forward_sweep
&& !into.compacted
&& forward_resume_after
.as_deref()
.is_some_and(|resume| scan_frontier_covers_path(&folder_item.name, resume))
&& self.old_cache.find(&folder_item.name).is_some()
&& !self.resume_subtree_has_failures(&folder_item.name)
{
let reuse_covered = self.can_reuse_covered_namespace(&folder_item.name, into.compacted);
let mut child_mode = if self.is_erasure_mode && is_erasure_candidate(&folder_item.name) {
FolderScanMode::ErasureCandidate {
reuse_namespace: reuse_covered,
}
} else {
FolderScanMode::Namespace
};
if child_mode == FolderScanMode::Namespace && reuse_covered {
self.new_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent);
into.add_child(&h);
continue;
@@ -2399,60 +2518,90 @@ impl FolderScanner {
{
let next_cycle = self.old_cache.info.next_cycle as u32;
if !h.mod_(next_cycle, data_usage_update_dir_cycles()) {
// Transfer and add as child...
self.new_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent);
into.add_child(&h);
self.record_scan_resume_hint(&folder_item.name);
continue;
if child_mode == FolderScanMode::Namespace {
self.new_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent);
into.add_child(&h);
self.record_scan_resume_hint(&folder_item.name);
continue;
}
// Sampling may skip namespace work, never the
// classification that protects its parent object.
child_mode = FolderScanMode::ErasureCandidate { reuse_namespace: true };
} else {
folder_item.object_heal_prob_div = data_usage_update_dir_cycles();
}
folder_item.object_heal_prob_div = data_usage_update_dir_cycles();
}
}
}
(self.update_current_path)(&folder_item.name).await;
let outcome = if into.compacted {
Box::pin(self.scan_folder_inner(ctx.clone(), folder_item.clone(), into, child_mode)).await?
} else {
let mut dst = DataUsageEntry::default();
match self
.scan_child_folder(ctx.clone(), folder_item.clone(), &mut this_hash, into, &mut dst, child_mode)
.await
{
Ok(outcome) => outcome,
Err(e) => {
if ctx.is_cancelled() {
self.preserve_partial_child_progress(&folder_item.parent, &h, into, &dst)
.await;
return Err(e);
}
if child_mode != FolderScanMode::Namespace {
return Err(e);
}
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
folder = %folder.name,
child_folder = %folder_item.name,
state = "child_scan_failed",
error = %e,
"Scanner child folder scan failed"
);
self.coverage_gap |= forward_sweep;
continue;
}
}
};
if outcome == FolderScanOutcome::Vanished {
self.new_cache.delete_recursive(&h);
self.update_cache.delete_recursive(&h);
into.children.remove(&h.key());
continue;
}
if outcome == FolderScanOutcome::ErasureData {
self.new_cache.delete_recursive(&h);
self.update_cache.delete_recursive(&h);
into.children.remove(&h.key());
// Earlier namespace siblings may have completed, but none
// can certify coverage of this unresolved parent object.
self.coverage_gap |= forward_sweep;
if self.coverage_frontier.as_deref().is_some_and(|frontier| {
frontier
.strip_prefix(&folder.name)
.is_some_and(|suffix| suffix.starts_with('/'))
}) {
self.coverage_frontier = None;
}
self.record_missing_erasure_metadata(&folder, into).await?;
break 'scan;
}
if into.compacted {
// In compacted mode child totals are accumulated directly into the parent entry.
let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), into));
fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
self.record_completed_child(&folder_item.name);
self.send_update_for_entry(&this_hash, &folder.parent, into).await;
tokio::task::yield_now().await;
} else {
let mut dst = DataUsageEntry::default();
// Use Box::pin for recursive async call
if let Err(e) = self
.scan_child_folder(ctx.clone(), folder_item.clone(), &mut this_hash, into, &mut dst)
.await
{
if ctx.is_cancelled() {
self.preserve_partial_child_progress(&folder_item.parent, &h, into, &dst)
.await;
return Err(e);
}
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
folder = %folder.name,
child_folder = %folder_item.name,
state = "child_scan_failed",
error = %e,
"Scanner child folder scan failed"
);
self.coverage_gap |= forward_sweep;
continue;
}
tokio::task::yield_now().await;
into.add_child(&h);
self.record_completed_child(&folder_item.name);
self.maybe_send_checkpoint(&folder, &this_hash, into);
// We scanned a folder, optionally send update.
self.update_cache.delete_recursive(&h);
self.update_cache.copy_with_children(&self.new_cache, &h, &folder_item.parent);
self.send_update().await;
@@ -2829,7 +2978,14 @@ impl FolderScanner {
// Use Box::pin for recursive async call
if let Err(e) = self
.scan_child_folder(ctx.clone(), folder_item.clone(), &mut this_hash, into, &mut dst)
.scan_child_folder(
ctx.clone(),
folder_item.clone(),
&mut this_hash,
into,
&mut dst,
FolderScanMode::Namespace,
)
.await
{
if ctx.is_cancelled() {
@@ -2924,7 +3080,7 @@ impl FolderScanner {
let scanned_objects = u64::try_from(into.objects).unwrap_or(u64::MAX);
emit_scanner_folder_trace(&self.root, &folder.name, scanned_objects, trace_started_at, "completed");
Ok(())
Ok(FolderScanOutcome::Namespace)
}
pub fn as_mut_new_cache(&mut self) -> &mut DataUsageCache {
crates/scanner/src/scanner_folder/item_actions.rs
+4 -42
View File
@@ -1300,48 +1300,10 @@ pub(super) fn classify_get_size_failure(item: &ScannerItem, err: &StorageError)
GetSizeFailureAction::RecordFailed
}
pub(super) async fn contains_erasure_part_file(
path: &str,
ctx: &CancellationToken,
sleeper: &DynamicSleeper,
) -> Result<bool, ScannerError> {
wait_for_scanner_io(ctx, sleeper.admit_disk_io(1, 0)).await?;
let mut entries = match wait_for_scanner_io(ctx, tokio::fs::read_dir(path)).await? {
Ok(entries) => entries,
Err(err) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) => return Ok(false),
Err(err) => return Err(ScannerError::Io(err)),
};
for _ in 0..ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT {
let entry = match next_scanner_dir_entry(ctx, sleeper, &mut entries).await {
Ok(Some(entry)) => entry,
Ok(None) => return Ok(false),
Err(ScannerError::Io(err)) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) => {
return Ok(false);
}
Err(err) => return Err(err),
};
let file_name = entry.file_name();
let Some(part_number) = file_name
.to_str()
.and_then(|name| name.strip_prefix("part."))
.and_then(|number| number.parse::<u32>().ok())
else {
continue;
};
if part_number == 0 {
continue;
}
match wait_for_scanner_io(ctx, entry.file_type()).await? {
Ok(file_type) if file_type.is_file() => return Ok(true),
Ok(_) => {}
Err(err) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::TooManyLinks) => {}
Err(err) => return Err(ScannerError::Io(err)),
}
}
Ok(false)
pub(super) fn is_erasure_part_name(name: &str) -> bool {
name.strip_prefix("part.")
.and_then(|number| number.parse::<u32>().ok())
.is_some_and(|number| number > 0)
}
#[cfg(test)]
crates/scanner/src/scanner_folder/tests.rs
+34 -123
View File
@@ -16,6 +16,7 @@ use crate::SCANNER_SLEEPER;
use super::*;
pub(super) mod erasure_probe;
mod mrf_ownership;
use crate::storage_api::VersionPurgeStatusType;
use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
@@ -151,6 +152,29 @@ struct CapturedLogWriter {
}
impl CapturedLogs {
fn capture<F: std::future::Future>(&self, level: tracing::Level, future: F) -> impl std::future::Future<Output = F::Output> {
use tracing::instrument::WithSubscriber as _;
let subscriber = tracing_subscriber::fmt()
.json()
.with_max_level(level)
.with_writer(self.clone())
.with_ansi(false)
.without_time()
.finish();
// Re-enter the dispatcher on every poll instead of leaving a
// thread-local default installed while this future is suspended.
let captured = future.with_subscriber(subscriber);
// Match ecstore's test_tracing guard: with a single dispatcher,
// first registration on an unsubscribed sibling can cache Never.
// Construct both dispatchers before polling and retain the inert one.
let interest_guard = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
async move {
let _interest_guard = interest_guard;
captured.await
}
}
fn contents(&self) -> String {
let buffer = self
.buffer
@@ -587,6 +611,7 @@ async fn checkpoint_ancestor_depth_limit_preserves_suspended_accounting() {
&mut hash,
&mut parent,
&mut child,
FolderScanMode::Namespace,
)
.await
.expect_err("unsupported depth must stop before recursion");
@@ -2448,113 +2473,6 @@ async fn scan_folder_reuses_symlink_size_without_bypassing_disk_validation() {
);
}
#[tokio::test(start_paused = true)]
async fn scanner_directory_probe_budget_wait_cancels_without_rate_delay() {
let path = std::env::temp_dir().join(format!("scanner-cancel-budget-{}", Uuid::new_v4()));
let sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests(
&path,
crate::sleeper::ScannerDiskIoBudgetLimits {
iops: Some(0),
bytes_per_sec: None,
},
);
sleeper.admit_disk_io(1, 0).await;
let ctx = CancellationToken::new();
let cancel = ctx.clone();
let task = tokio::spawn(async move {
tokio::task::yield_now().await;
cancel.cancel();
});
let started = tokio::time::Instant::now();
assert!(
contains_erasure_part_file(path.to_string_lossy().as_ref(), &ctx, &sleeper)
.await
.is_err()
);
task.await.expect("cancellation task");
assert_eq!(
started.elapsed(),
Duration::ZERO,
"cancellation must interrupt the one-second budget wait"
);
assert!(wait_for_scanner_io(&ctx, sleeper.admit_disk_io(1, 0)).await.is_err());
let status = crate::sleeper::scanner_disk_io_budget_status()
.into_iter()
.find(|status| status.disk == path.to_string_lossy())
.expect("bound disk budget");
assert_eq!(status.charged_ops, 2, "an already cancelled wait must not charge another operation");
}
#[tokio::test]
async fn erasure_directory_probe_respects_entry_limit_and_disk_budget() {
let temp_dir = tempfile::tempdir().expect("create probe directory");
for index in 0..=ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT {
tokio::fs::write(temp_dir.path().join(format!("namespace-{index}")), b"")
.await
.expect("create non-part entry");
}
let sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests(
temp_dir.path(),
crate::sleeper::ScannerDiskIoBudgetLimits {
iops: Some(1_000_000),
bytes_per_sec: None,
},
);
assert!(
!contains_erasure_part_file(temp_dir.path().to_string_lossy().as_ref(), &CancellationToken::new(), &sleeper)
.await
.expect("probe directory without erasure parts")
);
let status = crate::sleeper::scanner_disk_io_budget_status()
.into_iter()
.find(|status| status.disk == temp_dir.path().to_string_lossy())
.expect("bound disk budget");
assert_eq!(
status.charged_ops,
1 + u64::try_from(ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT).expect("entry limit fits u64"),
"one directory open plus at most the bounded probe entries"
);
assert_eq!(status.charged_bytes, 0);
}
#[tokio::test(start_paused = true)]
async fn scanner_directory_probe_cancels_during_entry_budget_wait() {
let temp_dir = tempfile::tempdir().expect("create probe directory");
tokio::fs::write(temp_dir.path().join("part.1"), b"shard")
.await
.expect("create erasure part");
let sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests(
temp_dir.path(),
crate::sleeper::ScannerDiskIoBudgetLimits {
iops: Some(0),
bytes_per_sec: None,
},
);
let ctx = CancellationToken::new();
let cancel = ctx.clone();
let disk = temp_dir.path().to_string_lossy().into_owned();
let task = tokio::spawn(async move {
loop {
if crate::sleeper::scanner_disk_io_budget_status()
.iter()
.any(|status| status.disk == disk && status.charged_ops == 2)
{
cancel.cancel();
return;
}
tokio::task::yield_now().await;
}
});
let started = tokio::time::Instant::now();
let result = contains_erasure_part_file(temp_dir.path().to_string_lossy().as_ref(), &ctx, &sleeper).await;
task.abort();
assert!(ctx.is_cancelled(), "probe must reach the entry admission after opening its directory");
task.await.expect("entry-wait cancellation task");
assert!(result.is_err(), "cancelled probe must not report an erasure part");
assert_eq!(started.elapsed(), Duration::ZERO, "entry admission must not wait out its one-second debt");
}
#[tokio::test]
#[serial]
async fn test_scan_folder_missing_bucket_root_still_reports_not_found() {
@@ -2745,15 +2663,11 @@ async fn test_scan_folder_xl_meta_named_directory_uses_namespace_descent() {
#[serial]
async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
let logs = CapturedLogs::default();
let subscriber = tracing_subscriber::fmt()
.json()
.with_max_level(tracing::Level::ERROR)
.with_writer(logs.clone())
.with_ansi(false)
.without_time()
.finish();
let _subscriber_guard = tracing::subscriber::set_default(subscriber);
logs.capture(tracing::Level::ERROR, scan_corrupt_metadata_with_captured_logs(&logs))
.await;
}
async fn scan_corrupt_metadata_with_captured_logs(logs: &CapturedLogs) {
let (mut scanner, temp_dir) = build_test_scanner().await;
// Canonicalize for the "drive" field comparison (scanner resolves symlinks).
let canonical_temp_dir = std::fs::canonicalize(&temp_dir).unwrap_or_else(|_| temp_dir.clone());
@@ -4215,14 +4129,11 @@ async fn test_scan_data_folder_success_clears_resume_hint() {
#[serial]
async fn test_scan_data_folder_keeps_unresolved_objects_partial() {
let logs = CapturedLogs::default();
let subscriber = tracing_subscriber::fmt()
.json()
.with_max_level(tracing::Level::DEBUG)
.with_writer(logs.clone())
.with_ansi(false)
.without_time()
.finish();
let _subscriber_guard = tracing::subscriber::set_default(subscriber);
logs.capture(tracing::Level::DEBUG, scan_unresolved_objects_with_captured_logs(&logs))
.await;
}
async fn scan_unresolved_objects_with_captured_logs(logs: &CapturedLogs) {
let (scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard {
temp_dir: Some(temp_dir.clone()),
crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs
+84 -8
View File
@@ -900,14 +900,11 @@ async fn checkpoint_fixture_hot_bucket_certifies_after_mixed_end_and_leader_hand
async fn hot_bucket_certifies_first_full_sweep_after_mixed_end(advance_epoch: bool) {
let logs = CapturedLogs::default();
let subscriber = tracing_subscriber::fmt()
.json()
.with_max_level(tracing::Level::DEBUG)
.with_writer(logs.clone())
.with_ansi(false)
.without_time()
.finish();
let _subscriber_guard = tracing::subscriber::set_default(subscriber);
logs.capture(tracing::Level::DEBUG, run_hot_bucket_certification(advance_epoch, &logs))
.await;
}
async fn run_hot_bucket_certification(advance_epoch: bool, logs: &CapturedLogs) {
let (scanner, root) = build_test_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
@@ -1056,6 +1053,49 @@ async fn hot_bucket_certifies_first_full_sweep_after_mixed_end(advance_epoch: bo
assert_eq!(fields["size_reconciliation_entries"], 0);
}
#[tokio::test(flavor = "current_thread")]
#[serial]
async fn checkpoint_admission_log_capture_survives_another_future_dispatcher() {
let logs = CapturedLogs::default();
let (mut cache, identity) = bound_checkpoint();
let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
let (done_tx, done_rx) = tokio::sync::oneshot::channel();
let captured = logs.capture(tracing::Level::DEBUG, async move {
entered_tx.send(()).expect("announce captured future");
ready_rx.await.expect("competing dispatcher is installed");
for cycle in 100..104 {
cache.prepare_bucket_checkpoint("bucket", cycle, 9, SOURCE, PLAN, identity);
assert_eq!(cache.info.next_cycle, cycle);
}
done_tx.send(()).expect("announce checkpoint completion");
});
let competing = async {
entered_rx.await.expect("captured future has yielded");
let subscriber = tracing_subscriber::fmt()
.with_max_level(tracing::Level::ERROR)
.with_writer(std::io::sink)
.finish();
// An unrelated future keeps a quieter thread-local default while
// the captured future resumes on this same runtime thread.
let _guard = tracing::subscriber::set_default(subscriber);
ready_tx.send(()).expect("announce competing dispatcher");
done_rx.await.expect("checkpoint preparation has completed");
};
tokio::join!(captured, competing);
let events = logs
.contents()
.lines()
.map(|line| serde_json::from_str::<serde_json::Value>(line).expect("scanner JSON event"))
.filter(|event| event["fields"]["state"] == "bucket_checkpoint_prepared")
.collect::<Vec<_>>();
assert_eq!(events.len(), 4, "all checkpoint admissions must use the captured dispatcher");
for (cycle, event) in (100..104).zip(events) {
assert_eq!(event["fields"]["cycle"], cycle);
}
}
async fn write_checkpoint_object(root: &std::path::Path, object: &str, versions: &[(Option<Uuid>, i64)]) {
let mut metadata = FileMeta::new();
for (index, (version_id, size)) in versions.iter().enumerate() {
@@ -1407,3 +1447,39 @@ where
}
panic!("finite stable fixture must converge using the same four-object budget without an unbounded final sweep");
}
fn emit_first_registration_capture_event() {
tracing::debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
state = "first_registration_capture",
"Scanner test capture exercised first callsite registration"
);
}
#[tokio::test(flavor = "current_thread")]
#[serial]
async fn checkpoint_log_capture_survives_first_registration_on_unsubscribed_thread() {
let logs = CapturedLogs::default();
// Construct the capture before polling it: its subscriber must be live
// when the other thread first registers this otherwise unused callsite.
let captured = logs.capture(tracing::Level::DEBUG, async {
emit_first_registration_capture_event();
});
std::thread::spawn(emit_first_registration_capture_event)
.join()
.expect("unsubscribed first-registration thread");
captured.await;
let events = logs
.contents()
.lines()
.map(|line| serde_json::from_str::<serde_json::Value>(line).expect("scanner JSON event"))
.filter(|event| event["fields"]["state"] == "first_registration_capture")
.count();
assert_eq!(
events, 1,
"first registration without a subscriber must not disable the captured callsite"
);
}
crates/scanner/src/scanner_folder/tests/checkpoint_fixture/periodic_snapshot.rs
+5 -8
View File
@@ -999,14 +999,11 @@ async fn periodic_resume_uses_traversal_order_for_punctuation_siblings() {
#[serial]
async fn periodic_failed_nested_entries_cannot_certify_a_bucket_with_diagnostics_disabled() {
let logs = CapturedLogs::default();
let subscriber = tracing_subscriber::fmt()
.json()
.with_max_level(tracing::Level::DEBUG)
.with_writer(logs.clone())
.with_ansi(false)
.without_time()
.finish();
let _subscriber_guard = tracing::subscriber::set_default(subscriber);
logs.capture(tracing::Level::DEBUG, periodic_failed_entries_with_captured_logs(&logs))
.await;
}
async fn periodic_failed_entries_with_captured_logs(logs: &CapturedLogs) {
let (loaded, _, original, _guard) = nested_periodic_snapshot(true, (0, 0)).await;
let result = temp_env::async_with_vars([(ENV_FAILED_OBJECT_TTL_SECS, Some("0"))], async {
let mut cache = loaded;
crates/scanner/src/scanner_folder/tests/erasure_probe.rs
+827
View File
@@ -0,0 +1,827 @@
// Copyright 2026 RustFS Team
// Licensed under the Apache License, Version 2.0.
use super::*;
use std::collections::{HashMap, VecDeque};
use std::sync::LazyLock;
#[derive(Clone, Copy)]
enum DirectoryOrder {
PartLast,
FailAfter(usize, std::io::ErrorKind),
FailOpen(std::io::ErrorKind),
}
pub(in crate::scanner_folder) type OrderedEntries = VecDeque<std::io::Result<tokio::fs::DirEntry>>;
static ORDERED_DIRECTORIES: LazyLock<Mutex<HashMap<String, DirectoryOrder>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
struct OrderedDirectory(String);
impl OrderedDirectory {
fn new(path: &std::path::Path) -> Self {
let path = path.to_string_lossy().into_owned();
ORDERED_DIRECTORIES
.lock()
.expect("ordered directory registry")
.insert(path.clone(), DirectoryOrder::PartLast);
Self(path)
}
fn failing(path: &std::path::Path, entries: usize, kind: std::io::ErrorKind) -> Self {
let guard = Self::new(path);
ORDERED_DIRECTORIES
.lock()
.expect("ordered directory registry")
.insert(guard.0.clone(), DirectoryOrder::FailAfter(entries, kind));
guard
}
fn fail_open(path: &std::path::Path, kind: std::io::ErrorKind) -> Self {
let guard = Self::new(path);
ORDERED_DIRECTORIES
.lock()
.expect("ordered directory registry")
.insert(guard.0.clone(), DirectoryOrder::FailOpen(kind));
guard
}
}
impl Drop for OrderedDirectory {
fn drop(&mut self) {
ORDERED_DIRECTORIES.lock().expect("ordered directory cleanup").remove(&self.0);
}
}
pub(in crate::scanner_folder) async fn ordered_entries(
path: &str,
reader: &mut tokio::fs::ReadDir,
) -> std::io::Result<Option<OrderedEntries>> {
let order = ORDERED_DIRECTORIES
.lock()
.expect("ordered directory registry")
.get(path)
.copied();
let Some(order) = order else { return Ok(None) };
if let DirectoryOrder::FailOpen(kind) = order {
return Err(std::io::Error::from(kind));
}
let mut entries = Vec::new();
while let Some(entry) = reader.next_entry().await? {
entries.push(entry);
}
// Real entries and file types, with an explicit adversarial order rather
// than an assumption about the filesystem's enumeration order.
entries.sort_by_key(|entry| (entry.file_name().to_str() == Some("part.1"), entry.file_name()));
let mut entries: OrderedEntries = entries.into_iter().map(Ok).collect();
if let DirectoryOrder::FailAfter(count, kind) = order {
entries.truncate(count);
entries.push_back(Err(std::io::Error::from(kind)));
}
Ok(Some(entries))
}
#[tokio::test]
#[serial]
async fn erasure_candidate_finds_late_part_without_scanning_data_children() {
let (mut scanner, root) = build_test_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
let data_dir = root.join("bucket/object").join(Uuid::new_v4().to_string());
tokio::fs::create_dir_all(&data_dir).await.expect("create candidate");
for index in 0..65 {
tokio::fs::write(data_dir.join(format!("part.{index}.integrity.orphan")), b"")
.await
.expect("write sidecar");
}
tokio::fs::write(data_dir.join("part.1"), b"shard")
.await
.expect("write late part");
tokio::fs::create_dir(data_dir.join("namespace-child"))
.await
.expect("stage a child before the part");
let _order = OrderedDirectory::new(&data_dir);
scanner.is_erasure_mode = true;
scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests(
&root,
crate::sleeper::ScannerDiskIoBudgetLimits {
iops: Some(1_000_000),
bytes_per_sec: None,
},
);
let parent = CancellationToken::new();
let budget = ScannerCycleBudget::new_with_progress_tracking(
&parent,
crate::scanner_budget::ScannerCycleBudgetConfig {
max_directories: Some(1),
..Default::default()
},
);
scanner.budget = budget.clone();
let mut into = DataUsageEntry::default();
scanner
.scan_folder(
budget.token(),
CachedFolder {
name: "bucket/object".into(),
parent: None,
object_heal_prob_div: 1,
},
&mut into,
)
.await
.expect("late part must establish the parent object boundary");
assert_eq!(into.failed_objects, 1);
assert_eq!(into.objects, 0);
assert!(
scanner
.new_cache
.info
.failed_objects
.contains_key(root.join("bucket/object/xl.meta").to_string_lossy().as_ref())
);
assert_eq!(budget.reason(), None, "payload directories must not consume namespace directory quota");
assert!(scanner.new_cache.info.pending_heals.iter().any(|pending| {
pending.kind == PendingScannerHealKind::Object
&& pending.bucket == "bucket"
&& pending.object.as_deref() == Some("object")
}));
}
async fn candidate_scanner() -> (FolderScanner, std::path::PathBuf, CachedFolder) {
let (mut scanner, root) = build_test_scanner().await;
for info in [
&mut scanner.old_cache.info,
&mut scanner.new_cache.info,
&mut scanner.update_cache.info,
] {
info.name = "bucket".into();
}
scanner.is_erasure_mode = true;
let name = format!("bucket/{}", Uuid::new_v4());
tokio::fs::create_dir_all(root.join(&name))
.await
.expect("create UUID namespace");
let folder = CachedFolder {
name,
parent: Some(hash_path("bucket")),
object_heal_prob_div: 1,
};
(scanner, root, folder)
}
fn bind_budget(scanner: &mut FolderScanner, root: &std::path::Path, iops: u64) {
scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests(
root,
crate::sleeper::ScannerDiskIoBudgetLimits {
iops: Some(iops),
bytes_per_sec: None,
},
);
}
fn charged_ops(root: &std::path::Path) -> u64 {
let status = crate::sleeper::scanner_disk_io_budget_status()
.into_iter()
.find(|status| status.disk == root.to_string_lossy())
.expect("bound candidate budget");
assert_eq!(status.charged_bytes, 0);
status.charged_ops
}
#[tokio::test]
#[serial]
async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
for entry_count in [0, 1, 63, 64, 65, 129] {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
for index in 0..entry_count {
tokio::fs::write(root.join(&folder.name).join(format!("namespace-{index}")), b"")
.await
.expect("create non-part entry");
}
bind_budget(&mut scanner, &root, 1_000_000);
let ctx = CancellationToken::new();
scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
let outcome = scanner
.scan_folder_inner(
ctx,
folder,
&mut DataUsageEntry::default(),
FolderScanMode::ErasureCandidate { reuse_namespace: false },
)
.await
.expect("classify full UUID namespace");
assert_eq!(outcome, FolderScanOutcome::Namespace);
assert_eq!(charged_ops(&root), 1 + entry_count, "one open and each entry once, never EOF");
assert_eq!(scanner.budget.entries_visited(), entry_count);
}
}
#[tokio::test]
#[serial]
async fn erasure_namespace_uses_one_listing_and_no_extra_probe_charge() {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
for index in 0..129 {
tokio::fs::write(root.join(&folder.name).join(format!("namespace-{index}")), b"")
.await
.expect("write namespace entry");
}
bind_budget(&mut scanner, &root, 1_000_000);
let mut into = DataUsageEntry::default();
scanner
.scan_folder(
CancellationToken::new(),
CachedFolder {
name: "bucket".into(),
parent: None,
object_heal_prob_div: 1,
},
&mut into,
)
.await
.expect("scan namespace through parent");
assert_eq!(
charged_ops(&root),
132,
"parent open/entry + candidate open/129 entries, no second listing"
);
assert_eq!(into.failed_objects, 0);
assert!(scanner.new_cache.info.failed_objects.is_empty());
}
#[tokio::test(start_paused = true)]
#[serial]
async fn scanner_directory_probe_budget_wait_cancels_without_rate_delay() {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
bind_budget(&mut scanner, &root, 0);
scanner.sleeper.admit_disk_io(1, 0).await;
let ctx = CancellationToken::new();
let cancel = ctx.clone();
let task = tokio::spawn(async move {
tokio::task::yield_now().await;
cancel.cancel();
});
let started = tokio::time::Instant::now();
let result = scanner
.scan_folder_inner(
ctx.clone(),
folder,
&mut DataUsageEntry::default(),
FolderScanMode::ErasureCandidate { reuse_namespace: false },
)
.await;
task.await.expect("open-wait cancellation");
assert!(result.is_err());
assert_eq!(started.elapsed(), Duration::ZERO);
assert!(wait_for_scanner_io(&ctx, scanner.sleeper.admit_disk_io(1, 0)).await.is_err());
assert_eq!(charged_ops(&root), 2, "already cancelled operations must not charge");
}
#[tokio::test(start_paused = true)]
#[serial]
async fn scanner_directory_probe_cancels_during_entry_budget_wait() {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
tokio::fs::write(root.join(&folder.name).join("part.1"), b"shard")
.await
.expect("write part");
bind_budget(&mut scanner, &root, 0);
let ctx = CancellationToken::new();
let cancel = ctx.clone();
let disk = root.clone();
let task = tokio::spawn(async move {
loop {
if charged_ops(&disk) == 2 {
cancel.cancel();
return;
}
tokio::task::yield_now().await;
}
});
let started = tokio::time::Instant::now();
let result = scanner
.scan_folder_inner(
ctx.clone(),
folder,
&mut DataUsageEntry::default(),
FolderScanMode::ErasureCandidate { reuse_namespace: false },
)
.await;
assert!(ctx.is_cancelled());
task.await.expect("entry-wait cancellation");
assert!(result.is_err(), "cancelled classification cannot become found or EOF");
assert_eq!(started.elapsed(), Duration::ZERO);
assert_eq!(charged_ops(&root), 2);
}
#[test]
fn erasure_part_names_require_a_positive_u32() {
for name in ["part.1", "part.10000", "part.4294967295"] {
assert!(is_erasure_part_name(name));
}
for name in [
"part.0",
"part.",
"part.-1",
"part.4294967296",
"part.1.integrity.orphan",
"Part.1",
] {
assert!(!is_erasure_part_name(name), "{name}");
}
}
fn enable_checkpoint(scanner: &mut FolderScanner) {
let plan = crate::DataUsageScanPlanDigest([11; 32]);
let identity = crate::DataUsageScanIdentity {
version: 1,
bucket_incarnation: Uuid::from_u128(7),
set_layout: crate::DataUsageScanPlanDigest([12; 32]),
publication_epoch: 3,
tier_registry_generation: 0,
scan_mode: HealScanMode::Normal,
};
let mut cache = DataUsageCache::default();
cache.prepare_bucket_checkpoint("bucket", 7, 3, crate::DataUsageCacheSource::new(1, 0), plan, identity);
cache.replace("bucket", "", DataUsageEntry::default());
scanner.old_cache = cache.clone();
scanner.new_cache = cache.clone();
scanner.update_cache = cache;
}
#[tokio::test]
#[serial]
async fn erasure_candidate_defers_metadata_until_part_classification_finishes() {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
let path = root.join(&folder.name);
for index in 0..65 {
tokio::fs::write(path.join(format!("part.{index}.integrity.orphan")), b"")
.await
.expect("write sidecar");
}
let object = folder.name.strip_prefix("bucket/").expect("candidate in bucket");
tokio::fs::write(path.join("xl.meta"), metadata_for_object("bucket", object))
.await
.expect("write candidate metadata");
tokio::fs::write(path.join("part.1"), b"shard")
.await
.expect("write late part");
let _order = OrderedDirectory::new(&path);
bind_budget(&mut scanner, &root, 1_000_000);
let ctx = CancellationToken::new();
scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
let mut into = DataUsageEntry::default();
let outcome = scanner
.scan_folder_inner(ctx, folder, &mut into, FolderScanMode::ErasureCandidate { reuse_namespace: false })
.await
.expect("classify candidate");
assert_eq!(outcome, FolderScanOutcome::ErasureData);
assert_eq!(into.objects, 0, "candidate xl.meta cannot start accounting or maintenance before EOF");
assert_eq!(scanner.budget.progress(), (0, 0));
assert_eq!(scanner.budget.entries_visited(), 67);
assert_eq!(charged_ops(&root), 68, "metadata entry only: no metadata stat or read reservation");
}
#[tokio::test]
#[serial]
async fn erasure_candidate_cancelled_batch_keeps_partial_checkpoint_and_reclassifies_on_resume() {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
let path = root.join(&folder.name);
for index in 0..65 {
tokio::fs::write(path.join(format!("part.{index}.integrity.orphan")), b"")
.await
.expect("write sidecar");
}
tokio::fs::write(path.join("part.1"), b"shard")
.await
.expect("write late part");
let _order = OrderedDirectory::new(&path);
enable_checkpoint(&mut scanner);
bind_budget(&mut scanner, &root, 1_000_000);
let ctx = CancellationToken::new();
let budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
scanner.budget = budget.clone();
let cancel_budget = budget.clone();
let cancel = tokio::spawn(async move {
loop {
if cancel_budget.entries_visited() >= 65 {
cancel_budget.cancel_for_runtime();
return;
}
tokio::task::yield_now().await;
}
});
let parent_folder = CachedFolder {
name: "bucket".into(),
parent: None,
object_heal_prob_div: 1,
};
let mut into = DataUsageEntry::default();
let result = scanner.scan_folder(budget.token(), parent_folder.clone(), &mut into).await;
cancel.await.expect("batch cancellation task");
assert!(result.is_err());
assert_eq!(
budget.entries_visited(),
65,
"parent entry and first full candidate batch are semantic progress"
);
assert_eq!(charged_ops(&root), 67, "two opens and 65 entries, cancellation before the next batch");
assert!(
scanner.new_cache.info.failed_objects.is_empty(),
"unfinished classification cannot manufacture missing metadata"
);
let snapshot = scanner
.checkpoint_snapshot(&parent_folder, &hash_path("bucket"), &into)
.expect("partial raw checkpoint");
assert!(!snapshot.info.snapshot_complete);
assert!(snapshot.info.scan_coverage_receipt.is_none());
let cursor = snapshot
.validated_raw_enumeration_cursor()
.expect("validated candidate cursor");
assert_eq!(cursor.parent, folder.name);
assert_eq!(cursor.entries_seen, 64);
let index = snapshot
.validated_raw_enumeration_page_index()
.expect("validated candidate index");
assert_eq!(index.indexed_entries().expect("index entries").len(), 64);
let bytes = serde_json::to_vec(&snapshot).expect("serialize partial checkpoint");
let reloaded: DataUsageCache = serde_json::from_slice(&bytes).expect("reload partial checkpoint");
scanner.old_cache = reloaded.clone();
scanner.new_cache = reloaded.clone();
scanner.update_cache = reloaded;
scanner.raw_enumeration_progress.clear();
let next_budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
scanner.budget = next_budget.clone();
scanner
.scan_folder(next_budget.token(), parent_folder, &mut DataUsageEntry::default())
.await
.expect("revalidate resumed candidate");
assert!(
scanner
.new_cache
.info
.failed_objects
.contains_key(root.join("bucket/xl.meta").to_string_lossy().as_ref())
);
assert_eq!(
charged_ops(&root),
136,
"resume recharges actual reads; the raw index is not a filesystem seek"
);
}
#[tokio::test]
#[serial]
async fn erasure_late_data_sibling_prevents_parent_coverage_certification() {
let (mut scanner, root) = build_test_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
enable_checkpoint(&mut scanner);
scanner.is_erasure_mode = true;
bind_budget(&mut scanner, &root, 1_000_000);
let first = format!("object/{}", Uuid::from_u128(1));
write_test_object_metadata(&root, "bucket", &first).await;
let data = root.join("bucket/object").join(Uuid::from_u128(2).to_string());
tokio::fs::create_dir_all(&data).await.expect("create later data sibling");
tokio::fs::write(data.join("part.1"), b"shard")
.await
.expect("write later part");
let mut into = DataUsageEntry::default();
scanner
.scan_folder(
CancellationToken::new(),
CachedFolder {
name: "bucket/object".into(),
parent: Some(hash_path("bucket")),
object_heal_prob_div: 1,
},
&mut into,
)
.await
.expect("scan mixed siblings");
assert_eq!(
scanner
.new_cache
.find(&format!("bucket/{first}"))
.expect("preceding UUID object retained")
.objects,
1,
"the preceding legitimate UUID object was scanned once"
);
assert_eq!(into.failed_objects, 1);
assert!(scanner.coverage_gap);
assert!(
scanner.coverage_frontier.is_none(),
"completed siblings cannot certify unresolved parent coverage"
);
assert!(
scanner
.new_cache
.info
.failed_objects
.contains_key(root.join("bucket/object/xl.meta").to_string_lossy().as_ref())
);
}
#[tokio::test]
#[serial]
async fn erasure_candidate_rechecks_data_below_a_restored_namespace_frontier() {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
enable_checkpoint(&mut scanner);
tokio::fs::write(root.join(&folder.name).join("part.1"), b"shard")
.await
.expect("write part");
let hash = hash_path(&folder.name);
scanner.old_cache.replace_hashed(
&hash,
&Some(hash_path("bucket")),
&DataUsageEntry {
objects: 1,
compacted: true,
..Default::default()
},
);
scanner.resume_frontier = Some(folder.name.clone());
bind_budget(&mut scanner, &root, 1_000_000);
let mut into = DataUsageEntry::default();
scanner
.scan_folder(
CancellationToken::new(),
CachedFolder {
name: "bucket".into(),
parent: None,
object_heal_prob_div: 1,
},
&mut into,
)
.await
.expect("reclassify covered UUID directory");
assert_eq!(into.failed_objects, 1);
assert_eq!(into.objects, 0, "stale namespace totals cannot bypass current part evidence");
assert!(scanner.new_cache.find(&folder.name).is_none());
}
#[cfg(unix)]
#[tokio::test]
#[serial]
async fn erasure_candidate_part_symlink_is_not_data_evidence() {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
let target = root.join("target");
tokio::fs::write(&target, b"shard").await.expect("write symlink target");
symlink(&target, root.join(&folder.name).join("part.1")).expect("create part symlink");
bind_budget(&mut scanner, &root, 1_000_000);
let outcome = scanner
.scan_folder_inner(
CancellationToken::new(),
folder,
&mut DataUsageEntry::default(),
FolderScanMode::ErasureCandidate { reuse_namespace: false },
)
.await
.expect("classify symlink namespace");
assert_eq!(outcome, FolderScanOutcome::Namespace);
assert_eq!(charged_ops(&root), 3, "open, entry, symlink target stat each once");
assert!(scanner.new_cache.info.failed_objects.is_empty());
}
#[tokio::test]
#[serial]
async fn erasure_candidate_covered_namespace_reuses_totals_after_classification() {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
let object = folder.name.strip_prefix("bucket/").expect("candidate in bucket");
write_test_object_metadata(&root, "bucket", object).await;
enable_checkpoint(&mut scanner);
scanner.old_cache.replace_hashed(
&hash_path(&folder.name),
&folder.parent,
&DataUsageEntry {
objects: 1,
..Default::default()
},
);
scanner.resume_frontier = Some(folder.name.clone());
bind_budget(&mut scanner, &root, 1_000_000);
let ctx = CancellationToken::new();
let budget = ScannerCycleBudget::new_with_progress_tracking(
&ctx,
crate::scanner_budget::ScannerCycleBudgetConfig {
max_objects: Some(1),
..Default::default()
},
);
scanner.budget = budget.clone();
scanner
.scan_folder(
budget.token(),
CachedFolder {
name: "bucket".into(),
parent: None,
object_heal_prob_div: 1,
},
&mut DataUsageEntry::default(),
)
.await
.expect("covered candidate must not exhaust object quota again");
assert_eq!(budget.reason(), None);
assert_eq!(budget.progress(), (0, 1), "only the parent consumes namespace quota");
assert_eq!(charged_ops(&root), 4, "list both directories once without re-reading covered metadata");
assert_eq!(scanner.new_cache.find(&folder.name).expect("reused candidate").objects, 1);
}
#[tokio::test]
#[serial]
async fn erasure_candidate_iteration_error_cannot_replay_deferred_metadata_as_eof() {
for kind in [std::io::ErrorKind::NotFound, std::io::ErrorKind::NotADirectory] {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
let path = root.join(&folder.name);
let object = folder.name.strip_prefix("bucket/").expect("candidate in bucket");
write_test_object_metadata(&root, "bucket", object).await;
tokio::fs::write(path.join("part.1"), b"shard")
.await
.expect("write unread part");
let _error = OrderedDirectory::failing(&path, 1, kind);
bind_budget(&mut scanner, &root, 1_000_000);
let result = scanner
.scan_folder(
CancellationToken::new(),
CachedFolder {
name: "bucket".into(),
parent: None,
object_heal_prob_div: 1,
},
&mut DataUsageEntry::default(),
)
.await;
assert!(
matches!(result, Err(ScannerError::Io(err)) if err.kind() == kind),
"candidate error must remain typed and cannot become namespace completion"
);
assert_eq!(
charged_ops(&root),
4,
"two opens, parent entry, deferred metadata entry; no metadata read"
);
assert!(
scanner.new_cache.info.failed_objects.is_empty(),
"unknown classification is not missing-metadata evidence"
);
assert!(scanner.new_cache.find(&folder.name).is_none());
}
}
#[tokio::test]
#[serial]
async fn erasure_candidate_preserves_compacted_namespace_sampling_after_classification() {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
let object = folder.name.strip_prefix("bucket/").expect("candidate in bucket");
write_test_object_metadata(&root, "bucket", object).await;
let hash = hash_path(&folder.name);
scanner.old_cache.replace_hashed(
&hash,
&folder.parent,
&DataUsageEntry {
objects: 1,
compacted: true,
..Default::default()
},
);
let skipped_cycle = (0..data_usage_update_dir_cycles())
.find(|cycle| !hash.mod_(*cycle, data_usage_update_dir_cycles()))
.expect("at least one cycle samples another namespace");
scanner.old_cache.info.next_cycle = u64::from(skipped_cycle);
bind_budget(&mut scanner, &root, 1_000_000);
let ctx = CancellationToken::new();
let budget = ScannerCycleBudget::new_with_progress_tracking(
&ctx,
crate::scanner_budget::ScannerCycleBudgetConfig {
max_objects: Some(1),
..Default::default()
},
);
scanner.budget = budget.clone();
scanner
.scan_folder(
budget.token(),
CachedFolder {
name: "bucket".into(),
parent: None,
object_heal_prob_div: 1,
},
&mut DataUsageEntry::default(),
)
.await
.expect("sampled candidate must not consume object quota");
assert_eq!(budget.reason(), None);
assert_eq!(budget.progress(), (0, 1));
assert_eq!(
charged_ops(&root),
4,
"sampling still requires full current classification, without metadata re-read"
);
assert_eq!(scanner.new_cache.find(&folder.name).expect("sampled candidate").objects, 1);
}
#[tokio::test]
#[serial]
async fn erasure_candidate_vanished_before_open_preserves_other_namespace_progress() {
for kind in [std::io::ErrorKind::NotFound, std::io::ErrorKind::NotADirectory] {
let (mut scanner, root, folder) = candidate_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
enable_checkpoint(&mut scanner);
scanner.old_cache.replace_hashed(
&hash_path(&folder.name),
&folder.parent,
&DataUsageEntry {
objects: 1,
..Default::default()
},
);
write_test_object_metadata(&root, "bucket", "remaining").await;
let _vanished = OrderedDirectory::fail_open(&root.join(&folder.name), kind);
bind_budget(&mut scanner, &root, 1_000_000);
let ctx = CancellationToken::new();
let budget = ScannerCycleBudget::new_with_progress_tracking(
&ctx,
crate::scanner_budget::ScannerCycleBudgetConfig {
max_directories: Some(2),
..Default::default()
},
);
scanner.budget = budget.clone();
let mut into = DataUsageEntry::default();
scanner
.scan_folder(
budget.token(),
CachedFolder {
name: "bucket".into(),
parent: None,
object_heal_prob_div: 1,
},
&mut into,
)
.await
.expect("concurrent candidate deletion or replacement must not fail the remaining namespace scan");
assert_eq!(budget.reason(), None);
assert_eq!(budget.progress().1, 2, "only parent and remaining namespace consume directory quota");
assert!(
scanner.new_cache.find(&folder.name).is_none(),
"vanished candidate cannot retain stale cached totals"
);
assert_eq!(
scanner
.new_cache
.find("bucket/remaining")
.expect("remaining object scanned")
.objects,
1
);
assert_eq!(
charged_ops(&root),
7,
"vanished open is charged but contributes no entry or metadata read"
);
assert!(scanner.new_cache.info.failed_objects.is_empty());
assert!(
scanner.new_cache.info.pending_heals.is_empty(),
"vanished directory is not missing-xl.meta evidence"
);
assert_eq!(
scanner.coverage_gap,
kind == std::io::ErrorKind::NotADirectory,
"a replaced directory retains the existing incomplete-coverage behavior"
);
}
}
crates/scanner/src/sleeper.rs
+1 -1
View File
@@ -225,7 +225,7 @@ impl DynamicSleeper {
self.admit_disk_io(1, 0).await;
}
async fn sleep_folder_preset(&self) {
pub(crate) async fn sleep_folder_preset(&self) {
if !Self::should_sleep() {
return;
}
docs/operations/scanner-runtime-controls.md
+3 -1
View File
@@ -142,7 +142,9 @@ The admission gate orders scanner I/O behind foreground requests but does not bo
Only `xl.meta` files enter object metadata accounting. Other payload files consume entry charges without extra size stats or metadata-read charges. An `xl.meta` symlink reuses its resolved target size for the byte charge. Cancellation interrupts folder pacing, budget waits, and directory/stat waits. Filesystem work already dispatched to a blocking worker can still finish.
Erasure data-directory probes for missing `xl.meta` use the same directory-open and entry budget. They retain their 64-entry limit and remain cancellation-aware without adding a preset folder delay per probe.
UUID candidates under a directory without `xl.meta` are classified during their normal directory traversal. Each directory open and entry is charged once; traversal yields every 64 entries and continues under the budget until a regular `part.N` is found or EOF is reached. A full batch is not evidence of absence. Candidate metadata accounting is deferred until classification finishes. A data directory retains the existing missing-parent-metadata and non-destructive heal path; a namespace continues from the same listing, without a second probe/open/entry pass. Data directories do not consume namespace-directory quota or a preset folder delay; a candidate proven to be a namespace consumes that quota and delay once.
Candidate entries use the normal raw-enumeration progress and checkpoint path. Cancellation or an iteration error leaves classification incomplete, never an EOF or successful coverage result. A child already absent or replaced by a non-directory at open is handled as vanished, without stale namespace totals or a missing-metadata heal request. A replacement retains the existing coverage gap; mid-iteration errors never become EOF. A clean namespace covered by a restored frontier reuses its prior totals only after current classification reaches EOF; metadata is not read or charged to object quota again. A later data candidate prevents completed namespace siblings from certifying the unresolved parent. Existing raw-page indexes are membership receipts, not portable filesystem seek offsets: a later cycle or process restart reopens and revalidates the directory, with actual reads charged again. This does not guarantee that a directory larger than the configured cycle budget can finish within that budget.
- Media is detected once per disk path from the backing block device's `rotational` flag on Linux, through partitions, `dm-*`, and software RAID; any spindle in the stack counts as rotational. Undetectable media, including non-Linux hosts and overlay filesystems without a bind-mounted volume, uses the rotational defaults because a spindle is the case that needs protection.
- `auto` resolves to `100` operations/s and `16 MiB/s` on rotational or unknown media and to `4000` operations/s and `256 MiB/s` on flash. The rotational values leave roughly half of a 7200 rpm spindle's random reads to foreground traffic; the flash values are a few percent of a SATA SSD. The `fastest` speed preset resolves `auto` to no cap; explicit numbers always apply.