// Copyright 2024 RustFS Team // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. use std::collections::HashSet; use std::fs::FileType; use std::io::ErrorKind; use std::sync::{Arc, Once}; use std::time::{Duration, SystemTime}; use crate::ReplTargetSizeSummary; use crate::data_usage_define::{DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, SizeSummary, hash_path}; use crate::error::ScannerError; use crate::scanner_io::ScannerIODisk as _; use crate::sleeper::DynamicSleeper; use metrics::{counter, describe_counter}; use rustfs_common::heal_channel::{ HEAL_DELETE_DANGLING, HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealScanMode, send_heal_request_with_admission, }; use rustfs_common::metrics::{IlmAction, Metric, Metrics, UpdateCurrentPathFn, current_path_updater}; use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc; use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::{GLOBAL_ExpiryState, apply_expiry_rule}; use rustfs_ecstore::bucket::lifecycle::evaluator::Evaluator; use rustfs_ecstore::bucket::lifecycle::{ bucket_lifecycle_ops::apply_transition_rule, lifecycle::{Event, Lifecycle, ObjectOpts}, }; use rustfs_ecstore::bucket::replication::{ReplicationConfig, ReplicationConfigurationExt as _, queue_replication_heal_internal}; use rustfs_ecstore::bucket::versioning::VersioningApi; use rustfs_ecstore::bucket::versioning_sys::BucketVersioningSys; use rustfs_ecstore::cache_value::metacache_set::{ListPathRawOptions, list_path_raw}; use rustfs_ecstore::disk::error::DiskError; use rustfs_ecstore::disk::{Disk, DiskAPI as _, DiskInfoOptions}; use rustfs_ecstore::error::StorageError; use rustfs_ecstore::global::is_erasure; use rustfs_ecstore::pools::{path2_bucket_object, path2_bucket_object_with_base_path}; use rustfs_ecstore::store_api::{ObjectInfo, ObjectToDelete}; use rustfs_ecstore::store_utils::is_reserved_or_invalid_bucket; use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicationStatusType}; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration}; use tokio::select; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; const DATA_USAGE_UPDATE_DIR_CYCLES: u32 = 16; const DATA_SCANNER_COMPACT_LEAST_OBJECT: usize = 500; const YIELD_EVERY_N_OBJECTS: u64 = 128; const DATA_SCANNER_COMPACT_AT_CHILDREN: usize = 10000; const DATA_SCANNER_COMPACT_AT_FOLDERS: usize = DATA_SCANNER_COMPACT_AT_CHILDREN / 4; const DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS: usize = 250_000; 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"; const ENV_FAILED_OBJECT_TTL_SECS: &str = "RUSTFS_DATA_USAGE_FAILED_OBJECT_TTL_SECS"; const ENV_FAILED_OBJECTS_MAX: &str = "RUSTFS_DATA_USAGE_FAILED_OBJECTS_MAX"; const DEFAULT_FAILED_OBJECT_TTL_SECS: u32 = 86_400; const DEFAULT_FAILED_OBJECTS_MAX: u32 = 10_000; const METRIC_SCANNER_INLINE_HEAL_TOTAL: &str = "rustfs_scanner_inline_heal_total"; static SCANNER_INLINE_HEAL_WARN_ONCE: Once = Once::new(); static SCANNER_INLINE_HEAL_METRICS_ONCE: Once = Once::new(); pub fn data_usage_update_dir_cycles() -> u32 { rustfs_utils::get_env_u32(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, DATA_USAGE_UPDATE_DIR_CYCLES) } pub fn heal_object_select_prob() -> u32 { rustfs_utils::get_env_u32(ENV_HEAL_OBJECT_SELECT_PROB, DEFAULT_HEAL_OBJECT_SELECT_PROB) } fn scanner_inline_heal_enabled() -> bool { scanner_inline_heal_enabled_from_value(std::env::var(rustfs_config::ENV_SCANNER_INLINE_HEAL_ENABLE).ok().as_deref()) } fn scanner_inline_heal_enabled_from_value(value: Option<&str>) -> bool { match value { Some(value) => matches!(value.trim().to_ascii_lowercase().as_str(), "1" | "true" | "on" | "yes"), None => rustfs_config::DEFAULT_SCANNER_INLINE_HEAL_ENABLE, } } fn ensure_scanner_inline_heal_metric_registered() { SCANNER_INLINE_HEAL_METRICS_ONCE.call_once(|| { describe_counter!( METRIC_SCANNER_INLINE_HEAL_TOTAL, "Total number of inline heal operations executed directly by scanner." ); counter!(METRIC_SCANNER_INLINE_HEAL_TOTAL).increment(0); }); } fn warn_inline_heal_compat_requested() { if !scanner_inline_heal_enabled() { return; } SCANNER_INLINE_HEAL_WARN_ONCE.call_once(|| { warn!( env = rustfs_config::ENV_SCANNER_INLINE_HEAL_ENABLE, "Inline scanner heal rollback is no longer supported; continuing to enqueue heal candidates asynchronously" ); }); } /// Cached folder information for scanning #[derive(Clone, Debug)] pub struct CachedFolder { pub name: String, pub parent: Option, pub object_heal_prob_div: u32, } /// Type alias for get size function pub type GetSizeFn = Box Result + Send + Sync>; fn build_bucket_heal_request(bucket: String, priority: HealChannelPriority) -> HealChannelRequest { HealChannelRequest { bucket, priority, ..Default::default() } } fn build_object_heal_request( bucket: String, object: String, version_id: Option, scan_mode: HealScanMode, priority: HealChannelPriority, ) -> HealChannelRequest { HealChannelRequest { bucket, object_prefix: Some(object), object_version_id: version_id, priority, scan_mode: Some(scan_mode), remove_corrupted: Some(HEAL_DELETE_DANGLING), ..Default::default() } } fn heal_priority_label(priority: HealChannelPriority) -> &'static str { match priority { HealChannelPriority::Low => "low", HealChannelPriority::Normal => "normal", HealChannelPriority::High => "high", HealChannelPriority::Critical => "critical", } } fn describe_heal_admission(result: HealAdmissionResult) -> String { match result { HealAdmissionResult::Accepted | HealAdmissionResult::Merged => result.result_label().to_string(), HealAdmissionResult::Full => "queue_full".to_string(), HealAdmissionResult::Dropped(reason) => format!("dropped:{}", reason.as_str()), } } fn record_high_priority_heal_escalation( candidate_type: &'static str, priority: HealChannelPriority, result: HealAdmissionResult, ) { counter!( "rustfs_heal_candidate_priority_reject_total", "type" => candidate_type.to_string(), "priority" => heal_priority_label(priority).to_string(), "result" => result.result_label().to_string(), "reason" => result.reason_label().to_string() ) .increment(1); } fn build_high_priority_heal_admission_error( candidate_type: &'static str, bucket: &str, object: Option<&str>, priority: HealChannelPriority, result: HealAdmissionResult, ) -> ScannerError { let object_text = object.map(|object| format!(", object='{object}'")).unwrap_or_default(); ScannerError::Other(format!( "high-priority heal request was not admitted: type={candidate_type}, bucket='{bucket}'{object_text}, priority={}, admission={}", heal_priority_label(priority), describe_heal_admission(result) )) } fn record_heal_candidate_admission(candidate_type: &'static str, priority: HealChannelPriority, result: HealAdmissionResult) { counter!( "rustfs_heal_candidate_enqueue_total", "type" => candidate_type.to_string(), "priority" => heal_priority_label(priority).to_string(), "result" => result.result_label().to_string() ) .increment(1); if matches!(result, HealAdmissionResult::Merged) { counter!( "rustfs_heal_candidate_merge_total", "type" => candidate_type.to_string() ) .increment(1); } if let HealAdmissionResult::Dropped(reason) = result { counter!( "rustfs_heal_candidate_drop_total", "type" => candidate_type.to_string(), "reason" => reason.as_str().to_string() ) .increment(1); } } async fn send_scanner_heal_request( candidate_type: &'static str, request: HealChannelRequest, ) -> Result { let priority = request.priority; match send_heal_request_with_admission(request).await { Ok(result) => { record_heal_candidate_admission(candidate_type, priority, result); Ok(result) } Err(err) => { counter!( "rustfs_heal_candidate_enqueue_total", "type" => candidate_type.to_string(), "priority" => heal_priority_label(priority).to_string(), "result" => "channel_error".to_string() ) .increment(1); Err(ScannerError::Other(err)) } } } async fn send_required_scanner_heal_request( candidate_type: &'static str, bucket: &str, object: Option<&str>, request: HealChannelRequest, ) -> Result<(), ScannerError> { let priority = request.priority; let result = send_scanner_heal_request(candidate_type, request).await?; if result.is_admitted() { return Ok(()); } record_high_priority_heal_escalation(candidate_type, priority, result); error!( candidate_type, bucket, object = object.unwrap_or(""), priority = heal_priority_label(priority), admission = result.result_label(), reason = result.reason_label(), "High-priority heal request was not admitted; escalating to scanner failure" ); Err(build_high_priority_heal_admission_error(candidate_type, bucket, object, priority, result)) } /// Scanner item representing a file during scanning #[derive(Clone, Debug)] pub struct ScannerItem { pub path: String, pub bucket: String, pub prefix: String, pub object_name: String, pub file_type: FileType, pub lifecycle: Option>, pub replication: Option>, pub heal_enabled: bool, pub heal_bitrot: bool, pub debug: bool, } impl ScannerItem { /// Get the object path (prefix + object_name) pub fn object_path(&self) -> String { if self.prefix.is_empty() { self.object_name.clone() } else { path_join_buf(&[&self.prefix, &self.object_name]) } } /// Transform meta directory by splitting prefix and extracting object name /// This converts a directory path like "bucket/dir1/dir2/file" to prefix="bucket/dir1/dir2" and object_name="file" pub fn transform_meta_dir(&mut self) { let prefix = self.prefix.clone(); // Clone to avoid borrow checker issues let split: Vec<&str> = prefix.split(SLASH_SEPARATOR).collect(); if split.len() > 1 { let prefix_parts: Vec<&str> = split[..split.len() - 1].to_vec(); self.prefix = path_join_buf(&prefix_parts); } else { self.prefix = String::new(); } // Object name is the last element self.object_name = split.last().unwrap_or(&"").to_string(); } pub async fn apply_actions( &mut self, object_infos: Vec, lock_retention: Option>, size_summary: &mut SizeSummary, ) { if object_infos.is_empty() { debug!("apply_actions: no object infos for object: {}", self.object_path()); return; } debug!("apply_actions: applying actions for object: {}", self.object_path()); let versioning_config = match BucketVersioningSys::get(&self.bucket).await { Ok(versioning_config) => versioning_config, Err(_) => { warn!("apply_actions: Failed to get versioning configuration for bucket {}", self.bucket); return; } }; let Some(lifecycle) = self.lifecycle.as_ref() else { let mut cumulative_size = 0; for oi in object_infos.iter() { let actual_size = match oi.get_actual_size() { Ok(size) => size, Err(_) => { warn!("apply_actions: Failed to get actual size for object {}", oi.name); continue; } }; let size = self.heal_actions(oi, actual_size, size_summary).await; size_summary.actions_accounting(oi, size, actual_size); cumulative_size += size; } self.alert_excessive_versions(object_infos.len(), cumulative_size); debug!("apply_actions: done for now no lifecycle config"); return; }; let object_opts = object_infos .iter() .map(ObjectOpts::from_object_info) .collect::>(); let events = match Evaluator::new(lifecycle.clone()) .with_lock_retention(lock_retention) .with_replication_config(self.replication.clone()) .eval(&object_opts) .await { Ok(events) => events, Err(e) => { warn!("apply_actions: Failed to evaluate lifecycle for object: {}", e); return; } }; let mut to_delete_objs: Vec = Vec::new(); let mut noncurrent_events: Vec = Vec::new(); let mut cumulative_size = 0; let mut remaining_versions = object_infos.len(); 'eventLoop: { for (i, event) in events.iter().enumerate() { let oi = &object_infos[i]; let actual_size = match oi.get_actual_size() { Ok(size) => size, Err(_) => { warn!("apply_actions: Failed to get actual size for object {}", oi.name); 0 } }; let mut size = actual_size; let done_ilm = Metrics::time_ilm(event.action); match event.action { IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => { remaining_versions = 0; debug!("apply_actions: applying expiry rule for object: {} {}", oi.name, event.action); apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await; done_ilm(1)(); break 'eventLoop; } IlmAction::DeleteAction | IlmAction::DeleteRestoredAction | IlmAction::DeleteRestoredVersionAction => { if !versioning_config.prefix_enabled(&self.object_path()) && event.action == IlmAction::DeleteAction { remaining_versions -= 1; size = 0; } debug!("apply_actions: applying expiry rule for object: {} {}", oi.name, event.action); apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await; done_ilm(1)(); } IlmAction::DeleteVersionAction => { remaining_versions -= 1; size = 0; if let Some(opt) = object_opts.get(i) { to_delete_objs.push(ObjectToDelete { object_name: opt.name.clone(), version_id: opt.version_id, ..Default::default() }); } noncurrent_events.push(event.clone()); done_ilm(1)(); } IlmAction::TransitionAction | IlmAction::TransitionVersionAction => { debug!("apply_actions: applying transition rule for object: {} {}", oi.name, event.action); apply_transition_rule(event, &LcEventSrc::Scanner, oi).await; done_ilm(1)(); } IlmAction::NoneAction | IlmAction::ActionCount => { size = self.heal_actions(oi, actual_size, size_summary).await; } } size_summary.actions_accounting(oi, size, actual_size); cumulative_size += size; } } if !to_delete_objs.is_empty() && let Some(event) = noncurrent_events.first().cloned() { GLOBAL_ExpiryState .write() .await .enqueue_by_newer_noncurrent(&self.bucket, to_delete_objs, event) .await; } self.alert_excessive_versions(remaining_versions, cumulative_size); } async fn heal_actions(&mut self, oi: &ObjectInfo, actual_size: i64, size_summary: &mut SizeSummary) -> i64 { if self.heal_enabled { warn_inline_heal_compat_requested(); self.enqueue_heal(oi).await; } self.heal_replication(oi, size_summary).await; actual_size } async fn heal_replication(&mut self, oi: &ObjectInfo, size_summary: &mut SizeSummary) { if oi.version_id.is_none_or(|v| v.is_nil()) { return; } let Some(replication) = self.replication.clone() else { return; }; let roi = queue_replication_heal_internal(&oi.bucket, oi.clone(), (*replication).clone(), 0).await; if oi.delete_marker || oi.version_purge_status.is_empty() { return; } for (arn, target_status) in roi.target_statuses.iter() { if !size_summary.repl_target_stats.contains_key(arn.as_str()) { size_summary .repl_target_stats .insert(arn.clone(), ReplTargetSizeSummary::default()); } if let Some(repl_target_size_summary) = size_summary.repl_target_stats.get_mut(arn.as_str()) { match target_status { ReplicationStatusType::Pending => { repl_target_size_summary.pending_size += roi.size; repl_target_size_summary.pending_count += 1; size_summary.pending_size += roi.size; size_summary.pending_count += 1; } ReplicationStatusType::Failed => { repl_target_size_summary.failed_size += roi.size; repl_target_size_summary.failed_count += 1; size_summary.failed_size += roi.size; size_summary.failed_count += 1; } ReplicationStatusType::Completed | ReplicationStatusType::CompletedLegacy => { repl_target_size_summary.replicated_size += roi.size; repl_target_size_summary.replicated_count += 1; size_summary.replicated_size += roi.size; size_summary.replicated_count += 1; } _ => {} } } } if oi.replication_status == ReplicationStatusType::Replica { size_summary.replica_size += roi.size; size_summary.replica_count += 1; } } async fn enqueue_heal(&mut self, oi: &ObjectInfo) { let done_heal = Metrics::time(Metric::HealAbandonedObject); debug!( "enqueue_heal: bucket: {}, object_path: {}, version_id: {}", self.bucket, self.object_path(), oi.version_id.unwrap_or_default() ); let scan_mode = if self.heal_bitrot { HealScanMode::Deep } else { HealScanMode::Normal }; let result = send_scanner_heal_request( "object", build_object_heal_request( self.bucket.clone(), self.object_path(), oi.version_id .and_then(|v| if v.is_nil() { None } else { Some(v.to_string()) }), scan_mode, HealChannelPriority::Low, ), ) .await; match result { Ok(HealAdmissionResult::Accepted | HealAdmissionResult::Merged) => {} Ok(result @ (HealAdmissionResult::Full | HealAdmissionResult::Dropped(_))) => { warn!( bucket = %self.bucket, object = %self.object_path(), admission = %describe_heal_admission(result), "enqueue_heal: low-priority heal request was not admitted" ); } Err(e) => warn!("enqueue_heal: failed to submit heal request: {}", e), } done_heal(); } fn alert_excessive_versions(&self, _object_infos_length: usize, _cumulative_size: i64) { // TODO: Implement alerting for excessive versions } } /// Folder scanner for scanning directory structures pub struct FolderScanner { root: String, old_cache: DataUsageCache, new_cache: DataUsageCache, update_cache: DataUsageCache, data_usage_scanner_debug: bool, heal_object_select: u32, scan_mode: HealScanMode, failed_object_ttl_secs: u64, failed_objects_max: usize, sleeper: DynamicSleeper, // should_heal: Arc bool + Send + Sync>, disks: Vec>, disks_quorum: usize, updates: Option>, last_update: SystemTime, update_current_path: UpdateCurrentPathFn, skip_heal: Arc, local_disk: Arc, } impl FolderScanner { fn now_secs() -> u64 { SystemTime::now() .duration_since(SystemTime::UNIX_EPOCH) .unwrap_or_default() .as_secs() } fn should_skip_failed(&self, path: &str) -> bool { let ttl = self.failed_object_ttl_secs; if ttl == 0 { return false; } let Some(last_failed) = self.new_cache.info.failed_objects.get(path) else { return false; }; let now = Self::now_secs(); now.saturating_sub(*last_failed) < ttl } fn record_failed(&mut self, path: &str) { let ttl = self.failed_object_ttl_secs; if ttl == 0 { return; } let now = Self::now_secs(); self.new_cache.info.failed_objects.insert(path.to_string(), now); let max_entries = self.failed_objects_max; if max_entries > 0 && self.new_cache.info.failed_objects.len() > max_entries { self.prune_failed_objects(now, ttl); } } fn prune_failed_objects_cache(&mut self) { let ttl = self.failed_object_ttl_secs; if ttl == 0 { return; } let now = Self::now_secs(); self.prune_failed_objects(now, ttl); } fn prune_failed_objects(&mut self, now: u64, ttl: u64) { let max_entries = self.failed_objects_max; let failed = &mut self.new_cache.info.failed_objects; if failed.is_empty() { return; } failed.retain(|_, ts| now.saturating_sub(*ts) < ttl); if max_entries == 0 { return; } if failed.len() <= max_entries { return; } let mut entries: Vec<(String, u64)> = failed.iter().map(|(k, v)| (k.clone(), *v)).collect(); entries.sort_by(|(k1, ts1), (k2, ts2)| ts1.cmp(ts2).then_with(|| k1.cmp(k2))); let remove_count = failed.len().saturating_sub(max_entries); for (key, _) in entries.into_iter().take(remove_count) { failed.remove(&key); } } pub async fn should_heal(&self) -> bool { if self.skip_heal.load(std::sync::atomic::Ordering::Relaxed) { return false; } if self.heal_object_select == 0 { return false; } if self .local_disk .disk_info(&DiskInfoOptions::default()) .await .unwrap_or_default() .healing { self.skip_heal.store(true, std::sync::atomic::Ordering::Relaxed); return false; } true } /// Set heal object select probability pub fn set_heal_object_select(&mut self, prob: u32) { self.heal_object_select = prob; } /// Set debug mode pub fn set_debug(&mut self, debug: bool) { self.data_usage_scanner_debug = debug; } /// Send update if enough time has passed /// Should be called on a regular basis when the new_cache contains more recent total than previously. /// May or may not send an update upstream. pub async fn send_update(&mut self) { // Send at most an update every minute. if self.updates.is_none() { return; } let elapsed = self.last_update.elapsed().unwrap_or(Duration::from_secs(0)); if elapsed < Duration::from_secs(60) { return; } if let Some(flat) = self.update_cache.size_recursive(&self.new_cache.info.name) && let Some(ref updates) = self.updates { // Try to send without blocking if let Err(e) = updates.send(flat.clone()).await { error!("send_update: failed to send update: {}", e); } self.last_update = SystemTime::now(); } } /// 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> { let done_folder = Metrics::time(Metric::ScanFolder); if ctx.is_cancelled() { return Err(ScannerError::Other("Operation cancelled".to_string())); } let this_hash = hash_path(&folder.name); // Store initial compaction state. let was_compacted = into.compacted; loop { if ctx.is_cancelled() { return Err(ScannerError::Other("Operation cancelled".to_string())); } self.prune_failed_objects_cache(); let mut abandoned_children: DataUsageHashMap = HashSet::new(); if !into.compacted { abandoned_children = self.old_cache.find_children_copy(this_hash.clone()); } debug!("scan_folder : {}/{}", &self.root, &folder.name); let (_, prefix) = path2_bucket_object_with_base_path(&self.root, &folder.name); let active_life_cycle = if self .old_cache .info .lifecycle .as_ref() .is_some_and(|v| v.has_active_rules(&prefix)) { self.old_cache.info.lifecycle.clone() } else { None }; let active_replication = if self.old_cache.info.replication.as_ref().is_some_and(|v| { !v.is_empty() && v.config.as_ref().is_some_and(|config| config.has_active_rules(&prefix, true)) }) { self.old_cache.info.replication.clone() } else { None }; self.sleeper.sleep_folder().await; let mut existing_folders: Vec = Vec::new(); let mut new_folders: Vec = Vec::new(); let mut found_objects = false; let mut object_count: u64 = 0; let dir_path = path_join_buf(&[&self.root, &folder.name]); debug!("scan_folder: dir_path: {:?}", dir_path); let mut dir_reader = match tokio::fs::read_dir(&dir_path).await { Ok(dir_reader) => dir_reader, Err(e) if e.kind() == ErrorKind::NotFound => { warn!("scan_folder: directory disappeared before read {}: {}", dir_path, e); return Ok(()); } Err(e) => return Err(ScannerError::Io(e)), }; loop { let entry = match dir_reader.next_entry().await { Ok(Some(entry)) => entry, Ok(None) => break, Err(e) if e.kind() == ErrorKind::NotFound => { warn!("scan_folder: directory disappeared during iteration {}: {}", dir_path, e); break; } Err(e) if e.kind() == ErrorKind::NotADirectory => { warn!("scan_folder: path became non-directory during iteration {}: {}", dir_path, e); break; } Err(e) => return Err(ScannerError::Io(e)), }; let file_name = entry.file_name().to_string_lossy().to_string(); if file_name.is_empty() || file_name == "." || file_name == ".." { continue; } let file_path = entry.path().to_string_lossy().to_string(); let trim_dir_name = file_path.strip_prefix(&dir_path).unwrap_or(&file_path); let entry_name = path_join_buf(&[&folder.name, trim_dir_name]); if entry_name.is_empty() || entry_name == folder.name { continue; } // Ignore entries that disappeared during traversal or hit symlink // loops, but propagate other walk errors. let mut entry_type = match entry.file_type().await { Ok(entry_type) => entry_type, Err(e) if e.kind() == ErrorKind::NotFound => { warn!("scan_folder: entry disappeared before type lookup {}: {}", entry_name, e); continue; } Err(e) if e.kind() == ErrorKind::TooManyLinks => { warn!("scan_folder: entry hit symlink loop before type lookup {}: {}", entry_name, e); continue; } Err(e) => return Err(ScannerError::Io(e)), }; if entry_type.is_symlink() { let metadata = match tokio::fs::metadata(&file_path).await { Ok(metadata) => metadata, Err(e) if e.kind() == ErrorKind::NotFound => { warn!("scan_folder: symlink target disappeared before metadata lookup {}: {}", file_path, e); continue; } Err(e) if e.kind() == ErrorKind::TooManyLinks => { warn!("scan_folder: symlink target hit loop before metadata lookup {}: {}", file_path, e); continue; } Err(e) => return Err(ScannerError::Io(e)), }; if metadata.is_dir() { warn!("scan_folder: ignoring symlinked directory {}", file_path); continue; } entry_type = metadata.file_type(); } // ok let (bucket, prefix) = path2_bucket_object_with_base_path(self.root.as_str(), &entry_name); if bucket.is_empty() { break; } if is_reserved_or_invalid_bucket(&bucket, false) { break; } if ctx.is_cancelled() { break; } if entry_type.is_dir() { let h = hash_path(&entry_name); if h == this_hash { continue; } let exists = self.old_cache.cache.contains_key(&h.key()); let this = CachedFolder { name: entry_name.clone(), parent: Some(this_hash.clone()), object_heal_prob_div: folder.object_heal_prob_div, }; abandoned_children.remove(&h.key()); if exists { existing_folders.push(this); self.update_cache .copy_with_children(&self.old_cache, &h, &Some(this_hash.clone())); } else { new_folders.push(this); } continue; } let timer = self.sleeper.timer(); let heal_enabled = this_hash.mod_alt( self.old_cache.info.next_cycle as u32 / folder.object_heal_prob_div, self.heal_object_select / folder.object_heal_prob_div, ) && self.should_heal().await; let mut item = ScannerItem { path: file_path, bucket, prefix: rustfs_utils::path::dir(&prefix), object_name: file_name, lifecycle: active_life_cycle.clone(), replication: active_replication.clone(), heal_enabled, heal_bitrot: self.scan_mode == HealScanMode::Deep, debug: self.data_usage_scanner_debug, file_type: entry_type, }; // If this path is already known as failed, just skip it. // We intentionally do NOT call `record_failed` or bump `failed_objects` here, // because the failure was recorded when the original error occurred // (e.g. in the get_size error branch below). This branch only accounts // for subsequent skips of already-failed paths. if self.should_skip_failed(&item.path) { continue; } let sz = match self.local_disk.get_size(item.clone()).await { Ok(sz) => sz, Err(e) => { let is_skip_file = matches!(e, StorageError::Io(ref io) if io.to_string() == "skip file"); if !is_skip_file { // Track failed objects to prevent infinite retry loops into.failed_objects += 1; self.record_failed(&item.path); // Only log non-skip errors to avoid noise warn!("scan_folder: failed to get size for item {}: {}", item.path, e); } timer.sleep().await; continue; } }; found_objects = true; item.transform_meta_dir(); abandoned_children.remove(&path_join_buf(&[&item.bucket, &item.object_path()])); into.add_sizes(&sz); into.objects += 1; object_count += 1; timer.sleep().await; if object_count.is_multiple_of(YIELD_EVERY_N_OBJECTS) { tokio::task::yield_now().await; } } if found_objects && is_erasure().await { // If we found an object in erasure mode, we skip subdirs (only datadirs)... info!("scan_folder: done for now found an object in erasure mode"); break; } // If we have many subfolders, compact ourself. let should_compact = (self.new_cache.info.name != folder.name && existing_folders.len() + new_folders.len() >= DATA_SCANNER_COMPACT_AT_FOLDERS) || existing_folders.len() + new_folders.len() >= DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS; // TODO: Check for excess folders and send events if !into.compacted && should_compact { into.compacted = true; new_folders.append(&mut existing_folders); existing_folders.clear(); if self.data_usage_scanner_debug { debug!("scan_folder: Preemptively compacting: {}, entries: {}", folder.name, new_folders.len()); } } if !into.compacted { for folder_item in &existing_folders { let h = hash_path(&folder_item.name); self.update_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent); } } // Scan new folders for folder_item in new_folders { if ctx.is_cancelled() { return Err(ScannerError::Other("Operation cancelled".to_string())); } let h = hash_path(&folder_item.name); // Add new folders to the update tree so totals update for these. if !into.compacted { let mut found_any = false; let mut parent = this_hash.clone(); let update_cache_name_hash = hash_path(&self.update_cache.info.name); while parent != update_cache_name_hash { let parent_key = parent.key(); let e = self.update_cache.find(&parent_key); if e.is_none_or(|v| v.compacted) { found_any = true; break; } if let Some(next) = self.update_cache.search_parent(&parent) { parent = next; } else { found_any = true; break; } } if !found_any { // Add non-compacted empty entry. self.update_cache .replace_hashed(&h, &Some(this_hash.clone()), &DataUsageEntry::default()); } } (self.update_current_path)(&folder_item.name).await; 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()))?; tokio::task::yield_now().await; } else { let mut dst = DataUsageEntry::default(); // Use Box::pin for recursive async call let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst)); if let Err(e) = fut.await { warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e); continue; } tokio::task::yield_now().await; let h = DataUsageHash(folder_item.name.clone()); into.add_child(&h); // 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; } if !into.compacted && self.update_cache.find(&this_hash.key()).is_some_and(|v| !v.compacted) { self.update_cache.delete_recursive(&h); self.update_cache .copy_with_children(&self.new_cache, &h, &Some(this_hash.clone())); } } // Scan existing folders for mut folder_item in existing_folders { if ctx.is_cancelled() { return Err(ScannerError::Other("Operation cancelled".to_string())); } let h = hash_path(&folder_item.name); if !into.compacted && self.old_cache.is_compacted(&h) { 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); continue; } folder_item.object_heal_prob_div = data_usage_update_dir_cycles(); } (self.update_current_path)(&folder_item.name).await; 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()))?; tokio::task::yield_now().await; } else { let mut dst = DataUsageEntry::default(); // Use Box::pin for recursive async call let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst)); if let Err(e) = fut.await { warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e); continue; } tokio::task::yield_now().await; let h = DataUsageHash(folder_item.name.clone()); into.add_child(&h); // 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; } } // Scan for healing if abandoned_children.is_empty() || !self.should_heal().await { debug!("scan_folder: done for now abandoned children are empty or we are not healing"); // If we are not heal scanning, return now. break; } if self.disks.is_empty() || self.disks_quorum == 0 { debug!("scan_folder: done for now disks are empty or quorum is 0"); break; } let mut resolver = MetadataResolutionParams { dir_quorum: self.disks_quorum, obj_quorum: self.disks_quorum, bucket: "".to_string(), strict: false, ..Default::default() }; for name in abandoned_children { if !self.should_heal().await { break; } let (bucket, prefix) = path2_bucket_object(name.as_str()); if bucket != resolver.bucket { send_required_scanner_heal_request( "bucket", &bucket, None, build_bucket_heal_request(bucket.clone(), HealChannelPriority::High), ) .await?; } resolver.bucket = bucket.clone(); let child_ctx = ctx.child_token(); let (agreed_tx, mut agreed_rx) = mpsc::channel::(1); let (partial_tx, mut partial_rx) = mpsc::channel::(1); let (finished_tx, mut finished_rx) = mpsc::channel::>>(1); let disks = self.disks.iter().cloned().map(Some).collect(); let disks_quorum = self.disks_quorum; let bucket_clone = bucket.clone(); let prefix_clone = prefix.clone(); let child_ctx_clone = child_ctx.clone(); tokio::spawn(async move { if let Err(e) = list_path_raw( child_ctx_clone.clone(), ListPathRawOptions { disks, bucket: bucket_clone.clone(), path: prefix_clone.clone(), recursive: true, report_not_found: true, min_disks: disks_quorum, agreed: Some(Box::new(move |entry: MetaCacheEntry| { let entry_name = entry.name.clone(); let agreed_tx = agreed_tx.clone(); Box::pin(async move { if let Err(e) = agreed_tx.send(entry_name).await { error!("scan_folder: list_path_raw: failed to send entry name: {}: {}", entry.name, e); } }) })), partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { let partial_tx = partial_tx.clone(); Box::pin(async move { if let Err(e) = partial_tx.send(entries).await { error!("scan_folder: list_path_raw: failed to send partial err: {}", e); } }) })), finished: Some(Box::new(move |errs: &[Option]| { let finished_tx = finished_tx.clone(); let errs_clone = errs.to_vec(); Box::pin(async move { if let Err(e) = finished_tx.send(errs_clone).await { error!("scan_folder: list_path_raw: failed to send finished errs: {}", e); } }) })), ..Default::default() }, ) .await { error!("scan_folder: failed to list path: {}/{}: {}", bucket_clone, prefix_clone, e); } }); let mut found_objects = false; let mut agreed_closed = false; let mut partial_closed = false; let mut finished_closed = false; loop { if agreed_closed && partial_closed && finished_closed { break; } select! { entry_name = agreed_rx.recv(), if !agreed_closed => { let Some(entry_name) = entry_name else { agreed_closed = true; continue; }; (self.update_current_path)(&entry_name).await; } entries = partial_rx.recv(), if !partial_closed => { let Some(entries) = entries else { partial_closed = true; continue; }; if !self.should_heal().await { child_ctx.cancel(); break; } let entry_option = match entries.resolve(resolver.clone()){ Some(entry) => { Some(entry) } None => { let (entry,_) = entries.first_found(); entry } }; let Some(entry) = entry_option else { break; }; (self.update_current_path)(&entry.name).await; if entry.is_dir() { continue; } let fivs = match entry.file_info_versions(&bucket) { Ok(fivs) => fivs, Err(e) => { error!("scan_folder: list_path_raw: failed to get file info versions: {}", e); send_required_scanner_heal_request( "object", &bucket, Some(&entry.name), build_object_heal_request( bucket.clone(), entry.name.clone(), None, self.scan_mode, HealChannelPriority::High, ), ) .await?; found_objects = true; continue; } }; for fiv in fivs.versions { send_required_scanner_heal_request( "object", &bucket, Some(&entry.name), build_object_heal_request( bucket.clone(), entry.name.clone(), fiv.version_id.and_then(|v| if v.is_nil() { None } else { Some(v.to_string()) }), self.scan_mode, HealChannelPriority::High, ), ) .await?; found_objects = true; } } errs = finished_rx.recv(), if !finished_closed => { let Some(errs) = errs else { finished_closed = true; continue; }; error!("scan_folder: list_path_raw: failed to get finished errs: {:?}", errs); child_ctx.cancel(); } _ = child_ctx.cancelled() => { break; } } } if found_objects { let folder_item = CachedFolder { name: name.clone(), parent: Some(this_hash.clone()), object_heal_prob_div: 1, }; 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()))?; tokio::task::yield_now().await; } else { let mut dst = DataUsageEntry::default(); // Use Box::pin for recursive async call let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst)); if let Err(e) = fut.await { warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e); continue; } tokio::task::yield_now().await; let h = DataUsageHash(folder_item.name.clone()); into.add_child(&h); // 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; } } } break; } if !was_compacted { self.new_cache.replace_hashed(&this_hash, &folder.parent, into); } if !into.compacted && self.new_cache.info.name != folder.name && let Some(mut flat) = self.new_cache.size_recursive(&this_hash.key()) { flat.compacted = true; let mut should_compact = false; if flat.objects < DATA_SCANNER_COMPACT_LEAST_OBJECT { should_compact = true; } else { // Compact if we only have objects as children... should_compact = true; for k in &into.children { if let Some(v) = self.new_cache.cache.get(k) && (!v.children.is_empty() || v.objects > 1) { should_compact = false; break; } } } if should_compact { self.new_cache.delete_recursive(&this_hash); self.new_cache.replace_hashed(&this_hash, &folder.parent, &flat); } } // Compact if too many children... if !into.compacted { let done_compact = Metrics::time(Metric::CompactFolder); self.new_cache.reduce_children_of( &this_hash, DATA_SCANNER_COMPACT_AT_CHILDREN, self.new_cache.info.name != folder.name, ); done_compact(); } if self.update_cache.cache.contains_key(&this_hash.key()) && !was_compacted { // Replace if existed before. if let Some(flat) = self.new_cache.size_recursive(&this_hash.key()) { self.update_cache.delete_recursive(&this_hash); self.update_cache.replace_hashed(&this_hash, &folder.parent, &flat); } } done_folder(); Ok(()) } pub fn as_mut_new_cache(&mut self) -> &mut DataUsageCache { &mut self.new_cache } } /// Scan a data folder /// This function scans the basepath+cache.info.name and returns an updated cache. /// The returned cache will always be valid, but may not be updated from the existing. /// Throttling between operations is controlled by the provided [`DynamicSleeper`]. /// If the supplied context is canceled the function will return at the first chance. #[allow(clippy::too_many_arguments)] pub async fn scan_data_folder( ctx: CancellationToken, disks: Vec>, local_disk: Arc, cache: DataUsageCache, updates: Option>, scan_mode: HealScanMode, sleeper: DynamicSleeper, ) -> Result { use crate::data_usage_define::DATA_USAGE_ROOT; ensure_scanner_inline_heal_metric_registered(); // Check that we're not trying to scan the root if cache.info.name.is_empty() || cache.info.name == DATA_USAGE_ROOT { return Err(ScannerError::Other("internal error: root scan attempted".to_string())); } // Get disk path let base_path = local_disk.path().to_string_lossy().to_string(); let (update_current_path, close_disk) = current_path_updater(&base_path, &cache.info.name); // Create skip_heal flag let is_erasure_mode = is_erasure().await; let skip_heal = Arc::new(std::sync::atomic::AtomicBool::new(!is_erasure_mode || cache.info.skip_healing)); // Create heal_object_select flag let heal_object_select = if is_erasure_mode && !cache.info.skip_healing { heal_object_select_prob() } else { 0 }; let disks_quorum = disks.len() / 2; let failed_object_ttl = rustfs_utils::get_env_u32(ENV_FAILED_OBJECT_TTL_SECS, DEFAULT_FAILED_OBJECT_TTL_SECS) as u64; let failed_objects_max = rustfs_utils::get_env_u32(ENV_FAILED_OBJECTS_MAX, DEFAULT_FAILED_OBJECTS_MAX) as usize; // Create folder scanner let mut scanner = FolderScanner { root: base_path, old_cache: cache.clone(), new_cache: DataUsageCache { info: cache.info.clone(), ..Default::default() }, update_cache: DataUsageCache { info: cache.info.clone(), ..Default::default() }, data_usage_scanner_debug: false, heal_object_select, scan_mode, failed_object_ttl_secs: failed_object_ttl, failed_objects_max, sleeper, disks, disks_quorum, updates, last_update: SystemTime::UNIX_EPOCH, update_current_path, skip_heal, local_disk, }; // Check if context is cancelled if ctx.is_cancelled() { return Err(ScannerError::Other("Operation cancelled".to_string())); } // Read top level in bucket let mut root = DataUsageEntry::default(); let folder = CachedFolder { name: cache.info.name.clone(), parent: None, object_heal_prob_div: 1, }; // Scan the folder match scanner.scan_folder(ctx, folder, &mut root).await { Ok(()) => { // Get the new cache and finalize it let new_cache = scanner.as_mut_new_cache(); new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN); new_cache.info.last_update = Some(SystemTime::now()); new_cache.info.next_cycle = cache.info.next_cycle; close_disk().await; Ok(new_cache.clone()) } Err(e) => { close_disk().await; // No useful information, return original cache Err(e) } } } #[cfg(test)] mod tests { use crate::SCANNER_SLEEPER; use super::*; use rustfs_ecstore::disk::{DiskOption, endpoint::Endpoint, new_disk}; use serial_test::serial; #[cfg(unix)] use std::os::unix::fs::{PermissionsExt, symlink}; use std::sync::atomic::AtomicBool; use uuid::Uuid; async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) { let temp_dir = std::env::temp_dir().join(format!("rustfs-scanner-test-{}", Uuid::new_v4())); tokio::fs::create_dir_all(&temp_dir) .await .expect("failed to create test directory"); let endpoint = Endpoint::try_from(temp_dir.to_string_lossy().as_ref()).expect("failed to create endpoint"); let disk = new_disk( &endpoint, &DiskOption { cleanup: false, health_check: false, }, ) .await .expect("failed to create disk"); let update_current_path: UpdateCurrentPathFn = Arc::new(|_: &str| Box::pin(async {})); let scanner = FolderScanner { root: temp_dir.to_string_lossy().to_string(), old_cache: DataUsageCache::default(), new_cache: DataUsageCache::default(), update_cache: DataUsageCache::default(), data_usage_scanner_debug: false, heal_object_select: 0, scan_mode: HealScanMode::Normal, failed_object_ttl_secs: u64::MAX, failed_objects_max: usize::MAX, sleeper: SCANNER_SLEEPER.clone(), disks: Vec::new(), disks_quorum: 0, updates: None, last_update: SystemTime::UNIX_EPOCH, update_current_path, skip_heal: Arc::new(AtomicBool::new(false)), local_disk: disk, }; (scanner, temp_dir) } struct TestGuard { temp_dir: Option, } impl TestGuard { fn new(ttl: u64, max: usize, scanner: &mut FolderScanner, temp_dir: std::path::PathBuf) -> Self { scanner.failed_object_ttl_secs = ttl; scanner.failed_objects_max = max; Self { temp_dir: Some(temp_dir), } } } impl Drop for TestGuard { fn drop(&mut self) { if let Some(temp_dir) = self.temp_dir.take() { let _ = std::fs::remove_dir_all(&temp_dir); } } } #[tokio::test] #[serial] async fn test_should_skip_failed_respects_ttl() { let (mut scanner, temp_dir) = build_test_scanner().await; let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir); let now = FolderScanner::now_secs(); scanner .new_cache .info .failed_objects .insert("recent".to_string(), now.saturating_sub(10)); scanner .new_cache .info .failed_objects .insert("expired".to_string(), now.saturating_sub(120)); assert!(scanner.should_skip_failed("recent")); assert!(!scanner.should_skip_failed("expired")); } #[tokio::test] #[serial] async fn test_record_failed_ttl_zero_noop() { let (mut scanner, temp_dir) = build_test_scanner().await; let _guard = TestGuard::new(0, 100, &mut scanner, temp_dir); scanner.record_failed("path1"); assert!(scanner.new_cache.info.failed_objects.is_empty()); let now = FolderScanner::now_secs(); scanner.new_cache.info.failed_objects.insert("path2".to_string(), now); assert!(!scanner.should_skip_failed("path2")); } #[tokio::test] #[serial] async fn test_record_failed_prunes_to_max_entries() { let (mut scanner, temp_dir) = build_test_scanner().await; let _guard = TestGuard::new(1000, 2, &mut scanner, temp_dir); let now = FolderScanner::now_secs(); scanner .new_cache .info .failed_objects .insert("old1".to_string(), now.saturating_sub(50)); scanner .new_cache .info .failed_objects .insert("old2".to_string(), now.saturating_sub(40)); scanner .new_cache .info .failed_objects .insert("old3".to_string(), now.saturating_sub(30)); scanner.record_failed("new"); assert_eq!(scanner.new_cache.info.failed_objects.len(), 2); assert!(scanner.new_cache.info.failed_objects.contains_key("new")); assert!(scanner.new_cache.info.failed_objects.contains_key("old3")); assert!(!scanner.new_cache.info.failed_objects.contains_key("old1")); assert!(!scanner.new_cache.info.failed_objects.contains_key("old2")); } #[tokio::test] #[serial] async fn test_prune_failed_objects_cache_drops_expired() { let (mut scanner, temp_dir) = build_test_scanner().await; let _guard = TestGuard::new(5, 10, &mut scanner, temp_dir); let now = FolderScanner::now_secs(); scanner .new_cache .info .failed_objects .insert("expired".to_string(), now.saturating_sub(10)); scanner .new_cache .info .failed_objects .insert("fresh".to_string(), now.saturating_sub(2)); scanner.prune_failed_objects_cache(); assert_eq!(scanner.new_cache.info.failed_objects.len(), 1); assert!(scanner.new_cache.info.failed_objects.contains_key("fresh")); } #[tokio::test] #[serial] async fn test_prune_failed_objects_max_zero_keeps_fresh() { let (mut scanner, temp_dir) = build_test_scanner().await; let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir); let now = FolderScanner::now_secs(); scanner .new_cache .info .failed_objects .insert("fresh1".to_string(), now.saturating_sub(5)); scanner .new_cache .info .failed_objects .insert("fresh2".to_string(), now.saturating_sub(10)); scanner .new_cache .info .failed_objects .insert("expired".to_string(), now.saturating_sub(120)); scanner.prune_failed_objects_cache(); assert_eq!(scanner.new_cache.info.failed_objects.len(), 2); assert!(scanner.new_cache.info.failed_objects.contains_key("fresh1")); assert!(scanner.new_cache.info.failed_objects.contains_key("fresh2")); assert!(!scanner.new_cache.info.failed_objects.contains_key("expired")); } #[test] fn test_scanner_inline_heal_enabled_defaults_to_false() { assert!(!scanner_inline_heal_enabled_from_value(None)); } #[test] fn test_scanner_inline_heal_enabled_reads_env_override() { assert!(scanner_inline_heal_enabled_from_value(Some("true"))); assert!(scanner_inline_heal_enabled_from_value(Some("YES"))); assert!(scanner_inline_heal_enabled_from_value(Some("1"))); assert!(!scanner_inline_heal_enabled_from_value(Some("false"))); } #[test] fn test_build_object_heal_request_omits_nil_version_id() { let request = build_object_heal_request( "bucket".to_string(), "path/to/object".to_string(), None, HealScanMode::Deep, HealChannelPriority::Low, ); assert_eq!(request.bucket, "bucket"); assert_eq!(request.object_prefix.as_deref(), Some("path/to/object")); assert!(request.object_version_id.is_none()); assert_eq!(request.scan_mode, Some(HealScanMode::Deep)); assert_eq!(request.priority, HealChannelPriority::Low); assert_eq!(request.remove_corrupted, Some(HEAL_DELETE_DANGLING)); } #[test] fn test_heal_priority_label_matches_priority_names() { assert_eq!(heal_priority_label(HealChannelPriority::Low), "low"); assert_eq!(heal_priority_label(HealChannelPriority::Normal), "normal"); assert_eq!(heal_priority_label(HealChannelPriority::High), "high"); assert_eq!(heal_priority_label(HealChannelPriority::Critical), "critical"); } #[test] fn test_describe_heal_admission_formats_unadmitted_results() { assert_eq!(describe_heal_admission(HealAdmissionResult::Accepted), "accepted"); assert_eq!(describe_heal_admission(HealAdmissionResult::Merged), "merged"); assert_eq!(describe_heal_admission(HealAdmissionResult::Full), "queue_full"); assert_eq!( describe_heal_admission(HealAdmissionResult::Dropped( rustfs_common::heal_channel::HealAdmissionDropReason::QueueFull )), "dropped:queue_full" ); } #[test] fn test_build_high_priority_heal_admission_error_contains_context() { let err = build_high_priority_heal_admission_error( "object", "bucket-a", Some("path/to/object"), HealChannelPriority::High, HealAdmissionResult::Full, ); let err_text = err.to_string(); assert!(err_text.contains("type=object")); assert!(err_text.contains("bucket='bucket-a'")); assert!(err_text.contains("object='path/to/object'")); assert!(err_text.contains("priority=high")); assert!(err_text.contains("admission=queue_full")); } #[tokio::test] async fn test_heal_actions_returns_actual_size_without_inline_heal() { let temp_dir = std::env::temp_dir(); let file_type = std::fs::metadata(&temp_dir).unwrap().file_type(); let mut item = ScannerItem { path: temp_dir.join("object").to_string_lossy().to_string(), bucket: "bucket".to_string(), prefix: "".to_string(), object_name: "object".to_string(), file_type, lifecycle: None, replication: None, heal_enabled: true, heal_bitrot: true, debug: false, }; let object_info = ObjectInfo { bucket: "bucket".to_string(), name: "object".to_string(), ..Default::default() }; let mut size_summary = SizeSummary::default(); let size = item.heal_actions(&object_info, 123, &mut size_summary).await; assert_eq!(size, 123); } #[tokio::test] #[serial] #[cfg(unix)] async fn test_scan_folder_skips_unreadable_child_directory() { let (mut scanner, temp_dir) = build_test_scanner().await; let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir.clone()); let bucket_dir = temp_dir.join("bucket"); let good_dir = bucket_dir.join("good"); let bad_dir = bucket_dir.join("bad"); std::fs::create_dir_all(&good_dir).expect("failed to create good dir"); std::fs::create_dir_all(&bad_dir).expect("failed to create bad dir"); std::fs::set_permissions(&bad_dir, std::fs::Permissions::from_mode(0o000)).expect("failed to remove bad dir permissions"); scanner.old_cache.info.name = "bucket".to_string(); scanner.new_cache.info.name = "bucket".to_string(); scanner.update_cache.info.name = "bucket".to_string(); let folder = CachedFolder { name: "bucket".to_string(), parent: None, object_heal_prob_div: 1, }; let mut into = DataUsageEntry::default(); let result = scanner.scan_folder(CancellationToken::new(), folder, &mut into).await; std::fs::set_permissions(&bad_dir, std::fs::Permissions::from_mode(0o755)) .expect("failed to restore bad dir permissions"); assert!(result.is_ok(), "expected unreadable child directory to be skipped"); } #[tokio::test] #[serial] async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() { let (mut scanner, temp_dir) = build_test_scanner().await; let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone()); let _heal_responder = rustfs_common::heal_channel::init_heal_channel().ok().map(|mut heal_rx| { tokio::spawn(async move { while let Some(command) = heal_rx.recv().await { if let rustfs_common::heal_channel::HealChannelCommand::Start { response_tx, .. } = command { let _ = response_tx.send(Ok(HealAdmissionResult::Accepted)); } } }) }); let bucket = "src-archive"; tokio::fs::create_dir_all(temp_dir.join(bucket)) .await .expect("failed to create bucket directory"); let mut disks = vec![scanner.local_disk.clone()]; for disk_name in ["disk2", "disk3", "disk4"] { let disk_root = temp_dir.join(disk_name); tokio::fs::create_dir_all(disk_root.join(bucket)) .await .expect("failed to create extra disk bucket directory"); let endpoint = Endpoint::try_from(disk_root.to_string_lossy().as_ref()).expect("failed to create extra disk endpoint"); let disk = new_disk( &endpoint, &DiskOption { cleanup: false, health_check: false, }, ) .await .expect("failed to create extra disk"); disks.push(disk); } scanner.heal_object_select = 1; scanner.disks = disks; scanner.disks_quorum = 2; scanner.old_cache.replace( "src-archive/snapshots/37b3f20d941e2f5e6d99114d9bb2f3e67a8a2e5c9c4c5a1b0d6e7f8091a2b3c4", bucket, DataUsageEntry { objects: 1, ..Default::default() }, ); let mut into = DataUsageEntry::default(); let folder = CachedFolder { name: bucket.to_string(), parent: None, object_heal_prob_div: 1, }; tokio::time::timeout( Duration::from_millis(200), scanner.scan_folder(CancellationToken::new(), folder, &mut into), ) .await .expect("scan_folder should not hang after list_path_raw finishes") .expect("scan_folder should finish successfully"); } #[tokio::test] #[serial] #[cfg(unix)] async fn test_scan_folder_ignores_symlinked_child_directory() { let (mut scanner, temp_dir) = build_test_scanner().await; let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir.clone()); let bucket_dir = temp_dir.join("bucket"); let target_dir = bucket_dir.join("target"); let link_dir = bucket_dir.join("link"); std::fs::create_dir_all(&target_dir).expect("failed to create target dir"); symlink(&target_dir, &link_dir).expect("failed to create symlinked dir"); scanner.old_cache.info.name = "bucket".to_string(); scanner.new_cache.info.name = "bucket".to_string(); scanner.update_cache.info.name = "bucket".to_string(); let folder = CachedFolder { name: "bucket".to_string(), parent: None, object_heal_prob_div: 1, }; let mut into = DataUsageEntry::default(); let result = scanner.scan_folder(CancellationToken::new(), folder, &mut into).await; assert!(result.is_ok(), "expected symlinked child directory to be ignored"); assert_eq!(into.failed_objects, 0, "expected ignored symlink not to count as a failed object"); } }