gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-27 15:37:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): remove total timeout from heal walks (#5502)

Browse Source 
Co-authored-by: cxymds <cxymds@gmail.com>
This commit is contained in:
Zhengchao An
2026-08-01 11:47:42 +08:00
committed by GitHub
parent 48c2fcb62b
commit 68e344bf03
2 changed files with 117 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/e2e_test/src/reliant/tiering.rs
+18 -46
View File
@@ -382,7 +382,6 @@ struct ManualTransitionRunReport {
skipped_delete_marker: u64, skipped_delete_marker: u64,
skipped_directory: u64, skipped_directory: u64,
skipped_replication: u64, skipped_replication: u64,
skipped_already_transitioned: u64,
skipped_already_in_flight: u64, skipped_already_in_flight: u64,
skipped_queue_full: u64, skipped_queue_full: u64,
skipped_queue_closed: 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)] #[derive(Debug, Deserialize)]
struct ManualTransitionQueueSnapshot { struct ManualTransitionQueueSnapshot {
queue_capacity: u64, 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?; cold_client.create_bucket().bucket(TIER_BUCKET).send().await?;
let mut hot = RustFSTestEnvironment::new().await?; let mut hot = RustFSTestEnvironment::new().await?;
hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) hot.start_rustfs_server_with_env(
.await?; vec![],
&[
("RUSTFS_SCANNER_ENABLED", "false"),
("RUSTFS_SCANNER_CYCLE", "3600"),
(MANUAL_TRANSITION_CANCEL_BARRIER_ENV, "1"),
],
)
.await?;
let hot_client = hot.create_s3_client(); let hot_client = hot.create_s3_client();
add_rustfs_tier(&hot, &cold).await?; 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_eq!(conflict.cancel_endpoint, status_endpoint);
assert!(!conflict.scope_key.is_empty()); 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?; 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.job_id, job_id);
assert_eq!(terminal.status, "cancelled", "terminal conflict winner response: {terminal:#?}");
assert!(!terminal.report.dry_run); assert!(!terminal.report.dry_run);
assert_eq!(terminal.report.bucket, MANUAL_ASYNC_CONFLICT_BUCKET); assert_eq!(terminal.report.bucket, MANUAL_ASYNC_CONFLICT_BUCKET);
assert_eq!(terminal.report.prefix, accepted.report.prefix); assert_eq!(terminal.report.prefix, accepted.report.prefix);
assert_conflict_winner_report( assert!(terminal.report.cancelled, "terminal conflict winner response: {terminal:#?}");
&terminal.status, assert_eq!(terminal.report.scanned, 0, "terminal conflict winner response: {terminal:#?}");
&terminal.report, assert_eq!(terminal.report.enqueued, 0, "terminal conflict winner response: {terminal:#?}");
MANUAL_ASYNC_CONFLICT_OBJECTS as u64, assert_eq!(
"terminal conflict winner response", 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?; 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);
assert!(after_remote_count <= before_remote_count + MANUAL_ASYNC_CONFLICT_OBJECTS); assert!(after_remote_count <= before_remote_count + MANUAL_ASYNC_CONFLICT_OBJECTS);
crates/scanner/src/scanner_folder.rs
+99 -68
View File
@@ -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_CHILDREN: usize = 10000;
const DATA_SCANNER_COMPACT_AT_FOLDERS: usize = DATA_SCANNER_COMPACT_AT_CHILDREN / 4; const DATA_SCANNER_COMPACT_AT_FOLDERS: usize = DATA_SCANNER_COMPACT_AT_CHILDREN / 4;
const DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS: usize = 250_000; 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_BATCH: u64 = 32;
const SCANNER_ENTRY_PROGRESS_INTERVAL: Duration = Duration::from_secs(30); const SCANNER_ENTRY_PROGRESS_INTERVAL: Duration = Duration::from_secs(30);
const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024; 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_INLINE_HEAL_METRICS_ONCE: Once = Once::new();
static SCANNER_ALERT_METRICS_ONCE: Once = Once::new(); static SCANNER_ALERT_METRICS_ONCE: Once = Once::new();
#[cfg(test)]
type ListPathRawTimeoutSnapshot = (bool, Option<Duration>, Option<Duration>);
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 { pub fn data_usage_update_dir_cycles() -> u32 {
rustfs_utils::get_env_u32(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, DATA_USAGE_UPDATE_DIR_CYCLES) 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<std::sync::atomic::AtomicBool>, skip_heal: Arc<std::sync::atomic::AtomicBool>,
local_disk: Arc<Disk>, local_disk: Arc<Disk>,
pending_heals_changed: bool, pending_heals_changed: bool,
#[cfg(test)]
list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
} }
impl FolderScanner { impl FolderScanner {
@@ -2410,75 +2427,79 @@ impl FolderScanner {
let bucket_clone = bucket.clone(); let bucket_clone = bucket.clone();
let prefix_clone = prefix.clone(); let prefix_clone = prefix.clone();
let child_ctx_clone = child_ctx.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 { tokio::spawn(async move {
if let Err(e) = list_path_raw( let options = ListPathRawOptions {
child_ctx_clone.clone(), disks,
ListPathRawOptions { bucket: bucket_clone.clone(),
disks, path: prefix_clone.clone(),
bucket: bucket_clone.clone(), recursive: true,
path: prefix_clone.clone(), report_not_found: true,
recursive: true, min_disks: disks_quorum,
report_not_found: true, agreed: Some(Box::new(move |entry: MetaCacheEntry| {
min_disks: disks_quorum, let entry_name = entry.name.clone();
walkdir_timeout: Some(SCANNER_LIST_PATH_RAW_TIMEOUT), let agreed_tx = agreed_tx.clone();
walkdir_stall_timeout: Some(SCANNER_LIST_PATH_RAW_TIMEOUT), Box::pin(async move {
agreed: Some(Box::new(move |entry: MetaCacheEntry| { if let Err(e) = agreed_tx.send(entry_name).await {
let entry_name = entry.name.clone(); error!(
let agreed_tx = agreed_tx.clone(); target: "rustfs::scanner::folder",
Box::pin(async move { event = EVENT_SCANNER_FOLDER_STATE,
if let Err(e) = agreed_tx.send(entry_name).await { component = LOG_COMPONENT_SCANNER,
error!( subsystem = LOG_SUBSYSTEM_FOLDER,
target: "rustfs::scanner::folder", entry = %entry.name,
event = EVENT_SCANNER_FOLDER_STATE, state = "list_path_agreed_send_failed",
component = LOG_COMPONENT_SCANNER, error = %e,
subsystem = LOG_SUBSYSTEM_FOLDER, "Scanner list_path_raw agreed callback failed"
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<DiskError>]| {
} let partial_tx = partial_tx.clone();
}) Box::pin(async move {
})), if let Err(e) = partial_tx.send(entries).await {
partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| { error!(
let partial_tx = partial_tx.clone(); target: "rustfs::scanner::folder",
Box::pin(async move { event = EVENT_SCANNER_FOLDER_STATE,
if let Err(e) = partial_tx.send(entries).await { component = LOG_COMPONENT_SCANNER,
error!( subsystem = LOG_SUBSYSTEM_FOLDER,
target: "rustfs::scanner::folder", state = "list_path_partial_send_failed",
event = EVENT_SCANNER_FOLDER_STATE, error = %e,
component = LOG_COMPONENT_SCANNER, "Scanner list_path_raw partial callback failed"
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<DiskError>]| {
} let finished_tx = finished_tx.clone();
}) let errs_clone = errs.to_vec();
})), Box::pin(async move {
finished: Some(Box::new(move |errs: &[Option<DiskError>]| { if let Err(e) = finished_tx.send(errs_clone).await {
let finished_tx = finished_tx.clone(); error!(
let errs_clone = errs.to_vec(); target: "rustfs::scanner::folder",
Box::pin(async move { event = EVENT_SCANNER_FOLDER_STATE,
if let Err(e) = finished_tx.send(errs_clone).await { component = LOG_COMPONENT_SCANNER,
error!( subsystem = LOG_SUBSYSTEM_FOLDER,
target: "rustfs::scanner::folder", state = "list_path_finished_send_failed",
event = EVENT_SCANNER_FOLDER_STATE, error = %e,
component = LOG_COMPONENT_SCANNER, "Scanner list_path_raw finished callback failed"
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 {
..Default::default() let _ = observer.send((
}, options.skip_walkdir_total_timeout,
) options.walkdir_timeout,
.await options.walkdir_stall_timeout,
{ ));
}
if let Err(e) = list_path_raw(child_ctx_clone.clone(), options).await {
if is_missing_path_disk_error(&e) { if is_missing_path_disk_error(&e) {
debug!( debug!(
target: "rustfs::scanner::folder", target: "rustfs::scanner::folder",
@@ -2820,6 +2841,8 @@ pub async fn scan_data_folder(
skip_heal, skip_heal,
local_disk, local_disk,
pending_heals_changed: false, pending_heals_changed: false,
#[cfg(test)]
list_path_raw_options_observer: None,
}; };
// Check if context is cancelled // Check if context is cancelled
@@ -3026,6 +3049,7 @@ mod tests {
skip_heal: Arc::new(AtomicBool::new(false)), skip_heal: Arc::new(AtomicBool::new(false)),
local_disk: disk, local_disk: disk,
pending_heals_changed: false, pending_heals_changed: false,
list_path_raw_options_observer: None,
}; };
(scanner, temp_dir) (scanner, temp_dir)
@@ -4210,6 +4234,8 @@ mod tests {
scanner.heal_object_select = 1; scanner.heal_object_select = 1;
scanner.disks = disks; scanner.disks = disks;
scanner.disks_quorum = 2; 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( scanner.old_cache.replace(
&format!("{bucket}/{object}"), &format!("{bucket}/{object}"),
bucket, bucket,
@@ -4231,6 +4257,11 @@ mod tests {
.expect("scan_folder should not hang after list_path_raw finishes") .expect("scan_folder should not hang after list_path_raw finishes")
.expect("scan_folder should finish successfully"); .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 let root = scanner
.new_cache .new_cache
.checked_flatten(bucket) .checked_flatten(bucket)