diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 22260a879..a969588c8 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -80,7 +80,11 @@ use rustfs_data_usage::TierStats; use rustfs_filemeta::{ FileInfo, FileInfoOpts, NULL_VERSION_ID, RestoreStatusOps, TRANSITION_COMPLETE, get_file_info, is_restored_object_on_disk, }; -use rustfs_utils::{get_env_i64, get_env_usize, path::encode_dir_object, string::strings_has_prefix_fold}; +use rustfs_utils::{ + get_env_i64, get_env_usize, + path::encode_dir_object, + string::{parse_bool, strings_has_prefix_fold}, +}; use s3s::dto::{ BucketLifecycleConfiguration, DefaultRetention, ExpirationStatus, ObjectLockConfiguration, RestoreRequest, RestoreRequestType, RestoreStatus, Timestamp, @@ -140,6 +144,7 @@ pub const AMZ_ENCRYPTION_KMS: &str = "aws:kms"; pub const ERR_INVALID_STORAGECLASS: &str = "invalid tier."; const ENV_STALE_UPLOADS_EXPIRY: &str = "RUSTFS_API_STALE_UPLOADS_EXPIRY"; const ENV_STALE_UPLOADS_CLEANUP_INTERVAL: &str = "RUSTFS_API_STALE_UPLOADS_CLEANUP_INTERVAL"; +const ENV_TIER_FREE_VERSION_RECOVERY_ENABLED: &str = "RUSTFS_TIER_FREE_VERSION_RECOVERY_ENABLED"; const DEFAULT_STALE_UPLOADS_EXPIRY: StdDuration = StdDuration::from_secs(24 * 60 * 60); const DEFAULT_STALE_UPLOADS_CLEANUP_INTERVAL: StdDuration = StdDuration::from_secs(6 * 60 * 60); const TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL: StdDuration = StdDuration::from_secs(60); @@ -2084,7 +2089,51 @@ async fn abandon_manual_transition_recovery_lease(api: Arc, job_id: Uui Ok(()) } +fn tier_free_version_recovery_enabled() -> bool { + resolve_tier_free_version_recovery_enabled(env::var(ENV_TIER_FREE_VERSION_RECOVERY_ENABLED)) +} + +fn resolve_tier_free_version_recovery_enabled(value: Result) -> bool { + match value { + Ok(value) => match parse_bool(value.trim()) { + Ok(enabled) => enabled, + Err(_) => { + warn!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + env = ENV_TIER_FREE_VERSION_RECOVERY_ENABLED, + reason = "invalid_boolean", + "Invalid tier free-version recovery setting; using enabled default" + ); + true + } + }, + Err(env::VarError::NotPresent) => true, + Err(env::VarError::NotUnicode(_)) => { + warn!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + env = ENV_TIER_FREE_VERSION_RECOVERY_ENABLED, + "Non-Unicode tier free-version recovery setting; using enabled default" + ); + true + } + } +} + fn spawn_tier_free_version_recovery_once(api: Arc, started: &OnceLock<()>) -> Option> { + if !tier_free_version_recovery_enabled() { + warn!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + env = ENV_TIER_FREE_VERSION_RECOVERY_ENABLED, + "Tier free-version recovery disabled by configuration" + ); + return None; + } if started.set(()).is_err() { return None; } @@ -2256,6 +2305,7 @@ fn jitter_tier_free_version_recovery_delay(delay: StdDuration) -> StdDuration { struct TierFreeVersionRecoverySchedule { next_delay: StdDuration, idle_interval: StdDuration, + failure_interval: StdDuration, previous_run_duration: StdDuration, jitter_next_delay: bool, bucket_marker: Option, @@ -2268,6 +2318,7 @@ impl Default for TierFreeVersionRecoverySchedule { Self { next_delay: StdDuration::ZERO, idle_interval: TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL, + failure_interval: TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL, previous_run_duration: StdDuration::ZERO, jitter_next_delay: false, bucket_marker: None, @@ -2292,12 +2343,19 @@ impl TierFreeVersionRecoverySchedule { self.reset_idle_interval(); } - fn record_failure(&mut self, run_duration: StdDuration) { - self.reset_idle_interval(); - self.previous_run_duration = run_duration; + fn record_failure(&mut self, _run_duration: StdDuration) { + self.idle_interval = TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL; + self.next_delay = self.failure_interval; + self.failure_interval = + std::cmp::min(self.failure_interval.saturating_mul(2), TIER_FREE_VERSION_RECOVERY_MAX_IDLE_INTERVAL); + // Keep the full backoff even after a long failed run, so a run whose + // duration exceeds the interval cannot restart immediately. + self.previous_run_duration = StdDuration::ZERO; + self.jitter_next_delay = false; } fn record_success(&mut self, stats: &FreeVersionRecoveryStats, run_duration: StdDuration) { + self.failure_interval = TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL; if stats.enqueued > 0 || stats.failed > 0 { self.follow_up_sweep = true; } @@ -4476,11 +4534,12 @@ mod tests { manual_transition_duration_elapsed, manual_transition_has_more_after_limit, manual_transition_version_marker, mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, persist_manual_transition_page_checkpoint, recover_manual_transition_jobs, recover_manual_transition_jobs_once, - replication_state_for_delete, resolve_transition_queue_capacity, resolve_transition_queue_send_timeout, - resolve_transition_worker_count, resolve_transition_workers_absolute_max, run_tier_free_version_recovery_loop, - select_restore_s3_location, set_lifecycle_observability_observer, set_recovered_free_version_enqueue_observer, - should_defer_date_expiry_for_recent_config_update, should_reuse_lifecycle_delete_replication_state, - transitioned_cleanup_tuple, transitioned_object_delete_opts, wait_for_tier_free_version_recovery, + replication_state_for_delete, resolve_tier_free_version_recovery_enabled, resolve_transition_queue_capacity, + resolve_transition_queue_send_timeout, resolve_transition_worker_count, resolve_transition_workers_absolute_max, + run_tier_free_version_recovery_loop, select_restore_s3_location, set_lifecycle_observability_observer, + set_recovered_free_version_enqueue_observer, should_defer_date_expiry_for_recent_config_update, + should_reuse_lifecycle_delete_replication_state, transitioned_cleanup_tuple, transitioned_object_delete_opts, + wait_for_tier_free_version_recovery, }; #[cfg(feature = "test-util")] use super::{delete_free_version_remote_object_then, encode_dir_object, get_transitioned_object_reader_with_tier_manager}; @@ -4588,6 +4647,30 @@ mod tests { assert_eq!(schedule.next_delay.as_secs() / TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL.as_secs(), 10); } + #[test] + fn tier_free_version_recovery_failure_backoff_waits_after_completion_and_resets_after_success() { + let mut schedule = TierFreeVersionRecoverySchedule::default(); + + for expected in [60, 120, 240, 480, 600, 600] { + schedule.record_failure(StdDuration::from_secs(75)); + assert_eq!(schedule.next_delay, StdDuration::from_secs(expected)); + assert_eq!(schedule.previous_run_duration, StdDuration::ZERO); + } + + schedule.record_success(&free_version_recovery_stats(0, 0, false), StdDuration::ZERO); + schedule.record_failure(StdDuration::from_secs(75)); + assert_eq!(schedule.next_delay, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL); + assert_eq!(schedule.previous_run_duration, StdDuration::ZERO); + } + + #[test] + fn tier_free_version_recovery_enabled_setting_is_opt_out_and_fails_open() { + assert!(resolve_tier_free_version_recovery_enabled(Err(env::VarError::NotPresent))); + assert!(resolve_tier_free_version_recovery_enabled(Ok("true".to_string()))); + assert!(!resolve_tier_free_version_recovery_enabled(Ok(" false ".to_string()))); + assert!(resolve_tier_free_version_recovery_enabled(Ok("invalid".to_string()))); + } + #[test] fn tier_free_version_recovery_pagination_preserves_full_sweep_backoff() { let idle = free_version_recovery_stats(0, 0, false); @@ -7875,6 +7958,21 @@ mod tests { } } + #[tokio::test] + #[serial] + async fn tier_free_version_recovery_disabled_does_not_consume_start_guard() { + let (_disk_paths, ecstore) = setup_test_env().await; + let started = OnceLock::new(); + + temp_env::async_with_vars([(super::ENV_TIER_FREE_VERSION_RECOVERY_ENABLED, Some("false"))], async { + let recovery = super::spawn_tier_free_version_recovery_once(Arc::clone(&ecstore), &started); + + assert!(recovery.is_none()); + assert!(started.get().is_none()); + }) + .await; + } + #[tokio::test] #[serial] async fn tier_free_version_recovery_production_entrypoint_enqueues_seeded_item() { 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 3187376b8..ac45de861 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs @@ -33,7 +33,6 @@ 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); #[cfg(not(test))] const BACKGROUND_WALK_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); #[cfg(test)] @@ -42,6 +41,21 @@ const BACKGROUND_WALK_SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(100); type ObjectInfoOrErr = StorageObjectInfoOrErr; type WalkOptions = StorageWalkOptions bool>; +fn recovery_walk_options(limit: usize, marker: Option) -> WalkOptions { + WalkOptions { + include_free_versions: true, + limit, + marker, + // Total walk time scales with bucket size, so it is left unbounded + // (Duration::ZERO disables the wall-clock budget). Per-call progress + // stalls stay bounded by the drive-level stall budget inherited from + // `RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS`. + walkdir_timeout: Some(Duration::ZERO), + walkdir_stall_timeout: None, + ..Default::default() + } +} + #[cfg(test)] pub(super) enum RecoveryWalkTestAction { SendItemsThenError(Vec, crate::error::Error), @@ -406,20 +420,8 @@ pub(super) async fn list_tier_free_versions( } } - api.walk( - cancel, - &bucket_name, - "", - tx, - WalkOptions { - include_free_versions: true, - limit: walk_scan_limit, - marker: object_marker, - ..Default::default() - } - .with_walkdir_timeouts(BACKGROUND_WALKDIR_TIMEOUT), - ) - .await + api.walk(cancel, &bucket_name, "", tx, recovery_walk_options(walk_scan_limit, object_marker)) + .await } }); @@ -695,6 +697,16 @@ mod tests { ); } + #[test] + fn recovery_walk_disables_total_timeout_and_inherits_stall_timeout() { + let opts = recovery_walk_options(123, Some("marker".to_string())); + + assert_eq!(opts.limit, 123); + assert_eq!(opts.marker.as_deref(), Some("marker")); + assert_eq!(opts.walkdir_timeout, Some(Duration::ZERO)); + assert_eq!(opts.walkdir_stall_timeout, None); + } + #[test] fn scan_truncation_keeps_marker_after_nonrecoverable_window() { let mut page = FreeVersionRecoveryPage { diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index b48fd6ae0..dfce7dbf3 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -267,6 +267,23 @@ impl Clone for ListPathRawOptions { } } +fn walk_dir_options(opts: &ListPathRawOptions) -> WalkDirOptions { + WalkDirOptions { + bucket: opts.bucket.clone(), + base_dir: opts.path.clone(), + recursive: opts.recursive, + incl_deleted: opts.incl_deleted, + report_notfound: opts.report_not_found, + filter_prefix: opts.filter_prefix.clone(), + forward_to: opts.forward_to.clone(), + limit: opts.per_disk_limit, + skip_total_timeout: opts.skip_walkdir_total_timeout, + timeout_ms: opts.walkdir_timeout.map(duration_millis), + stall_timeout_ms: opts.walkdir_stall_timeout.map(duration_millis), + ..Default::default() + } +} + pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> disk::error::Result<()> { let rx = rx.child_token(); let _cancel_guard = rx.clone().drop_guard(); @@ -373,20 +390,7 @@ async fn list_path_raw_inner( None }; - let wakl_opts = WalkDirOptions { - bucket: opts_clone.bucket.clone(), - base_dir: opts_clone.path.clone(), - recursive: opts_clone.recursive, - incl_deleted: opts_clone.incl_deleted, - report_notfound: opts_clone.report_not_found, - filter_prefix: opts_clone.filter_prefix.clone(), - 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() - }; + let wakl_opts = walk_dir_options(&opts_clone); let mut need_fallback = false; let mut last_err = None; @@ -559,27 +563,7 @@ async fn list_path_raw_inner( } let fallback_walk_started = std::time::Instant::now(); - match disk - .as_ref() - .walk_dir( - WalkDirOptions { - bucket: opts_clone.bucket.clone(), - base_dir: opts_clone.path.clone(), - recursive: opts_clone.recursive, - incl_deleted: opts_clone.incl_deleted, - report_notfound: opts_clone.report_not_found, - filter_prefix: opts_clone.filter_prefix.clone(), - 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, - ) - .await - { + match disk.as_ref().walk_dir(walk_dir_options(&opts_clone), &mut wr).await { Ok(_r) => { rustfs_io_metrics::record_stage_duration( "metacache_walk_dir_fallback", @@ -1091,6 +1075,19 @@ mod tests { assert!(!is_benign_not_found_listing_failure(&[DiskError::DiskNotFound])); } + #[test] + fn walk_dir_options_preserve_zero_total_and_inherited_stall_timeouts() { + let options = walk_dir_options(&ListPathRawOptions { + walkdir_timeout: Some(Duration::ZERO), + walkdir_stall_timeout: None, + ..Default::default() + }); + + assert_eq!(options.timeout_ms, Some(0)); + assert_eq!(options.stall_timeout_ms, None); + assert!(!options.skip_total_timeout); + } + #[tokio::test] async fn list_path_raw_empty_disks_returns_read_quorum() { let err = list_path_raw(CancellationToken::new(), ListPathRawOptions::default()) diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 7f37be8d7..2ae0ed8d1 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -2063,7 +2063,7 @@ mod tests { } #[tokio::test] - async fn walk_dir_skip_total_timeout_keeps_stream_pending() { + async fn walk_dir_total_timeout_disable_modes_keep_stream_pending() { temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("1"))], async { let dir = tempfile::tempdir().expect("temp dir should be created"); let endpoint = @@ -2086,24 +2086,34 @@ mod tests { .await .expect("object metadata should be written"); - let mut writer = PendingWriter; - let result = tokio::time::timeout( - Duration::from_millis(20), - wrapper.walk_dir( + for (reason, options) in [ + ( + "skip_total_timeout", WalkDirOptions { bucket: bucket.to_string(), recursive: true, skip_total_timeout: true, ..Default::default() }, - &mut writer, ), - ) - .await; + ( + "zero per-request timeout", + WalkDirOptions { + bucket: bucket.to_string(), + recursive: true, + timeout_ms: Some(0), + stall_timeout_ms: None, + ..Default::default() + }, + ), + ] { + let mut writer = PendingWriter; + let result = tokio::time::timeout(Duration::from_millis(1_100), wrapper.walk_dir(options, &mut writer)).await; - assert!(result.is_err(), "skip_total_timeout should leave backpressured walk pending"); - assert_eq!(wrapper.runtime_state(), RuntimeDriveHealthState::Online); - assert!(!wrapper.health.is_faulty()); + assert!(result.is_err(), "{reason} should leave backpressured walk pending"); + assert_eq!(wrapper.runtime_state(), RuntimeDriveHealthState::Online); + assert!(!wrapper.health.is_faulty()); + } }) .await; }