From 627c396649fa6b2bf858548057273e6e7d511edd Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sat, 18 Jul 2026 00:04:43 +0800 Subject: [PATCH] fix(scanner): allow usage snapshot save when existing timestamp is future-dated beyond clock tolerance (#4982) --- crates/data-usage/src/data_usage.rs | 36 +++++++++- crates/ecstore/src/data_usage/mod.rs | 101 +++++++++++++++++++++++++-- crates/scanner/src/scanner.rs | 92 +++++++++++++++++++++++- 3 files changed, 221 insertions(+), 8 deletions(-) diff --git a/crates/data-usage/src/data_usage.rs b/crates/data-usage/src/data_usage.rs index 86474139c..0373c8fd2 100644 --- a/crates/data-usage/src/data_usage.rs +++ b/crates/data-usage/src/data_usage.rs @@ -18,9 +18,27 @@ use std::{ collections::{HashMap, HashSet}, hash::{DefaultHasher, Hash, Hasher}, path::Path, - time::SystemTime, + time::{Duration, SystemTime}, }; +/// Maximum amount a persisted `last_update` may lead the local wall clock before the +/// persisted timestamp is treated as untrustworthy. +/// +/// Invariant: the "skip stale usage update" monotonicity check (incoming `last_update` +/// <= existing `last_update` => skip persisting) is only valid while the existing +/// timestamp could plausibly have been produced by a healthy clock. If the on-disk +/// snapshot is future-dated beyond this tolerance (NTP step-back, or scanner +/// leadership moving to a node with a slower clock), the comparison would skip every +/// save forever and freeze admin usage stats; callers must bypass the skip instead. +pub const USAGE_LAST_UPDATE_FUTURE_TOLERANCE: Duration = Duration::from_secs(5 * 60); + +/// Returns true when `existing_last_update` is ahead of `now` by more than +/// [`USAGE_LAST_UPDATE_FUTURE_TOLERANCE`], i.e. the persisted timestamp cannot be +/// trusted for staleness comparisons and a fresh snapshot save must be allowed. +pub fn usage_last_update_is_untrusted_future(existing_last_update: SystemTime, now: SystemTime) -> bool { + existing_last_update > now + USAGE_LAST_UPDATE_FUTURE_TOLERANCE +} + #[derive(Clone, Copy, Default, Debug, Serialize, Deserialize, PartialEq)] pub struct TierStats { pub total_size: u64, @@ -1297,6 +1315,22 @@ pub struct CompressionTotalInfo { mod tests { use super::*; + #[test] + fn test_usage_last_update_future_tolerance_boundary() { + let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + + // Within tolerance (including the exact boundary) the timestamp is trusted. + assert!(!usage_last_update_is_untrusted_future(now, now)); + assert!(!usage_last_update_is_untrusted_future(now - Duration::from_secs(60), now)); + assert!(!usage_last_update_is_untrusted_future(now + USAGE_LAST_UPDATE_FUTURE_TOLERANCE, now)); + + // Beyond tolerance the persisted timestamp is untrustworthy. + assert!(usage_last_update_is_untrusted_future( + now + USAGE_LAST_UPDATE_FUTURE_TOLERANCE + Duration::from_secs(1), + now + )); + } + #[test] fn test_data_usage_info_creation() { let mut info = DataUsageInfo::new(); diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index 7c7cf809e..c27e7a599 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -147,18 +147,36 @@ lazy_static::lazy_static! { ); } +/// Decide whether an incoming usage snapshot must be skipped as stale, given the local +/// wall clock `now`. Mirrors `stale_data_usage_update_reason` in +/// `crates/scanner/src/scanner.rs` — keep the two consistent. +/// +/// If the persisted `existing.last_update` is future-dated beyond +/// [`rustfs_data_usage::USAGE_LAST_UPDATE_FUTURE_TOLERANCE`] (clock step-back or a +/// slower-clock scanner leader), it is untrustworthy: the save is allowed so usage +/// stats cannot freeze forever. +fn stale_data_usage_persist_reason(incoming: &DataUsageInfo, existing: &DataUsageInfo, now: SystemTime) -> Option<&'static str> { + match (incoming.last_update, existing.last_update) { + (Some(new_ts), Some(existing_ts)) + if new_ts <= existing_ts && !rustfs_data_usage::usage_last_update_is_untrusted_future(existing_ts, now) => + { + Some("older_or_equal_last_update") + } + _ => None, + } +} + /// Store data usage info to backend storage #[instrument(skip(store))] pub async fn store_data_usage_in_backend(data_usage_info: DataUsageInfo, store: Arc) -> Result<(), Error> { // Prevent older data from overwriting newer persisted stats if let Ok(buf) = read_config(store.clone(), &DATA_USAGE_OBJ_NAME_PATH).await && let Ok(existing) = serde_json::from_slice::(&buf) - && let (Some(new_ts), Some(existing_ts)) = (data_usage_info.last_update, existing.last_update) - && new_ts <= existing_ts + && let Some(reason) = stale_data_usage_persist_reason(&data_usage_info, &existing, SystemTime::now()) { info!( - "Skip persisting data usage: incoming last_update {:?} <= existing {:?}", - new_ts, existing_ts + "Skip persisting data usage ({reason}): incoming last_update {:?} <= existing {:?}", + data_usage_info.last_update, existing.last_update ); return Ok(()); } @@ -1372,6 +1390,81 @@ mod tests { info } + fn usage_with_last_update(last_update: Option) -> DataUsageInfo { + DataUsageInfo { + last_update, + ..Default::default() + } + } + + #[test] + fn stale_data_usage_persist_reason_allows_newer_incoming() { + let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + let incoming = usage_with_last_update(Some(now)); + let existing = usage_with_last_update(Some(now - Duration::from_secs(60))); + assert_eq!(stale_data_usage_persist_reason(&incoming, &existing, now), None); + } + + #[test] + fn stale_data_usage_persist_reason_skips_older_or_equal_incoming() { + let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + let existing = usage_with_last_update(Some(now - Duration::from_secs(60))); + + let older = usage_with_last_update(Some(now - Duration::from_secs(120))); + assert_eq!( + stale_data_usage_persist_reason(&older, &existing, now), + Some("older_or_equal_last_update") + ); + + let equal = usage_with_last_update(existing.last_update); + assert_eq!( + stale_data_usage_persist_reason(&equal, &existing, now), + Some("older_or_equal_last_update") + ); + } + + #[test] + fn stale_data_usage_persist_reason_allows_save_when_existing_is_future_dated() { + // Existing snapshot timestamp beyond the clock tolerance is untrustworthy + // (clock step-back / slower-clock leader): the save must be allowed even + // though incoming <= existing, otherwise usage stats freeze forever. + let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + let existing = + usage_with_last_update(Some(now + rustfs_data_usage::USAGE_LAST_UPDATE_FUTURE_TOLERANCE + Duration::from_secs(1))); + let incoming = usage_with_last_update(Some(now)); + assert_eq!(stale_data_usage_persist_reason(&incoming, &existing, now), None); + } + + #[test] + fn stale_data_usage_persist_reason_skips_at_exact_tolerance_boundary() { + // Exactly at now + tolerance is still within the trusted window. + let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + let existing = usage_with_last_update(Some(now + rustfs_data_usage::USAGE_LAST_UPDATE_FUTURE_TOLERANCE)); + let incoming = usage_with_last_update(Some(now)); + assert_eq!( + stale_data_usage_persist_reason(&incoming, &existing, now), + Some("older_or_equal_last_update") + ); + } + + #[test] + fn stale_data_usage_persist_reason_preserves_none_handling() { + // This call site (unlike the scanner sibling) allows saves when either + // timestamp is missing — pin that behavior. + let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + + let incoming_none = usage_with_last_update(None); + let existing_some = usage_with_last_update(Some(now - Duration::from_secs(60))); + assert_eq!(stale_data_usage_persist_reason(&incoming_none, &existing_some, now), None); + + let incoming_some = usage_with_last_update(Some(now)); + let existing_none = usage_with_last_update(None); + assert_eq!(stale_data_usage_persist_reason(&incoming_some, &existing_none, now), None); + + let both_none = usage_with_last_update(None); + assert_eq!(stale_data_usage_persist_reason(&both_none, &usage_with_last_update(None), now), None); + } + #[tokio::test] async fn compute_usage_preserves_same_object_across_1000_entry_page_boundary() { let first_page = UsageVersionPage { diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index cdcaabcde..26cf044f3 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1041,9 +1041,25 @@ impl Drop for ScannerScanModeGuard { } } -fn stale_data_usage_update_reason(incoming: &DataUsageInfo, existing: &DataUsageInfo) -> Option<&'static str> { +/// Decide whether an incoming usage snapshot must be skipped as stale, given the local +/// wall clock `now`. Mirrors `stale_data_usage_persist_reason` in +/// `crates/ecstore/src/data_usage/mod.rs` — keep the two consistent. +/// +/// If the persisted `existing.last_update` is future-dated beyond +/// [`rustfs_data_usage::USAGE_LAST_UPDATE_FUTURE_TOLERANCE`] (clock step-back or a +/// slower-clock scanner leader), it is untrustworthy: the save is allowed so usage +/// stats cannot freeze forever. +fn stale_data_usage_update_reason( + incoming: &DataUsageInfo, + existing: &DataUsageInfo, + now: std::time::SystemTime, +) -> Option<&'static str> { match (incoming.last_update, existing.last_update) { - (Some(new_ts), Some(existing_ts)) if new_ts <= existing_ts => Some("older_or_equal_last_update"), + (Some(new_ts), Some(existing_ts)) + if new_ts <= existing_ts && !rustfs_data_usage::usage_last_update_is_untrusted_future(existing_ts, now) => + { + Some("older_or_equal_last_update") + } (None, Some(_)) => Some("missing_incoming_last_update"), _ => None, } @@ -1066,7 +1082,7 @@ pub async fn store_data_usage_in_backend( if let Ok(buf) = read_config(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await && let Ok(existing) = serde_json::from_slice::(&buf) - && let Some(reason) = stale_data_usage_update_reason(&data_usage_info, &existing) + && let Some(reason) = stale_data_usage_update_reason(&data_usage_info, &existing, std::time::SystemTime::now()) { debug!( target: "rustfs::scanner", @@ -1592,6 +1608,76 @@ mod tests { assert_eq!(saved.last_update, Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20))); } + fn usage_with_last_update(last_update: Option) -> DataUsageInfo { + DataUsageInfo { + last_update, + ..Default::default() + } + } + + #[test] + fn test_stale_data_usage_update_reason_allows_newer_incoming() { + let now = std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + let incoming = usage_with_last_update(Some(now)); + let existing = usage_with_last_update(Some(now - Duration::from_secs(60))); + assert_eq!(stale_data_usage_update_reason(&incoming, &existing, now), None); + } + + #[test] + fn test_stale_data_usage_update_reason_skips_older_or_equal_incoming() { + let now = std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + let existing = usage_with_last_update(Some(now - Duration::from_secs(60))); + + let older = usage_with_last_update(Some(now - Duration::from_secs(120))); + assert_eq!(stale_data_usage_update_reason(&older, &existing, now), Some("older_or_equal_last_update")); + + let equal = usage_with_last_update(existing.last_update); + assert_eq!(stale_data_usage_update_reason(&equal, &existing, now), Some("older_or_equal_last_update")); + } + + #[test] + fn test_stale_data_usage_update_reason_allows_save_when_existing_is_future_dated() { + // Existing snapshot timestamp beyond the clock tolerance is untrustworthy + // (clock step-back / slower-clock leader): the save must be allowed even + // though incoming <= existing, otherwise usage stats freeze forever. + let now = std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + let existing = + usage_with_last_update(Some(now + rustfs_data_usage::USAGE_LAST_UPDATE_FUTURE_TOLERANCE + Duration::from_secs(1))); + let incoming = usage_with_last_update(Some(now)); + assert_eq!(stale_data_usage_update_reason(&incoming, &existing, now), None); + } + + #[test] + fn test_stale_data_usage_update_reason_skips_at_exact_tolerance_boundary() { + // Exactly at now + tolerance is still within the trusted window. + let now = std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + let existing = usage_with_last_update(Some(now + rustfs_data_usage::USAGE_LAST_UPDATE_FUTURE_TOLERANCE)); + let incoming = usage_with_last_update(Some(now)); + assert_eq!( + stale_data_usage_update_reason(&incoming, &existing, now), + Some("older_or_equal_last_update") + ); + } + + #[test] + fn test_stale_data_usage_update_reason_preserves_none_handling() { + let now = std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + + let incoming_none = usage_with_last_update(None); + let existing_some = usage_with_last_update(Some(now - Duration::from_secs(60))); + assert_eq!( + stale_data_usage_update_reason(&incoming_none, &existing_some, now), + Some("missing_incoming_last_update") + ); + + let incoming_some = usage_with_last_update(Some(now)); + let existing_none = usage_with_last_update(None); + assert_eq!(stale_data_usage_update_reason(&incoming_some, &existing_none, now), None); + + let both_none = usage_with_last_update(None); + assert_eq!(stale_data_usage_update_reason(&both_none, &usage_with_last_update(None), now), None); + } + #[tokio::test] async fn test_store_data_usage_in_backend_keeps_backup_when_primary_save_fails() { let store = Arc::new(MemoryConfigStore::default());