test(scanner): add bounded cache cost microprofile (#7261)

crates/scanner/src/data_usage_define/tests.rs

@@ -29,6 +29,8 @@ use tokio::sync::Mutex;
 
 const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([3; 32]);
 
+mod cache_cost;
+
 #[test]
 fn scoped_scan_coverage_metadata_preserves_map_compatibility() {
     #[derive(serde::Deserialize)]

crates/scanner/src/data_usage_define/tests/cache_cost.rs

@@ -0,0 +1,346 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::*;
+use std::hint::black_box;
+use std::sync::atomic::AtomicU64;
+use std::time::Instant as WallInstant;
+
+const MAX_WIRE_BYTES: u64 = 32 * 1024 * 1024;
+const CACHE_NAME: &str = "bucket/cache-cost.bin";
+
+/// Two bounded memory slots model revision preconditions and count the bytes
+/// consumed by the real save entry point, not disk writes or fsync latency.
+#[derive(Debug, Default)]
+struct CountingStore {
+    slots: Mutex<[(u64, Vec<u8>); 2]>,
+    puts: AtomicU64,
+    bytes: AtomicU64,
+    ingest_ns: AtomicU64,
+}
+
+impl CountingStore {
+    fn slot(object: &str) -> usize {
+        let main = path_join_buf(&[BUCKET_META_PREFIX, CACHE_NAME]);
+        if object == main {
+            0
+        } else {
+            assert_eq!(object, format!("{main}.bkp"), "only two fixture cache paths are permitted");
+            1
+        }
+    }
+
+    fn reset_counts(&self) {
+        self.puts.store(0, Ordering::Relaxed);
+        self.bytes.store(0, Ordering::Relaxed);
+        self.ingest_ns.store(0, Ordering::Relaxed);
+    }
+}
+
+#[async_trait::async_trait]
+impl ObjectIO for CountingStore {
+    type Error = Error;
+    type RangeSpec = HTTPRangeSpec;
+    type HeaderMap = HeaderMap;
+    type ObjectOptions = ObjectOptions;
+    type ObjectInfo = ObjectInfo;
+    type GetObjectReader = ScannerGetObjectReader;
+    type PutObjectReader = ScannerPutObjReader;
+
+    async fn get_object_reader(
+        &self,
+        bucket: &str,
+        object: &str,
+        _range: Option<Self::RangeSpec>,
+        _headers: Self::HeaderMap,
+        _options: &Self::ObjectOptions,
+    ) -> StorageResult<Self::GetObjectReader> {
+        // The real loader may probe the legacy metadata bucket on a miss.
+        if bucket != RUSTFS_META_BUCKET {
+            return Err(Error::FileNotFound);
+        }
+        let slots = self.slots.lock().await;
+        let (revision, bytes) = &slots[Self::slot(object)];
+        if *revision == 0 {
+            return Err(Error::FileNotFound);
+        }
+        Ok(CacheReadStore::reader(CacheReadBody::Bytes(bytes.clone()), &revision.to_string()))
+    }
+
+    async fn put_object(
+        &self,
+        bucket: &str,
+        object: &str,
+        data: &mut Self::PutObjectReader,
+        options: &Self::ObjectOptions,
+    ) -> StorageResult<Self::ObjectInfo> {
+        assert_eq!(bucket, RUSTFS_META_BUCKET);
+        let started = WallInstant::now();
+        let mut bytes = Vec::new();
+        (&mut data.stream).take(MAX_WIRE_BYTES + 1).read_to_end(&mut bytes).await?;
+        assert!(u64::try_from(bytes.len()).expect("wire length") <= MAX_WIRE_BYTES);
+        let mut slots = self.slots.lock().await;
+        let (revision, stored) = &mut slots[Self::slot(object)];
+        let preconditions = options.http_preconditions.as_ref().expect("profile saves must use CAS");
+        let expected = revision.to_string();
+        if (*revision == 0 && preconditions.if_none_match_value() != Some("*"))
+            || (*revision != 0 && preconditions.if_match_value() != Some(expected.as_str()))
+        {
+            return Err(Error::PreconditionFailed);
+        }
+        self.bytes
+            .fetch_add(u64::try_from(bytes.len()).expect("save length"), Ordering::Relaxed);
+        self.puts.fetch_add(1, Ordering::Relaxed);
+        *stored = bytes;
+        *revision += 1;
+        self.ingest_ns.fetch_add(elapsed_ns(started), Ordering::Relaxed);
+        Ok(ObjectInfo {
+            etag: Some(revision.to_string()),
+            ..Default::default()
+        })
+    }
+}
+
+#[async_trait::async_trait]
+impl crate::ScannerConfigObjectDelete for CountingStore {
+    async fn delete_config_object(
+        &self,
+        _bucket: &str,
+        _object: &str,
+        _options: crate::ScannerObjectOptions,
+    ) -> crate::EcstoreResult<crate::ScannerObjectInfo> {
+        Err(Error::NotImplemented)
+    }
+
+    async fn scanner_data_usage_publication_admission(&self) -> Option<crate::ScannerDataUsagePublicationAdmission> {
+        Some(crate::ScannerDataUsagePublicationAdmission::unfenced())
+    }
+}
+
+fn elapsed_ns(started: WallInstant) -> u64 {
+    u64::try_from(started.elapsed().as_nanos()).expect("bounded profile duration")
+}
+
+fn fixture(objects: usize) -> DataUsageCache {
+    assert!((1..=16384).contains(&objects));
+    let mut cache = DataUsageCache::default();
+    cache.info.name = "bucket".to_string();
+    cache.info.snapshot_complete = true;
+    cache.replace("bucket", "", DataUsageEntry::default());
+    for index in 0..objects {
+        cache.replace(
+            &format!("bucket/object-{index:05}"),
+            "bucket",
+            DataUsageEntry {
+                objects: 1,
+                versions: 2,
+                size: 4096,
+                ..Default::default()
+            },
+        );
+    }
+    cache
+}
+
+fn canonical_cache_value(mut value: Value) -> Value {
+    for entry in value["cache"].as_object_mut().expect("cache entry map").values_mut() {
+        let children = entry["children"].as_array_mut().expect("entry children set");
+        // Sort only the set representation. Do not deduplicate or reorder
+        // histograms and other arrays whose element positions carry meaning.
+        children.sort_unstable_by(|left, right| {
+            left.as_str()
+                .expect("child key string")
+                .cmp(right.as_str().expect("child key string"))
+        });
+    }
+    value
+}
+
+fn same_cache(actual: &DataUsageCache, expected: &DataUsageCache) {
+    assert_eq!(
+        canonical_cache_value(serde_json::to_value(actual).expect("actual cache structure")),
+        canonical_cache_value(serde_json::to_value(expected).expect("expected cache structure")),
+        "every cache field and map entry must be retained"
+    );
+}
+
+#[test]
+fn cache_cost_comparison_preserves_set_and_ordered_field_semantics() {
+    let forward = fixture(2);
+    let mut reverse = forward.clone();
+    let children = &mut reverse.cache.get_mut(&hash_path("bucket").key()).expect("root").children;
+    children.clear();
+    for index in (0..2).rev() {
+        children.insert(hash_path(&format!("bucket/object-{index:05}")).key());
+    }
+    same_cache(&forward, &reverse);
+
+    let original = serde_json::json!({"cache": {"root": {"children": ["a", "b"], "size": 1, "histogram": [1, 2]}}});
+    let mut reordered = original.clone();
+    reordered["cache"]["root"]["children"] = serde_json::json!(["b", "a"]);
+    assert_eq!(canonical_cache_value(original.clone()), canonical_cache_value(reordered));
+    for children in [serde_json::json!(["a"]), serde_json::json!(["a", "b", "b"])] {
+        let mut changed = original.clone();
+        changed["cache"]["root"]["children"] = children;
+        assert_ne!(canonical_cache_value(original.clone()), canonical_cache_value(changed));
+    }
+    for (field, replacement) in [("size", serde_json::json!(2)), ("histogram", serde_json::json!([2, 1]))] {
+        let mut changed = original.clone();
+        changed["cache"]["root"][field] = replacement;
+        assert_ne!(canonical_cache_value(original.clone()), canonical_cache_value(changed));
+    }
+}
+
+fn quantiles(mut samples: Vec<u64>) -> Value {
+    assert!(!samples.is_empty() && samples.len() <= 5);
+    samples.sort_unstable();
+    serde_json::json!({"p50_ns": samples[samples.len() / 2], "max_ns": samples[samples.len() - 1]})
+}
+
+async fn profile_case(objects: usize, scenario: &str, samples: usize) {
+    let baseline = fixture(objects);
+    let mut cache = baseline.clone();
+    let dirty = match scenario {
+        "unchanged" => 0,
+        "small_dirty" => (objects / 100).max(1),
+        "all_dirty" => objects,
+        _ => panic!("unknown fixed scenario"),
+    };
+    let mut changed_entry_wire_bytes = 0;
+    for index in 0..dirty {
+        let entry = cache
+            .cache
+            .get_mut(&hash_path(&format!("bucket/object-{index:05}")).key())
+            .expect("dirty leaf");
+        entry.size += 1;
+        entry.versions += 1;
+        changed_entry_wire_bytes += rmp_serde::to_vec(entry).expect("changed entry wire bytes").len();
+    }
+    if scenario == "small_dirty" {
+        cache.info.snapshot_complete = false;
+        cache.info.scan_resume_after = Some("bucket/object-00000".to_string());
+    }
+    let expected_wire = cache.marshal_msg().expect("fixture encoding");
+    assert!(u64::try_from(expected_wire.len()).expect("fixture bytes") <= MAX_WIRE_BYTES);
+    let store = Arc::new(CountingStore::default());
+    let mut loaded = DataUsageCache::default();
+    let initial = loaded
+        .load_with_revisions(store.clone(), CACHE_NAME)
+        .await
+        .expect("initial revisions");
+    baseline
+        .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &initial, 0)
+        .await
+        .expect("baseline save");
+
+    let mut clone_ns = Vec::new();
+    let mut copy_ns = Vec::new();
+    let mut flatten_ns = Vec::new();
+    let mut encode_ns = Vec::new();
+    let mut save_ns = Vec::new();
+    let mut ingest_ns = Vec::new();
+    for _ in 0..samples {
+        let started = WallInstant::now();
+        let cloned = black_box(cache.clone());
+        clone_ns.push(elapsed_ns(started));
+        same_cache(&cloned, &cache);
+        drop(cloned);
+
+        let mut copied = DataUsageCache {
+            info: cache.info.clone(),
+            ..Default::default()
+        };
+        let started = WallInstant::now();
+        copied.copy_with_children(black_box(&cache), &hash_path("bucket"), &None);
+        copy_ns.push(elapsed_ns(started));
+        same_cache(&copied, &cache);
+        drop(copied);
+
+        let started = WallInstant::now();
+        let aggregate = black_box(cache.checked_flatten("bucket").expect("valid fixture tree"));
+        flatten_ns.push(elapsed_ns(started));
+        assert_eq!(
+            (aggregate.objects, aggregate.versions, aggregate.size),
+            (objects, objects * 2 + dirty, objects * 4096 + dirty)
+        );
+
+        let started = WallInstant::now();
+        let encoded = black_box(cache.marshal_msg().expect("measured encoding"));
+        encode_ns.push(elapsed_ns(started));
+        assert_eq!(encoded, expected_wire);
+        same_cache(&DataUsageCache::unmarshal(&encoded).expect("measured wire reload"), &cache);
+
+        let revisions = loaded
+            .load_with_revisions(store.clone(), CACHE_NAME)
+            .await
+            .expect("current revisions");
+        store.reset_counts();
+        let started = WallInstant::now();
+        cache
+            .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
+            .await
+            .expect("measured save");
+        save_ns.push(elapsed_ns(started));
+        ingest_ns.push(store.ingest_ns.load(Ordering::Relaxed));
+        assert_eq!(store.puts.load(Ordering::Relaxed), 2, "main and backup writes must both occur");
+        assert_eq!(
+            store.bytes.load(Ordering::Relaxed),
+            u64::try_from(expected_wire.len() * 2).expect("two saved bodies")
+        );
+        loaded
+            .load_with_revisions(store.clone(), CACHE_NAME)
+            .await
+            .expect("saved cache reload");
+        same_cache(&loaded, &cache);
+    }
+
+    let before_rejected = store.slots.lock().await[0].1.clone();
+    let mut conflicting = cache.clone();
+    conflicting.info.next_cycle += 1;
+    assert!(matches!(
+        conflicting
+            .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &initial, 0)
+            .await,
+        Err(Error::PreconditionFailed)
+    ));
+    assert_eq!(
+        store.slots.lock().await[0].1,
+        before_rejected,
+        "stale CAS must not replace the retained checkpoint"
+    );
+    println!(
+        "CACHE_COST {}",
+        serde_json::json!({
+            "schema": 1, "scenario": scenario, "objects": objects, "dirty_objects": dirty, "samples": samples,
+            "build": {
+                "debug_assertions": cfg!(debug_assertions),
+                "test_opt_level_override": option_env!("CARGO_PROFILE_TEST_OPT_LEVEL"),
+                "dev_opt_level_override": option_env!("CARGO_PROFILE_DEV_OPT_LEVEL"),
+                "rustflags_visible_to_rustc": option_env!("RUSTFLAGS"),
+                "encoded_rustflags_visible_to_rustc": option_env!("CARGO_ENCODED_RUSTFLAGS"),
+                "source_revision": option_env!("RUSTFS_CACHE_COST_SOURCE"),
+                "source_tree": option_env!("RUSTFS_CACHE_COST_TREE"),
+            },
+            "retained_cache_entries": cache.cache.len(), "cache_wire_bytes": expected_wire.len(),
+            "changed_entry_wire_bytes": changed_entry_wire_bytes, "save_body_bytes_per_sample": expected_wire.len() * 2,
+            "snapshot_complete": cache.info.snapshot_complete,
+            "clone": quantiles(clone_ns), "copy_with_children": quantiles(copy_ns), "checked_flatten": quantiles(flatten_ns),
+            "encode": quantiles(encode_ns), "save_inclusive": quantiles(save_ns), "memory_backend_ingest": quantiles(ingest_ns),
+        })
+    );
+}
+
+#[tokio::test]
+async fn cache_cost_profile_preserves_checkpoint_and_counts() {
+    let profile = match std::env::var("RUSTFS_CACHE_COST_PROFILE") {
+        Err(std::env::VarError::NotPresent) => false,
+        Ok(value) if value == "1" => true,
+        _ => panic!("RUSTFS_CACHE_COST_PROFILE must be absent or 1"),
+    };
+    let (sizes, samples): (&[usize], usize) = if profile { (&[1024, 4096, 16384], 5) } else { (&[64], 1) };
+    for &objects in sizes {
+        for scenario in ["unchanged", "small_dirty", "all_dirty"] {
+            profile_case(objects, scenario, samples).await;
+        }
+    }
+}