fix(scanner): surface per-tier usage in the data-usage snapshot (#5623)

SizeSummary::tier_stats was populated for every scanned object but
apply_scanner_size_summary dropped it, so per-tier usage never reached
DataUsageInfo. Wire it through the same merge chain repl_target_stats
already uses, up to DataUsageInfo::tier_stats.

DataUsageEntry used the derived MessagePack encoding, which serialises
structs as arrays: appending a field turns the whole cache into a decode
error for older readers, so mixed-version nodes would invalidate each
other's cache every scan cycle. Give it the same hand-written
map-encoded Serialize DataUsageCacheInfo already carries, and record the
invariant in AGENTS.md.

Widen TierStats counters from i32 to u64 so a tier past 2^31 versions
cannot make checked_merge reject an entire usage snapshot, and drop the
duplicate TierStats/AllTierStats definitions in the scanner crate in
favour of the data-usage ones.

crates/scanner/src/data_usage_define.rs

@@ -27,8 +27,8 @@ use rustfs_common::heal_channel::HealScanMode;
 #[cfg(test)]
 use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
 pub use rustfs_data_usage::{
-    BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DataUsageEntry, DataUsageHash, DataUsageHashMap,
-    DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, hash_path,
+    AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DataUsageEntry, DataUsageHash,
+    DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, TierStats, hash_path,
 };
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
 use tokio::time::{Duration, Instant, sleep, timeout};
@@ -184,69 +184,6 @@ pub static BACKGROUND_HEAL_INFO_PATH: LazyLock<String> =
 
 const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
 
-#[derive(Clone, Copy, Default, Debug, Serialize, Deserialize, PartialEq)]
-pub struct TierStats {
-    pub total_size: u64,
-    pub num_versions: i32,
-    pub num_objects: i32,
-}
-
-impl TierStats {
-    pub fn add(&self, u: &TierStats) -> TierStats {
-        TierStats {
-            total_size: self.total_size + u.total_size,
-            num_versions: self.num_versions + u.num_versions,
-            num_objects: self.num_objects + u.num_objects,
-        }
-    }
-
-    pub fn from_object_info(oi: &ObjectInfo) -> Self {
-        TierStats {
-            total_size: oi.size as u64,
-            num_versions: 1,
-            num_objects: if oi.is_latest { 1 } else { 0 },
-        }
-    }
-}
-
-#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
-pub struct AllTierStats {
-    pub tiers: HashMap<String, TierStats>,
-}
-
-impl AllTierStats {
-    pub fn new() -> Self {
-        Self { tiers: HashMap::new() }
-    }
-
-    pub fn add_sizes(&mut self, tiers: HashMap<String, TierStats>) {
-        for (tier, st) in tiers {
-            self.tiers
-                .insert(tier.clone(), self.tiers.get(&tier).copied().unwrap_or_default().add(&st));
-        }
-    }
-
-    pub fn merge(&mut self, other: AllTierStats) {
-        for (tier, st) in other.tiers {
-            self.tiers
-                .insert(tier.clone(), self.tiers.get(&tier).copied().unwrap_or_default().add(&st));
-        }
-    }
-
-    pub fn populate_stats(&self, stats: &mut HashMap<String, TierStats>) {
-        for (tier, st) in &self.tiers {
-            stats.insert(
-                tier.clone(),
-                TierStats {
-                    total_size: st.total_size,
-                    num_versions: st.num_versions,
-                    num_objects: st.num_objects,
-                },
-            );
-        }
-    }
-}
-
 /// Size summary for a single object or group of objects
 #[derive(Debug, Default, Clone)]
 pub struct SizeSummary {
@@ -301,7 +238,11 @@ impl SizeSummary {
         }
 
         if let Some(tier_stats) = self.tier_stats.get_mut(&tier) {
-            *tier_stats = tier_stats.add(&TierStats::from_object_info(oi));
+            *tier_stats = tier_stats.add(&TierStats {
+                total_size: u64::try_from(oi.size).unwrap_or(0),
+                num_versions: 1,
+                num_objects: u64::from(oi.is_latest),
+            });
         }
     }
 }
@@ -963,6 +904,7 @@ impl DataUsageCache {
             versions_total_count: flat.versions as u64,
             delete_markers_total_count: flat.delete_markers as u64,
             objects_total_size: flat.size as u64,
+            tier_stats: flat.all_tier_stats.filter(|tiers| !tiers.is_empty()),
             buckets_count: u64::try_from(buckets.len()).unwrap_or(u64::MAX),
             buckets_usage,
             ..Default::default()

crates/scanner/src/scanner_folder.rs

@@ -461,6 +461,8 @@ fn apply_scanner_size_summary(into: &mut DataUsageEntry, summary: &SizeSummary)
             .pending_count
             .saturating_add(u64::try_from(st.pending_count).unwrap_or(u64::MAX));
     }
+
+    into.add_tier_sizes(&summary.tier_stats);
 }
 
 /// Cached folder information for scanning
