fix(object-capacity): harden write timing, interval bounds, and scope expiry (#4575)

fix(object-capacity): harden clock handling in write window, background intervals and scope registry

Three low-severity robustness gaps (S32+S33+#35):

- WriteRecord keyed its 60-bucket write window by wall-clock unix
  seconds while the debounce used Instant. An NTP step backwards marked
  recent buckets as future and silently suppressed write-triggered
  refreshes until the wall clock caught up; a forward step aged the
  whole window at once. Bucket keys now derive from a monotonic
  per-record epoch, immune to clock steps.
- Background refresh/metrics intervals were only clamped at zero;
  RUSTFS_CAPACITY_SCHEDULED_INTERVAL=u64::MAX overflowed
  Instant + Duration and panicked the spawned task, silently killing
  scheduled refreshes. Intervals now clamp into [1s, 30 days] via one
  helper with a structured warn.
- record_capacity_scope merged new disks into an expired entry and
  refreshed its recorded_at, resurrecting a scope take_capacity_scope
  would have discarded (and the merge path never pruned). Expired
  entries are now replaced, not merged into.

Ref: rustfs/backlog#1022 (S32+S33+#35 from audit rustfs/backlog#1010)

crates/object-capacity/src/capacity_manager.rs

@@ -37,7 +37,7 @@ use std::collections::{HashMap, HashSet};
 use std::future::Future;
 use std::panic::AssertUnwindSafe;
 use std::sync::Arc;
-use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+use std::time::{Duration, Instant};
 use tokio::sync::{Mutex, RwLock, watch};
 use tracing::{debug, info, warn};
 
@@ -445,6 +445,11 @@ const WRITE_WINDOW_BUCKETS: usize = WRITE_WINDOW_SECS as usize;
 /// leader wedges in a way the guards don't cover (backlog#1017).
 const REFRESH_JOINER_WAIT_TIMEOUT: Duration = Duration::from_secs(300);
 
+/// Upper bound for background refresh/metrics intervals: 30 days. Far beyond
+/// any sane configuration, but small enough that `Instant + Duration` can
+/// never overflow.
+const MAX_BACKGROUND_INTERVAL: Duration = Duration::from_secs(30 * 24 * 60 * 60);
+
 #[derive(Clone, Copy, Debug, Default)]
 struct WriteBucket {
     second: u64,
@@ -460,14 +465,24 @@ pub struct WriteRecord {
     pub write_count: usize,
     /// Fixed-size time buckets for the recent write window.
     write_buckets: [WriteBucket; WRITE_WINDOW_BUCKETS],
+    /// Monotonic origin for bucket keys. Wall-clock keys made an NTP step
+    /// backwards mark recent buckets as "future" and silently suppress
+    /// write-triggered refreshes until the clock catches up (backlog#1022 S32).
+    epoch: Instant,
 }
 
 impl WriteRecord {
-    fn current_unix_second() -> u64 {
-        SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .unwrap_or(Duration::ZERO)
-            .as_secs()
+    fn new() -> Self {
+        Self {
+            last_write_time: None,
+            write_count: 0,
+            write_buckets: [WriteBucket::default(); WRITE_WINDOW_BUCKETS],
+            epoch: Instant::now(),
+        }
+    }
+
+    fn monotonic_second(&self) -> u64 {
+        self.epoch.elapsed().as_secs()
     }
 
     fn recent_write_count(&self, now_second: u64) -> usize {
@@ -481,7 +496,7 @@ impl WriteRecord {
     }
 
     fn record_write(&mut self, now: Instant) -> usize {
-        let now_second = Self::current_unix_second();
+        let now_second = self.monotonic_second();
         let bucket_idx = (now_second % WRITE_WINDOW_BUCKETS as u64) as usize;
         let bucket = &mut self.write_buckets[bucket_idx];
 
@@ -665,11 +680,7 @@ impl HybridCapacityManager {
     pub fn new(config: HybridStrategyConfig) -> Self {
         Self {
             cache: Arc::new(RwLock::new(None)),
-            write_record: Arc::new(RwLock::new(WriteRecord {
-                last_write_time: None,
-                write_count: 0,
-                write_buckets: [WriteBucket::default(); WRITE_WINDOW_BUCKETS],
-            })),
+            write_record: Arc::new(RwLock::new(WriteRecord::new())),
             dirty_disks: Arc::new(RwLock::new(HashMap::new())),
             disk_cache: Arc::new(RwLock::new(HashMap::new())),
             disk_cache_complete: Arc::new(RwLock::new(false)),
@@ -863,7 +874,7 @@ impl HybridCapacityManager {
             }
 
             let write_record = self.write_record.read().await;
-            let write_frequency = write_record.recent_write_count(WriteRecord::current_unix_second());
+            let write_frequency = write_record.recent_write_count(write_record.monotonic_second());
             if write_frequency <= self.config.write_frequency_threshold {
                 return false;
             }
@@ -913,7 +924,7 @@ impl HybridCapacityManager {
     #[allow(dead_code)]
     pub async fn get_write_frequency(&self) -> usize {
         let record = self.write_record.read().await;
-        record.recent_write_count(WriteRecord::current_unix_second())
+        record.recent_write_count(record.monotonic_second())
     }
 
     /// Snapshot the currently dirty disks recorded from write-side scope propagation.
@@ -1209,6 +1220,28 @@ pub fn create_isolated_manager(config: HybridStrategyConfig) -> Arc<HybridCapaci
 }
 
 /// Start background update task
+/// Clamp a background interval into [1s, 30 days].
+///
+/// Zero panics tokio::time::interval; absurd values (e.g. u64::MAX seconds)
+/// overflow `Instant + Duration` and panic the spawned task, silently killing
+/// scheduled refreshes and runtime metrics (backlog#1022 S33).
+fn clamp_background_interval(value: Duration, env_var: &'static str) -> Duration {
+    let clamped = value.clamp(Duration::from_secs(1), MAX_BACKGROUND_INTERVAL);
+    if clamped != value {
+        warn!(
+            event = EVENT_CAPACITY_REFRESH_INTERVAL_CLAMPED,
+            component = LOG_COMPONENT_CAPACITY,
+            subsystem = LOG_SUBSYSTEM_RUNTIME,
+            result = "clamped",
+            env_var,
+            configured_secs = value.as_secs(),
+            effective_secs = clamped.as_secs(),
+            "capacity refresh interval clamped"
+        );
+    }
+    clamped
+}
+
 pub async fn start_background_task(disks: Vec<CapacityDiskRef>) {
     let manager = get_capacity_manager();
     let manager_for_refresh = manager.clone();
@@ -1216,35 +1249,8 @@ pub async fn start_background_task(disks: Vec<CapacityDiskRef>) {
     let mut refresh_interval = manager.get_config().scheduled_update_interval;
     let mut metrics_interval = manager.get_config().metrics_interval;
 
-    // Prevent panic in tokio::time::interval when misconfigured to 0
-    if refresh_interval.is_zero() {
-        warn!(
-            event = EVENT_CAPACITY_REFRESH_INTERVAL_CLAMPED,
-            component = LOG_COMPONENT_CAPACITY,
-            subsystem = LOG_SUBSYSTEM_RUNTIME,
-            result = "clamped",
-            env_var = ENV_CAPACITY_SCHEDULED_INTERVAL,
-            configured_secs = 0,
-            effective_secs = 1,
-            reason = "zero_interval",
-            "capacity refresh interval clamped"
-        );
-        refresh_interval = Duration::from_secs(1);
-    }
-    if metrics_interval.is_zero() {
-        warn!(
-            event = EVENT_CAPACITY_REFRESH_INTERVAL_CLAMPED,
-            component = LOG_COMPONENT_CAPACITY,
-            subsystem = LOG_SUBSYSTEM_RUNTIME,
-            result = "clamped",
-            env_var = ENV_CAPACITY_METRICS_INTERVAL,
-            configured_secs = 0,
-            effective_secs = 1,
-            reason = "zero_interval",
-            "capacity refresh interval clamped"
-        );
-        metrics_interval = Duration::from_secs(1);
-    }
+    refresh_interval = clamp_background_interval(refresh_interval, ENV_CAPACITY_SCHEDULED_INTERVAL);
+    metrics_interval = clamp_background_interval(metrics_interval, ENV_CAPACITY_METRICS_INTERVAL);
 
     tokio::spawn(async move {
         let mut timer = tokio::time::interval_at(tokio::time::Instant::now() + refresh_interval, refresh_interval);
@@ -1436,6 +1442,26 @@ mod tests {
         assert_eq!(manager.get_write_frequency().await, 20);
     }
 
+    #[test]
+    fn test_clamp_background_interval_bounds() {
+        // Zero panics tokio interval; u64::MAX seconds overflows Instant +
+        // Duration and used to panic the spawned background task.
+        assert_eq!(
+            clamp_background_interval(Duration::ZERO, ENV_CAPACITY_SCHEDULED_INTERVAL),
+            Duration::from_secs(1)
+        );
+        assert_eq!(
+            clamp_background_interval(Duration::from_secs(u64::MAX), ENV_CAPACITY_SCHEDULED_INTERVAL),
+            MAX_BACKGROUND_INTERVAL
+        );
+        assert_eq!(
+            clamp_background_interval(Duration::from_secs(120), ENV_CAPACITY_SCHEDULED_INTERVAL),
+            Duration::from_secs(120)
+        );
+        // The cap itself must be safely addable to a tokio Instant.
+        let _ = tokio::time::Instant::now() + MAX_BACKGROUND_INTERVAL;
+    }
+
     #[test]
     #[serial]
     fn test_recent_write_count_ignores_future_buckets() {
@@ -1443,6 +1469,7 @@ mod tests {
             last_write_time: None,
             write_count: 1,
             write_buckets: [WriteBucket::default(); WRITE_WINDOW_BUCKETS],
+            epoch: Instant::now(),
         };
 
         record.write_buckets[0] = WriteBucket { second: 120, count: 3 };

crates/object-capacity/src/capacity_scope.rs

@@ -83,8 +83,15 @@ pub fn record_capacity_scope(token: Uuid, scope: CapacityScope) {
         enforce_hard_limit(&mut entries, CAPACITY_SCOPE_REGISTRY_HARD_LIMIT);
     }
     if let Some(entry) = entries.get_mut(&token) {
-        merge_capacity_scopes(&mut entry.scope, scope);
-        entry.recorded_at = now;
+        // An expired entry would already be discarded by take_capacity_scope;
+        // merging into it (and refreshing recorded_at) would resurrect stale
+        // disks. Replace it with the fresh scope instead (backlog#1022 #35).
+        if now.duration_since(entry.recorded_at) > CAPACITY_SCOPE_TTL {
+            *entry = CapacityScopeEntry { scope, recorded_at: now };
+        } else {
+            merge_capacity_scopes(&mut entry.scope, scope);
+            entry.recorded_at = now;
+        }
     } else {
         entries.insert(token, CapacityScopeEntry { scope, recorded_at: now });
     }
@@ -156,6 +163,41 @@ mod tests {
         .join();
     }
 
+    #[test]
+    fn record_capacity_scope_replaces_expired_entry_instead_of_merging() {
+        let _guard = test_lock().lock().expect("test lock poisoned");
+        clear_capacity_scope_registry_for_test();
+        let token = Uuid::new_v4();
+        let stale_disk = CapacityScopeDisk {
+            endpoint: "node-old".to_string(),
+            drive_path: "/tmp/disk-old".to_string(),
+        };
+        let fresh_disk = CapacityScopeDisk {
+            endpoint: "node-new".to_string(),
+            drive_path: "/tmp/disk-new".to_string(),
+        };
+
+        record_capacity_scope(token, CapacityScope { disks: vec![stale_disk] });
+        // Backdate the entry beyond the TTL: take_capacity_scope would drop
+        // it, so a new record for the same token must not resurrect it.
+        capacity_scope_registry()
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .get_mut(&token)
+            .expect("entry must exist")
+            .recorded_at = Instant::now() - CAPACITY_SCOPE_TTL - Duration::from_secs(1);
+
+        record_capacity_scope(
+            token,
+            CapacityScope {
+                disks: vec![fresh_disk.clone()],
+            },
+        );
+
+        assert_eq!(take_capacity_scope(token), Some(CapacityScope { disks: vec![fresh_disk] }));
+        clear_capacity_scope_registry_for_test();
+    }
+
     #[test]
     fn record_and_take_capacity_scope_round_trips() {
         let _guard = test_lock().lock().expect("test lock poisoned");