fix(capacity): skip idle scheduled disk scans (#6541)

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>

crates/object-capacity/README.md

@@ -171,7 +171,9 @@ Refreshing only dirty disks is safe only when:
 - which means the system has already completed at least one full refresh without partial errors
 - and the per-disk cache is fully populated
 
-If the per-disk cache is incomplete, or there are no dirty disks, the system falls back to a full refresh.
+If no aggregate cache exists yet, the system performs a full refresh to establish an initial value. Once an aggregate cache exists, a scheduled refresh with no dirty disks stays idle instead of repeatedly walking unchanged disks. If disks are dirty while the per-disk cache is incomplete, the system performs a full refresh because a subset cannot be merged safely.
+
+A full refresh that reaches its time budget may publish an estimated aggregate without establishing a complete per-disk baseline. That bounded estimate acknowledges dirty marks that predate the scan, while marks recorded during the scan remain pending. This prevents one old dirty mark from causing an endless timeout loop without treating the estimate as exact.
 
 ### Merge Rules After a Subset Refresh
 

crates/object-capacity/README_ZH.md

@@ -171,7 +171,9 @@ crate 不是“超时就直接失败”的设计：
 - 也就是系统已经完成过一次“无部分错误”的全盘刷新
 - 并且成功拿到了每盘缓存
 
-若当前还没有完整 per-disk cache，或者脏盘集合为空，就会回退到全盘刷新。
+若当前还没有任何聚合缓存，系统会先执行一次全盘刷新来建立初始值。已有聚合缓存后，定时刷新在没有脏盘时会保持 idle，不再重复遍历未变化的磁盘。若 per-disk cache 尚不完整但存在脏盘，系统仍执行全盘刷新，因为此时不能安全合并子集结果。
+
+全盘刷新达到时间预算后，可以发布估算聚合值，但不会借此建立完整 per-disk 基线。该有界估算会确认扫描开始前的脏标记；扫描过程中记录的新标记仍会保留。这样既不会把估算值当成精确值，也不会让一个旧脏标记永久触发超时循环。
 
 ### 子集刷新后的合并规则
 

crates/object-capacity/src/capacity_manager.rs

@@ -14,7 +14,7 @@
 
 //! Hybrid Capacity Manager for efficient capacity statistics
 
-use super::scan::refresh_capacity_with_scope;
+use super::scan::{ScheduledCapacityRefresh, refresh_capacity_with_scope, select_scheduled_capacity_refresh};
 use super::types::CapacityDiskRef;
 use crate::capacity_scope::{CapacityScope, CapacityScopeDisk, drain_global_dirty_scopes, take_capacity_scope};
 use futures::FutureExt;
@@ -1021,6 +1021,7 @@ impl HybridCapacityManager {
     /// remote or removed disks would otherwise stay marked forever and keep
     /// the dirty-disk gauge permanently non-zero (backlog#1020 S30).
     pub async fn retain_dirty_disks_within(&self, local: &HashSet<CapacityScopeDisk>) {
+        self.sync_global_dirty_scopes().await;
         let mut dirty_disks = self.dirty_disks.write().await;
         let before = dirty_disks.len();
         dirty_disks.retain(|disk, _| local.contains(disk));
@@ -1378,6 +1379,44 @@ where
     }
 }
 
+async fn run_scheduled_capacity_refresh(manager: Arc<HybridCapacityManager>, disks: Vec<CapacityDiskRef>) -> bool {
+    let start = Instant::now();
+    match select_scheduled_capacity_refresh(manager.as_ref(), &disks).await {
+        ScheduledCapacityRefresh::Idle => {
+            debug!(
+                event = EVENT_CAPACITY_REFRESH_SCHEDULED,
+                component = LOG_COMPONENT_CAPACITY,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                state = "skipped",
+                source = DataSource::Scheduled.as_metric_label(),
+                reason = "no_dirty_disks",
+                disk_count = disks.len(),
+                "capacity refresh scheduled"
+            );
+            true
+        }
+        ScheduledCapacityRefresh::Scan { disks, dirty_subset } => {
+            debug!(
+                event = EVENT_CAPACITY_REFRESH_SCHEDULED,
+                component = LOG_COMPONENT_CAPACITY,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                state = "started",
+                source = DataSource::Scheduled.as_metric_label(),
+                refresh_scope = if dirty_subset { "dirty_subset" } else { "full" },
+                disk_count = disks.len(),
+                enqueue_latency_ms = start.elapsed().as_millis() as u64,
+                "capacity refresh scheduled"
+            );
+            let result = manager
+                .refresh_or_join(DataSource::Scheduled, move || async move {
+                    refresh_capacity_with_scope(disks, dirty_subset).await
+                })
+                .await;
+            scheduled_refresh_was_clean(&result)
+        }
+    }
+}
+
 /// Owned capacity scheduler tasks for one server runtime.
 #[must_use = "capacity background tasks stop when their lifecycle handle is dropped"]
 pub struct CapacityBackgroundTasks {
@@ -1429,29 +1468,7 @@ pub async fn start_background_tasks(disks: Vec<CapacityDiskRef>) -> CapacityBack
 
     tasks.spawn(async move {
         run_scheduled_refresh_loop(refresh_interval, refresh_shutdown, move || {
-            let start = Instant::now();
+            run_scheduled_capacity_refresh(manager_for_refresh.clone(), disks.clone())
-            let manager = manager_for_refresh.clone();
-            let disks = disks.clone();
-            let disk_count = disks.len();
-            async move {
-                debug!(
-                    event = EVENT_CAPACITY_REFRESH_SCHEDULED,
-                    component = LOG_COMPONENT_CAPACITY,
-                    subsystem = LOG_SUBSYSTEM_RUNTIME,
-                    state = "started",
-                    source = DataSource::Scheduled.as_metric_label(),
-                    disk_count,
-                    enqueue_latency_ms = start.elapsed().as_millis() as u64,
-                    "capacity refresh scheduled"
-                );
-                let result = manager
-                    .refresh_or_join(
-                        DataSource::Scheduled,
-                        move || async move { refresh_capacity_with_scope(disks, false).await },
-                    )
-                    .await;
-                scheduled_refresh_was_clean(&result)
-            }
         })
         .await;
     });
@@ -1526,6 +1543,39 @@ mod tests {
         assert!(!scheduled_refresh_was_clean(&Err("scan failed".to_string())));
     }
 
+    #[tokio::test]
+    async fn test_scheduled_capacity_refresh_skips_clean_cache_then_scans_dirty_disk() {
+        let temp_dir = tempfile::TempDir::new().expect("capacity test directory should be created");
+        std::fs::write(temp_dir.path().join("object.bin"), b"capacity-bytes").expect("capacity fixture should be written");
+        let disk = CapacityDiskRef {
+            endpoint: "node-a".to_string(),
+            drive_path: temp_dir.path().display().to_string(),
+        };
+        let manager = create_isolated_manager(HybridStrategyConfig::default());
+        manager
+            .update_capacity(CapacityUpdate::estimated(123, 1), DataSource::RealTime)
+            .await;
+
+        assert!(run_scheduled_capacity_refresh(manager.clone(), vec![disk.clone()]).await);
+        let cached = manager.get_capacity().await.expect("cached capacity should remain available");
+        assert_eq!(cached.total_used, 123);
+        assert_eq!(cached.source, DataSource::RealTime);
+
+        manager
+            .mark_dirty_scope(&CapacityScope {
+                disks: vec![CapacityScopeDisk {
+                    endpoint: disk.endpoint.clone(),
+                    drive_path: disk.drive_path.clone(),
+                }],
+            })
+            .await;
+
+        assert!(run_scheduled_capacity_refresh(manager.clone(), vec![disk]).await);
+        let cached = manager.get_capacity().await.expect("dirty refresh should update the cache");
+        assert_eq!(cached.source, DataSource::Scheduled);
+        assert!(manager.get_dirty_disks().await.is_empty());
+    }
+
     #[tokio::test(start_paused = true)]
     async fn test_scheduled_refresh_loop_applies_backoff_and_reset() {
         use std::collections::VecDeque;

crates/object-capacity/src/scan.rs

@@ -73,6 +73,15 @@ struct CapacityScanReport {
     per_disk: Vec<DiskCapacityScanResult>,
 }
 
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) enum ScheduledCapacityRefresh {
+    Idle,
+    Scan {
+        disks: Vec<CapacityDiskRef>,
+        dirty_subset: bool,
+    },
+}
+
 impl CapacityScanReport {
     fn into_capacity_update(self, expected_disk_count: usize, replaces_disk_cache: bool) -> CapacityUpdate {
         let mut update = if self.summary.is_estimated {
@@ -94,7 +103,13 @@ impl CapacityScanReport {
                 .collect();
             // Skipped metadata or timeout fallback estimates can update an
             // existing complete cache, but must not establish a new baseline.
-            if !self.summary.timed_out && !self.summary.metadata_incomplete {
+            if self.summary.timed_out {
+                // A timeout estimate is still the best bounded refresh for a
+                // disk too large to enumerate. Acknowledge dirty marks that
+                // predate this attempt so an idle disk does not loop forever;
+                // update_capacity preserves marks recorded during the scan.
+                update.clear_dirty_disks = update.per_disk.iter().map(|entry| entry.disk.clone()).collect();
+            } else if !self.summary.metadata_incomplete {
                 update.expected_disk_count = Some(expected_disk_count);
                 update.replaces_disk_cache = replaces_disk_cache;
                 update.clear_dirty_disks = update.per_disk.iter().map(|entry| entry.disk.clone()).collect();
@@ -329,23 +344,33 @@ pub(crate) async fn calculate_data_dir_used_capacity(
     Ok(calculate_data_dir_used_capacity_report(disks).await?.summary)
 }
 
-pub async fn select_capacity_refresh_disks(
+pub(crate) async fn select_scheduled_capacity_refresh(
     capacity_manager: &HybridCapacityManager,
     disks: &[CapacityDiskRef],
-) -> (Vec<CapacityDiskRef>, bool) {
+) -> ScheduledCapacityRefresh {
     // The write side marks every disk of an EC set dirty, including remote
     // peers, but only local disks are ever scanned and cleared — drop ghost
     // entries so the dirty gauge reflects local pending work (backlog#1020).
     let local_set: HashSet<CapacityScopeDisk> = disks.iter().map(disk_scope_key).collect();
     capacity_manager.retain_dirty_disks_within(&local_set).await;
 
-    if !capacity_manager.can_refresh_dirty_subset().await {
-        return (disks.to_vec(), false);
-    }
-
     let dirty_disks = capacity_manager.get_dirty_disks().await;
     if dirty_disks.is_empty() {
-        return (disks.to_vec(), false);
+        return if disks.is_empty() || capacity_manager.get_capacity().await.is_some() {
+            ScheduledCapacityRefresh::Idle
+        } else {
+            ScheduledCapacityRefresh::Scan {
+                disks: disks.to_vec(),
+                dirty_subset: false,
+            }
+        };
+    }
+
+    if !capacity_manager.can_refresh_dirty_subset().await {
+        return ScheduledCapacityRefresh::Scan {
+            disks: disks.to_vec(),
+            dirty_subset: false,
+        };
     }
 
     let dirty_set: HashSet<CapacityScopeDisk> = dirty_disks.into_iter().collect();
@@ -356,9 +381,27 @@ pub async fn select_capacity_refresh_disks(
         .collect();
 
     if selected.is_empty() || selected.len() >= disks.len() {
-        (disks.to_vec(), false)
+        ScheduledCapacityRefresh::Scan {
+            disks: disks.to_vec(),
+            dirty_subset: false,
+        }
     } else {
-        (selected, true)
+        ScheduledCapacityRefresh::Scan {
+            disks: selected,
+            dirty_subset: true,
+        }
+    }
+}
+
+pub async fn select_capacity_refresh_disks(
+    capacity_manager: &HybridCapacityManager,
+    disks: &[CapacityDiskRef],
+) -> (Vec<CapacityDiskRef>, bool) {
+    match select_scheduled_capacity_refresh(capacity_manager, disks).await {
+        // Preserve the public selector's historical contract. The background
+        // scheduler consumes the richer internal plan and can remain idle.
+        ScheduledCapacityRefresh::Idle => (disks.to_vec(), false),
+        ScheduledCapacityRefresh::Scan { disks, dirty_subset } => (disks, dirty_subset),
     }
 }
 
@@ -1606,21 +1649,27 @@ mod tests {
 
     #[test]
     fn test_into_capacity_update_incomplete_results_do_not_replace_disk_cache() {
-        for scan in [
+        for (scan, clears_dirty) in [
-            CapacityScanResult {
+            (
-                used_bytes: 100,
+                CapacityScanResult {
-                file_count: 10,
+                    used_bytes: 100,
-                is_estimated: true,
+                    file_count: 10,
-                metadata_incomplete: true,
+                    is_estimated: true,
-                ..Default::default()
+                    metadata_incomplete: true,
-            },
+                    ..Default::default()
-            CapacityScanResult {
+                },
-                used_bytes: 100,
+                false,
-                file_count: 10,
+            ),
-                is_estimated: true,
+            (
-                timed_out: true,
+                CapacityScanResult {
-                ..Default::default()
+                    used_bytes: 100,
-            },
+                    file_count: 10,
+                    is_estimated: true,
+                    timed_out: true,
+                    ..Default::default()
+                },
+                true,
+            ),
         ] {
             let disk = CapacityScopeDisk {
                 endpoint: "node-a".to_string(),
@@ -1641,7 +1690,11 @@ mod tests {
             assert_eq!(update.per_disk[0].disk, disk);
             assert_eq!(update.expected_disk_count, None);
             assert!(!update.replaces_disk_cache);
-            assert!(update.clear_dirty_disks.is_empty());
+            if clears_dirty {
+                assert_eq!(update.clear_dirty_disks, vec![disk]);
+            } else {
+                assert!(update.clear_dirty_disks.is_empty());
+            }
         }
     }
 
@@ -1712,6 +1765,29 @@ mod tests {
         assert_eq!(selected.len(), 2);
     }
 
+    #[tokio::test]
+    async fn test_scheduled_capacity_refresh_returns_idle_with_cached_incomplete_baseline() {
+        let manager = create_isolated_manager(HybridStrategyConfig::default());
+        manager
+            .update_capacity(CapacityUpdate::estimated(100, 10), DataSource::Scheduled)
+            .await;
+
+        let disks = vec![CapacityDiskRef {
+            endpoint: "disk-1".to_string(),
+            drive_path: "/tmp/disk-1".to_string(),
+        }];
+
+        assert_eq!(
+            select_scheduled_capacity_refresh(manager.as_ref(), &disks).await,
+            ScheduledCapacityRefresh::Idle
+        );
+
+        let (selected, dirty_subset) = select_capacity_refresh_disks(manager.as_ref(), &disks).await;
+        assert_eq!(selected, disks);
+        assert!(!dirty_subset);
+        assert!(!manager.can_refresh_dirty_subset().await);
+    }
+
     #[tokio::test]
     async fn test_select_capacity_refresh_disks_returns_dirty_subset_when_cache_complete() {
         let manager = create_isolated_manager(HybridStrategyConfig::default());