fix(object-capacity): bound wedged scans with a hard outer timeout and cap joiner waits (#4533)

The whole disk traversal runs inside spawn_blocking with no timeout
anywhere on the async side; the in-scan ProgressMonitor checks are
cooperative and only run between walker entries. A stat/readdir blocked
on a dying disk or hung NFS mount therefore never returns: the blocking
thread leaks, the refresh singleflight stays running forever, every
subsequent scheduled refresh is skipped as inflight, and every admin
capacity query joins an unbounded wait until process restart (S02).

- Wrap each disk scan in tokio::time::timeout with a hard wall-clock
  ceiling of 2x the cooperative budget (min 5s). On expiry the caller
  fails the disk scan (releasing the singleflight through the normal
  error path, where the degraded/partial machinery from backlog#1014
  keeps the failed disk's last-known value) and a shared AtomicBool asks
  the blocking walker to exit at its next entry, bounding the thread
  leak to the single wedged syscall.
- Bound refresh_or_join joiner waits at 5 minutes so admin queries
  degrade into a clear error instead of hanging if the leader wedges in
  a way the drop/panic guards don't cover.

Ref: rustfs/backlog#1017 (S02 from audit rustfs/backlog#1010)

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::<Result<CapacityUpdate, String>>().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() {

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<CapacityScanResult, std::io::Error> {
     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<CapacityScanResult, std::io::Error> {
+fn scan_dir_blocking(path: &Path, limits: &ScanLimits, cancelled: &AtomicBool) -> Result<CapacityScanResult, std::io::Error> {
     let ScanLimits {
         max_files_threshold,
         base_timeout,
@@ -745,6 +783,13 @@ fn scan_dir_blocking(path: &Path, limits: &ScanLimits) -> Result<CapacityScanRes
             .into_iter();
 
         for entry_result in walker {
+            if cancelled.load(Ordering::Relaxed) {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::TimedOut,
+                    format!("capacity scan cancelled after hard budget expiry ({file_count} files seen)"),
+                ));
+            }
+
             let entry = match entry_result {
                 Ok(entry) => 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);