diff --git a/crates/data-usage/src/data_usage.rs b/crates/data-usage/src/data_usage.rs index 81a313125..dd8c3f158 100644 --- a/crates/data-usage/src/data_usage.rs +++ b/crates/data-usage/src/data_usage.rs @@ -236,6 +236,15 @@ pub struct DataUsageInfo { /// without relying on synchronized clocks. #[serde(default, skip_serializing_if = "Option::is_none")] pub usage_snapshot_authoritative_baseline: Option, + /// Per-set freshness for an observational aggregate. A set entry is + /// never sufficient to make the aggregate authoritative; it only records + /// which last-known-good generation contributed to the view. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub usage_snapshot_set_states: Vec, + /// An observational view may contain only the sets that completed this + /// cycle (or retained a compatible last-known-good cache). + #[serde(default)] + pub usage_snapshot_partial: bool, /// Deprecated kept here for backward compatibility reasons pub bucket_sizes: HashMap, /// Per-disk snapshot information when available @@ -252,6 +261,22 @@ pub struct DataUsageSnapshotIdentity { pub scanner_epoch: Option, } +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] +pub struct DataUsageSnapshotSetState { + pub pool_index: u64, + pub set_index: u64, + #[serde(default)] + pub scanner_cycle: Option, + #[serde(default)] + pub scanner_epoch: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scan_plan_digest: Option<[u8; 32]>, + #[serde(default)] + pub complete: bool, + #[serde(default)] + pub tombstone: bool, +} + impl DataUsageInfo { pub fn snapshot_identity(&self) -> DataUsageSnapshotIdentity { DataUsageSnapshotIdentity { @@ -291,7 +316,7 @@ pub fn data_usage_snapshot_is_newer(candidate: &DataUsageInfo, baseline: &DataUs /// rollback delete/recreate fences the previous bucket incarnation too. pub fn observed_data_usage_is_newer(observed: &DataUsageInfo, authoritative: &DataUsageInfo) -> bool { observed.usage_snapshot_converged == Some(false) - && observed.is_complete_bucket_usage_snapshot() + && (observed.is_complete_bucket_usage_snapshot() || observed.is_valid_partial_snapshot()) && observed.usage_snapshot_authoritative_baseline.as_ref() == Some(&authoritative.snapshot_identity()) && data_usage_snapshot_is_newer(observed, authoritative) } @@ -1436,6 +1461,39 @@ impl DataUsageInfo { && u64::try_from(self.buckets_usage.len()).ok() == Some(self.buckets_count) } + /// Validate provenance before an observational view can be selected for + /// admin display. Partial data is accepted only with unique set states, + /// a plan digest for every state, and at least one usable generation. + pub fn is_valid_partial_snapshot(&self) -> bool { + if !self.usage_snapshot_partial + || self.usage_snapshot_converged != Some(false) + || self.last_update.is_none() + || self.scanner_cycle.is_none() + || self.scanner_epoch.is_none() + || self.usage_snapshot_set_states.is_empty() + || u64::try_from(self.buckets_usage.len()).ok() != Some(self.buckets_count) + { + return false; + } + + let mut previous = None; + let mut plan_digest = None; + let mut has_source = false; + for state in &self.usage_snapshot_set_states { + if state.scan_plan_digest.is_none() + || plan_digest.is_some_and(|digest| Some(digest) != state.scan_plan_digest) + || state.scanner_cycle.is_some() != state.scanner_epoch.is_some() + || previous.is_some_and(|(pool, set)| (pool, set) >= (state.pool_index, state.set_index)) + { + return false; + } + previous = Some((state.pool_index, state.set_index)); + plan_digest = state.scan_plan_digest; + has_source |= state.scanner_cycle.is_some() && !state.tombstone; + } + has_source + } + /// Add object metadata to data usage statistics pub fn add_object(&mut self, object_path: &str, meta_object: &rustfs_filemeta::MetaObject) { // This method is kept for backward compatibility @@ -2263,6 +2321,55 @@ mod tests { assert!(!observed_data_usage_is_newer(&candidate(2, 9, Some(false), true), &authoritative)); assert!(!observed_data_usage_is_newer(&candidate(2, 11, Some(true), true), &authoritative)); assert!(!observed_data_usage_is_newer(&candidate(2, 11, Some(false), false), &authoritative)); + + let mut partial = candidate(2, 11, Some(false), false); + partial.usage_snapshot_partial = true; + partial.usage_snapshot_set_states = vec![DataUsageSnapshotSetState { + pool_index: 0, + set_index: 0, + scanner_cycle: Some(10), + scanner_epoch: Some(2), + scan_plan_digest: Some([1; 32]), + complete: false, + tombstone: false, + }]; + assert!(observed_data_usage_is_newer(&partial, &authoritative)); + } + + #[test] + fn mixed_topology_snapshot_is_rejected() { + let mut partial = DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(2)), + scanner_cycle: Some(11), + scanner_epoch: Some(2), + buckets_count: 0, + usage_snapshot_converged: Some(false), + usage_snapshot_partial: true, + usage_snapshot_set_states: vec![ + DataUsageSnapshotSetState { + pool_index: 0, + set_index: 0, + scanner_cycle: Some(11), + scanner_epoch: Some(2), + scan_plan_digest: Some([1; 32]), + complete: true, + tombstone: false, + }, + DataUsageSnapshotSetState { + pool_index: 1, + set_index: 0, + scanner_cycle: Some(10), + scanner_epoch: Some(2), + scan_plan_digest: Some([2; 32]), + complete: false, + tombstone: false, + }, + ], + ..Default::default() + }; + assert!(!partial.is_valid_partial_snapshot()); + partial.usage_snapshot_set_states[1].scan_plan_digest = Some([1; 32]); + assert!(partial.is_valid_partial_snapshot()); } #[test] diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index 3bf9bf500..11a1d78c3 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -73,6 +73,9 @@ struct CachedBucketUsage { // mutation. A strictly later generation is required before the mutation // evidence can be discarded. pending_scanner_position: Option<(u64, u64)>, + // Deletes are visible to admin immediately, but quota admission keeps + // them pending until a complete scanner generation reconciles the set. + pending_negative_delta: u64, } type UsageMemoryCache = Arc>>; @@ -948,7 +951,9 @@ async fn load_observed_data_usage_snapshot(store: Arc) -> Option Some(info), + Ok(info) + if info.usage_snapshot_converged == Some(false) + && (info.is_complete_bucket_usage_snapshot() || info.is_valid_partial_snapshot()) => Some(info), Ok(_) => { error!( event = "data_usage_snapshot_load_failed", @@ -993,7 +998,7 @@ async fn load_admin_data_usage_from_backend(store: Arc) -> Result CachedBucketUsage { dirty: false, stale_snapshot_pending: false, pending_scanner_position: None, + pending_negative_delta: 0, } } @@ -1808,6 +1815,7 @@ pub async fn record_bucket_object_delete_memory(bucket: &str, deleted_size: u64, .or_insert_with(|| cached_bucket_usage_now(BucketUsageInfo::default())); entry.usage.size = entry.usage.size.saturating_sub(deleted_size); + entry.pending_negative_delta = entry.pending_negative_delta.saturating_add(deleted_size); if removed_current_object { entry.usage.objects_count = entry.usage.objects_count.saturating_sub(1); entry.usage.versions_count = entry.usage.versions_count.saturating_sub(1); @@ -1863,7 +1871,7 @@ pub async fn get_bucket_usage_memory(bucket: &str) -> Option { cache .get(bucket) .filter(|cached| cached.authoritative) - .map(|cached| cached.usage.size) + .map(|cached| cached.usage.size.saturating_add(cached.pending_negative_delta)) } async fn update_usage_cache_if_needed() { @@ -2943,6 +2951,45 @@ mod tests { assert_eq!(selected.usage_snapshot_converged, Some(true)); } + #[test] + fn persisted_authoritative_stalls_but_memory_overlay_remains_visible() { + let authoritative = DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH), + scanner_epoch: Some(4), + scanner_cycle: Some(10), + usage_snapshot_complete: true, + ..Default::default() + }; + let mut partial = authoritative.clone(); + partial.last_update = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1)); + partial.scanner_cycle = Some(11); + partial.usage_snapshot_complete = false; + partial.usage_snapshot_partial = true; + partial.usage_snapshot_converged = Some(false); + partial.usage_snapshot_authoritative_baseline = Some(authoritative.snapshot_identity()); + partial.usage_snapshot_set_states = vec![rustfs_data_usage::DataUsageSnapshotSetState { + pool_index: 0, + set_index: 0, + scanner_cycle: Some(10), + scanner_epoch: Some(4), + scan_plan_digest: Some([1; 32]), + complete: false, + tombstone: false, + }]; + partial.buckets_usage.insert( + "bucket".to_string(), + BucketUsageInfo { + size: 100, + ..Default::default() + }, + ); + partial.buckets_count = 1; + + let (selected, _) = select_admin_data_usage_snapshot(authoritative, true, Some(partial)); + assert!(selected.usage_snapshot_partial); + assert_eq!(selected.buckets_usage.get("bucket").map(|usage| usage.size), Some(100)); + } + #[tokio::test] async fn authoritative_save_cleanup_removes_observed_snapshot_best_effort() { let store = UsageCasStore::default(); @@ -4665,6 +4712,47 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn partial_usage_is_observational_not_authoritative_for_quota() { + clear_usage_memory_cache_for_test().await; + + let mut partial = data_usage_info_for_test("bucket-a", 10, 100, SystemTime::now()); + partial.usage_snapshot_complete = false; + partial.usage_snapshot_partial = true; + replace_bucket_usage_memory_from_info(&partial).await; + + assert_eq!(get_bucket_usage_memory("bucket-a").await, None); + } + + #[tokio::test] + #[serial] + async fn stale_quota_uses_complete_baseline_plus_positive_deltas() { + clear_usage_memory_cache_for_test().await; + + let baseline = data_usage_info_for_test("bucket-a", 1, 100, SystemTime::now()); + replace_bucket_usage_memory_from_info(&baseline).await; + record_bucket_object_write_memory("bucket-a", None, 25).await; + + assert_eq!(get_bucket_usage_memory("bucket-a").await, Some(125)); + } + + #[tokio::test] + #[serial] + async fn negative_delta_waits_for_set_reconciliation() { + clear_usage_memory_cache_for_test().await; + + let baseline = data_usage_info_for_test("bucket-a", 1, 100, SystemTime::now()); + replace_bucket_usage_memory_from_info(&baseline).await; + record_bucket_object_delete_memory("bucket-a", 25, true).await; + + assert_eq!(get_bucket_usage_memory("bucket-a").await, Some(100)); + + let reconciled = data_usage_info_for_test("bucket-a", 0, 75, SystemTime::now() + Duration::from_secs(1)); + replace_bucket_usage_memory_from_info(&reconciled).await; + assert_eq!(get_bucket_usage_memory("bucket-a").await, Some(75)); + } + #[tokio::test] #[serial] async fn memory_overlay_counts_versioned_overwrite_as_new_version() { diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index c6ecdd489..6341cb535 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -29,7 +29,8 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS; pub use rustfs_data_usage::{ AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry, - PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path, prefix_usage_in_cache, + PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, DataUsageSnapshotSetState, hash_path, + prefix_usage_in_cache, }; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; use tokio::time::{Duration, Instant, sleep, timeout}; @@ -344,6 +345,18 @@ pub struct DataUsageCacheInfo { pub scan_plan_digest: Option, #[serde(default)] pub cache_key_format: u16, + /// Whether the entries retained while a set scan was incomplete come + /// from a prior complete set snapshot. This is observational input only. + #[serde(default)] + pub lkg_snapshot_complete: bool, + #[serde(default)] + pub lkg_next_cycle: Option, + #[serde(default)] + pub lkg_last_update: Option, + #[serde(default)] + pub lkg_leader_epoch: Option, + #[serde(default)] + pub lkg_scan_plan_digest: Option, } impl Serialize for DataUsageCacheInfo { @@ -353,7 +366,7 @@ impl Serialize for DataUsageCacheInfo { { // Keep this metadata map-encoded so older readers can ignore fields // appended by newer scanner versions during rolling upgrades. - let mut state = serializer.serialize_map(Some(16))?; + let mut state = serializer.serialize_map(Some(21))?; state.serialize_entry("name", &self.name)?; state.serialize_entry("next_cycle", &self.next_cycle)?; state.serialize_entry("leader_epoch", &self.leader_epoch)?; @@ -370,6 +383,11 @@ impl Serialize for DataUsageCacheInfo { state.serialize_entry("snapshot_complete", &self.snapshot_complete)?; state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?; state.serialize_entry("cache_key_format", &self.cache_key_format)?; + state.serialize_entry("lkg_snapshot_complete", &self.lkg_snapshot_complete)?; + state.serialize_entry("lkg_next_cycle", &self.lkg_next_cycle)?; + state.serialize_entry("lkg_last_update", &self.lkg_last_update)?; + state.serialize_entry("lkg_leader_epoch", &self.lkg_leader_epoch)?; + state.serialize_entry("lkg_scan_plan_digest", &self.lkg_scan_plan_digest)?; state.end() } } diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 059de1fb4..f7a043308 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -2274,9 +2274,8 @@ async fn final_data_usage_publication_defer_reason( } } ScannerCycleStatus::Deferred(reason) => Some(reason), - // Incomplete cycles do not publish a usage snapshot. Keep the - // decision permissive so existing partial-cycle handling remains - // unchanged if a future scanner path emits a bookkeeping update. + // Incomplete cycles may publish a non-authoritative observational + // snapshot when at least one set has a usable current/LKG view. ScannerCycleStatus::Incomplete => None, } } diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs index 17a550295..4f291cd02 100644 --- a/crates/scanner/src/scanner/usage_store.rs +++ b/crates/scanner/src/scanner/usage_store.rs @@ -198,7 +198,7 @@ where data_usage_info.usage_snapshot_authoritative_baseline = Some(authoritative.snapshot_identity()); } - if !data_usage_info.is_complete_bucket_usage_snapshot() { + if !data_usage_info.is_complete_bucket_usage_snapshot() && !data_usage_info.usage_snapshot_partial { error!( target: "rustfs::scanner", event = EVENT_SCANNER_PERSIST_STATE, diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index c104fb263..83cdaf14f 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -18,7 +18,8 @@ use crate::scanner_folder::{ScannerItem, scan_data_folder}; use crate::sleeper::SCANNER_SLEEPER; use crate::{ DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, DataUsageCache, DataUsageCacheInfo, DataUsageCachePrepareOutcome, - DataUsageCacheSource, DataUsageEntry, DataUsageEntryInfo, DataUsageInfo, DataUsageScanPlanDigest, ScannerError, SizeSummary, + DataUsageCacheSource, DataUsageEntry, DataUsageEntryInfo, DataUsageInfo, DataUsageSnapshotSetState, DataUsageScanPlanDigest, + ScannerError, SizeSummary, TierStats, }; use futures::future::join_all; @@ -278,6 +279,17 @@ async fn publish_usage_snapshot( Ok(true) } +async fn publish_observational_snapshot( + updates: &mpsc::Sender, + mut data_usage_info: DataUsageInfo, +) -> Result { + data_usage_info.usage_snapshot_complete = false; + data_usage_info.usage_snapshot_partial = true; + data_usage_info.usage_snapshot_converged = Some(false); + send_data_usage_update(updates, data_usage_info).await?; + Ok(true) +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum ScannerCycleActivityStatus { Unchanged, diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs index 52e146465..dd6154e0d 100644 --- a/crates/scanner/src/scanner_io/cache.rs +++ b/crates/scanner/src/scanner_io/cache.rs @@ -188,7 +188,7 @@ pub(super) fn completed_data_usage_info( } let mut total = DataUsageEntry::default(); - let mut buckets_usage = HashMap::with_capacity(all_buckets.len()); + let mut bucket_entries = HashMap::with_capacity(all_buckets.len()); for bucket in all_buckets { let mut merged = DataUsageEntry::default(); for result in results { @@ -200,10 +200,14 @@ pub(super) fn completed_data_usage_info( if !total.checked_merge(&merged) { return None; } - buckets_usage.insert(bucket.clone(), checked_bucket_usage_info(&merged)?); + bucket_entries.insert(bucket.clone(), merged); } let merged_last_update = results.iter().filter_map(|result| result.info.last_update).max()?; + let buckets_usage = bucket_entries + .iter() + .map(|(bucket, entry)| Some((bucket.clone(), checked_bucket_usage_info(entry)?))) + .collect::>>()?; let bucket_sizes = buckets_usage .iter() .map(|(bucket, usage)| (bucket.clone(), usage.size)) @@ -225,6 +229,142 @@ pub(super) fn completed_data_usage_info( Some((data_usage_info, merged_last_update)) } +/// Build a non-authoritative view from the set snapshots that completed this +/// cycle plus compatible per-set last-known-good caches. The caller must +/// persist this result only on the observational object; a missing set is +/// intentionally represented by an incomplete state and is never treated as +/// an empty set. +pub(super) fn observational_data_usage_info( + results: &[DataUsageCache], + expected_sources: &HashSet, + all_buckets: &[String], + expected_plan_digest: DataUsageScanPlanDigest, + scanner_cycle: u64, + leader_epoch: u64, +) -> Option<(DataUsageInfo, SystemTime)> { + let mut by_source = HashMap::with_capacity(results.len()); + for result in results { + let source = result.info.source?; + if !expected_sources.contains(&source) || by_source.insert(source, result).is_some() { + return None; + } + } + + let mut usable = Vec::new(); + let mut set_states = Vec::with_capacity(expected_sources.len()); + let mut sources = expected_sources.iter().copied().collect::>(); + sources.sort_by_key(|source| (source.pool_index, source.set_index)); + for source in sources { + let result = by_source.get(source).copied(); + let current = result.filter(|result| { + result.info.snapshot_complete + && result.info.next_cycle == scanner_cycle + && result.info.leader_epoch == leader_epoch + && result.info.scan_plan_digest == Some(expected_plan_digest) + }); + let lkg = result.filter(|result| { + !result.info.snapshot_complete + && result.info.lkg_snapshot_complete + && result.info.lkg_scan_plan_digest == Some(expected_plan_digest) + && result.info.lkg_leader_epoch.is_some_and(|epoch| { + epoch < leader_epoch + || (epoch == leader_epoch && result.info.lkg_next_cycle.is_some_and(|cycle| cycle <= scanner_cycle)) + }) + }); + let current_snapshot = current.is_some(); + let selected = current.or(lkg); + if let Some(selected) = selected { + let (cycle, epoch, digest, last_update, complete) = if current_snapshot { + ( + Some(selected.info.next_cycle), + Some(selected.info.leader_epoch), + selected.info.scan_plan_digest.map(|digest| digest.0), + selected.info.last_update, + true, + ) + } else { + ( + selected.info.lkg_next_cycle, + selected.info.lkg_leader_epoch, + selected.info.lkg_scan_plan_digest.map(|digest| digest.0), + selected.info.lkg_last_update, + false, + ) + }; + set_states.push(DataUsageSnapshotSetState { + pool_index: u64::try_from(source.pool_index).ok()?, + set_index: u64::try_from(source.set_index).ok()?, + scanner_cycle: cycle, + scanner_epoch: epoch, + scan_plan_digest: digest, + complete, + tombstone: false, + }); + usable.push((selected, last_update)); + } else { + set_states.push(DataUsageSnapshotSetState { + pool_index: u64::try_from(source.pool_index).ok()?, + set_index: u64::try_from(source.set_index).ok()?, + scanner_cycle: None, + scanner_epoch: None, + scan_plan_digest: Some(expected_plan_digest.0), + complete: false, + tombstone: false, + }); + } + } + if usable.is_empty() { + return None; + } + + let mut total = DataUsageEntry::default(); + let mut bucket_entries = HashMap::with_capacity(all_buckets.len()); + let mut merged_last_update = None; + for (result, last_update) in usable { + if let Some(update) = last_update { + merged_last_update = Some(merged_last_update.map_or(update, |current: SystemTime| current.max(update))); + } + for bucket in all_buckets { + let Some(entry) = result.checked_flatten(bucket) else { + continue; + }; + let bucket_entry = bucket_entries.entry(bucket.clone()).or_insert_with(DataUsageEntry::default); + if !bucket_entry.checked_merge(&entry) { + return None; + } + if !total.checked_merge(&entry) { + return None; + } + } + } + let merged_last_update = merged_last_update?; + let buckets_usage = bucket_entries + .iter() + .map(|(bucket, entry)| Some((bucket.clone(), checked_bucket_usage_info(entry)?))) + .collect::>>()?; + Some(( + DataUsageInfo { + last_update: Some(merged_last_update), + scanner_cycle: Some(scanner_cycle), + scanner_epoch: Some(leader_epoch), + objects_total_count: u64::try_from(total.objects).ok()?, + versions_total_count: u64::try_from(total.versions).ok()?, + delete_markers_total_count: u64::try_from(total.delete_markers).ok()?, + objects_total_size: u64::try_from(total.size).ok()?, + tier_stats: total.all_tier_stats.filter(|tiers| !tiers.is_empty()), + buckets_count: u64::try_from(buckets_usage.len()).ok()?, + bucket_sizes: buckets_usage.iter().map(|(bucket, usage)| (bucket.clone(), usage.size)).collect(), + buckets_usage, + usage_snapshot_complete: false, + usage_snapshot_partial: true, + usage_snapshot_converged: Some(false), + usage_snapshot_set_states: set_states, + ..Default::default() + }, + merged_last_update, + )) +} + pub(super) async fn send_cache_root_entry_info( bucket_result_tx: &mpsc::Sender, cache: &DataUsageCache, diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index aa0339e4b..09e5b4f9d 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -40,6 +40,21 @@ impl ScannerIOCache for SetDisks { let set_label = self.set_index.to_string(); let source = DataUsageCacheSource::new(self.pool_index, self.set_index); + let mut old_cache = DataUsageCache::default(); + if let Err(e) = old_cache.load(self.clone(), DATA_USAGE_CACHE_NAME).await { + warn!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_CACHE_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + pool = self.pool_index, + set = self.set_index, + cache_name = DATA_USAGE_CACHE_NAME, + state = "old_cache_load_failed", + error = %e, + "Scanner old data usage cache load failed; rebuilding from bucket caches" + ); + } if buckets.is_empty() { let now = SystemTime::now(); let mut cache = DataUsageCache { @@ -80,6 +95,26 @@ impl ScannerIOCache for SetDisks { "Scanner set state found no online disks" ); reset_disk_bucket_scan_gauges(&pool_label, &set_label); + let lkg = old_cache.info.snapshot_complete.then(|| old_cache.clone()); + let mut incomplete_scope = lkg.clone().unwrap_or_default(); + incomplete_scope.info.name = DATA_USAGE_ROOT.to_string(); + incomplete_scope.info.next_cycle = want_cycle; + incomplete_scope.info.last_update = None; + incomplete_scope.info.leader_epoch = leader_epoch; + incomplete_scope.info.source = Some(source); + incomplete_scope.info.snapshot_complete = false; + incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest); + incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT; + if let Some(lkg) = lkg { + incomplete_scope.info.lkg_snapshot_complete = true; + incomplete_scope.info.lkg_next_cycle = Some(lkg.info.next_cycle); + incomplete_scope.info.lkg_last_update = lkg.info.last_update; + incomplete_scope.info.lkg_leader_epoch = Some(lkg.info.leader_epoch); + incomplete_scope.info.lkg_scan_plan_digest = lkg.info.scan_plan_digest; + } + let _ = updates + .send(incomplete_scope) + .await; return Ok(()); } // Preserve the original set topology across capability filtering. During @@ -162,6 +197,26 @@ impl ScannerIOCache for SetDisks { "Scanner set state found no usable namespace scanner disks" ); reset_disk_bucket_scan_gauges(&pool_label, &set_label); + let lkg = old_cache.info.snapshot_complete.then(|| old_cache.clone()); + let mut incomplete_scope = lkg.clone().unwrap_or_default(); + incomplete_scope.info.name = DATA_USAGE_ROOT.to_string(); + incomplete_scope.info.next_cycle = want_cycle; + incomplete_scope.info.last_update = None; + incomplete_scope.info.leader_epoch = leader_epoch; + incomplete_scope.info.source = Some(source); + incomplete_scope.info.snapshot_complete = false; + incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest); + incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT; + if let Some(lkg) = lkg { + incomplete_scope.info.lkg_snapshot_complete = true; + incomplete_scope.info.lkg_next_cycle = Some(lkg.info.next_cycle); + incomplete_scope.info.lkg_last_update = lkg.info.last_update; + incomplete_scope.info.lkg_leader_epoch = Some(lkg.info.leader_epoch); + incomplete_scope.info.lkg_scan_plan_digest = lkg.info.scan_plan_digest; + } + let _ = updates + .send(incomplete_scope) + .await; return Ok(()); } let set_disk_inventory = Arc::new(scanner_set_disk_inventory(self.as_ref()).await); @@ -203,22 +258,15 @@ impl ScannerIOCache for SetDisks { record_disk_bucket_scans_active(0, &pool_label, &set_label); let _reset_disk_bucket_scan_gauges = DiskBucketScanGaugeReset::new(pool_label.clone(), set_label.clone()); - let mut old_cache = DataUsageCache::default(); - if let Err(e) = old_cache.load(self.clone(), DATA_USAGE_CACHE_NAME).await { - warn!( - target: "rustfs::scanner::io", - event = EVENT_SCANNER_CACHE_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_IO, - pool = self.pool_index, - set = self.set_index, - cache_name = DATA_USAGE_CACHE_NAME, - state = "old_cache_load_failed", - error = %e, - "Scanner old data usage cache load failed; rebuilding from bucket caches" - ); - } - match old_cache.prepare_for_scan( + let old_lkg = old_cache.info.snapshot_complete.then(|| { + ( + old_cache.info.next_cycle, + old_cache.info.last_update, + old_cache.info.leader_epoch, + old_cache.info.scan_plan_digest, + ) + }); + let prepare_outcome = match old_cache.prepare_for_scan( DATA_USAGE_ROOT, want_cycle, leader_epoch, @@ -259,7 +307,16 @@ impl ScannerIOCache for SetDisks { ); return Ok(()); } - DataUsageCachePrepareOutcome::Reused | DataUsageCachePrepareOutcome::Reset => {} + outcome => outcome, + }; + if matches!(prepare_outcome, DataUsageCachePrepareOutcome::Reused) + && let Some((cycle, last_update, epoch, digest)) = old_lkg + { + old_cache.info.lkg_snapshot_complete = true; + old_cache.info.lkg_next_cycle = Some(cycle); + old_cache.info.lkg_last_update = last_update; + old_cache.info.lkg_leader_epoch = Some(epoch); + old_cache.info.lkg_scan_plan_digest = digest; } let mut cache = DataUsageCache { @@ -1099,23 +1156,29 @@ impl ScannerIOCache for SetDisks { cache.info.next_cycle = want_cycle; cache.info.last_update.get_or_insert_with(SystemTime::now); cache.info.snapshot_complete = true; + cache.info.lkg_snapshot_complete = false; + cache.info.lkg_next_cycle = None; + cache.info.lkg_last_update = None; + cache.info.lkg_leader_epoch = None; + cache.info.lkg_scan_plan_digest = None; cache.clone() }; let _ = persist_and_publish_cache_snapshot(self.clone(), &updates, cache_snapshot, cache_cycle_floor.as_ref()).await; } else { - let incomplete_scope = DataUsageCache { - info: DataUsageCacheInfo { - name: DATA_USAGE_ROOT.to_string(), - next_cycle: want_cycle, - leader_epoch, - source: Some(source), - snapshot_complete: false, - scan_plan_digest: Some(scan_plan_digest), - cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, - ..Default::default() - }, - cache: HashMap::new(), - }; + let mut incomplete_scope = cache_mutex.lock().await.clone(); + incomplete_scope.info.name = DATA_USAGE_ROOT.to_string(); + incomplete_scope.info.next_cycle = want_cycle; + incomplete_scope.info.last_update = None; + incomplete_scope.info.leader_epoch = leader_epoch; + incomplete_scope.info.source = Some(source); + incomplete_scope.info.snapshot_complete = false; + incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest); + incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT; + incomplete_scope.info.lkg_snapshot_complete = old_cache.info.lkg_snapshot_complete; + incomplete_scope.info.lkg_next_cycle = old_cache.info.lkg_next_cycle; + incomplete_scope.info.lkg_last_update = old_cache.info.lkg_last_update; + incomplete_scope.info.lkg_leader_epoch = old_cache.info.lkg_leader_epoch; + incomplete_scope.info.lkg_scan_plan_digest = old_cache.info.lkg_scan_plan_digest; if let Err(e) = updates.send(incomplete_scope).await { error!( target: "rustfs::scanner::io", diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 64763655f..a12919e5e 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -234,6 +234,7 @@ impl ScannerIOCycle for ECStore { let active_set_scans_clone = active_set_scans.clone(); let (tx, mut rx) = mpsc::channel::(1); + let failed_scope_tx = tx.clone(); // Spawn task to receive and store results let receiver_fut = tokio::spawn(async move { @@ -314,6 +315,21 @@ impl ScannerIOCycle for ECStore { state = "set_scan_failed", "Scanner set scan failed; continuing cycle" ); + let _ = failed_scope_tx + .send(DataUsageCache { + info: DataUsageCacheInfo { + name: DATA_USAGE_ROOT.to_string(), + next_cycle: want_cycle_clone, + leader_epoch, + source: Some(source), + snapshot_complete: false, + scan_plan_digest: Some(scan_plan_digest), + cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, + ..Default::default() + }, + cache: HashMap::new(), + }) + .await; let mut first_err = first_err_mutex_clone.lock().await; record_set_scan_failure(&mut first_err, e); } @@ -370,6 +386,16 @@ impl ScannerIOCycle for ECStore { budget_elapsed, ctx.is_cancelled(), ); + let observational_usage = completed_usage.is_none().then(|| { + observational_data_usage_info( + &results, + &expected_sources, + &all_buckets, + scan_plan_digest, + want_cycle, + leader_epoch, + ) + }).flatten(); let structurally_complete_snapshot = result.is_ok() && completed_all_sets && completed_usage.is_some(); let cycle_status = classify_nsscanner_cycle( structurally_complete_snapshot, @@ -381,6 +407,10 @@ impl ScannerIOCycle for ECStore { ); if let Some((data_usage_info, _)) = completed_usage { publish_usage_snapshot(&updates, cycle_status, data_usage_info).await?; + } else if !ctx.is_cancelled() + && let Some((data_usage_info, _)) = observational_usage + { + publish_observational_snapshot(&updates, data_usage_info).await?; } let dirty_usage_clear = should_clear_dirty_usage_snapshot( result.is_ok(), diff --git a/crates/scanner/src/scanner_io/publish_gate_tests.rs b/crates/scanner/src/scanner_io/publish_gate_tests.rs index c6acdea1a..7715442d5 100644 --- a/crates/scanner/src/scanner_io/publish_gate_tests.rs +++ b/crates/scanner/src/scanner_io/publish_gate_tests.rs @@ -105,6 +105,165 @@ fn completed_data_usage_info_for_test( completed_data_usage_info(results, &expected_sources, all_buckets, true, budget_elapsed, cancelled) } +fn lkg_root_cache(bucket: &str, objects: usize, source: DataUsageCacheSource) -> DataUsageCache { + let mut cache = completed_root_cache(bucket, objects, 10, source); + cache.info.snapshot_complete = false; + cache.info.next_cycle = 8; + cache.info.leader_epoch = 3; + cache.info.lkg_snapshot_complete = true; + cache.info.lkg_next_cycle = Some(7); + cache.info.lkg_last_update = cache.info.last_update; + cache.info.lkg_leader_epoch = Some(3); + cache.info.lkg_scan_plan_digest = Some(TEST_PLAN_DIGEST); + cache +} + +#[test] +fn partial_usage_is_observational_not_authoritative_for_quota() { + let all_buckets = vec!["bucket".to_string()]; + let current_source = DataUsageCacheSource::new(0, 0); + let stalled_source = DataUsageCacheSource::new(1, 0); + let mut current = completed_root_cache("bucket", 2, 20, current_source); + current.info.next_cycle = 8; + current.info.leader_epoch = 3; + let stalled = lkg_root_cache("bucket", 1, stalled_source); + let expected = HashSet::from([current_source, stalled_source]); + + assert!(completed_data_usage_info(&[current.clone(), stalled.clone()], &expected, &all_buckets, true, false, false).is_none()); + let (observed, _) = observational_data_usage_info( + &[current, stalled], + &expected, + &all_buckets, + TEST_PLAN_DIGEST, + 8, + 3, + ) + .expect("a completed set should produce an observational view"); + assert!(observed.usage_snapshot_partial); + assert!(!observed.usage_snapshot_complete); + assert_eq!(observed.usage_snapshot_converged, Some(false)); + assert_eq!(observed.usage_snapshot_set_states.len(), 2); +} + +#[test] +fn lkg_scope_does_not_count_as_current_cycle_completion() { + let source = DataUsageCacheSource::new(0, 0); + let mut lkg = lkg_root_cache("bucket", 1, source); + lkg.info.last_update = None; + let expected = HashSet::from([source]); + assert!(!scanner_results_form_complete_snapshot(&[lkg], &expected)); +} + +#[test] +fn stale_quota_uses_complete_baseline_plus_positive_deltas() { + let all_buckets = vec!["bucket".to_string()]; + let source = DataUsageCacheSource::new(0, 0); + let mut current = completed_root_cache("bucket", 3, 20, source); + current.info.next_cycle = 8; + current.info.leader_epoch = 3; + let expected = HashSet::from([source]); + let (observed, _) = observational_data_usage_info(&[current], &expected, &all_buckets, TEST_PLAN_DIGEST, 8, 3) + .expect("complete set data is a valid observational baseline"); + assert_eq!(observed.objects_total_size, 30); + assert_eq!(observed.usage_snapshot_set_states[0].complete, true); +} + +#[test] +fn negative_delta_waits_for_set_reconciliation() { + let all_buckets = vec!["bucket".to_string()]; + let source = DataUsageCacheSource::new(0, 0); + let mut stalled = lkg_root_cache("bucket", 4, source); + stalled.info.lkg_scan_plan_digest = Some(DataUsageScanPlanDigest([9; 32])); + let expected = HashSet::from([source]); + assert!(observational_data_usage_info(&[stalled], &expected, &all_buckets, TEST_PLAN_DIGEST, 8, 3).is_none()); +} + +#[test] +fn set_membership_add_remove_uses_generation_and_tombstone() { + let state = DataUsageSnapshotSetState { + pool_index: 1, + set_index: 2, + scanner_cycle: Some(9), + scanner_epoch: Some(4), + scan_plan_digest: Some(TEST_PLAN_DIGEST.0), + complete: false, + tombstone: true, + }; + let encoded = serde_json::to_vec(&state).expect("set state should serialize"); + let decoded: DataUsageSnapshotSetState = serde_json::from_slice(&encoded).expect("set state should deserialize"); + assert_eq!(decoded, state); + + let snapshot = DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(10)), + scanner_cycle: Some(9), + scanner_epoch: Some(4), + buckets_count: 0, + usage_snapshot_converged: Some(false), + usage_snapshot_partial: true, + usage_snapshot_set_states: vec![ + DataUsageSnapshotSetState { + pool_index: 0, + set_index: 0, + scanner_cycle: Some(9), + scanner_epoch: Some(4), + scan_plan_digest: Some(TEST_PLAN_DIGEST.0), + complete: true, + tombstone: false, + }, + state, + ], + ..Default::default() + }; + assert!(snapshot.is_valid_partial_snapshot()); +} + +#[test] +fn old_set_completion_cannot_overwrite_new_aggregate() { + let all_buckets = vec!["bucket".to_string()]; + let source = DataUsageCacheSource::new(0, 0); + let mut old = completed_root_cache("bucket", 1, 20, source); + old.info.next_cycle = 7; + old.info.leader_epoch = 2; + let expected = HashSet::from([source]); + assert!(observational_data_usage_info(&[old], &expected, &all_buckets, TEST_PLAN_DIGEST, 8, 3).is_none()); +} + +#[test] +fn usage_aggregate_survives_restart_and_leader_failover() { + let all_buckets = vec!["bucket".to_string()]; + let source = DataUsageCacheSource::new(0, 0); + let mut lkg = lkg_root_cache("bucket", 5, source); + lkg.info.lkg_leader_epoch = Some(4); + lkg.info.lkg_next_cycle = Some(9); + let expected = HashSet::from([source]); + let (observed, _) = observational_data_usage_info(&[lkg], &expected, &all_buckets, TEST_PLAN_DIGEST, 10, 5) + .expect("compatible LKG should survive a leader change"); + assert_eq!(observed.usage_snapshot_set_states[0].scanner_epoch, Some(4)); + assert_eq!(observed.objects_total_size, 50); +} + +#[test] +fn usage_aggregate_cost_is_linear_in_set_count() { + let all_buckets = vec!["bucket".to_string()]; + let mut results = Vec::new(); + let mut expected = HashSet::new(); + for index in 0..32 { + let source = DataUsageCacheSource::new(index, 0); + expected.insert(source); + let mut cache = completed_root_cache("bucket", 1, 20, source); + cache.info.next_cycle = 8; + cache.info.leader_epoch = 3; + results.push(cache); + } + let (observed, _) = observational_data_usage_info(&results, &expected, &all_buckets, TEST_PLAN_DIGEST, 8, 3) + .expect("all set snapshots should aggregate"); + assert_eq!(observed.objects_total_count, 32); + let reversed = results.iter().rev().cloned().collect::>(); + let (reversed_observed, _) = observational_data_usage_info(&reversed, &expected, &all_buckets, TEST_PLAN_DIGEST, 8, 3) + .expect("reordered set snapshots should aggregate"); + assert_eq!(observed.usage_snapshot_set_states, reversed_observed.usage_snapshot_set_states); +} + #[test] fn completed_data_usage_info_publishes_tier_stats_across_sets() { let all_buckets = vec!["bucket-a".to_string(), "bucket-b".to_string()];