diff --git a/crates/object-capacity/src/capacity_manager.rs b/crates/object-capacity/src/capacity_manager.rs index c4c8c49b5..4cf689ee4 100644 --- a/crates/object-capacity/src/capacity_manager.rs +++ b/crates/object-capacity/src/capacity_manager.rs @@ -50,6 +50,7 @@ const EVENT_CAPACITY_REFRESH_CACHE_UPDATED: &str = "capacity_refresh_cache_updat const EVENT_CAPACITY_REFRESH_WRITE_RECORDED: &str = "capacity_refresh_write_recorded"; const EVENT_CAPACITY_REFRESH_DEBOUNCE_STATE: &str = "capacity_refresh_debounce_state"; const EVENT_CAPACITY_REFRESH_PANIC: &str = "capacity_refresh_panic"; +const EVENT_CAPACITY_REFRESH_JOINER_TIMEOUT: &str = "capacity_refresh_joiner_timeout"; const EVENT_CAPACITY_REFRESH_CANCELLED: &str = "capacity_refresh_cancelled"; const EVENT_CAPACITY_REFRESH_RUNTIME_SUMMARY: &str = "capacity_refresh_runtime_summary"; const EVENT_CAPACITY_REFRESH_INTERVAL_CLAMPED: &str = "capacity_refresh_interval_clamped"; @@ -439,6 +440,14 @@ impl DataSource { const WRITE_WINDOW_SECS: u64 = 60; const WRITE_WINDOW_BUCKETS: usize = WRITE_WINDOW_SECS as usize; +/// Upper bound on how long a joiner waits for the in-flight refresh leader. +/// +/// A healthy full refresh finishes well within this (each disk scan is capped +/// by the outer wall-clock budget); the bound only exists so admin queries +/// return a clear error instead of hanging until process restart if the +/// leader wedges in a way the guards don't cover (backlog#1017). +const REFRESH_JOINER_WAIT_TIMEOUT: Duration = Duration::from_secs(300); + #[derive(Clone, Copy, Debug, Default)] struct WriteBucket { second: u64, @@ -927,11 +936,25 @@ impl HybridCapacityManager { if let Some(mut result_rx) = maybe_rx { // Wait until the leader publishes Some(result). Because we subscribed before - // releasing the mutex, we cannot miss the notification. - if result_rx.wait_for(|v| v.is_some()).await.is_err() { + // releasing the mutex, we cannot miss the notification. The wait is bounded so + // a wedged leader degrades admin queries into a clear error, never a hang. + match tokio::time::timeout(REFRESH_JOINER_WAIT_TIMEOUT, result_rx.wait_for(|v| v.is_some())).await { + Err(_) => { + warn!( + event = EVENT_CAPACITY_REFRESH_JOINER_TIMEOUT, + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_REFRESH, + result = "timeout", + source = source.as_metric_label(), + waited_ms = REFRESH_JOINER_WAIT_TIMEOUT.as_millis() as u64, + "capacity refresh joiner timed out waiting for the leader" + ); + return Err("timed out waiting for the in-flight capacity refresh to publish a result".to_string()); + } // The leader's sender was dropped (e.g. due to a panic) without publishing // a result. Surface a clear error rather than silently returning the default. - return Err("capacity refresh leader exited without publishing a result".to_string()); + Ok(Err(_)) => return Err("capacity refresh leader exited without publishing a result".to_string()), + Ok(Ok(_)) => {} } return result_rx .borrow() @@ -1896,6 +1919,34 @@ mod tests { assert!(!manager.can_refresh_dirty_subset().await); } + #[tokio::test(start_paused = true)] + #[serial] + async fn test_refresh_or_join_joiner_times_out_when_leader_wedges() { + let manager = create_isolated_manager(HybridStrategyConfig::default()); + + // A leader that never publishes (models a refresh wedged beyond what + // the drop/panic guards cover). + let leader_manager = manager.clone(); + let leader = tokio::spawn(async move { + leader_manager + .refresh_or_join(DataSource::Scheduled, || async { + futures::future::pending::>().await + }) + .await + }); + // Let the leader claim the singleflight slot before joining. + tokio::task::yield_now().await; + + // Paused time auto-advances past REFRESH_JOINER_WAIT_TIMEOUT: the + // joiner must surface a clear error instead of hanging forever. + let err = manager + .refresh_or_join(DataSource::RealTime, || async { Ok(CapacityUpdate::exact(1, 0)) }) + .await + .unwrap_err(); + assert!(err.contains("timed out"), "unexpected joiner error: {err}"); + leader.abort(); + } + #[tokio::test] #[serial] async fn test_refresh_or_join_returns_cluster_total_for_dirty_subset() { diff --git a/crates/object-capacity/src/scan.rs b/crates/object-capacity/src/scan.rs index ef0760a0c..f9f383fbf 100644 --- a/crates/object-capacity/src/scan.rs +++ b/crates/object-capacity/src/scan.rs @@ -27,6 +27,8 @@ use rustfs_io_metrics::capacity_metrics::{ use std::collections::HashSet; use std::ffi::OsStr; use std::path::{Component, Path, PathBuf}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; use tracing::{debug, info, warn}; use walkdir::WalkDir; @@ -49,6 +51,7 @@ const EVENT_CAPACITY_SCAN_TRAVERSAL_FAILED: &str = "capacity_scan_traversal_fail const EVENT_CAPACITY_SCAN_METADATA_FAILED: &str = "capacity_scan_metadata_failed"; const EVENT_CAPACITY_SCAN_SAMPLING_APPLIED: &str = "capacity_scan_sampling_applied"; const EVENT_CAPACITY_SCAN_EXACT_COMPLETED: &str = "capacity_scan_exact_completed"; +const EVENT_CAPACITY_SCAN_HARD_TIMEOUT: &str = "capacity_scan_hard_timeout"; const RUSTFS_META_BUCKET: &str = ".rustfs.sys"; const RUSTFS_META_TMP_DIR: &str = "tmp"; const RUSTFS_META_TMP_TRASH_DIR: &str = ".trash"; @@ -695,16 +698,51 @@ fn timeout_fallback_estimate( }) } +/// Hard wall-clock ceiling for one disk scan, enforced from async context. +/// +/// The in-scan `ProgressMonitor` checks are cooperative — they only run +/// between walker entries, so a stat/readdir blocked on a dying disk or hung +/// NFS mount never returns control to them. Twice the cooperative budget +/// leaves room for a slow-but-alive walker to reach its own timeout fallback +/// first; only a truly wedged scan trips the outer ceiling. +fn outer_scan_budget(limits: &ScanLimits) -> Duration { + limits.max_timeout.saturating_mul(2).max(Duration::from_secs(5)) +} + async fn get_dir_size_async(path: &Path) -> Result { let path = path.to_path_buf(); let limits = ScanLimits::from_env(); + let budget = outer_scan_budget(&limits); - tokio::task::spawn_blocking(move || scan_dir_blocking(&path, &limits)) - .await - .map_err(std::io::Error::other)? + // The blocking thread cannot be killed; on ceiling expiry the caller (and + // with it the refresh singleflight) is released with an error while + // `cancelled` asks the walker to exit at its next entry, bounding the + // thread leak to the single wedged syscall (backlog#1017 S02). + let cancelled = Arc::new(AtomicBool::new(false)); + let scan_cancelled = cancelled.clone(); + let scan = tokio::task::spawn_blocking(move || scan_dir_blocking(&path, &limits, &scan_cancelled)); + + match tokio::time::timeout(budget, scan).await { + Ok(join_result) => join_result.map_err(std::io::Error::other)?, + Err(_) => { + cancelled.store(true, Ordering::Relaxed); + warn!( + event = EVENT_CAPACITY_SCAN_HARD_TIMEOUT, + component = LOG_COMPONENT_CAPACITY, + subsystem = LOG_SUBSYSTEM_SCAN, + result = "hard_timeout", + budget_ms = budget.as_millis() as u64, + "capacity scan exceeded hard wall-clock budget; blocking walker asked to stop" + ); + Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + format!("capacity scan exceeded hard wall-clock budget of {budget:?}"), + )) + } + } } -fn scan_dir_blocking(path: &Path, limits: &ScanLimits) -> Result { +fn scan_dir_blocking(path: &Path, limits: &ScanLimits, cancelled: &AtomicBool) -> Result { let ScanLimits { max_files_threshold, base_timeout, @@ -745,6 +783,13 @@ fn scan_dir_blocking(path: &Path, limits: &ScanLimits) -> Result entry, Err(err) => { @@ -1329,10 +1374,34 @@ mod tests { // Zero budget times out at the first progress check with no samples; // a tempdir is not a dedicated mount, so no estimator exists and the // original timeout error must still surface (last-resort behavior). - let result = scan_dir_blocking(temp_dir.path(), &tight_limits(Duration::ZERO)); + let result = scan_dir_blocking(temp_dir.path(), &tight_limits(Duration::ZERO), &AtomicBool::new(false)); assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::TimedOut); } + #[test] + fn test_scan_dir_blocking_stops_when_cancelled() { + use std::fs::File; + use std::io::Write; + + let temp_dir = tempfile::TempDir::new().unwrap(); + let mut file = File::create(temp_dir.path().join("f0")).unwrap(); + file.write_all(b"x").unwrap(); + + // A pre-cancelled scan must exit at the first walker entry instead of + // completing, so a wedged-then-released walker doesn't keep scanning + // long after the outer budget already failed the refresh. + let result = scan_dir_blocking(temp_dir.path(), &tight_limits(Duration::from_secs(600)), &AtomicBool::new(true)); + assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::TimedOut); + } + + #[test] + fn test_outer_scan_budget_bounds() { + // Twice the cooperative ceiling, and never below the 5s floor even for + // degenerate max_timeout values. + assert_eq!(outer_scan_budget(&tight_limits(Duration::from_secs(15))), Duration::from_secs(30)); + assert_eq!(outer_scan_budget(&tight_limits(Duration::ZERO)), Duration::from_secs(5)); + } + #[test] fn test_scan_dir_blocking_generous_budget_still_exact() { use std::fs::File; @@ -1344,7 +1413,8 @@ mod tests { file.write_all(b"hello").unwrap(); } - let result = scan_dir_blocking(temp_dir.path(), &tight_limits(Duration::from_secs(600))).unwrap(); + let result = + scan_dir_blocking(temp_dir.path(), &tight_limits(Duration::from_secs(600)), &AtomicBool::new(false)).unwrap(); assert_eq!(result.used_bytes, 25); assert_eq!(result.file_count, 5); assert!(!result.is_estimated);