feat(scanner): add scanner budgets and progress metrics (#3145)

* fix(scanner): preserve maintenance scan cadence

* feat(scanner): add scanner concurrency budget

* feat(scanner): expose scanner runtime progress

* fix(scanner): address scanner review feedback

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>

crates/common/src/metrics.rs

@@ -370,6 +370,8 @@ pub struct ScannerMetricsReport {
     pub current_started: DateTime<Utc>,
     pub cycles_completed_at: Vec<DateTime<Utc>>,
     pub ongoing_buckets: usize,
+    #[serde(default)]
+    pub active_scan_paths: usize,
     pub life_time_ops: HashMap<String, u64>,
     pub life_time_ilm: HashMap<String, u64>,
     pub last_minute: ScannerLastMinute,
@@ -600,18 +602,22 @@ impl Metrics {
     pub async fn report(&self) -> ScannerMetricsReport {
         let mut m = ScannerMetricsReport::default();
 
-        if let Some(cycle) = self.get_cycle().await {
+        let has_cycle = if let Some(cycle) = self.get_cycle().await {
             m.current_cycle = cycle.current;
             m.cycles_completed_at = cycle.cycle_completed;
             m.current_started = cycle.started;
-        }
+            true
+        } else {
+            false
+        };
 
-        if let Some(init_time) = crate::get_global_init_time().await {
+        if !has_cycle && let Some(init_time) = crate::get_global_init_time().await {
             m.current_started = init_time;
         }
 
         m.collected_at = Utc::now();
         m.active_paths = self.get_current_paths().await;
+        m.active_scan_paths = m.active_paths.len();
 
         // Lifetime operation counts
         for i in 0..Metric::Last as usize {
@@ -747,3 +753,46 @@ impl Drop for CloseDiskGuard {
         // If there is no runtime we are in a test or shutdown path; skip cleanup.
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn report_counts_active_scan_paths() {
+        let metrics = Metrics::new();
+        metrics
+            .current_paths
+            .write()
+            .await
+            .insert("disk-a".to_string(), Arc::new(CurrentPathTracker::new("bucket-a".to_string())));
+
+        let report = metrics.report().await;
+
+        assert_eq!(report.active_scan_paths, 1);
+        assert_eq!(report.ongoing_buckets, 0);
+        assert_eq!(report.active_paths, vec!["disk-a/bucket-a".to_string()]);
+    }
+
+    #[tokio::test]
+    async fn report_preserves_current_cycle_started_time() {
+        let previous_init_time = *crate::globals::GLOBAL_INIT_TIME.read().await;
+        let init_time = Utc::now() - chrono::Duration::hours(1);
+        let cycle_started = Utc::now();
+        *crate::globals::GLOBAL_INIT_TIME.write().await = Some(init_time);
+
+        let metrics = Metrics::new();
+        metrics
+            .set_cycle(Some(CurrentCycle {
+                current: 7,
+                started: cycle_started,
+                ..Default::default()
+            }))
+            .await;
+
+        let report = metrics.report().await;
+        *crate::globals::GLOBAL_INIT_TIME.write().await = previous_init_time;
+
+        assert_eq!(report.current_started, cycle_started);
+    }
+}

crates/config/src/constants/scanner.rs

@@ -83,9 +83,27 @@ pub const ENV_SCANNER_IDLE_MODE: &str = "RUSTFS_SCANNER_IDLE_MODE";
 /// - Example: `export RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS=30`
 pub const ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS: &str = "RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS";
 
+/// Environment variable that caps concurrent scanner set tasks.
+/// A value of `0` keeps the existing topology-based concurrency.
+/// - Example: `export RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS=2`
+pub const ENV_SCANNER_MAX_CONCURRENT_SET_SCANS: &str = "RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS";
+
+/// Environment variable that caps concurrent scanner disk bucket walks per set.
+/// A value of `0` keeps the existing disk-count-based concurrency.
+/// - Example: `export RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS=1`
+pub const ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS: &str = "RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS";
+
 /// Default scanner idle mode.
 pub const DEFAULT_SCANNER_IDLE_MODE: bool = true;
 
+/// Default set scan concurrency budget.
+/// `0` means no additional limit beyond deployment topology.
+pub const DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS: usize = 0;
+
+/// Default disk scan concurrency budget.
+/// `0` means no additional limit beyond available disks in the set.
+pub const DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS: usize = 0;
+
 /// Compatibility flag kept for Patch 3 rollback windows.
 ///
 /// Inline scanner heal execution has been removed in favor of heal-candidate enqueue.

crates/obs/src/metrics/collectors/scanner.rs

@@ -21,7 +21,7 @@
 
 use crate::metrics::report::PrometheusMetric;
 use crate::metrics::schema::scanner::{
-    SCANNER_BUCKET_SCANS_FINISHED_MD, SCANNER_BUCKET_SCANS_STARTED_MD, SCANNER_DIRECTORIES_SCANNED_MD,
+    SCANNER_ACTIVE_PATHS_MD, SCANNER_BUCKET_SCANS_FINISHED_MD, SCANNER_BUCKET_SCANS_STARTED_MD, SCANNER_DIRECTORIES_SCANNED_MD,
     SCANNER_LAST_ACTIVITY_SECONDS_MD, SCANNER_OBJECTS_SCANNED_MD, SCANNER_VERSIONS_SCANNED_MD,
 };
 
@@ -40,6 +40,8 @@ pub struct ScannerStats {
     pub versions_scanned: u64,
     /// Seconds since last scanner activity
     pub last_activity_seconds: u64,
+    /// Number of scanner paths currently being processed
+    pub active_paths: u64,
 }
 
 /// Collects scanner metrics from the given stats.
@@ -54,6 +56,7 @@ pub fn collect_scanner_metrics(stats: &ScannerStats) -> Vec<PrometheusMetric> {
         PrometheusMetric::from_descriptor(&SCANNER_OBJECTS_SCANNED_MD, stats.objects_scanned as f64),
         PrometheusMetric::from_descriptor(&SCANNER_VERSIONS_SCANNED_MD, stats.versions_scanned as f64),
         PrometheusMetric::from_descriptor(&SCANNER_LAST_ACTIVITY_SECONDS_MD, stats.last_activity_seconds as f64),
+        PrometheusMetric::from_descriptor(&SCANNER_ACTIVE_PATHS_MD, stats.active_paths as f64),
     ]
 }
 
@@ -71,18 +74,24 @@ mod tests {
             objects_scanned: 1000000,
             versions_scanned: 1500000,
             last_activity_seconds: 30,
+            active_paths: 4,
         };
 
         let metrics = collect_scanner_metrics(&stats);
         report_metrics(&metrics);
 
-        assert_eq!(metrics.len(), 6);
+        assert_eq!(metrics.len(), 7);
 
         let objects = metrics.iter().find(|m| m.value == 1000000.0);
         assert!(objects.is_some());
 
         let last_activity = metrics.iter().find(|m| m.value == 30.0);
         assert!(last_activity.is_some());
+
+        let active_paths = metrics
+            .iter()
+            .find(|m| m.name == SCANNER_ACTIVE_PATHS_MD.get_full_metric_name());
+        assert_eq!(active_paths.map(|m| m.value), Some(4.0));
     }
 
     #[test]
@@ -90,7 +99,7 @@ mod tests {
         let stats = ScannerStats::default();
         let metrics = collect_scanner_metrics(&stats);
 
-        assert_eq!(metrics.len(), 6);
+        assert_eq!(metrics.len(), 7);
         for metric in &metrics {
             assert_eq!(metric.value, 0.0);
             assert!(metric.labels.is_empty());

crates/obs/src/metrics/schema/entry/metric_name.rs

@@ -278,6 +278,7 @@ pub enum MetricName {
     ScannerObjectsScanned,
     ScannerVersionsScanned,
     ScannerLastActivitySeconds,
+    ScannerActivePaths,
 
     // CPU system-related metrics
     SysCPUAvgIdle,
@@ -618,6 +619,7 @@ impl MetricName {
             Self::ScannerObjectsScanned => "objects_scanned".to_string(),
             Self::ScannerVersionsScanned => "versions_scanned".to_string(),
             Self::ScannerLastActivitySeconds => "last_activity_seconds".to_string(),
+            Self::ScannerActivePaths => "active_paths".to_string(),
 
             // CPU system-related metrics
             Self::SysCPUAvgIdle => "avg_idle".to_string(),

crates/obs/src/metrics/schema/scanner.rs

@@ -70,3 +70,12 @@ pub static SCANNER_LAST_ACTIVITY_SECONDS_MD: LazyLock<MetricDescriptor> = LazyLo
         subsystems::SCANNER,
     )
 });
+
+pub static SCANNER_ACTIVE_PATHS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_gauge_md(
+        MetricName::ScannerActivePaths,
+        "Current number of scanner paths being processed.",
+        &[],
+        subsystems::SCANNER,
+    )
+});

crates/obs/src/metrics/stats_collector.rs

@@ -875,6 +875,7 @@ pub async fn collect_scanner_metric_stats() -> Option<ScannerStats> {
     let versions_scanned = metrics.life_time_ilm.values().copied().sum();
     let reference_time = metrics.cycles_completed_at.last().copied().unwrap_or(metrics.current_started);
     let last_activity_seconds = Utc::now().signed_duration_since(reference_time).num_seconds().max(0) as u64;
+    let active_paths = metrics.active_scan_paths as u64;
 
     Some(ScannerStats {
         bucket_scans_finished,
@@ -886,6 +887,7 @@ pub async fn collect_scanner_metric_stats() -> Option<ScannerStats> {
         objects_scanned,
         versions_scanned,
         last_activity_seconds,
+        active_paths,
     })
 }
 

crates/scanner/src/scanner.rs

@@ -31,11 +31,15 @@ use rustfs_config::{
     ENV_SCANNER_START_DELAY_SECS,
 };
 use rustfs_ecstore::StorageAPI as _;
+use rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle as _;
+use rustfs_ecstore::bucket::metadata_sys::{get_lifecycle_config, get_replication_config};
+use rustfs_ecstore::bucket::replication::ReplicationConfigurationExt as _;
 use rustfs_ecstore::config::com::{read_config, save_config};
 use rustfs_ecstore::disk::RUSTFS_META_BUCKET;
 use rustfs_ecstore::error::Error as EcstoreError;
 use rustfs_ecstore::global::is_erasure_sd;
 use rustfs_ecstore::store::ECStore;
+use rustfs_ecstore::store_api::{BucketOperations, BucketOptions};
 use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc;
 use tokio::time::{Duration, Instant};
@@ -108,7 +112,7 @@ fn initial_scanner_delay_for(start_delay_secs: Option<u64>) -> Duration {
 }
 
 pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
-    configure_scanner_defaults().await;
+    configure_scanner_defaults(&storeapi).await;
     // Force init global sleeper so config is read once at startup.
     let _ = &*SCANNER_SLEEPER;
 
@@ -136,16 +140,105 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
     });
 }
 
-async fn configure_scanner_defaults() {
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+struct ScannerMaintenanceFeatures {
+    lifecycle: bool,
+    replication: bool,
+    inspection_failed: bool,
+}
+
+impl ScannerMaintenanceFeatures {
+    fn needs_regular_cycle(self) -> bool {
+        self.lifecycle || self.replication || self.inspection_failed
+    }
+}
+
+fn single_disk_default_cycle_secs(features: ScannerMaintenanceFeatures) -> Option<u64> {
+    if features.needs_regular_cycle() {
+        None
+    } else {
+        Some(SINGLE_DISK_SCANNER_CYCLE_SECS)
+    }
+}
+
+async fn detect_scanner_maintenance_features(storeapi: &Arc<ECStore>) -> ScannerMaintenanceFeatures {
+    let mut features = ScannerMaintenanceFeatures::default();
+    let buckets = match storeapi
+        .list_bucket(&BucketOptions {
+            no_metadata: true,
+            ..Default::default()
+        })
+        .await
+    {
+        Ok(buckets) => buckets,
+        Err(err) => {
+            warn!(
+                error = %err,
+                "Failed to inspect scanner maintenance features; preserving speed-based scanner cycle"
+            );
+            features.inspection_failed = true;
+            return features;
+        }
+    };
+
+    for bucket in buckets {
+        if !features.lifecycle {
+            match get_lifecycle_config(&bucket.name).await {
+                Ok((lifecycle, _)) => {
+                    features.lifecycle = lifecycle.has_active_rules("");
+                }
+                Err(EcstoreError::ConfigNotFound) => {}
+                Err(err) => {
+                    warn!(
+                        bucket = %bucket.name,
+                        error = %err,
+                        "Failed to inspect bucket lifecycle config; preserving speed-based scanner cycle"
+                    );
+                    features.inspection_failed = true;
+                }
+            }
+        }
+
+        if !features.replication {
+            match get_replication_config(&bucket.name).await {
+                Ok((replication, _)) => {
+                    features.replication = replication.has_active_rules("", true);
+                }
+                Err(EcstoreError::ConfigNotFound) => {}
+                Err(err) => {
+                    warn!(
+                        bucket = %bucket.name,
+                        error = %err,
+                        "Failed to inspect bucket replication config; preserving speed-based scanner cycle"
+                    );
+                    features.inspection_failed = true;
+                }
+            }
+        }
+
+        if features.needs_regular_cycle() {
+            break;
+        }
+    }
+
+    features
+}
+
+async fn configure_scanner_defaults(storeapi: &Arc<ECStore>) {
     if is_erasure_sd().await {
+        let features = detect_scanner_maintenance_features(storeapi).await;
+        let default_cycle_secs = single_disk_default_cycle_secs(features);
         set_scanner_default_speed(ScannerSpeed::Slowest);
-        set_scanner_default_cycle_secs(Some(SINGLE_DISK_SCANNER_CYCLE_SECS));
+        set_scanner_default_cycle_secs(default_cycle_secs);
         info!(
             env_speed = ENV_SCANNER_SPEED,
             env_cycle = ENV_SCANNER_CYCLE,
             env_start_delay = ENV_SCANNER_START_DELAY_SECS,
-            default_cycle_secs = SINGLE_DISK_SCANNER_CYCLE_SECS,
-            "Using slower scanner defaults for single-disk deployments; explicit scanner cycle or start-delay settings still take precedence"
+            ?default_cycle_secs,
+            lifecycle_active = features.lifecycle,
+            replication_active = features.replication,
+            feature_inspection_failed = features.inspection_failed,
+            "Using slower scanner defaults for single-disk deployments; regular scanner cycles are preserved when maintenance features are active"
         );
     } else {
         set_scanner_default_speed(ScannerSpeed::Default);
@@ -641,6 +734,47 @@ mod tests {
         });
     }
 
+    #[test]
+    fn test_single_disk_default_cycle_uses_long_interval_without_maintenance_features() {
+        assert_eq!(
+            single_disk_default_cycle_secs(ScannerMaintenanceFeatures::default()),
+            Some(SINGLE_DISK_SCANNER_CYCLE_SECS)
+        );
+    }
+
+    #[test]
+    fn test_single_disk_default_cycle_preserves_regular_cycle_for_lifecycle() {
+        assert_eq!(
+            single_disk_default_cycle_secs(ScannerMaintenanceFeatures {
+                lifecycle: true,
+                ..Default::default()
+            }),
+            None
+        );
+    }
+
+    #[test]
+    fn test_single_disk_default_cycle_preserves_regular_cycle_for_replication() {
+        assert_eq!(
+            single_disk_default_cycle_secs(ScannerMaintenanceFeatures {
+                replication: true,
+                ..Default::default()
+            }),
+            None
+        );
+    }
+
+    #[test]
+    fn test_single_disk_default_cycle_preserves_regular_cycle_on_inspection_failure() {
+        assert_eq!(
+            single_disk_default_cycle_secs(ScannerMaintenanceFeatures {
+                inspection_failed: true,
+                ..Default::default()
+            }),
+            None
+        );
+    }
+
     #[test]
     #[serial]
     fn test_cycle_interval_keeps_single_disk_cycle_with_explicit_speed() {

crates/scanner/src/scanner_io.rs

@@ -23,6 +23,10 @@ use metrics::counter;
 use rand::seq::SliceRandom as _;
 use rustfs_common::heal_channel::HealScanMode;
 use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete};
+use rustfs_config::{
+    DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS,
+    ENV_SCANNER_MAX_CONCURRENT_SET_SCANS,
+};
 use rustfs_ecstore::bucket::bucket_target_sys::BucketTargetSys;
 use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_ExpiryState;
 use rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle;
@@ -44,14 +48,143 @@ use rustfs_utils::path::path_join_buf;
 use s3s::dto::{BucketLifecycleConfiguration, ReplicationConfiguration};
 use std::collections::HashMap;
 use std::path::Path;
-use std::time::SystemTime;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::{Instant, SystemTime};
 use std::{fmt::Debug, sync::Arc};
 use time::OffsetDateTime;
-use tokio::sync::{Mutex, mpsc};
+use tokio::sync::{Mutex, Semaphore, mpsc};
 use tokio::time::Duration;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, warn};
 
+const METRIC_SCANNER_SET_SCAN_CONCURRENCY_LIMIT: &str = "rustfs_scanner_set_scan_concurrency_limit";
+const METRIC_SCANNER_DISK_SCAN_CONCURRENCY_LIMIT: &str = "rustfs_scanner_disk_scan_concurrency_limit";
+const METRIC_SCANNER_SET_SCAN_WAIT_SECONDS: &str = "rustfs_scanner_set_scan_wait_seconds";
+const METRIC_SCANNER_DISK_SCAN_WAIT_SECONDS: &str = "rustfs_scanner_disk_scan_wait_seconds";
+const METRIC_SCANNER_SET_SCANS_ACTIVE: &str = "rustfs_scanner_set_scans_active";
+const METRIC_SCANNER_SET_SCANS_QUEUED: &str = "rustfs_scanner_set_scans_queued";
+const METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE: &str = "rustfs_scanner_disk_bucket_scans_active";
+const METRIC_SCANNER_DISK_BUCKET_SCANS_QUEUED: &str = "rustfs_scanner_disk_bucket_scans_queued";
+
+struct SetScanActiveGuard {
+    active: Arc<AtomicUsize>,
+}
+
+impl SetScanActiveGuard {
+    fn new(active: Arc<AtomicUsize>) -> Self {
+        let active_count = active.fetch_add(1, Ordering::Relaxed) + 1;
+        metrics::gauge!(METRIC_SCANNER_SET_SCANS_ACTIVE).set(active_count as f64);
+        Self { active }
+    }
+}
+
+impl Drop for SetScanActiveGuard {
+    fn drop(&mut self) {
+        let active_count = decrement_atomic_usize(&self.active);
+        metrics::gauge!(METRIC_SCANNER_SET_SCANS_ACTIVE).set(active_count as f64);
+    }
+}
+
+struct DiskBucketScanActiveGuard {
+    active: Arc<AtomicUsize>,
+    pool: String,
+    set: String,
+}
+
+impl DiskBucketScanActiveGuard {
+    fn new(active: Arc<AtomicUsize>, pool: String, set: String) -> Self {
+        let active_count = active.fetch_add(1, Ordering::Relaxed) + 1;
+        metrics::gauge!(
+            METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE,
+            "pool" => pool.clone(),
+            "set" => set.clone()
+        )
+        .set(active_count as f64);
+        Self { active, pool, set }
+    }
+}
+
+impl Drop for DiskBucketScanActiveGuard {
+    fn drop(&mut self) {
+        let active_count = decrement_atomic_usize(&self.active);
+        metrics::gauge!(
+            METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE,
+            "pool" => self.pool.clone(),
+            "set" => self.set.clone()
+        )
+        .set(active_count as f64);
+    }
+}
+
+fn decrement_atomic_usize(counter: &AtomicUsize) -> usize {
+    counter
+        .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(1)))
+        .map(|previous| previous.saturating_sub(1))
+        .unwrap_or_else(|current| current)
+}
+
+fn record_disk_bucket_scans_queued(count: usize, pool: &str, set: &str) {
+    metrics::gauge!(
+        METRIC_SCANNER_DISK_BUCKET_SCANS_QUEUED,
+        "pool" => pool.to_owned(),
+        "set" => set.to_owned()
+    )
+    .set(count as f64);
+}
+
+fn decrement_disk_bucket_scans_queued(counter: &AtomicUsize, pool: &str, set: &str) {
+    let queued_count = decrement_atomic_usize(counter);
+    record_disk_bucket_scans_queued(queued_count, pool, set);
+}
+
+fn reset_set_scan_gauges() {
+    metrics::gauge!(METRIC_SCANNER_SET_SCAN_CONCURRENCY_LIMIT).set(0.0);
+    metrics::gauge!(METRIC_SCANNER_SET_SCANS_QUEUED).set(0.0);
+    metrics::gauge!(METRIC_SCANNER_SET_SCANS_ACTIVE).set(0.0);
+}
+
+fn reset_disk_bucket_scan_gauges(pool: &str, set: &str) {
+    metrics::gauge!(
+        METRIC_SCANNER_DISK_SCAN_CONCURRENCY_LIMIT,
+        "pool" => pool.to_owned(),
+        "set" => set.to_owned()
+    )
+    .set(0.0);
+    record_disk_bucket_scans_queued(0, pool, set);
+    metrics::gauge!(
+        METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE,
+        "pool" => pool.to_owned(),
+        "set" => set.to_owned()
+    )
+    .set(0.0);
+}
+
+fn scanner_concurrency_limit(configured: usize, available: usize) -> usize {
+    if available == 0 {
+        return 0;
+    }
+
+    if configured == 0 {
+        available
+    } else {
+        configured.min(available).max(1)
+    }
+}
+
+fn scanner_max_concurrent_set_scans(available: usize) -> usize {
+    scanner_concurrency_limit(
+        rustfs_utils::get_env_usize(ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS),
+        available,
+    )
+}
+
+fn scanner_max_concurrent_disk_scans(available: usize) -> usize {
+    scanner_concurrency_limit(
+        rustfs_utils::get_env_usize(ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS),
+        available,
+    )
+}
+
 fn record_set_scan_failure(first_err: &mut Option<Error>, err: Error) {
     if first_err.is_none() {
         *first_err = Some(err);
@@ -146,6 +279,7 @@ impl ScannerIO for ECStore {
         let all_buckets = self.list_bucket(&BucketOptions::default()).await?;
 
         if all_buckets.is_empty() {
+            reset_set_scan_gauges();
             if let Err(e) = updates.send(DataUsageInfo::default()).await {
                 error!("Failed to send data usage info: {}", e);
             }
@@ -156,6 +290,24 @@ impl ScannerIO for ECStore {
         for pool in self.pools.iter() {
             total_results += pool.disk_set.len();
         }
+        if total_results == 0 {
+            warn!("nsscanner: no disk sets available for non-empty bucket list");
+            reset_set_scan_gauges();
+            return Ok(());
+        }
+
+        let set_scan_limit = scanner_max_concurrent_set_scans(total_results);
+        metrics::gauge!(METRIC_SCANNER_SET_SCAN_CONCURRENCY_LIMIT).set(set_scan_limit as f64);
+        debug!(
+            total_sets = total_results,
+            concurrency_limit = set_scan_limit,
+            "nsscanner: scanner set concurrency budget"
+        );
+        let set_scan_semaphore = Arc::new(Semaphore::new(set_scan_limit));
+        let queued_set_scans = Arc::new(AtomicUsize::new(total_results));
+        let active_set_scans = Arc::new(AtomicUsize::new(0));
+        metrics::gauge!(METRIC_SCANNER_SET_SCANS_QUEUED).set(total_results as f64);
+        metrics::gauge!(METRIC_SCANNER_SET_SCANS_ACTIVE).set(0.0);
 
         let results = vec![DataUsageCache::default(); total_results];
         let results_mutex: Arc<Mutex<Vec<DataUsageCache>>> = Arc::new(Mutex::new(results));
@@ -178,6 +330,9 @@ impl ScannerIO for ECStore {
                 let scan_mode_clone = scan_mode;
                 let results_mutex_clone = results_mutex.clone();
                 let first_err_mutex_clone = first_err_mutex.clone();
+                let set_scan_semaphore_clone = set_scan_semaphore.clone();
+                let queued_set_scans_clone = queued_set_scans.clone();
+                let active_set_scans_clone = active_set_scans.clone();
 
                 let (tx, mut rx) = mpsc::channel::<DataUsageCache>(1);
 
@@ -193,6 +348,25 @@ impl ScannerIO for ECStore {
                 let all_buckets_clone = all_buckets.clone();
                 // Spawn task to run the scanner
                 let scanner_fut = tokio::spawn(async move {
+                    let permit_wait = child_token_clone.clone();
+                    let permit_wait_start = Instant::now();
+                    let _permit = tokio::select! {
+                        permit = set_scan_semaphore_clone.acquire_owned() => match permit {
+                            Ok(permit) => permit,
+                            Err(_) => return,
+                        },
+                        _ = permit_wait.cancelled() => return,
+                    };
+                    metrics::histogram!(
+                        METRIC_SCANNER_SET_SCAN_WAIT_SECONDS,
+                        "pool" => pool_label.clone(),
+                        "set" => set_label.clone()
+                    )
+                    .record(permit_wait_start.elapsed().as_secs_f64());
+                    let queued_count = decrement_atomic_usize(&queued_set_scans_clone);
+                    metrics::gauge!(METRIC_SCANNER_SET_SCANS_QUEUED).set(queued_count as f64);
+                    let _active_guard = SetScanActiveGuard::new(active_set_scans_clone);
+
                     if let Err(e) = set_clone
                         .nsscanner_cache(child_token_clone.clone(), all_buckets_clone, tx, want_cycle_clone, scan_mode_clone)
                         .await
@@ -280,6 +454,8 @@ impl ScannerIO for ECStore {
         });
 
         let _ = join_all(wait_futs).await;
+        metrics::gauge!(METRIC_SCANNER_SET_SCANS_QUEUED).set(0.0);
+        metrics::gauge!(METRIC_SCANNER_SET_SCANS_ACTIVE).set(0.0);
 
         let _ = update_tx.send(());
 
@@ -300,15 +476,44 @@ impl ScannerIOCache for SetDisks {
         want_cycle: u64,
         scan_mode: HealScanMode,
     ) -> Result<()> {
+        let pool_label = self.pool_index.to_string();
+        let set_label = self.set_index.to_string();
+
         if buckets.is_empty() {
+            reset_disk_bucket_scan_gauges(&pool_label, &set_label);
             return Ok(());
         }
 
         let (disks, healing) = self.get_online_disks_with_healing(false).await;
         if disks.is_empty() {
             debug!("nsscanner_cache: no online disks available for set");
+            reset_disk_bucket_scan_gauges(&pool_label, &set_label);
             return Ok(());
         }
+        let disk_scan_limit = scanner_max_concurrent_disk_scans(disks.len());
+        metrics::gauge!(
+            METRIC_SCANNER_DISK_SCAN_CONCURRENCY_LIMIT,
+            "pool" => self.pool_index.to_string(),
+            "set" => self.set_index.to_string()
+        )
+        .set(disk_scan_limit as f64);
+        debug!(
+            pool = self.pool_index,
+            set = self.set_index,
+            online_disks = disks.len(),
+            concurrency_limit = disk_scan_limit,
+            "nsscanner_cache: scanner disk concurrency budget"
+        );
+        let disk_scan_semaphore = Arc::new(Semaphore::new(disk_scan_limit));
+        let queued_disk_bucket_scans = Arc::new(AtomicUsize::new(buckets.len()));
+        let active_disk_bucket_scans = Arc::new(AtomicUsize::new(0));
+        record_disk_bucket_scans_queued(buckets.len(), &pool_label, &set_label);
+        metrics::gauge!(
+            METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE,
+            "pool" => pool_label.clone(),
+            "set" => set_label.clone()
+        )
+        .set(0.0);
 
         let mut old_cache = DataUsageCache::default();
         old_cache.load(self.clone(), DATA_USAGE_CACHE_NAME).await?;
@@ -412,12 +617,58 @@ impl ScannerIOCache for SetDisks {
             let store_clone_clone = self.clone();
             let bucket_result_tx_clone_clone = bucket_result_tx_clone.clone();
             let disk_clone = disk.clone();
+            let disk_scan_semaphore_clone = disk_scan_semaphore.clone();
+            let queued_disk_bucket_scans_clone = queued_disk_bucket_scans.clone();
+            let active_disk_bucket_scans_clone = active_disk_bucket_scans.clone();
+            let pool_label_clone = pool_label.clone();
+            let set_label_clone = set_label.clone();
             futs.push(tokio::spawn(async move {
-                while let Some(bucket) = bucket_rx_mutex_clone.lock().await.recv().await {
+                loop {
+                    let Some(bucket) = bucket_rx_mutex_clone.lock().await.recv().await else {
+                        break;
+                    };
+
                     if ctx_clone.is_cancelled() {
+                        decrement_disk_bucket_scans_queued(&queued_disk_bucket_scans_clone, &pool_label_clone, &set_label_clone);
                         break;
                     }
 
+                    let permit_wait = ctx_clone.clone();
+                    let permit_wait_start = Instant::now();
+                    let _permit = tokio::select! {
+                        permit = disk_scan_semaphore_clone.clone().acquire_owned() => match permit {
+                            Ok(permit) => permit,
+                            Err(_) => {
+                                decrement_disk_bucket_scans_queued(
+                                    &queued_disk_bucket_scans_clone,
+                                    &pool_label_clone,
+                                    &set_label_clone,
+                                );
+                                break;
+                            },
+                        },
+                        _ = permit_wait.cancelled() => {
+                            decrement_disk_bucket_scans_queued(
+                                &queued_disk_bucket_scans_clone,
+                                &pool_label_clone,
+                                &set_label_clone,
+                            );
+                            break;
+                        },
+                    };
+                    metrics::histogram!(
+                        METRIC_SCANNER_DISK_SCAN_WAIT_SECONDS,
+                        "pool" => pool_label_clone.clone(),
+                        "set" => set_label_clone.clone()
+                    )
+                    .record(permit_wait_start.elapsed().as_secs_f64());
+                    decrement_disk_bucket_scans_queued(&queued_disk_bucket_scans_clone, &pool_label_clone, &set_label_clone);
+                    let _active_guard = DiskBucketScanActiveGuard::new(
+                        active_disk_bucket_scans_clone.clone(),
+                        pool_label_clone.clone(),
+                        set_label_clone.clone(),
+                    );
+
                     debug!("nsscanner_disk: got bucket: {}", bucket.name);
 
                     let cache_name = path_join_buf(&[&bucket.name, DATA_USAGE_CACHE_NAME]);
@@ -532,6 +783,13 @@ impl ScannerIOCache for SetDisks {
         }
 
         let _ = join_all(futs).await;
+        record_disk_bucket_scans_queued(0, &pool_label, &set_label);
+        metrics::gauge!(
+            METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE,
+            "pool" => pool_label.clone(),
+            "set" => set_label.clone()
+        )
+        .set(0.0);
 
         drop(bucket_result_tx_clone);
 
@@ -721,6 +979,8 @@ impl ScannerIODisk for Disk {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use serial_test::serial;
+    use temp_env::with_var;
 
     #[test]
     fn record_set_scan_failure_preserves_first_error() {
@@ -750,6 +1010,49 @@ mod tests {
         assert!(err.to_string().contains("set failed"));
     }
 
+    #[test]
+    fn scanner_concurrency_limit_preserves_available_when_unconfigured() {
+        assert_eq!(scanner_concurrency_limit(0, 4), 4);
+    }
+
+    #[test]
+    fn scanner_concurrency_limit_caps_to_configured_value() {
+        assert_eq!(scanner_concurrency_limit(2, 4), 2);
+    }
+
+    #[test]
+    fn scanner_concurrency_limit_never_exceeds_available_work() {
+        assert_eq!(scanner_concurrency_limit(8, 4), 4);
+    }
+
+    #[test]
+    fn scanner_concurrency_limit_handles_no_available_work() {
+        assert_eq!(scanner_concurrency_limit(2, 0), 0);
+    }
+
+    #[test]
+    fn decrement_atomic_usize_saturates_at_zero() {
+        let counter = AtomicUsize::new(1);
+        assert_eq!(decrement_atomic_usize(&counter), 0);
+        assert_eq!(decrement_atomic_usize(&counter), 0);
+    }
+
+    #[test]
+    #[serial]
+    fn scanner_max_concurrent_set_scans_uses_env_cap() {
+        with_var(ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, Some("2"), || {
+            assert_eq!(scanner_max_concurrent_set_scans(4), 2);
+        });
+    }
+
+    #[test]
+    #[serial]
+    fn scanner_max_concurrent_disk_scans_uses_env_cap() {
+        with_var(ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, Some("1"), || {
+            assert_eq!(scanner_max_concurrent_disk_scans(4), 1);
+        });
+    }
+
     #[test]
     #[cfg(windows)]
     fn is_xl_meta_path_accepts_windows_separator() {