From 506cd156bb4aea221fd81bf45bb5ac9057e14574 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 8 Jul 2026 17:06:51 +0800 Subject: [PATCH] fix(scanner): scope long walk timeouts (#4376) * fix(scanner): scope long walk timeouts * fix(scanner): bound IAM config walks --------- Co-authored-by: Henry Guo --- .../lifecycle/tier_free_version_recovery.rs | 5 +- .../replication/replication_resyncer.rs | 10 +- .../ecstore/src/cache_value/metacache_set.rs | 29 +++++- crates/ecstore/src/cluster/rpc/remote_disk.rs | 32 ++++++- crates/ecstore/src/disk/disk_store.rs | 92 ++++++++++++++++++- crates/ecstore/src/disk/mod.rs | 22 +++++ crates/ecstore/src/store/list_objects.rs | 32 ++++++- crates/iam/src/store/object.rs | 10 +- .../protocols/src/swift/expiration_worker.rs | 6 +- crates/scanner/src/scanner_folder.rs | 3 + crates/storage-api/src/object.rs | 13 +++ 11 files changed, 236 insertions(+), 18 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs b/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs index 94cde8108..392aeb5d4 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs @@ -13,6 +13,7 @@ // limitations under the License. use std::sync::Arc; +use std::time::Duration; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; @@ -29,6 +30,7 @@ use rustfs_filemeta::FileInfo; pub const DEFAULT_FREE_VERSION_RECOVERY_LIMIT: usize = 1_000; const DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT: usize = 10_000; +const BACKGROUND_WALKDIR_TIMEOUT: Duration = Duration::from_secs(60); type ObjectInfoOrErr = StorageObjectInfoOrErr; type WalkOptions = StorageWalkOptions bool>; @@ -222,7 +224,8 @@ async fn list_tier_free_versions( limit: walk_scan_limit, marker: object_marker, ..Default::default() - }, + } + .with_walkdir_timeouts(BACKGROUND_WALKDIR_TIMEOUT), ) .await } diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 5ed81a5f6..146e2f8c0 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -84,6 +84,8 @@ use tokio::time::Duration as TokioDuration; use tokio_util::io::ReaderStream; use tokio_util::sync::CancellationToken; use tracing::{debug, error, instrument, trace, warn}; + +const BACKGROUND_WALKDIR_TIMEOUT: TokioDuration = TokioDuration::from_secs(60); use uuid::Uuid; const EVENT_RESYNC_STATUS_UPDATE_SKIPPED: &str = "replication_resync_status_update_skipped"; @@ -645,7 +647,13 @@ impl ReplicationResyncer { if let Err(err) = storage .clone() - .walk(cancellation_token.clone(), &opts.bucket, "", tx.clone(), WalkOptions::default()) + .walk( + cancellation_token.clone(), + &opts.bucket, + "", + tx.clone(), + WalkOptions::default().with_walkdir_timeouts(BACKGROUND_WALKDIR_TIMEOUT), + ) .await { error!( diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index ce120301f..d56cdc8bd 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -91,6 +91,10 @@ async fn take_fallback_candidate(fallback_items: &Arc> fallback_items.lock().await.pop_front() } +fn duration_millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + #[cfg(test)] #[derive(Clone)] pub(crate) enum TestReaderBehavior { @@ -117,6 +121,8 @@ pub struct ListPathRawOptions { pub report_not_found: bool, pub per_disk_limit: i32, pub skip_walkdir_total_timeout: bool, + pub walkdir_timeout: Option, + pub walkdir_stall_timeout: Option, pub agreed: Option, pub partial: Option, pub finished: Option, @@ -146,6 +152,8 @@ impl Clone for ListPathRawOptions { report_not_found: self.report_not_found, per_disk_limit: self.per_disk_limit, skip_walkdir_total_timeout: self.skip_walkdir_total_timeout, + walkdir_timeout: self.walkdir_timeout, + walkdir_stall_timeout: self.walkdir_stall_timeout, #[cfg(test)] test_reader_behaviors: self.test_reader_behaviors.clone(), #[cfg(test)] @@ -259,6 +267,8 @@ async fn list_path_raw_inner( forward_to: opts_clone.forward_to.clone(), limit: opts_clone.per_disk_limit, skip_total_timeout: opts_clone.skip_walkdir_total_timeout, + timeout_ms: opts_clone.walkdir_timeout.map(duration_millis), + stall_timeout_ms: opts_clone.walkdir_stall_timeout.map(duration_millis), ..Default::default() }; @@ -419,6 +429,8 @@ async fn list_path_raw_inner( forward_to: opts_clone.forward_to.clone(), limit: opts_clone.per_disk_limit, skip_total_timeout: opts_clone.skip_walkdir_total_timeout, + timeout_ms: opts_clone.walkdir_timeout.map(duration_millis), + stall_timeout_ms: opts_clone.walkdir_stall_timeout.map(duration_millis), ..Default::default() }, &mut wr, @@ -485,10 +497,19 @@ async fn list_path_raw_inner( } let revjob = spawn(async move { - #[cfg(test)] - let peek_timeout = opts.peek_timeout.unwrap_or_else(get_drive_walkdir_stall_timeout); - #[cfg(not(test))] - let peek_timeout = get_drive_walkdir_stall_timeout(); + let peek_timeout = opts + .walkdir_stall_timeout + .or({ + #[cfg(test)] + { + opts.peek_timeout + } + #[cfg(not(test))] + { + None + } + }) + .unwrap_or_else(get_drive_walkdir_stall_timeout); let mut errs: Vec> = Vec::with_capacity(readers.len()); for _ in 0..readers.len() { errs.push(None); diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 177f39f3d..0c63d624e 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -2018,14 +2018,14 @@ impl DiskAPI for RemoteDisk { let disk = self.disk_ref().await; let body = serde_json::to_vec(&opts)?; - let stall_timeout = get_drive_walkdir_stall_timeout(); + let stall_timeout = opts.stall_timeout_duration().unwrap_or_else(get_drive_walkdir_stall_timeout); let bucket = opts.bucket.clone(); let base_dir = opts.base_dir.clone(); let disk_for_log = disk.clone(); let timeout_duration = if opts.skip_total_timeout { Duration::ZERO } else { - get_drive_walkdir_timeout() + opts.timeout_duration().unwrap_or_else(get_drive_walkdir_timeout) }; self.execute_with_timeout_for_op_and_health_action( @@ -3771,6 +3771,34 @@ mod tests { } } + #[tokio::test] + async fn test_remote_disk_walk_dir_uses_per_request_stall_timeout() { + let transport = RecordingInternodeDataTransport::default(); + let remote_disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await; + let opts = WalkDirOptions { + bucket: "bucket".to_string(), + base_dir: "prefix".to_string(), + recursive: true, + stall_timeout_ms: Some(60_000), + ..Default::default() + }; + let mut writer = Vec::new(); + + remote_disk + .walk_dir(opts, &mut writer) + .await + .expect("walk_dir should be sent through configured data transport"); + + let calls = transport.calls(); + assert_eq!(calls.len(), 1); + match &calls[0] { + RecordedTransportCall::WalkDir(request) => { + assert_eq!(request.stall_timeout, Some(Duration::from_secs(60))); + } + other => panic!("expected walk-dir transport call, got {other:?}"), + } + } + #[tokio::test] async fn test_remote_disk_walk_dir_retries_once_on_retryable_transport_error() { let transport = RetryingWalkDirInternodeDataTransport::with_steps(vec![ diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index c984718b8..9c71b0c6c 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -1221,7 +1221,7 @@ impl DiskAPI for LocalDiskWrapper { let timeout_duration = if opts.skip_total_timeout { Duration::ZERO } else { - get_drive_walkdir_timeout() + opts.timeout_duration().unwrap_or_else(get_drive_walkdir_timeout) }; self.track_disk_health_with_op_and_timeout_action( @@ -1545,6 +1545,52 @@ mod tests { }); } + #[test] + fn drive_walkdir_timeout_uses_default_when_unset() { + temp_env::with_var_unset(rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, || { + temp_env::with_var_unset(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, || { + temp_env::with_var_unset(rustfs_config::ENV_DRIVE_TIMEOUT_PROFILE, || { + assert_eq!( + get_drive_walkdir_timeout(), + Duration::from_secs(rustfs_config::DEFAULT_DRIVE_WALKDIR_TIMEOUT_SECS) + ); + }); + }); + }); + } + + #[test] + fn drive_walkdir_stall_timeout_uses_default_when_unset() { + temp_env::with_var_unset(rustfs_config::ENV_DRIVE_WALKDIR_STALL_TIMEOUT_SECS, || { + temp_env::with_var_unset(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, || { + temp_env::with_var_unset(rustfs_config::ENV_DRIVE_TIMEOUT_PROFILE, || { + assert_eq!( + get_drive_walkdir_stall_timeout(), + Duration::from_secs(rustfs_config::DEFAULT_DRIVE_WALKDIR_STALL_TIMEOUT_SECS) + ); + }); + }); + }); + } + + #[test] + fn drive_walkdir_timeout_prefers_canonical_over_legacy() { + temp_env::with_var(rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("11"), || { + temp_env::with_var(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("17"), || { + assert_eq!(get_drive_walkdir_timeout(), Duration::from_secs(11)); + }); + }); + } + + #[test] + fn drive_walkdir_stall_timeout_prefers_canonical_over_legacy() { + temp_env::with_var(rustfs_config::ENV_DRIVE_WALKDIR_STALL_TIMEOUT_SECS, Some("13"), || { + temp_env::with_var(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("17"), || { + assert_eq!(get_drive_walkdir_stall_timeout(), Duration::from_secs(13)); + }); + }); + } + #[test] fn object_disk_read_timeout_uses_default_when_unset() { temp_env::with_var_unset(rustfs_config::ENV_OBJECT_DISK_READ_TIMEOUT, || { @@ -1795,6 +1841,50 @@ mod tests { .await; } + #[tokio::test] + async fn walk_dir_uses_per_request_timeout_before_env_default() { + temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("60"))], async { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = + Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8")).expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + let wrapper = LocalDiskWrapper::new(disk, false); + let bucket = "test-bucket"; + let object = "test-object"; + + wrapper.make_volume(bucket).await.expect("bucket should be created"); + + let mut file_info = FileInfo::new(&format!("{bucket}/{object}"), 1, 0); + file_info.volume = bucket.to_string(); + file_info.name = object.to_string(); + file_info.mod_time = Some(::time::OffsetDateTime::now_utc()); + file_info.erasure.index = 1; + + wrapper + .write_metadata("", bucket, object, file_info) + .await + .expect("object metadata should be written"); + + let mut writer = PendingWriter; + let result = wrapper + .walk_dir( + WalkDirOptions { + bucket: bucket.to_string(), + recursive: true, + timeout_ms: Some(10), + ..Default::default() + }, + &mut writer, + ) + .await; + + assert_eq!(result.expect_err("walk_dir should use per-request timeout"), DiskError::Timeout); + assert_eq!(wrapper.runtime_state(), RuntimeDriveHealthState::Online); + assert!(!wrapper.health.is_faulty()); + }) + .await; + } + #[tokio::test] async fn walk_dir_skip_total_timeout_keeps_stream_pending() { temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("1"))], async { diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index af630dd16..9e253f4c8 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -813,6 +813,24 @@ pub struct WalkDirOptions { // Skip the wrapper-level total timeout for long streaming walks. #[serde(default)] pub skip_total_timeout: bool, + + // Override the wrapper-level total timeout for long background walks. + #[serde(default)] + pub timeout_ms: Option, + + // Override the remote stream stall timeout for long background walks. + #[serde(default)] + pub stall_timeout_ms: Option, +} + +impl WalkDirOptions { + pub fn timeout_duration(&self) -> Option { + self.timeout_ms.map(std::time::Duration::from_millis) + } + + pub fn stall_timeout_duration(&self) -> Option { + self.stall_timeout_ms.map(std::time::Duration::from_millis) + } } #[derive(Clone, Debug, Default)] @@ -1046,6 +1064,8 @@ mod tests { limit: 100, disk_id: "disk-123".to_string(), skip_total_timeout: false, + timeout_ms: Some(10_000), + stall_timeout_ms: Some(20_000), }; assert_eq!(opts.bucket, "test-bucket"); @@ -1058,6 +1078,8 @@ mod tests { assert_eq!(opts.limit, 100); assert_eq!(opts.disk_id, "disk-123"); assert!(!opts.skip_total_timeout); + assert_eq!(opts.timeout_duration(), Some(std::time::Duration::from_secs(10))); + assert_eq!(opts.stall_timeout_duration(), Some(std::time::Duration::from_secs(20))); } /// Test DeleteOptions structure diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index f97db9eef..d6937c777 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -54,7 +54,7 @@ use std::sync::{ Arc, atomic::{AtomicU64, Ordering}, }; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::io::duplex; use tokio::sync::broadcast::{self}; use tokio::sync::mpsc::{self, Receiver, Sender}; @@ -70,6 +70,10 @@ const MAX_OBJECT_LIST: i32 = 1000; const METACACHE_SHARE_PREFIX: bool = false; +fn duration_millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + type ListObjectsInfo = StorageListObjectsInfo; type ListObjectsV2Info = StorageListObjectsV2Info; type ListObjectVersionsInfo = StorageListObjectVersionsInfo; @@ -252,6 +256,8 @@ pub struct ListPathOptions { pub set_idx: Option, pub cursor_source: Option, pub cursor_generation: Option, + pub walkdir_timeout: Option, + pub walkdir_stall_timeout: Option, } const MARKER_TAG_VERSION: &str = "v2"; @@ -2488,6 +2494,8 @@ struct ListingSupplementOptions { forward_to: Option, per_disk_limit: i32, skip_total_timeout: bool, + walkdir_timeout: Option, + walkdir_stall_timeout: Option, } struct ListingSupplement { @@ -2618,6 +2626,8 @@ async fn read_fallback_listing_disk( forward_to: options.forward_to, limit: options.per_disk_limit, skip_total_timeout: options.skip_total_timeout, + timeout_ms: options.walkdir_timeout.map(duration_millis), + stall_timeout_ms: options.walkdir_stall_timeout.map(duration_millis), ..Default::default() }, &mut wr, @@ -4224,6 +4234,8 @@ impl ECStore { forward_to: opts.marker.clone(), per_disk_limit: bounded_usize_to_i32(opts.limit), skip_total_timeout: false, + walkdir_timeout: opts.walkdir_timeout, + walkdir_stall_timeout: opts.walkdir_stall_timeout, }, fallback_disks.clone(), claim_tracker.clone(), @@ -4244,6 +4256,8 @@ impl ECStore { forward_to: opts.marker.clone(), min_disks: raw_min_disks, per_disk_limit: bounded_usize_to_i32(opts.limit), + walkdir_timeout: opts.walkdir_timeout, + walkdir_stall_timeout: opts.walkdir_stall_timeout, agreed: Some(Box::new(move |entry: MetaCacheEntry| { Box::pin({ let value = tx1.clone(); @@ -5364,6 +5378,8 @@ impl Sets { forward_to: opts.marker.clone(), per_disk_limit: bounded_usize_to_i32(opts.limit), skip_total_timeout: false, + walkdir_timeout: opts.walkdir_timeout, + walkdir_stall_timeout: opts.walkdir_stall_timeout, }, fallback_disks.clone(), claim_tracker.clone(), @@ -5384,6 +5400,8 @@ impl Sets { forward_to: opts.marker.clone(), min_disks: raw_min_disks, per_disk_limit: bounded_usize_to_i32(opts.limit), + walkdir_timeout: opts.walkdir_timeout, + walkdir_stall_timeout: opts.walkdir_stall_timeout, agreed: Some(Box::new(move |entry: MetaCacheEntry| { Box::pin({ let value = tx1.clone(); @@ -5953,6 +5971,8 @@ impl SetDisks { incl_deleted: true, recursive: true, versioned: true, + walkdir_timeout: opts.walkdir_timeout, + walkdir_stall_timeout: opts.walkdir_stall_timeout, ..Default::default() }, entry_tx, @@ -6215,6 +6235,8 @@ impl SetDisks { forward_to: opts.marker.clone(), per_disk_limit: limit, skip_total_timeout: false, + walkdir_timeout: opts.walkdir_timeout, + walkdir_stall_timeout: opts.walkdir_stall_timeout, }, fallback_disks.clone(), claim_tracker.clone(), @@ -6235,6 +6257,8 @@ impl SetDisks { forward_to: opts.marker, min_disks: raw_min_disks, per_disk_limit: limit, + walkdir_timeout: opts.walkdir_timeout, + walkdir_stall_timeout: opts.walkdir_stall_timeout, agreed: Some(Box::new(move |entry: MetaCacheEntry| { Box::pin({ let value = tx1.clone(); @@ -6597,6 +6621,8 @@ mod test { forward_to: None, per_disk_limit: 100, skip_total_timeout: true, + walkdir_timeout: None, + walkdir_stall_timeout: None, }, Arc::new(Vec::new()), FallbackClaimTracker::default(), @@ -6611,6 +6637,8 @@ mod test { forward_to: None, per_disk_limit: 0, skip_total_timeout: false, + walkdir_timeout: None, + walkdir_stall_timeout: None, }, Arc::new(Vec::new()), FallbackClaimTracker::default(), @@ -6658,6 +6686,8 @@ mod test { forward_to: None, per_disk_limit: 0, skip_total_timeout: false, + walkdir_timeout: None, + walkdir_stall_timeout: None, }, Arc::new(vec![disk]), FallbackClaimTracker::default(), diff --git a/crates/iam/src/store/object.rs b/crates/iam/src/store/object.rs index 3de123cc1..18cedfec9 100644 --- a/crates/iam/src/store/object.rs +++ b/crates/iam/src/store/object.rs @@ -38,7 +38,7 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, error, warn}; use crate::storage_api::object_store::{ - HTTPPreconditions, ListOperations as _, ObjectInfoOrErr as StorageObjectInfoOrErr, ObjectOperations, + HTTPPreconditions, ListOperations, ObjectInfoOrErr as StorageObjectInfoOrErr, ObjectOperations, }; pub static IAM_CONFIG_PREFIX: LazyLock = LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam")); @@ -65,6 +65,7 @@ type IamObjectOptions = ::ObjectOptions; const IAM_IDENTITY_FILE: &str = "identity.json"; const IAM_POLICY_FILE: &str = "policy.json"; const IAM_GROUP_MEMBERS_FILE: &str = "members.json"; +const BACKGROUND_WALKDIR_TIMEOUT: Duration = Duration::from_secs(60); fn get_user_identity_path(user: &str, user_type: UserType) -> String { let base_path: &str = match user_type { @@ -414,10 +415,9 @@ impl ObjectStore { let path = prefix.to_owned(); let sender_on_error = sender.clone(); tokio::spawn(async move { - if let Err(err) = store - .walk(ctx.clone(), Self::BUCKET_NAME, &path, tx, Default::default()) - .await - { + let walk_opts = + ::WalkOptions::default().with_walkdir_timeouts(BACKGROUND_WALKDIR_TIMEOUT); + if let Err(err) = store.walk(ctx.clone(), Self::BUCKET_NAME, &path, tx, walk_opts).await { let reason = classify_iam_system_path_failure_reason(&err); record_system_path_failure("iam_config", "walk", reason); error!( diff --git a/crates/protocols/src/swift/expiration_worker.rs b/crates/protocols/src/swift/expiration_worker.rs index e9f409404..b1952e7e3 100644 --- a/crates/protocols/src/swift/expiration_worker.rs +++ b/crates/protocols/src/swift/expiration_worker.rs @@ -718,13 +718,13 @@ mod tests { use std::sync::Mutex; type MetadataResult = SwiftResult>>; + type ResultQueue = Mutex>; #[derive(Default)] - #[allow(clippy::type_complexity)] struct MockExpirationObjectBackend { expiring_objects: Mutex>, - metadata_results: Mutex>, - delete_results: Mutex>>, + metadata_results: ResultQueue, + delete_results: ResultQueue>, deleted_objects: Mutex>, } diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index fdacc2479..706d74f4d 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -75,6 +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 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"; @@ -2397,6 +2398,8 @@ impl FolderScanner { 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(); diff --git a/crates/storage-api/src/object.rs b/crates/storage-api/src/object.rs index 6c68e33ca..ad88d620c 100644 --- a/crates/storage-api/src/object.rs +++ b/crates/storage-api/src/object.rs @@ -14,6 +14,7 @@ use std::fmt; use std::sync::Arc; +use std::time::Duration; use crate::replication::{ ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_statuses_map, version_purge_statuses_map, @@ -238,6 +239,8 @@ pub struct WalkOptions { pub versions_sort: WalkVersionsSortOrder, pub limit: usize, pub include_free_versions: bool, + pub walkdir_timeout: Option, + pub walkdir_stall_timeout: Option, } impl Default for WalkOptions { @@ -250,10 +253,20 @@ impl Default for WalkOptions { versions_sort: WalkVersionsSortOrder::default(), limit: 0, include_free_versions: false, + walkdir_timeout: None, + walkdir_stall_timeout: None, } } } +impl WalkOptions { + pub fn with_walkdir_timeouts(mut self, timeout: Duration) -> Self { + self.walkdir_timeout = Some(timeout); + self.walkdir_stall_timeout = Some(timeout); + self + } +} + #[async_trait::async_trait] pub trait ObjectIO: Send + Sync + fmt::Debug + 'static { type Error: std::error::Error + Send + Sync + 'static;