test: enable scanner attribution in validation build

.github/workflows/build.yml

@@ -398,7 +398,7 @@ jobs:
 
           FEATURE_ARGS=()
           if [[ -n "${{ matrix.features }}" ]]; then
-            FEATURE_ARGS=(--features "${{ matrix.features }}")
+            FEATURE_ARGS=(--features "${{ matrix.features }},diagnostic-prefix-attribution")
           fi
 
           # gnu targets pin an explicit glibc baseline (`<target>.<glibc>` for
@@ -455,7 +455,7 @@ jobs:
         shell: bash
         run: |
           cargo metadata --locked --format-version 1 --filter-platform '${{ matrix.target }}' \
-            --features rustfs/pyroscope > "$RUNNER_TEMP/cpu-catalog-metadata.json"
+            --features rustfs/pyroscope,rustfs/diagnostic-prefix-attribution > "$RUNNER_TEMP/cpu-catalog-metadata.json"
           features=$(jq -ce '
             (.packages[] | select(.name == "rustfs") | .id) as $id |
             .resolve.nodes[] | select(.id == $id) | .features | map(select(. != "default")) | sort

Cargo.lock

@@ -10642,6 +10642,9 @@ dependencies = [
  "hotpath",
  "http 1.5.0",
  "metrics",
+ "opentelemetry-proto",
+ "opentelemetry_sdk",
+ "prost 0.14.4",
  "rand 0.10.3",
  "rmp-serde",
  "rustfs-common",
@@ -10655,6 +10658,7 @@ dependencies = [
  "rustfs-heal-contracts",
  "rustfs-lifecycle",
  "rustfs-lock",
+ "rustfs-obs",
  "rustfs-s3-types",
  "rustfs-scanner-metrics",
  "rustfs-storage-api",

crates/scanner/Cargo.toml

@@ -31,6 +31,7 @@ workspace = true
 
 [features]
 default = []
+diagnostic-prefix-attribution = []
 diagnostic-read-version-caller = ["rustfs-ecstore/diagnostic-read-version-caller"]
 hotpath = [
     "hotpath/hotpath",
@@ -107,7 +108,11 @@ bytes.workspace = true
 hex-simd.workspace = true
 
 [dev-dependencies]
+opentelemetry-proto.workspace = true
+opentelemetry_sdk.workspace = true
+prost.workspace = true
 rustfs-heal.workspace = true
+rustfs-obs.workspace = true
 rustfs-test-utils.workspace = true
 tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] }
 serial_test = { workspace = true }

crates/scanner/src/scanner_folder.rs

@@ -807,6 +807,8 @@ pub struct FolderScanner {
     checkpoint_depth: usize,
     checkpoint_objects: u64,
     object_metadata_visits: metrics::Counter,
+    #[cfg(feature = "diagnostic-prefix-attribution")]
+    prefix_attribution: Option<prefix_attribution::VisitAttribution>,
     last_checkpoint_objects: u64,
     last_checkpoint_at: Instant,
     last_checkpoint_build_time: Duration,
@@ -2309,9 +2311,21 @@ impl FolderScanner {
 
                 let size_result = run_scanner_node_operation(self.node_admission.as_ref(), &ctx, async {
                     self.object_metadata_visits.increment(1);
-                    self.local_disk
+                    #[cfg(feature = "diagnostic-prefix-attribution")]
+                    if let Some(attribution) = &self.prefix_attribution {
+                        attribution.record(&item.bucket, &item.prefix);
+                    }
+                    let result = self
+                        .local_disk
                         .get_size_with_tier_names(item.clone(), &self.tier_registry.names)
-                        .await
+                        .await;
+                    #[cfg(feature = "diagnostic-prefix-attribution")]
+                    if result.is_ok()
+                        && let Some(attribution) = &self.prefix_attribution
+                    {
+                        attribution.record_success(&item.bucket, &item.prefix);
+                    }
+                    result
                 })
                 .await?;
                 if self.node_admission.is_some() && ctx.is_cancelled() {
@@ -3227,6 +3241,12 @@ pub(crate) async fn scan_data_folder_scoped(
         "bucket" => bucket_name.clone(),
         "disk" => base_path.clone()
     );
+    #[cfg(feature = "diagnostic-prefix-attribution")]
+    let prefix_attribution = prefix_attribution::configured_for(&bucket_name, &base_path);
+    #[cfg(feature = "diagnostic-prefix-attribution")]
+    if prefix_attribution.is_some() {
+        object_metadata_visits.increment(0);
+    }
     let next_cycle = cache.info.next_cycle;
     let resume_position_on_entry = cache.has_scan_position();
     let had_scan_checkpoint = cache.info.scan_checkpoint.is_some();
@@ -3264,6 +3284,8 @@ pub(crate) async fn scan_data_folder_scoped(
         checkpoint_depth: 0,
         checkpoint_objects: 0,
         object_metadata_visits,
+        #[cfg(feature = "diagnostic-prefix-attribution")]
+        prefix_attribution,
         last_checkpoint_objects: 0,
         last_checkpoint_at: Instant::now(),
         last_checkpoint_build_time: Duration::ZERO,
@@ -3471,6 +3493,8 @@ pub(crate) async fn scan_data_folder_scoped(
 
 mod item_actions;
 mod ledger;
+#[cfg(feature = "diagnostic-prefix-attribution")]
+mod prefix_attribution;
 
 use item_actions::*;
 pub use item_actions::{GetSizeFn, ScannerItem};

crates/scanner/src/scanner_folder/prefix_attribution.rs

@@ -0,0 +1,235 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+//! Optional, fixed-cardinality attribution of scanner size-call attempts on stored paths.
+
+use std::sync::{Arc, OnceLock};
+
+use serde::Deserialize;
+use sha2::{Digest, Sha256};
+
+const ENV_PREFIX_ATTRIBUTION: &str = "RUSTFS_SCANNER_PREFIX_ATTRIBUTION";
+const MAX_CONFIG_BYTES: usize = 4096;
+const MAX_PREFIX_BYTES: usize = 1024;
+const VISITS: &str = "rustfs_scanner_object_metadata_prefix_visits_total";
+const SUCCESSES: &str = "rustfs_scanner_object_metadata_prefix_successes_total";
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Input {
+    bucket: String,
+    hot_prefix: String,
+    cold_prefix: String,
+}
+
+pub(super) struct Scope {
+    bucket: String,
+    hot_prefix: String,
+    cold_prefix: String,
+    id: String,
+}
+
+impl Scope {
+    #[cfg(test)]
+    pub(super) fn id(&self) -> &str {
+        &self.id
+    }
+
+    fn parse(raw: &str) -> Option<Self> {
+        if raw.len() > MAX_CONFIG_BYTES {
+            return None;
+        }
+        let input: Input = serde_json::from_str(raw).ok()?;
+        if input.bucket.is_empty()
+            || input.bucket.len() > 63
+            || input.bucket.contains('/')
+            || input.bucket.chars().any(char::is_control)
+            || crate::is_reserved_or_invalid_bucket(&input.bucket, false)
+            || !valid_prefix(&input.hot_prefix)
+            || !valid_prefix(&input.cold_prefix)
+            || input.hot_prefix.starts_with(&input.cold_prefix)
+            || input.cold_prefix.starts_with(&input.hot_prefix)
+        {
+            return None;
+        }
+        // Length prefixes prevent ambiguous identities. The configuration is
+        // never used as a metric label, only this stable digest is exported.
+        let mut hash = Sha256::new();
+        for value in [
+            "scanner-prefix-attribution-v2",
+            &input.bucket,
+            &input.hot_prefix,
+            &input.cold_prefix,
+        ] {
+            hash.update(u32::try_from(value.len()).ok()?.to_be_bytes());
+            hash.update(value.as_bytes());
+        }
+        let mut id = String::with_capacity(64);
+        const HEX: &[u8; 16] = b"0123456789abcdef";
+        for byte in hash.finalize() {
+            id.push(HEX[(byte >> 4) as usize] as char);
+            id.push(HEX[(byte & 15) as usize] as char);
+        }
+        Some(Self {
+            bucket: input.bucket,
+            hot_prefix: input.hot_prefix,
+            cold_prefix: input.cold_prefix,
+            id,
+        })
+    }
+}
+
+fn valid_prefix(value: &str) -> bool {
+    if value.is_empty()
+        || value.len() > MAX_PREFIX_BYTES
+        || !value.ends_with('/')
+        || value.starts_with('/')
+        || value.chars().any(char::is_control)
+    {
+        return false;
+    }
+    value[..value.len() - 1]
+        .split('/')
+        .all(|part| !part.is_empty() && part != "." && part != "..")
+}
+
+#[cfg(test)]
+tokio::task_local! {
+    pub(super) static TEST_SCOPE: Option<Arc<Scope>>;
+}
+
+fn scope() -> Option<Arc<Scope>> {
+    #[cfg(test)]
+    if let Ok(scope) = TEST_SCOPE.try_with(Clone::clone) {
+        return scope;
+    }
+    static SCOPE: OnceLock<Option<Arc<Scope>>> = OnceLock::new();
+    SCOPE
+        .get_or_init(|| {
+            std::env::var(ENV_PREFIX_ATTRIBUTION)
+                .ok()
+                .and_then(|raw| Scope::parse(&raw))
+                .map(Arc::new)
+        })
+        .clone()
+}
+
+pub(super) struct VisitAttribution {
+    scope: Arc<Scope>,
+    hot: metrics::Counter,
+    cold: metrics::Counter,
+    other: metrics::Counter,
+    ambiguous: metrics::Counter,
+    hot_success: metrics::Counter,
+    cold_success: metrics::Counter,
+    other_success: metrics::Counter,
+    ambiguous_success: metrics::Counter,
+}
+
+impl VisitAttribution {
+    fn new(scope: Arc<Scope>, bucket: &str, disk: &str) -> Option<Self> {
+        if scope.bucket != bucket {
+            return None;
+        }
+        let counter = |name: &'static str, class: &'static str| metrics::counter!(name, "disk" => disk.to_string(), "class" => class, "scope" => scope.id.clone());
+        let hot = counter(VISITS, "hot");
+        let cold = counter(VISITS, "cold");
+        let other = counter(VISITS, "other");
+        let ambiguous = counter(VISITS, "ambiguous");
+        let hot_success = counter(SUCCESSES, "hot");
+        let cold_success = counter(SUCCESSES, "cold");
+        let other_success = counter(SUCCESSES, "other");
+        let ambiguous_success = counter(SUCCESSES, "ambiguous");
+        // Make an observed zero distinguishable from a missing series.
+        for counter in [
+            &hot,
+            &cold,
+            &other,
+            &ambiguous,
+            &hot_success,
+            &cold_success,
+            &other_success,
+            &ambiguous_success,
+        ] {
+            counter.increment(0);
+        }
+        Some(Self {
+            scope,
+            hot,
+            cold,
+            other,
+            ambiguous,
+            hot_success,
+            cold_success,
+            other_success,
+            ambiguous_success,
+        })
+    }
+
+    fn record_inner(&self, bucket: &str, object_path: &str, success: bool) {
+        if bucket != self.scope.bucket {
+            return;
+        }
+        // An encoded directory marker and a literal suffix can share the
+        // same storage path. Never guess which original key produced it.
+        let counter = if object_path
+            .split(|ch| ch == '/' || (cfg!(windows) && ch == '\\'))
+            .any(|component| rustfs_utils::path::has_suffix(component, rustfs_utils::path::GLOBAL_DIR_SUFFIX))
+        {
+            if success { &self.ambiguous_success } else { &self.ambiguous }
+        } else if object_path.starts_with(&self.scope.hot_prefix) {
+            if success { &self.hot_success } else { &self.hot }
+        } else if object_path.starts_with(&self.scope.cold_prefix) {
+            if success { &self.cold_success } else { &self.cold }
+        } else if success {
+            &self.other_success
+        } else {
+            &self.other
+        };
+        counter.increment(1);
+    }
+
+    pub(super) fn record(&self, bucket: &str, object_path: &str) {
+        self.record_inner(bucket, object_path, false);
+    }
+
+    pub(super) fn record_success(&self, bucket: &str, object_path: &str) {
+        self.record_inner(bucket, object_path, true);
+    }
+}
+
+pub(super) fn configured_for(bucket: &str, disk: &str) -> Option<VisitAttribution> {
+    VisitAttribution::new(scope()?, bucket, disk)
+}
+
+#[cfg(test)]
+pub(super) fn test_scope(raw: &str) -> Option<Arc<Scope>> {
+    Scope::parse(raw).map(Arc::new)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn scopes_require_disjoint_canonical_nonempty_prefixes() {
+        let valid = r#"{"bucket":"bucket","hot_prefix":"hot/","cold_prefix":"cold/"}"#;
+        let scope = Scope::parse(valid).expect("valid scope");
+        assert_eq!(scope.id, "48a1cfe7ac0b0e2e0e1a1df5285bf0b1ff2261c9fd17eb9e9acfb53ea1d540ed");
+        assert_eq!(scope.id, Scope::parse(valid).expect("same scope").id);
+        let changed = r#"{"bucket":"bucket","hot_prefix":"other/","cold_prefix":"cold/"}"#;
+        assert_ne!(scope.id, Scope::parse(changed).expect("different scope").id);
+        for invalid in [
+            r#"{"bucket":"bucket","hot_prefix":"hot","cold_prefix":"cold/"}"#,
+            r#"{"bucket":"bucket","hot_prefix":"hot/","cold_prefix":"hot/deep/"}"#,
+            r#"{"bucket":"bucket","hot_prefix":"/hot/","cold_prefix":"cold/"}"#,
+            r#"{"bucket":"bucket","hot_prefix":"hot//nested/","cold_prefix":"cold/"}"#,
+            r#"{"bucket":"bucket","hot_prefix":"../hot/","cold_prefix":"cold/"}"#,
+            r#"{"bucket":"bucket","hot_prefix":"hot/","cold_prefix":"cold/","extra":0}"#,
+            r#"{"bucket":"bucket","hot_prefix":"hot/","hot_prefix":"other/","cold_prefix":"cold/"}"#,
+        ] {
+            assert!(Scope::parse(invalid).is_none(), "invalid scope accepted");
+        }
+        assert!(Scope::parse(&"x".repeat(MAX_CONFIG_BYTES + 1)).is_none());
+    }
+}

crates/scanner/src/scanner_folder/tests.rs

@@ -33,6 +33,8 @@ mod incremental_enumeration;
 mod listing_completion;
 mod metadata_visits;
 mod node_admission;
+#[cfg(feature = "diagnostic-prefix-attribution")]
+mod prefix_attribution_tests;
 mod usage_updates;
 
 /// Reset the process-global alert cooldown map; test-only.
@@ -375,6 +377,8 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         checkpoint_depth: 0,
         checkpoint_objects: 0,
         object_metadata_visits: metrics::Counter::noop(),
+        #[cfg(feature = "diagnostic-prefix-attribution")]
+        prefix_attribution: None,
         last_checkpoint_objects: 0,
         last_checkpoint_at: Instant::now(),
         last_checkpoint_build_time: Duration::ZERO,

crates/scanner/src/scanner_folder/tests/prefix_attribution_tests.rs

@@ -0,0 +1,437 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::super::prefix_attribution::{TEST_SCOPE, test_scope};
+use super::*;
+use std::sync::atomic::AtomicU64;
+
+const TOTAL: &str = "rustfs_scanner_object_metadata_visits_total";
+const ATTRIBUTED: &str = "rustfs_scanner_object_metadata_prefix_visits_total";
+const SUCCESSES: &str = "rustfs_scanner_object_metadata_prefix_successes_total";
+const PLAN: &str = r#"{"bucket":"bucket","hot_prefix":"hot/","cold_prefix":"cold/"}"#;
+
+#[derive(Default)]
+struct Recorder(Mutex<HashMap<metrics::Key, Arc<AtomicU64>>>);
+
+impl metrics::Recorder for Recorder {
+    fn describe_counter(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_gauge(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_histogram(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
+        if !matches!(key.name(), TOTAL | ATTRIBUTED | SUCCESSES) {
+            return metrics::Counter::noop();
+        }
+        metrics::Counter::from_arc(self.0.lock().expect("recorder").entry(key.clone()).or_default().clone())
+    }
+    fn register_gauge(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge {
+        metrics::Gauge::noop()
+    }
+    fn register_histogram(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Histogram {
+        metrics::Histogram::noop()
+    }
+}
+
+fn attributed(name: &'static str, disk: &str, id: &str, class: &'static str) -> metrics::Key {
+    metrics::Key::from_parts(
+        name,
+        vec![
+            metrics::Label::new("disk", disk.to_owned()),
+            metrics::Label::new("class", class),
+            metrics::Label::new("scope", id.to_owned()),
+        ],
+    )
+}
+
+async fn run_scan(
+    disk: Arc<Disk>,
+    bucket: &str,
+    cancelled: bool,
+    configured: Option<Arc<super::super::prefix_attribution::Scope>>,
+    recorder: &impl metrics::Recorder,
+) -> Result<DataUsageCache, ScannerError> {
+    let parent = CancellationToken::new();
+    if cancelled {
+        parent.cancel();
+    }
+    let budget = ScannerCycleBudget::new(&parent, Default::default());
+    let cache = DataUsageCache {
+        info: crate::data_usage_define::DataUsageCacheInfo {
+            name: bucket.to_string(),
+            ..Default::default()
+        },
+        ..Default::default()
+    };
+    TEST_SCOPE
+        .scope(configured, async {
+            let scan = scan_data_folder(
+                parent,
+                budget,
+                vec![disk.clone()],
+                disk,
+                cache,
+                None,
+                HealScanMode::Normal,
+                DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest),
+            );
+            let mut scan = std::pin::pin!(scan);
+            std::future::poll_fn(|cx| metrics::with_local_recorder(recorder, || scan.as_mut().poll(cx))).await
+        })
+        .await
+}
+
+#[derive(Clone, Debug)]
+struct ExportedCounter {
+    name: String,
+    labels: Vec<(String, String)>,
+    value: u64,
+}
+
+#[derive(Clone, Debug, Default)]
+struct CounterExporter(Arc<Mutex<Vec<ExportedCounter>>>);
+
+impl opentelemetry_sdk::metrics::exporter::PushMetricExporter for CounterExporter {
+    async fn export(
+        &self,
+        resource: &opentelemetry_sdk::metrics::data::ResourceMetrics,
+    ) -> opentelemetry_sdk::error::OTelSdkResult {
+        use opentelemetry_proto::tonic::{
+            collector::metrics::v1::ExportMetricsServiceRequest,
+            common::v1::any_value,
+            metrics::v1::{AggregationTemporality, metric, number_data_point},
+        };
+        use prost::Message;
+
+        // Use the same SDK-to-OTLP conversion as the production HTTP exporter.
+        let encoded = ExportMetricsServiceRequest::from(resource).encode_to_vec();
+        let decoded = ExportMetricsServiceRequest::decode(encoded.as_slice()).expect("decode exported OTLP protobuf");
+        let mut observed = self.0.lock().expect("lock exported counters");
+        observed.clear();
+        for resource in decoded.resource_metrics {
+            for scope in resource.scope_metrics {
+                for metric in scope.metrics {
+                    if !matches!(
+                        metric.name.as_str(),
+                        TOTAL | ATTRIBUTED | SUCCESSES | "rustfs_scanner_zero_export_test_beacon_total"
+                    ) {
+                        continue;
+                    }
+                    let Some(metric::Data::Sum(sum)) = metric.data else {
+                        panic!("counter exported with non-sum type");
+                    };
+                    assert!(sum.is_monotonic);
+                    assert_eq!(
+                        AggregationTemporality::try_from(sum.aggregation_temporality).expect("known counter temporality"),
+                        AggregationTemporality::Cumulative
+                    );
+                    for point in sum.data_points {
+                        assert!(point.start_time_unix_nano > 0);
+                        assert!(point.time_unix_nano >= point.start_time_unix_nano);
+                        assert_eq!(point.flags, 0);
+                        let Some(number_data_point::Value::AsInt(value)) = point.value else {
+                            panic!("counter point has no integer value");
+                        };
+                        let labels = point
+                            .attributes
+                            .into_iter()
+                            .map(|attr| {
+                                let Some(any_value::Value::StringValue(value)) = attr.value.and_then(|value| value.value) else {
+                                    panic!("counter label is not a string");
+                                };
+                                (attr.key, value)
+                            })
+                            .collect();
+                        observed.push(ExportedCounter {
+                            name: metric.name.clone(),
+                            labels,
+                            value: u64::try_from(value).expect("counter value must be nonnegative"),
+                        });
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
+    fn force_flush(&self) -> opentelemetry_sdk::error::OTelSdkResult {
+        Ok(())
+    }
+
+    fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> opentelemetry_sdk::error::OTelSdkResult {
+        Ok(())
+    }
+
+    fn temporality(&self) -> opentelemetry_sdk::metrics::Temporality {
+        opentelemetry_sdk::metrics::Temporality::Cumulative
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn configured_zero_total_is_exported_without_manufacturing_unconfigured_points() {
+    use opentelemetry_sdk::metrics::PeriodicReader;
+
+    let (scanner, _retained_temp_dir) = build_test_scanner().await;
+    let disk = scanner.local_disk.clone();
+    let disk_label = disk.path().to_string_lossy().into_owned();
+    let plan = test_scope(PLAN).expect("valid scope");
+    for (bucket, scope, configured) in [
+        ("bucket", Some(plan.clone()), true),
+        ("bucket", None, false),
+        ("other-bucket", Some(plan.clone()), false),
+    ] {
+        let exporter = CounterExporter::default();
+        let reader = PeriodicReader::builder(exporter.clone())
+            .with_interval(std::time::Duration::from_secs(3600))
+            .build();
+        let (provider, recorder) = rustfs_obs::Recorder::builder("scanner-zero-export-test")
+            .with_meter_provider(|builder| builder.with_reader(reader))
+            .build();
+        assert!(matches!(
+            run_scan(disk.clone(), bucket, true, scope, &recorder).await,
+            Err(ScannerError::Other(reason)) if reason == "Operation cancelled"
+        ));
+        metrics::with_local_recorder(&recorder, || {
+            metrics::counter!("rustfs_scanner_zero_export_test_beacon_total").increment(1);
+        });
+        provider.force_flush().expect("export scanner counters");
+        let rows = exporter.0.lock().expect("lock exported counters").clone();
+        assert!(
+            rows.iter()
+                .any(|row| row.name == "rustfs_scanner_zero_export_test_beacon_total" && row.value == 1)
+        );
+        let has_label = |row: &ExportedCounter, key: &str, value: &str| {
+            row.labels.iter().any(|(label, actual)| label == key && actual == value)
+        };
+        let totals: Vec<_> = rows
+            .iter()
+            .filter(|row| row.name == TOTAL && has_label(row, "bucket", bucket) && has_label(row, "disk", &disk_label))
+            .collect();
+        assert_eq!(totals.len(), usize::from(configured), "total presence requires matching diagnostic scope");
+        if configured {
+            assert_eq!(totals[0].value, 0);
+            assert_eq!(totals[0].labels.len(), 2);
+        }
+        for name in [ATTRIBUTED, SUCCESSES] {
+            for class in ["hot", "cold", "other", "ambiguous"] {
+                let points: Vec<_> = rows
+                    .iter()
+                    .filter(|row| {
+                        row.name == name
+                            && has_label(row, "disk", &disk_label)
+                            && has_label(row, "scope", plan.id())
+                            && has_label(row, "class", class)
+                    })
+                    .collect();
+                assert_eq!(points.len(), usize::from(configured), "attributed series presence");
+                if configured {
+                    assert_eq!(points[0].value, 0);
+                    assert_eq!(points[0].labels.len(), 3);
+                }
+            }
+        }
+        provider.shutdown().expect("stop test exporter");
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn actual_metadata_attempts_split_same_bucket_without_prefix_labels() {
+    let (scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(temp_dir.clone()),
+    };
+    for name in ["hot/object", "cold/object", "hotter/object"] {
+        write_test_object_metadata(&temp_dir, "bucket", name).await;
+    }
+    let disk = scanner.local_disk.clone();
+    let disk_label = disk.path().to_string_lossy().into_owned();
+    let plan = test_scope(PLAN).expect("valid scope");
+    let recorder = Recorder::default();
+    for expected in [1, 2] {
+        run_scan(disk.clone(), "bucket", false, Some(plan.clone()), &recorder)
+            .await
+            .expect("actual bucket scan");
+        let rows = recorder.0.lock().expect("recorder");
+        assert_eq!(rows.len(), 9, "one existing total plus four classes for attempts and successes");
+        let total = metrics::Key::from_parts(
+            TOTAL,
+            vec![
+                metrics::Label::new("bucket", "bucket"),
+                metrics::Label::new("disk", disk_label.clone()),
+            ],
+        );
+        assert_eq!(rows[&total].load(Ordering::Relaxed), expected * 3);
+        for class in ["hot", "cold", "other"] {
+            assert_eq!(
+                rows[&attributed(ATTRIBUTED, &disk_label, plan.id(), class)].load(Ordering::Relaxed),
+                expected
+            );
+            assert_eq!(
+                rows[&attributed(SUCCESSES, &disk_label, plan.id(), class)].load(Ordering::Relaxed),
+                expected
+            );
+        }
+        assert_eq!(
+            rows[&attributed(ATTRIBUTED, &disk_label, plan.id(), "ambiguous")].load(Ordering::Relaxed),
+            0
+        );
+    }
+    assert!(
+        run_scan(disk.clone(), "bucket", true, Some(plan.clone()), &recorder)
+            .await
+            .is_err()
+    );
+    let rows = recorder.0.lock().expect("recorder");
+    for class in ["hot", "cold", "other"] {
+        assert_eq!(rows[&attributed(ATTRIBUTED, &disk_label, plan.id(), class)].load(Ordering::Relaxed), 2);
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_or_wrong_bucket_produces_no_attributed_series() {
+    let (scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(temp_dir.clone()),
+    };
+    write_test_object_metadata(&temp_dir, "bucket", "hot/object").await;
+    write_test_object_metadata(&temp_dir, "other-bucket", "hot/object").await;
+    let disk = scanner.local_disk.clone();
+    let recorder = Recorder::default();
+    run_scan(disk.clone(), "bucket", false, None, &recorder)
+        .await
+        .expect("disabled scan");
+    run_scan(disk, "other-bucket", false, test_scope(PLAN), &recorder)
+        .await
+        .expect("unselected bucket scan");
+    assert!(
+        recorder
+            .0
+            .lock()
+            .expect("recorder")
+            .keys()
+            .all(|key| !matches!(key.name(), ATTRIBUTED | SUCCESSES))
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn configured_absent_class_is_a_registered_zero() {
+    let (scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(temp_dir.clone()),
+    };
+    write_test_object_metadata(&temp_dir, "bucket", "hot/object").await;
+    let disk = scanner.local_disk.clone();
+    let disk_label = disk.path().to_string_lossy().into_owned();
+    let plan = test_scope(PLAN).expect("valid scope");
+    let recorder = Recorder::default();
+    run_scan(disk, "bucket", false, Some(plan.clone()), &recorder)
+        .await
+        .expect("actual bucket scan");
+    let rows = recorder.0.lock().expect("recorder");
+    assert_eq!(rows[&attributed(ATTRIBUTED, &disk_label, plan.id(), "hot")].load(Ordering::Relaxed), 1);
+    assert_eq!(rows[&attributed(ATTRIBUTED, &disk_label, plan.id(), "cold")].load(Ordering::Relaxed), 0);
+    assert_eq!(rows[&attributed(ATTRIBUTED, &disk_label, plan.id(), "other")].load(Ordering::Relaxed), 0);
+}
+
+#[tokio::test]
+#[serial]
+async fn failed_metadata_attempt_is_not_reported_as_success() {
+    let (scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(temp_dir.clone()),
+    };
+    write_test_object_metadata(&temp_dir, "bucket", "hot/object").await;
+    write_test_object_metadata_bytes(&temp_dir, "bucket", "cold/object", b"invalid-metadata").await;
+    let disk = scanner.local_disk.clone();
+    let disk_label = disk.path().to_string_lossy().into_owned();
+    let plan = test_scope(PLAN).expect("valid scope");
+    let recorder = Recorder::default();
+    assert!(matches!(
+        run_scan(disk, "bucket", false, Some(plan.clone()), &recorder).await,
+        Err(ScannerError::PartialCache(_))
+    ));
+    let rows = recorder.0.lock().expect("recorder");
+    assert_eq!(rows[&attributed(ATTRIBUTED, &disk_label, plan.id(), "hot")].load(Ordering::Relaxed), 1);
+    assert_eq!(rows[&attributed(SUCCESSES, &disk_label, plan.id(), "hot")].load(Ordering::Relaxed), 1);
+    assert_eq!(rows[&attributed(ATTRIBUTED, &disk_label, plan.id(), "cold")].load(Ordering::Relaxed), 1);
+    assert_eq!(rows[&attributed(SUCCESSES, &disk_label, plan.id(), "cold")].load(Ordering::Relaxed), 0);
+}
+
+#[tokio::test]
+#[serial]
+async fn closed_node_admission_does_not_record_prefix_attempts() {
+    use rustfs_concurrency::adaptive::NodeAdmission;
+    use std::num::NonZeroU16;
+
+    let (mut scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(temp_dir.clone()),
+    };
+    write_test_object_metadata(&temp_dir, "bucket", "cold/object").await;
+    let disk_label = scanner.local_disk.path().to_string_lossy().into_owned();
+    let plan = test_scope(PLAN).expect("valid scope");
+    let recorder = Recorder::default();
+    let gate = NodeAdmission::new(NonZeroU16::new(1).expect("one unit"));
+    scanner.node_admission = Some(gate.clone());
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
+    TEST_SCOPE
+        .scope(Some(plan.clone()), async {
+            scanner.prefix_attribution = metrics::with_local_recorder(&recorder, || {
+                super::super::prefix_attribution::configured_for("bucket", &disk_label)
+            });
+            let result = scanner
+                .scan_folder(
+                    CancellationToken::new(),
+                    CachedFolder {
+                        name: "bucket".to_string(),
+                        parent: None,
+                        object_heal_prob_div: 1,
+                    },
+                    &mut DataUsageEntry::default(),
+                )
+                .await;
+            assert!(matches!(result, Err(ScannerError::ResourceDeferred)));
+        })
+        .await;
+    let rows = recorder.0.lock().expect("recorder");
+    for name in [ATTRIBUTED, SUCCESSES] {
+        for class in ["hot", "cold", "other", "ambiguous"] {
+            assert_eq!(rows[&attributed(name, &disk_label, plan.id(), class)].load(Ordering::Relaxed), 0);
+        }
+    }
+    assert_eq!(gate.snapshot().active, 0);
+    assert_eq!(gate.completed(), Some(0));
+}
+
+#[tokio::test]
+#[serial]
+async fn encoded_directory_paths_are_not_attributed_to_a_raw_prefix() {
+    let (scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(temp_dir.clone()),
+    };
+    for (stored, logical) in [
+        ("hot__XLDIR__", "hot/"),
+        ("cold__XLDIR__", "cold/"),
+        ("hot/inner__XLDIR__", "hot/inner/"),
+        ("hot__XLDIR____XLDIR__", "hot__XLDIR__/"),
+    ] {
+        write_test_object_metadata_bytes(&temp_dir, "bucket", stored, &metadata_for_object("bucket", logical)).await;
+    }
+    let disk = scanner.local_disk.clone();
+    let disk_label = disk.path().to_string_lossy().into_owned();
+    let plan = test_scope(PLAN).expect("valid scope");
+    let recorder = Recorder::default();
+    run_scan(disk, "bucket", false, Some(plan.clone()), &recorder)
+        .await
+        .expect("scan encoded metadata paths");
+    let rows = recorder.0.lock().expect("recorder");
+    for name in [ATTRIBUTED, SUCCESSES] {
+        assert_eq!(rows[&attributed(name, &disk_label, plan.id(), "ambiguous")].load(Ordering::Relaxed), 4);
+        for class in ["hot", "cold", "other"] {
+            assert_eq!(rows[&attributed(name, &disk_label, plan.id(), class)].load(Ordering::Relaxed), 0);
+        }
+    }
+}

docs/operations/scanner-runtime-controls.md

@@ -891,3 +891,13 @@ config:
 ```
 
 Use `extraEnv` for environment variables that are not represented by chart values, including every heal knob above.
+
+## Diagnostic same-bucket prefix attribution
+
+Build with `diagnostic-prefix-attribution` to enable bounded diagnostic counters. The feature is disabled in ordinary builds. Set `RUSTFS_SCANNER_PREFIX_ATTRIBUTION` before starting RustFS to a JSON object with one bucket and two disjoint prefixes, for example `{"bucket":"bench","hot_prefix":"hot/","cold_prefix":"cold/"}`. Prefixes must end in `/` and have no empty, `.` or `..` path component. A missing, malformed or overlapping scope disables attribution; it does not emit zero. Restart to change the scope.
+
+`rustfs_scanner_object_metadata_prefix_visits_total` classifies stored Scanner paths by disk into the fixed classes `hot`, `cold`, `other` and `ambiguous`. Its scope digest uses the `scanner-prefix-attribution-v2` domain; raw bucket and prefix text are not labels, but the digest is not a secrecy guarantee. A path containing a component ending in the encoded directory suffix is `ambiguous`, because a directory marker and a literal key can share that representation. Never count it as hot or cold. Other path normalization and filesystem aliases can also prevent recovery of the original S3 key, so this is not general raw-key attribution.
+
+An attempt is recorded immediately before calling `get_size_with_tier_names` after node admission. A concurrent cancellation can still prevent its internal metadata read. Attempts include failures and repeated visits, and are not unique objects, native I/O or durable progress. A separate `rustfs_scanner_object_metadata_prefix_successes_total` increments only when that call returns success; it does not certify publication. An operation crossing a measurement boundary can produce a success in one window and an attempt in another. The class handles are registered for a valid matching bucket so an observed zero can be distinguished from a missing series.
+
+Use a controlled canonical dataset and independently verify its inventory and per-disk scope before interpreting these counts. Require complete per-disk series for all four classes, unchanged scope, no resets and genuinely matching source windows; missing data or nonzero ambiguous counts cannot establish cold attribution. A zero ambiguous count does not prove a complete scan or rule out other path aliases. Build both comparison legs with the same feature and configuration, and separately measure its overhead. This diagnostic alone does not establish clone-byte attribution, physical cold cache, foreground latency or SSD acceptance.

rustfs/Cargo.toml

@@ -81,6 +81,8 @@ dial9 = ["rustfs-obs/dial9"]
 diagnostic-allocator-live = []
 # Diagnostic-only S3-operation attribution for internode ReadVersion and list-walker metrics.
 diagnostic-read-version-caller = ["rustfs-ecstore/diagnostic-read-version-caller", "rustfs-io-metrics/diagnostic-read-version-caller", "rustfs-scanner/diagnostic-read-version-caller"]
+# Diagnostic-only fixed-prefix attribution for actual Scanner metadata visits.
+diagnostic-prefix-attribution = ["rustfs-scanner/diagnostic-prefix-attribution"]
 http3 = ["dep:h3", "dep:h3-quinn", "dep:quinn"]
 hotpath = [
     "hotpath/hotpath",

scripts/scanner_process_receipt.py

@@ -17,7 +17,9 @@ SAFE_ENVIRONMENT = frozenset({
     "RUSTFS_ADDRESS", "RUSTFS_CONSOLE_ENABLE", "RUSTFS_ERASURE_SET_DRIVE_COUNT",
     "RUSTFS_STORAGE_CLASS_STANDARD", "RUSTFS_DURABILITY_MODE", "RUSTFS_NEW_BUCKET_DURABILITY_MODE",
     "RUSTFS_SCANNER_ENABLED", "RUSTFS_HEAL_ENABLED", "RUSTFS_SCANNER_SPEED", "RUSTFS_SCANNER_IDLE_MODE",
-    "RUSTFS_SCANNER_START_DELAY_SECS", "RUSTFS_SCANNER_CYCLE",
+    "RUSTFS_SCANNER_START_DELAY_SECS", "RUSTFS_SCANNER_CYCLE", "RUSTFS_SCANNER_PREFIX_ATTRIBUTION",
+    "HOTPATH_FOCUS", "HOTPATH_ALLOC_CUMULATIVE", "HOTPATH_PROMETHEUS_HOST", "HOTPATH_PROMETHEUS_PORT",
+    "HOTPATH_METRICS_PORT", "HOTPATH_OUTPUT_PATH", "HOTPATH_DRAIN_INTERVAL",
     "RUSTFS_HEAL_ADAPTIVE_ADMISSION",
     "RUSTFS_OBJECT_LOCK_DIAG_ENABLE", "RUSTFS_OBS_LOGS_EXPORT_ENABLED", "RUSTFS_OBS_TRACES_EXPORT_ENABLED",
     "RUSTFS_OBS_METRICS_EXPORT_ENABLED", "RUST_LOG", "NO_PROXY", "no_proxy",
@@ -66,7 +68,7 @@ def environment_digest(raw, expected):
         key = key.decode()
         if key in PRIVATE_ENVIRONMENT:
             continue
-        if key.startswith("RUSTFS_") or key in SAFE_ENVIRONMENT:
+        if key.startswith(("RUSTFS_", "HOTPATH_")) or key in SAFE_ENVIRONMENT:
             require(key in SAFE_ENVIRONMENT and key not in observed, "unrecognized or duplicate benchmark environment")
             observed[key] = value.decode()
     require(observed == expected, "benchmark environment mismatch")

scripts/test_scanner_process_receipt.py

@@ -89,6 +89,20 @@ class ProcessReceiptTest(unittest.TestCase):
         self.assertEqual(receipt.environment_digest(changed, self.public),
                          receipt.environment_digest(self.environment, self.public))
 
+    def test_prefix_and_profiler_environment_is_bound_without_revealing_values(self):
+        options = {
+            "RUSTFS_SCANNER_PREFIX_ATTRIBUTION": '{"bucket":"bucket","hot_prefix":"hot/","cold_prefix":"cold/"}',
+            "HOTPATH_ALLOC_CUMULATIVE": "0",
+            "HOTPATH_PROMETHEUS_PORT": "19191",
+        }
+        raw = self.environment + b"".join((key + "=" + value).encode() + b"\0" for key, value in options.items())
+        expected = {**self.public, **options}
+        self.assertEqual(receipt.environment_digest(raw, expected), receipt.fingerprint(expected))
+        with self.assertRaises(ValueError):
+            receipt.environment_digest(raw, {**expected, "HOTPATH_ALLOC_CUMULATIVE": "1"})
+        with self.assertRaises(ValueError):
+            receipt.environment_digest(raw + b"HOTPATH_UNRECOGNIZED=1\0", expected)
+
     def test_scanner_schedule_is_bound_to_actual_process_environment(self):
         schedule = {"RUSTFS_SCANNER_START_DELAY_SECS": "1", "RUSTFS_SCANNER_CYCLE": "5"}
         self.public.update(schedule)