From 68e344bf03f54ba88c1ab19e2ddc7e734f514df0 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sat, 1 Aug 2026 11:47:42 +0800 Subject: [PATCH] fix(scanner): remove total timeout from heal walks (#5502) Co-authored-by: cxymds --- crates/e2e_test/src/reliant/tiering.rs | 64 +++------- crates/scanner/src/scanner_folder.rs | 167 +++++++++++++++---------- 2 files changed, 117 insertions(+), 114 deletions(-) diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs index 872a594bc..4b58392e3 100644 --- a/crates/e2e_test/src/reliant/tiering.rs +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -382,7 +382,6 @@ struct ManualTransitionRunReport { skipped_delete_marker: u64, skipped_directory: u64, skipped_replication: u64, - skipped_already_transitioned: u64, skipped_already_in_flight: u64, skipped_queue_full: u64, skipped_queue_closed: u64, @@ -408,41 +407,6 @@ fn assert_completed_or_in_flight_partial(state: &str, report: &ManualTransitionR } } -fn assert_conflict_winner_report(state: &str, report: &ManualTransitionRunReport, expected_objects: u64, context: &str) { - assert_completed_or_in_flight_partial(state, report, context); - if report.skipped_already_in_flight > 0 { - assert!( - report.scanned <= expected_objects, - "{context}: scanned more objects than the conflict scope contains: {report:#?}" - ); - assert!( - report.eligible <= expected_objects, - "{context}: marked more objects eligible than the conflict scope contains: {report:#?}" - ); - assert_eq!( - report.enqueued + report.skipped_already_in_flight, - report.eligible, - "{context}: partial in-flight accounting must cover every eligible object: {report:#?}" - ); - } else { - assert_eq!(report.scanned, expected_objects, "{context}: {report:#?}"); - assert_eq!( - report.eligible + report.skipped_already_transitioned, - expected_objects, - "{context}: {report:#?}" - ); - assert_eq!( - report.enqueued + report.skipped_already_in_flight, - expected_objects, - "{context}: {report:#?}" - ); - } - assert_eq!( - report.transition_completed, report.enqueued, - "{context}: winner must wait for all queued transitions: {report:#?}" - ); -} - #[derive(Debug, Deserialize)] struct ManualTransitionQueueSnapshot { queue_capacity: u64, @@ -1276,8 +1240,15 @@ async fn test_manual_transition_async_scope_conflicts_report_active_job() -> Tes cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) - .await?; + hot.start_rustfs_server_with_env( + vec![], + &[ + ("RUSTFS_SCANNER_ENABLED", "false"), + ("RUSTFS_SCANNER_CYCLE", "3600"), + (MANUAL_TRANSITION_CANCEL_BARRIER_ENV, "1"), + ], + ) + .await?; let hot_client = hot.create_s3_client(); add_rustfs_tier(&hot, &cold).await?; @@ -1347,20 +1318,21 @@ async fn test_manual_transition_async_scope_conflicts_report_active_job() -> Tes assert_eq!(conflict.cancel_endpoint, status_endpoint); assert!(!conflict.scope_key.is_empty()); + manual_transition_job_cancel(&hot, cancel_endpoint).await?; + let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, MANUAL_ASYNC_CONFLICT_TERMINAL_TIMEOUT).await?; assert_eq!(terminal.job_id, job_id); + assert_eq!(terminal.status, "cancelled", "terminal conflict winner response: {terminal:#?}"); assert!(!terminal.report.dry_run); assert_eq!(terminal.report.bucket, MANUAL_ASYNC_CONFLICT_BUCKET); assert_eq!(terminal.report.prefix, accepted.report.prefix); - assert_conflict_winner_report( - &terminal.status, - &terminal.report, - MANUAL_ASYNC_CONFLICT_OBJECTS as u64, - "terminal conflict winner response", + assert!(terminal.report.cancelled, "terminal conflict winner response: {terminal:#?}"); + assert_eq!(terminal.report.scanned, 0, "terminal conflict winner response: {terminal:#?}"); + assert_eq!(terminal.report.enqueued, 0, "terminal conflict winner response: {terminal:#?}"); + assert_eq!( + terminal.report.transition_completed, 0, + "terminal conflict winner response: {terminal:#?}" ); - assert_eq!(terminal.report.dry_run_eligible, 0, "terminal conflict winner response: {terminal:#?}"); - assert_eq!(terminal.report.transition_failed, 0, "terminal conflict winner response: {terminal:#?}"); - assert_eq!(terminal.report.tier_failure, 0, "terminal conflict winner response: {terminal:#?}"); let after_remote_count = cold_tier_object_count(&cold_client).await?; assert!(after_remote_count >= before_remote_count); assert!(after_remote_count <= before_remote_count + MANUAL_ASYNC_CONFLICT_OBJECTS); diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 7d2ea6608..2a6181b7e 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -75,7 +75,7 @@ const DATA_SCANNER_COMPACT_LEAST_OBJECT: usize = 500; 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 SCANNER_LIST_PATH_RAW_TIMEOUT: Duration = Duration::from_secs(60); +const SCANNER_LIST_PATH_RAW_STALL_TIMEOUT: Duration = Duration::from_secs(60); const SCANNER_ENTRY_PROGRESS_BATCH: u64 = 32; const SCANNER_ENTRY_PROGRESS_INTERVAL: Duration = Duration::from_secs(30); const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024; @@ -100,6 +100,21 @@ static SCANNER_INLINE_HEAL_WARN_ONCE: Once = Once::new(); static SCANNER_INLINE_HEAL_METRICS_ONCE: Once = Once::new(); static SCANNER_ALERT_METRICS_ONCE: Once = Once::new(); +#[cfg(test)] +type ListPathRawTimeoutSnapshot = (bool, Option, Option); + +fn scanner_abandoned_child_list_options() -> ListPathRawOptions { + // A complete heal walk scales with bucket size and may legitimately take + // longer than a fixed wall-clock budget. Keep the total duration unbounded; + // Retain the scanner's per-read stall budget and keep cancellation controlled + // by the scanner cycle token. + ListPathRawOptions { + skip_walkdir_total_timeout: true, + walkdir_stall_timeout: Some(SCANNER_LIST_PATH_RAW_STALL_TIMEOUT), + ..Default::default() + } +} + pub fn data_usage_update_dir_cycles() -> u32 { rustfs_utils::get_env_u32(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, DATA_USAGE_UPDATE_DIR_CYCLES) } @@ -1281,6 +1296,8 @@ pub struct FolderScanner { skip_heal: Arc, local_disk: Arc, pending_heals_changed: bool, + #[cfg(test)] + list_path_raw_options_observer: Option>, } impl FolderScanner { @@ -2410,75 +2427,79 @@ impl FolderScanner { let bucket_clone = bucket.clone(); let prefix_clone = prefix.clone(); let child_ctx_clone = child_ctx.clone(); + #[cfg(test)] + let list_path_raw_options_observer = self.list_path_raw_options_observer.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, - walkdir_timeout: Some(SCANNER_LIST_PATH_RAW_TIMEOUT), - walkdir_stall_timeout: Some(SCANNER_LIST_PATH_RAW_TIMEOUT), - 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!( - target: "rustfs::scanner::folder", - event = EVENT_SCANNER_FOLDER_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_FOLDER, - entry = %entry.name, - state = "list_path_agreed_send_failed", - error = %e, - "Scanner list_path_raw agreed callback failed" - ); - } - }) - })), - 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!( - target: "rustfs::scanner::folder", - event = EVENT_SCANNER_FOLDER_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_FOLDER, - state = "list_path_partial_send_failed", - error = %e, - "Scanner list_path_raw partial callback failed" - ); - } - }) - })), - 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!( - target: "rustfs::scanner::folder", - event = EVENT_SCANNER_FOLDER_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_FOLDER, - state = "list_path_finished_send_failed", - error = %e, - "Scanner list_path_raw finished callback failed" - ); - } - }) - })), - ..Default::default() - }, - ) - .await - { + let options = 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!( + target: "rustfs::scanner::folder", + event = EVENT_SCANNER_FOLDER_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_FOLDER, + entry = %entry.name, + state = "list_path_agreed_send_failed", + error = %e, + "Scanner list_path_raw agreed callback failed" + ); + } + }) + })), + 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!( + target: "rustfs::scanner::folder", + event = EVENT_SCANNER_FOLDER_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_FOLDER, + state = "list_path_partial_send_failed", + error = %e, + "Scanner list_path_raw partial callback failed" + ); + } + }) + })), + 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!( + target: "rustfs::scanner::folder", + event = EVENT_SCANNER_FOLDER_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_FOLDER, + state = "list_path_finished_send_failed", + error = %e, + "Scanner list_path_raw finished callback failed" + ); + } + }) + })), + ..scanner_abandoned_child_list_options() + }; + #[cfg(test)] + if let Some(observer) = list_path_raw_options_observer { + let _ = observer.send(( + options.skip_walkdir_total_timeout, + options.walkdir_timeout, + options.walkdir_stall_timeout, + )); + } + if let Err(e) = list_path_raw(child_ctx_clone.clone(), options).await { if is_missing_path_disk_error(&e) { debug!( target: "rustfs::scanner::folder", @@ -2820,6 +2841,8 @@ pub async fn scan_data_folder( skip_heal, local_disk, pending_heals_changed: false, + #[cfg(test)] + list_path_raw_options_observer: None, }; // Check if context is cancelled @@ -3026,6 +3049,7 @@ mod tests { skip_heal: Arc::new(AtomicBool::new(false)), local_disk: disk, pending_heals_changed: false, + list_path_raw_options_observer: None, }; (scanner, temp_dir) @@ -4210,6 +4234,8 @@ mod tests { scanner.heal_object_select = 1; scanner.disks = disks; scanner.disks_quorum = 2; + let (options_tx, mut options_rx) = mpsc::unbounded_channel(); + scanner.list_path_raw_options_observer = Some(options_tx); scanner.old_cache.replace( &format!("{bucket}/{object}"), bucket, @@ -4231,6 +4257,11 @@ mod tests { .expect("scan_folder should not hang after list_path_raw finishes") .expect("scan_folder should finish successfully"); + let observed_options = tokio::time::timeout(Duration::from_secs(1), options_rx.recv()) + .await + .expect("abandoned-child listing options should be observed promptly") + .expect("abandoned-child listing options channel should remain open"); + assert_eq!(observed_options, (true, None, Some(SCANNER_LIST_PATH_RAW_STALL_TIMEOUT))); let root = scanner .new_cache .checked_flatten(bucket)