@@ -3049,7 +3051,7 @@ mod tests {
 
     use super::*;
     use crate::storage_api::VersionPurgeStatusType;
-    use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, new_disk};
+    use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
     use rustfs_filemeta::{FileInfo, FileMeta};
     use serial_test::serial;
     #[cfg(unix)]
@@ -3128,6 +3130,56 @@ mod tests {
         assert_eq!(target_stats.pending_count, u64::MAX);
     }
 
+    #[test]
+    fn scanner_size_summary_application_accumulates_tier_stats() {
+        let mut entry = DataUsageEntry::default();
+        let mut summary = SizeSummary::default();
+        summary.tier_stats.insert(
+            "WARM".to_string(),
+            TierStats {
+                total_size: 100,
+                num_versions: 2,
+                num_objects: 1,
+            },
+        );
+        // Scanners seed a zeroed entry for every configured tier; those must not
+        // reach the cache as empty keys.
+        summary
+            .tier_stats
+            .insert(storageclass::STANDARD.to_string(), TierStats::default());
+
+        apply_scanner_size_summary(&mut entry, &summary);
+        apply_scanner_size_summary(&mut entry, &summary);
+
+        let tiers = entry.all_tier_stats.as_ref().expect("tier stats should be recorded");
+        assert_eq!(
+            tiers.tiers.get("WARM"),
+            Some(&TierStats {
+                total_size: 200,
+                num_versions: 4,
+                num_objects: 2,
+            })
+        );
+        assert!(!tiers.tiers.contains_key(storageclass::STANDARD));
+    }
+
+    #[test]
+    fn scanner_size_summary_application_skips_untiered_summaries() {
+        let mut entry = DataUsageEntry::default();
+        let mut summary = SizeSummary {
+            total_size: 10,
+            ..Default::default()
+        };
+        summary
+            .tier_stats
+            .insert(storageclass::STANDARD.to_string(), TierStats::default());
+        summary.tier_stats.insert(storageclass::RRS.to_string(), TierStats::default());
+
+        apply_scanner_size_summary(&mut entry, &summary);
+
+        assert!(entry.all_tier_stats.is_none(), "zero-only tier maps must not allocate cache state");
+    }
+
     async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         let temp_dir = std::env::temp_dir().join(format!("rustfs-scanner-test-{}", Uuid::new_v4()));
         tokio::fs::create_dir_all(&temp_dir)

crates/scanner/src/scanner_io.rs

@@ -1146,6 +1146,7 @@ fn completed_data_usage_info(
         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(all_buckets.len()).ok()?,
         bucket_sizes,
         buckets_usage,
@@ -1250,6 +1251,57 @@ mod publish_gate_tests {
         completed_data_usage_info(results, &expected_sources, all_buckets, true, budget_elapsed, cancelled)
     }
 
+    #[test]
+    fn completed_data_usage_info_publishes_tier_stats_across_sets() {
+        let all_buckets = vec!["bucket-a".to_string(), "bucket-b".to_string()];
+        let warm = |total_size, num_versions, num_objects| {
+            HashMap::from([(
+                "WARM".to_string(),
+                TierStats {
+                    total_size,
+                    num_versions,
+                    num_objects,
+                },
+            )])
+        };
+
+        let mut first_set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
+        let mut tiered = DataUsageEntry::default();
+        tiered.add_tier_sizes(&warm(100, 2, 1));
+        first_set.replace("bucket-b", DATA_USAGE_ROOT, tiered);
+
+        let mut second_set = completed_root_cache("bucket-b", 2, 20, DataUsageCacheSource::new(1, 0));
+        let mut tiered = DataUsageEntry::default();
+        tiered.add_tier_sizes(&warm(50, 1, 1));
+        second_set.replace("bucket-a", DATA_USAGE_ROOT, tiered);
+
+        let (data_usage_info, _) = completed_data_usage_info_for_test(&[first_set, second_set], &all_buckets, false, false)
+            .expect("completed sets should publish a snapshot");
+
+        assert_eq!(
+            data_usage_info
+                .tier_stats
+                .expect("tier usage should reach the snapshot")
+                .tiers["WARM"],
+            TierStats {
+                total_size: 150,
+                num_versions: 3,
+                num_objects: 2,
+            }
+        );
+    }
+
+    #[test]
+    fn completed_data_usage_info_omits_tier_stats_without_tiered_objects() {
+        let all_buckets = vec!["bucket-a".to_string()];
+        let set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
+
+        let (data_usage_info, _) = completed_data_usage_info_for_test(&[set], &all_buckets, false, false)
+            .expect("completed set should publish a snapshot");
+
+        assert!(data_usage_info.tier_stats.is_none());
+    }
+
     #[test]
     fn completed_data_usage_info_requires_every_set_before_publish() {
         let all_buckets = vec!["bucket-a".to_string(), "bucket-b".to_string(), "bucket-empty".to_string()];