fix(scanner): account remote progress and preserve dispatch limits (#8485)

* fix(scanner): account remote progress and preserve dispatch limits

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(scanner): isolate remote progress by cycle

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/scanner-metrics/Cargo.toml

@@ -38,7 +38,7 @@ tokio = { workspace = true, features = ["sync","rt"] }
 
 [dev-dependencies]
 serde_json = { workspace = true }
-tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] }
 uuid = { workspace = true, features = ["v4"] }
 
 [lib]

crates/scanner-metrics/src/metrics.rs

@@ -880,6 +880,24 @@ pub struct TierRequestCount {
 
 const TIER_REQUEST_COUNTER_SLOTS: usize = TierRequestOperation::ALL.len() * TierRequestOutcome::ALL.len();
 
+/// Remote work belongs to one cycle; retained worker handles cannot update a later cycle.
+#[derive(Default)]
+pub struct RemoteScanProgress {
+    objects: AtomicU64,
+    directories: AtomicU64,
+}
+
+impl RemoteScanProgress {
+    /// Record authenticated per-walk deltas without changing process-local operation counters.
+    pub fn record(&self, objects: u64, directories: u64) {
+        for (counter, delta) in [(&self.objects, objects), (&self.directories, directories)] {
+            if delta != 0 {
+                let _ = counter.try_update(Ordering::Relaxed, Ordering::Relaxed, |value| Some(value.saturating_add(delta)));
+            }
+        }
+    }
+}
+
 pub struct Metrics {
     operations: Vec<AtomicU64>,
     latency: Vec<LockedLastMinuteLatency>,
@@ -889,6 +907,10 @@ pub struct Metrics {
     cycle_info: Arc<RwLock<Option<CurrentCycle>>>,
     current_scan_mode: AtomicU8,
     current_scan_cycle_work_active: AtomicBool,
+    // Cycle transitions and reports lock cycle_info before remote_scan_progress.
+    // Cycle transitions take cycle_info before this lock; never hold it across awaits
+    // or while acquiring another metrics lock. Workers update their own handle directly.
+    remote_scan_progress: Mutex<Arc<RemoteScanProgress>>,
     current_scan_cycle_objects_start: AtomicU64,
     current_scan_cycle_directories_start: AtomicU64,
     current_scan_cycle_bucket_drive_scans_start: AtomicU64,
@@ -1963,6 +1985,7 @@ impl Metrics {
             cycle_info: Arc::new(RwLock::new(None)),
             current_scan_mode: AtomicU8::new(HealScanMode::Unknown as u8),
             current_scan_cycle_work_active: AtomicBool::new(false),
+            remote_scan_progress: Mutex::new(Arc::new(RemoteScanProgress::default())),
             current_scan_cycle_objects_start: AtomicU64::new(0),
             current_scan_cycle_directories_start: AtomicU64::new(0),
             current_scan_cycle_bucket_drive_scans_start: AtomicU64::new(0),
@@ -2794,8 +2817,18 @@ impl Metrics {
 
     /// Publish a scanner cycle and its work-accounting baseline as one state transition.
     pub async fn start_scan_cycle_work_with_cycle(&self, cycle: CurrentCycle) -> ScanCycleWorkSnapshot {
+        self.start_scan_cycle_work_with_cycle_and_remote_progress(cycle, Arc::new(RemoteScanProgress::default()))
+            .await
+    }
+
+    /// Install the cycle's remote work handle together with its cycle identity.
+    pub async fn start_scan_cycle_work_with_cycle_and_remote_progress(
+        &self,
+        cycle: CurrentCycle,
+        remote_progress: Arc<RemoteScanProgress>,
+    ) -> ScanCycleWorkSnapshot {
         let mut current_cycle = self.cycle_info.write().await;
-        let snapshot = self.start_scan_cycle_work();
+        let snapshot = self.begin_scan_cycle_work(remote_progress);
         *current_cycle = Some(cycle);
         snapshot
     }
@@ -2910,6 +2943,15 @@ impl Metrics {
     }
 
     pub fn start_scan_cycle_work(&self) -> ScanCycleWorkSnapshot {
+        self.begin_scan_cycle_work(Arc::new(RemoteScanProgress::default()))
+    }
+
+    fn begin_scan_cycle_work(&self, remote_progress: Arc<RemoteScanProgress>) -> ScanCycleWorkSnapshot {
+        // The cycle transition serializes replacement; workers keep their own Arc.
+        *self
+            .remote_scan_progress
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) = remote_progress;
         let snapshot = self.scan_cycle_work_snapshot();
         let source_snapshot = self.scanner_source_work_values();
         let replication_repair_snapshot = self.scanner_replication_repair_work_values();
@@ -3027,9 +3069,22 @@ impl Metrics {
 
     fn scan_cycle_work_since(&self, start: ScanCycleWorkSnapshot) -> ScanCycleWorkSnapshot {
         let current = self.scan_cycle_work_snapshot();
+        let (remote_objects, remote_directories) = {
+            let remote = self
+                .remote_scan_progress
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            (remote.objects.load(Ordering::Relaxed), remote.directories.load(Ordering::Relaxed))
+        };
         ScanCycleWorkSnapshot {
-            objects_scanned: current.objects_scanned.saturating_sub(start.objects_scanned),
-            directories_scanned: current.directories_scanned.saturating_sub(start.directories_scanned),
+            objects_scanned: current
+                .objects_scanned
+                .saturating_sub(start.objects_scanned)
+                .saturating_add(remote_objects),
+            directories_scanned: current
+                .directories_scanned
+                .saturating_sub(start.directories_scanned)
+                .saturating_add(remote_directories),
             bucket_drive_scans: current.bucket_drive_scans.saturating_sub(start.bucket_drive_scans),
             bucket_drive_failures: current.bucket_drive_failures.saturating_sub(start.bucket_drive_failures),
             yield_events: current.yield_events.saturating_sub(start.yield_events),
@@ -3650,14 +3705,18 @@ pub type CloseDiskFn = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>>
 /// Register a new disk in the global path tracker and return two callbacks:
 /// one to update the current path and one to deregister the disk when done.
 pub async fn current_path_updater(disk: &str, initial: &str) -> (UpdateCurrentPathFn, CloseDiskFn) {
+    register_current_path(&global_metrics().current_paths, disk, initial).await
+}
+
+async fn register_current_path(
+    current_paths: &Arc<RwLock<HashMap<String, Arc<CurrentPathTracker>>>>,
+    disk: &str,
+    initial: &str,
+) -> (UpdateCurrentPathFn, CloseDiskFn) {
     let tracker = Arc::new(CurrentPathTracker::new(initial.to_string()));
     let disk_name = disk.to_string();
 
-    global_metrics()
-        .current_paths
-        .write()
-        .await
-        .insert(disk_name.clone(), Arc::clone(&tracker));
+    current_paths.write().await.insert(disk_name.clone(), Arc::clone(&tracker));
 
     let update_fn: UpdateCurrentPathFn = {
         let tracker = Arc::clone(&tracker);
@@ -3670,10 +3729,17 @@ pub async fn current_path_updater(disk: &str, initial: &str) -> (UpdateCurrentPa
 
     let done_fn: CloseDiskFn = {
         let disk = disk_name;
+        let current_paths = Arc::clone(current_paths);
+        let tracker = Arc::clone(&tracker);
         Arc::new(move || {
             let disk = disk.clone();
+            let current_paths = Arc::clone(&current_paths);
+            let tracker = Arc::clone(&tracker);
             Box::pin(async move {
-                global_metrics().current_paths.write().await.remove(&disk);
+                let mut paths = current_paths.write().await;
+                if paths.get(&disk).is_some_and(|current| Arc::ptr_eq(current, &tracker)) {
+                    paths.remove(&disk);
+                }
             })
         })
     };
@@ -3857,6 +3923,33 @@ mod tests {
         assert!(!global_metrics().current_paths.read().await.contains_key(&disk));
     }
 
+    #[tokio::test]
+    async fn stale_path_cleanup_preserves_a_replacement_scan() {
+        let metrics = Metrics::new();
+        let (_, close_first) = register_current_path(&metrics.current_paths, "remote-node/disk", "first-bucket").await;
+        let (_, close_second) = register_current_path(&metrics.current_paths, "remote-node/disk", "second-bucket").await;
+        close_first().await;
+        assert_eq!(metrics.get_current_paths().await, vec!["remote-node/disk/second-bucket"]);
+        close_second().await;
+        assert!(metrics.get_current_paths().await.is_empty());
+    }
+
+    #[tokio::test]
+    async fn cancelled_path_guard_cleans_up_its_remote_scan() {
+        let metrics = Metrics::new();
+        let (_, close_path) = register_current_path(&metrics.current_paths, "remote-node/disk", "bucket").await;
+        let guard = CloseDiskGuard::new(close_path);
+        assert_eq!(metrics.report().await.active_paths, vec!["remote-node/disk/bucket"]);
+        drop(guard);
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while !metrics.get_current_paths().await.is_empty() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("cancelled remote path guard must remove its registration");
+    }
+
     #[tokio::test]
     async fn report_counts_active_scan_paths() {
         let metrics = Metrics::new();
@@ -5061,6 +5154,86 @@ mod tests {
         assert_eq!(report.usage_freshness.last_cycle_cleared_dirty_buckets, 0);
     }
 
+    #[test]
+    fn remote_scan_progress_saturates_instead_of_wrapping() {
+        let remote = RemoteScanProgress::default();
+        remote.record(u64::MAX - 1, u64::MAX - 2);
+        remote.record(3, 4);
+        remote.record(0, 0);
+        assert_eq!(remote.objects.load(Ordering::Relaxed), u64::MAX);
+        assert_eq!(remote.directories.load(Ordering::Relaxed), u64::MAX);
+    }
+
+    #[tokio::test]
+    async fn saturated_remote_cycle_does_not_hide_later_remote_and_local_work() {
+        let metrics = Metrics::new();
+        let old_remote = Arc::new(RemoteScanProgress::default());
+        let previous = metrics
+            .start_scan_cycle_work_with_cycle_and_remote_progress(CurrentCycle::default(), old_remote.clone())
+            .await;
+        old_remote.record(u64::MAX, u64::MAX);
+        metrics.finish_scan_cycle_work(previous);
+
+        let remote = Arc::new(RemoteScanProgress::default());
+        let next = metrics
+            .start_scan_cycle_work_with_cycle_and_remote_progress(CurrentCycle::default(), remote.clone())
+            .await;
+        old_remote.record(9, 7);
+        let empty = metrics.report().await;
+        assert_eq!(empty.current_cycle_objects_scanned, 0);
+        assert_eq!(empty.current_cycle_directories_scanned, 0);
+        remote.record(2, 1);
+        metrics.operations[Metric::ScanObject as usize].fetch_add(3, Ordering::Relaxed);
+        metrics.operations[Metric::ScanFolder as usize].fetch_add(2, Ordering::Relaxed);
+        let active = metrics.report().await;
+        assert_eq!(active.current_cycle_objects_scanned, 5);
+        assert_eq!(active.current_cycle_directories_scanned, 3);
+        metrics.finish_scan_cycle_work(next);
+        let finished = metrics.report().await;
+        assert_eq!(finished.last_cycle_objects_scanned, 5);
+        assert_eq!(finished.last_cycle_directories_scanned, 3);
+    }
+
+    #[tokio::test]
+    async fn cycle_work_combines_local_and_remote_progress_without_changing_local_operations() {
+        let metrics = Metrics::new();
+        metrics.operations[Metric::ScanObject as usize].store(10, Ordering::Relaxed);
+        metrics.operations[Metric::ScanFolder as usize].store(5, Ordering::Relaxed);
+        let remote = Arc::new(RemoteScanProgress::default());
+        let start = metrics
+            .start_scan_cycle_work_with_cycle_and_remote_progress(CurrentCycle::default(), remote.clone())
+            .await;
+        metrics.operations[Metric::ScanObject as usize].fetch_add(3, Ordering::Relaxed);
+        metrics.operations[Metric::ScanFolder as usize].fetch_add(2, Ordering::Relaxed);
+        remote.record(7, 4);
+        let active = metrics.report().await;
+        assert_eq!(active.current_cycle_objects_scanned, 10);
+        assert_eq!(active.current_cycle_directories_scanned, 6);
+        assert_eq!(metrics.lifetime(Metric::ScanObject), 13);
+        assert_eq!(metrics.lifetime(Metric::ScanFolder), 7);
+        metrics.finish_scan_cycle_work(start);
+        let finished = metrics.report().await;
+        assert_eq!(finished.current_cycle_objects_scanned, 0);
+        assert_eq!(finished.current_cycle_directories_scanned, 0);
+        assert_eq!(finished.last_cycle_objects_scanned, 10);
+        assert_eq!(finished.last_cycle_directories_scanned, 6);
+
+        remote.record(9, 3);
+        let next_remote = Arc::new(RemoteScanProgress::default());
+        let next = metrics
+            .start_scan_cycle_work_with_cycle_and_remote_progress(CurrentCycle::default(), next_remote.clone())
+            .await;
+        let empty = metrics.report().await;
+        assert_eq!(empty.current_cycle_objects_scanned, 0);
+        assert_eq!(empty.current_cycle_directories_scanned, 0);
+        remote.record(11, 5);
+        next_remote.record(2, 1);
+        metrics.finish_scan_cycle_work(next);
+        let finished = metrics.report().await;
+        assert_eq!(finished.last_cycle_objects_scanned, 2);
+        assert_eq!(finished.last_cycle_directories_scanned, 1);
+    }
+
     #[tokio::test]
     async fn report_includes_active_scan_cycle_work() {
         let metrics = Metrics::new();

crates/scanner/src/remote_scanner/stream.rs

@@ -29,7 +29,7 @@ use crate::{
 use hmac::{Hmac, KeyInit, Mac};
 use rustfs_credentials::try_get_rpc_token;
 use rustfs_heal_contracts::heal_channel::HealScanMode;
-use rustfs_scanner_metrics::metrics::{Metric, Metrics};
+use rustfs_scanner_metrics::metrics::{CloseDiskGuard, Metric, Metrics, current_path_updater};
 use rustfs_utils::path::path_join_buf;
 use serde::{Deserialize, Serialize};
 use sha2::Sha256;
@@ -1017,71 +1017,79 @@ pub(crate) async fn scan_remote_bucket(
         )));
     }
 
-    let rpc_deadline = Instant::now() + NS_SCANNER_MAX_RPC_LIFETIME;
-    let open_deadline = (Instant::now() + NS_SCANNER_STALL_TIMEOUT).min(rpc_deadline);
-    let open_stream = disk.open_ns_scanner_stream(crate::NsScannerOpenRequest {
-        request_id,
-        server_epoch,
-        session_id,
-        session_sequence,
-        next_cycle,
-        leader_epoch,
-        body,
-        stall_timeout: Some(NS_SCANNER_STALL_TIMEOUT),
-    });
-    tokio::pin!(open_stream);
-    let reader = tokio::select! {
-        _ = ctx.cancelled() => {
-            budget.cancel_after_unreported_remote_progress();
-            return Err(RemoteScannerFailure::transport(StorageError::other(
-                "remote namespace scanner cancelled while opening stream",
-            )));
-        }
-        result = tokio::time::timeout_at(open_deadline, &mut open_stream) => {
-            match result {
-                Err(_) => {
-                    budget.cancel_after_unreported_remote_progress();
-                    return Err(RemoteScannerFailure::transport(StorageError::other(
-                        "remote namespace scanner response headers timed out",
-                    )));
-                }
-                Ok(Err(err)) if err.is_internode_http_status(429) => {
-                    return Err(RemoteScannerFailure::retry_bucket(StorageError::other(format!(
-                        "remote namespace scanner worker rejected zero-progress work: {err}"
-                    ))));
-                }
-                Ok(Err(err)) => {
-                    budget.cancel_after_unreported_remote_progress();
-                    return Err(RemoteScannerFailure::transport(StorageError::other(format!(
-                        "failed to open remote namespace scanner stream: {err}"
-                    ))));
-                }
-                Ok(Ok(reader)) => reader,
-            }
-        }
-    };
-    let authenticator = FrameAuthenticator::from_rpc_secret(request_id).map_err(|err| {
-        RemoteScannerFailure::transport(StorageError::other(format!(
-            "failed to authenticate remote namespace scanner stream: {err}"
-        )))
-    })?;
-
-    let stream_result = consume_remote_scanner_stream_until(
-        reader,
-        ctx,
-        budget.clone(),
-        RemoteScannerResponseExpectation {
-            bucket,
-            source: expected_source,
+    let disk_name = format!("{}{}", disk.host_name(), disk.path().display());
+    let (_, close_path) = current_path_updater(&disk_name, bucket).await;
+    let mut path_guard = CloseDiskGuard::new(close_path);
+    let result = async {
+        let rpc_deadline = Instant::now() + NS_SCANNER_MAX_RPC_LIFETIME;
+        let open_deadline = (Instant::now() + NS_SCANNER_STALL_TIMEOUT).min(rpc_deadline);
+        let open_stream = disk.open_ns_scanner_stream(crate::NsScannerOpenRequest {
+            request_id,
+            server_epoch,
+            session_id,
+            session_sequence,
             next_cycle,
-            scan_plan_digest,
-            tier_registry_generation,
-        },
-        authenticator,
-        rpc_deadline,
-    )
+            leader_epoch,
+            body,
+            stall_timeout: Some(NS_SCANNER_STALL_TIMEOUT),
+        });
+        tokio::pin!(open_stream);
+        let reader = tokio::select! {
+            _ = ctx.cancelled() => {
+                budget.cancel_after_unreported_remote_progress();
+                return Err(RemoteScannerFailure::transport(StorageError::other(
+                    "remote namespace scanner cancelled while opening stream",
+                )));
+            }
+            result = tokio::time::timeout_at(open_deadline, &mut open_stream) => {
+                match result {
+                    Err(_) => {
+                        budget.cancel_after_unreported_remote_progress();
+                        return Err(RemoteScannerFailure::transport(StorageError::other(
+                            "remote namespace scanner response headers timed out",
+                        )));
+                    }
+                    Ok(Err(err)) if err.is_internode_http_status(429) => {
+                        return Err(RemoteScannerFailure::retry_bucket(StorageError::other(format!(
+                            "remote namespace scanner worker rejected zero-progress work: {err}"
+                        ))));
+                    }
+                    Ok(Err(err)) => {
+                        budget.cancel_after_unreported_remote_progress();
+                        return Err(RemoteScannerFailure::transport(StorageError::other(format!(
+                            "failed to open remote namespace scanner stream: {err}"
+                        ))));
+                    }
+                    Ok(Ok(reader)) => reader,
+                }
+            }
+        };
+        let authenticator = FrameAuthenticator::from_rpc_secret(request_id).map_err(|err| {
+            RemoteScannerFailure::transport(StorageError::other(format!(
+                "failed to authenticate remote namespace scanner stream: {err}"
+            )))
+        })?;
+
+        let stream_result = consume_remote_scanner_stream_until(
+            reader,
+            ctx,
+            budget.clone(),
+            RemoteScannerResponseExpectation {
+                bucket,
+                source: expected_source,
+                next_cycle,
+                scan_plan_digest,
+                tier_registry_generation,
+            },
+            authenticator,
+            rpc_deadline,
+        )
+        .await;
+        finish_remote_scanner_stream(stream_result, budget.as_ref())
+    }
     .await;
-    finish_remote_scanner_stream(stream_result, budget.as_ref())
+    path_guard.close().await;
+    result
 }
 
 fn finish_remote_scanner_stream(

crates/scanner/src/remote_scanner/stream/tests.rs

@@ -917,18 +917,20 @@ async fn complete_terminal_frame_reconciles_progress_and_usage() {
     let (mut writer, reader) = tokio::io::duplex(4096);
     tokio::spawn(async move {
         let mut sequence = 0;
-        write_frame(
-            &mut writer,
-            &writer_auth,
-            &mut sequence,
-            &RemoteScannerFrame::progress(RemoteScannerProgress {
-                objects_scanned: 2,
-                directories_started: 1,
-                ..Default::default()
-            }),
-        )
-        .await
-        .expect("progress should write");
+        for _ in 0..2 {
+            write_frame(
+                &mut writer,
+                &writer_auth,
+                &mut sequence,
+                &RemoteScannerFrame::progress(RemoteScannerProgress {
+                    objects_scanned: 2,
+                    directories_started: 1,
+                    ..Default::default()
+                }),
+            )
+            .await
+            .expect("progress should write");
+        }
         write_frame(
             &mut writer,
             &writer_auth,
@@ -961,10 +963,26 @@ async fn complete_terminal_frame_reconciles_progress_and_usage() {
             ..Default::default()
         },
     );
-    let outcome =
-        consume_remote_scanner_stream(reader, parent, budget.clone(), "bucket", TEST_SOURCE, TEST_PLAN_DIGEST, reader_auth)
-            .await
-            .expect("stream should complete");
+    let metrics = Metrics::new();
+    let start = metrics
+        .start_scan_cycle_work_with_cycle_and_remote_progress(Default::default(), budget.remote_scan_progress())
+        .await;
+    let outcome = consume_remote_scanner_stream_until(
+        reader,
+        parent,
+        budget.clone(),
+        RemoteScannerResponseExpectation {
+            bucket: "bucket",
+            source: TEST_SOURCE,
+            next_cycle: TEST_NEXT_CYCLE,
+            scan_plan_digest: TEST_PLAN_DIGEST,
+            tier_registry_generation: 0,
+        },
+        reader_auth,
+        Instant::now() + NS_SCANNER_MAX_RPC_LIFETIME,
+    )
+    .await
+    .expect("stream should complete");
 
     let RemoteScannerOutcome::Complete {
         usage,
@@ -976,6 +994,15 @@ async fn complete_terminal_frame_reconciles_progress_and_usage() {
     assert_eq!(usage.entry.objects, 3);
     assert!(pending_maintenance_work);
     assert_eq!(budget.progress(), (3, 2));
+    let active = metrics.report().await;
+    assert_eq!(active.current_cycle_objects_scanned, 3);
+    assert_eq!(active.current_cycle_directories_scanned, 2);
+    assert_eq!(metrics.lifetime(Metric::ScanObject), 0);
+    assert_eq!(metrics.lifetime(Metric::ScanFolder), 0);
+    metrics.finish_scan_cycle_work(start);
+    let finished = metrics.report().await;
+    assert_eq!(finished.last_cycle_objects_scanned, 3);
+    assert_eq!(finished.last_cycle_directories_scanned, 2);
 }
 
 #[tokio::test]
@@ -1271,6 +1298,39 @@ fn uncertain_stream_failure_cancels_count_budget() {
     assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Objects));
 }
 
+#[tokio::test]
+async fn remote_progress_rejects_regression_and_counts_new_attempts_once() {
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, ScannerCycleBudgetConfig::default());
+    let metrics = Metrics::new();
+    let _start = metrics
+        .start_scan_cycle_work_with_cycle_and_remote_progress(Default::default(), budget.remote_scan_progress())
+        .await;
+    let mut last = RemoteScannerProgress::default();
+    let progress = RemoteScannerProgress {
+        objects_scanned: 4,
+        directories_started: 2,
+        entries_visited: 6,
+    };
+    assert!(apply_remote_progress(&budget, &mut last, progress).expect("first progress must advance"));
+    assert!(!apply_remote_progress(&budget, &mut last, progress).expect("duplicate progress must be accepted"));
+    let regression = RemoteScannerProgress {
+        objects_scanned: 3,
+        ..progress
+    };
+    assert!(apply_remote_progress(&budget, &mut last, regression).is_err());
+    assert_eq!(last, progress);
+    assert_eq!(metrics.report().await.current_cycle_objects_scanned, 4);
+    assert_eq!(metrics.report().await.current_cycle_directories_scanned, 2);
+
+    let mut retry = RemoteScannerProgress::default();
+    apply_remote_progress(&budget, &mut retry, progress).expect("new attempt counts repeated work");
+    assert_eq!(budget.progress(), (8, 4));
+    let report = metrics.report().await;
+    assert_eq!(report.current_cycle_objects_scanned, 8);
+    assert_eq!(report.current_cycle_directories_scanned, 4);
+}
+
 #[test]
 fn directory_entry_activity_counts_as_semantic_progress() {
     let parent = CancellationToken::new();

crates/scanner/src/scanner.rs

@@ -51,9 +51,9 @@ use rustfs_data_usage::observed_data_usage_is_newer;
 use rustfs_heal_contracts::heal_channel::HealScanMode;
 use rustfs_lock::{NamespaceLockGuard, error::LockError};
 use rustfs_scanner_metrics::metrics::{
-    CurrentCycle, Metric, Metrics, ScanCyclePartialReason, ScanCycleWorkSnapshot, ScannerUsageSaveResult, ScannerWorkSource,
-    emit_scan_cycle_complete, emit_scan_cycle_deferred, emit_scan_cycle_partial_with_source, emit_scan_cycle_superseded,
-    global_metrics,
+    CurrentCycle, Metric, Metrics, RemoteScanProgress, ScanCyclePartialReason, ScanCycleWorkSnapshot, ScannerUsageSaveResult,
+    ScannerWorkSource, emit_scan_cycle_complete, emit_scan_cycle_deferred, emit_scan_cycle_partial_with_source,
+    emit_scan_cycle_superseded, global_metrics,
 };
 use serde::{Deserialize, Serialize};
 use sha2::{Digest as _, Sha256};
@@ -1777,7 +1777,7 @@ where
     let now = Instant::now();
     cycle_info.started = Utc::now();
 
-    let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
+    let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone(), cycle_budget.remote_scan_progress()).await;
 
     // Refresh the storage-owned movement snapshot before reading background
     // heal state. A missing heal object yields an in-memory default; do not
@@ -2598,9 +2598,13 @@ struct ScannerCycleMetricsGuard {
 }
 
 impl ScannerCycleMetricsGuard {
-    async fn new(cycle: CurrentCycle) -> Self {
+    async fn new(cycle: CurrentCycle, remote_progress: Arc<RemoteScanProgress>) -> Self {
         Self {
-            start: Some(global_metrics().start_scan_cycle_work_with_cycle(cycle).await),
+            start: Some(
+                global_metrics()
+                    .start_scan_cycle_work_with_cycle_and_remote_progress(cycle, remote_progress)
+                    .await,
+            ),
         }
     }
 

crates/scanner/src/scanner/tests.rs

@@ -714,7 +714,7 @@ async fn checkpoint_runtime_handoff(store: Arc<MemoryConfigStore>, expected_cycl
     assert_eq!((cycle.next, saved_epoch), (expected_cycle, epoch));
     cycle.current = cycle.next;
     let ctx = CancellationToken::new();
-    let mut metrics = ScannerCycleMetricsGuard::new(cycle.clone()).await;
+    let mut metrics = ScannerCycleMetricsGuard::new(cycle.clone(), Arc::new(RemoteScanProgress::default())).await;
     tokio::time::pause();
     let budget = ScannerCycleBudget::new(
         &ctx,
@@ -1289,7 +1289,8 @@ async fn test_mark_scan_cycle_idle_clears_published_cycle_state() {
     };
 
     global_metrics().set_current_scan_mode(HealScanMode::Deep);
-    let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
+    let mut cycle_metrics_guard =
+        ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await;
 
     mark_scan_cycle_idle(&mut cycle_info, &mut cycle_metrics_guard).await;
 
@@ -1307,6 +1308,33 @@ async fn test_mark_scan_cycle_idle_clears_published_cycle_state() {
     global_metrics().set_cycle(None).await;
 }
 
+#[tokio::test]
+#[serial]
+async fn scanner_cycle_metrics_guard_installs_budget_remote_progress() {
+    let parent = CancellationToken::new();
+    let old_budget = ScannerCycleBudget::new(&parent, ScannerCycleBudgetConfig::default());
+    let mut guard = ScannerCycleMetricsGuard::new(CurrentCycle::default(), old_budget.remote_scan_progress()).await;
+    old_budget.record_remote_progress(3, 2);
+    guard.finish(CurrentCycle::default()).await;
+    let previous = global_metrics().report().await;
+    assert_eq!(previous.last_cycle_objects_scanned, 3);
+    assert_eq!(previous.last_cycle_directories_scanned, 2);
+
+    let budget = ScannerCycleBudget::new(&parent, ScannerCycleBudgetConfig::default());
+    let mut guard = ScannerCycleMetricsGuard::new(CurrentCycle::default(), budget.remote_scan_progress()).await;
+    budget.record_remote_progress(2, 1);
+    old_budget.record_remote_progress(9, 7);
+    let active = global_metrics().report().await;
+    assert_eq!(active.current_cycle_objects_scanned, 2);
+    assert_eq!(active.current_cycle_directories_scanned, 1);
+    guard.finish(CurrentCycle::default()).await;
+    old_budget.record_remote_progress(11, 5);
+    let finished = global_metrics().report().await;
+    assert_eq!(finished.last_cycle_objects_scanned, 2);
+    assert_eq!(finished.last_cycle_directories_scanned, 1);
+    global_metrics().set_cycle(None).await;
+}
+
 #[tokio::test]
 #[serial]
 async fn scanner_cycle_metrics_guard_covers_published_first_cycle_lifetime() {
@@ -1317,7 +1345,7 @@ async fn scanner_cycle_metrics_guard_covers_published_first_cycle_lifetime() {
         started: cycle_started,
         ..Default::default()
     };
-    let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
+    let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await;
     let setup_report = global_metrics().report().await;
     assert!(setup_report.current_cycle_active);
     assert_eq!(setup_report.current_cycle, 0);
@@ -1343,7 +1371,7 @@ async fn scanner_cycle_metrics_guard_keeps_active_cycle_published_during_finaliz
         started: Utc::now(),
         ..Default::default()
     };
-    let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
+    let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await;
 
     cycle_info.current = 0;
     tokio::task::yield_now().await;
@@ -1362,12 +1390,15 @@ async fn scanner_cycle_metrics_guard_keeps_active_cycle_published_during_finaliz
 #[tokio::test]
 #[serial]
 async fn scanner_cycle_metrics_guard_drop_clears_activity() {
-    let guard = ScannerCycleMetricsGuard::new(CurrentCycle {
-        current: 12,
-        next: 13,
-        started: Utc::now(),
-        ..Default::default()
-    })
+    let guard = ScannerCycleMetricsGuard::new(
+        CurrentCycle {
+            current: 12,
+            next: 13,
+            started: Utc::now(),
+            ..Default::default()
+        },
+        Arc::new(RemoteScanProgress::default()),
+    )
     .await;
     assert!(global_metrics().report().await.current_cycle_active);
 
@@ -1639,7 +1670,8 @@ async fn test_finalize_partial_scan_cycle_advances_and_persists_counter() {
         cycle_completed: vec![],
         started: Utc::now(),
     };
-    let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
+    let mut cycle_metrics_guard =
+        ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await;
 
     assert!(finalize_partial_scan_cycle(&ctx, store.clone(), &mut cycle_info, &mut revision, 1, &mut cycle_metrics_guard,).await);
 
@@ -1675,7 +1707,8 @@ async fn scanner_cycle_recovers_to_newer_durable_cache_floor() {
         cycle_completed: vec![],
         started: Utc::now(),
     };
-    let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
+    let mut cycle_metrics_guard =
+        ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await;
 
     assert!(
         persist_required_scanner_cycle_floor(
@@ -1715,7 +1748,8 @@ async fn scanner_cycle_rejects_invalid_cache_floor() {
         cycle_completed: vec![],
         started: Utc::now(),
     };
-    let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
+    let mut cycle_metrics_guard =
+        ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await;
 
     assert!(
         !persist_required_scanner_cycle_floor(
@@ -1736,7 +1770,8 @@ async fn scanner_cycle_rejects_invalid_cache_floor() {
         next: 12,
         ..Default::default()
     };
-    let mut max_cycle_metrics_guard = ScannerCycleMetricsGuard::new(max_cycle_info.clone()).await;
+    let mut max_cycle_metrics_guard =
+        ScannerCycleMetricsGuard::new(max_cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await;
     assert!(
         !persist_required_scanner_cycle_floor(
             &ctx,
@@ -4600,7 +4635,8 @@ async fn test_finalize_partial_scan_cycle_reports_persist_failure() {
         cycle_completed: vec![],
         started: Utc::now(),
     };
-    let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
+    let mut cycle_metrics_guard =
+        ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await;
 
     assert!(!finalize_partial_scan_cycle(&ctx, store, &mut cycle_info, &mut revision, 1, &mut cycle_metrics_guard,).await);
     assert_eq!(cycle_info.next, 13);

crates/scanner/src/scanner_budget.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use rustfs_scanner_metrics::metrics::RemoteScanProgress;
 use std::sync::{
     Arc,
     atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
@@ -73,6 +74,7 @@ pub struct ScannerCycleBudget {
     entries_visited: AtomicU64,
     last_progress_millis: AtomicU64,
     cycle_state_persisted: AtomicBool,
+    remote_scan_progress: Arc<RemoteScanProgress>,
 }
 
 impl ScannerCycleBudget {
@@ -137,6 +139,7 @@ impl ScannerCycleBudget {
             entries_visited: AtomicU64::new(0),
             last_progress_millis: AtomicU64::new(0),
             cycle_state_persisted: AtomicBool::new(false),
+            remote_scan_progress: Arc::new(RemoteScanProgress::default()),
         })
     }
 
@@ -237,7 +240,12 @@ impl ScannerCycleBudget {
         }
     }
 
+    pub(crate) fn remote_scan_progress(&self) -> Arc<RemoteScanProgress> {
+        Arc::clone(&self.remote_scan_progress)
+    }
+
     pub(crate) fn record_remote_progress(&self, objects_scanned: u64, directories_started: u64) {
+        self.remote_scan_progress.record(objects_scanned, directories_started);
         if self.track_progress || self.max_objects.is_some() {
             let objects = saturating_fetch_add(&self.objects_scanned, objects_scanned);
             if self.track_progress {

crates/scanner/src/scanner_io/guards.rs

@@ -546,15 +546,13 @@ pub(super) fn reset_disk_bucket_scan_gauges(pool: &str, set: &str) {
     record_disk_bucket_scans_active(0, pool, set);
 }
 
+// Foreground pressure is handled during work by disk admission and the sleeper.
+// Dispatch limits must not freeze a momentary request count for an entire cycle.
 pub(super) fn scanner_concurrency_limit(configured: usize, available: usize) -> usize {
     if available == 0 {
         return 0;
     }
 
-    if crate::workload_admission::foreground_workload_activity() > 0 {
-        return 1;
-    }
-
     if configured == 0 {
         available
     } else {

crates/scanner/src/scanner_io/io_cache.rs

@@ -469,21 +469,22 @@ impl ScannerIOCache for SetDisks {
                 "Scanner set assigned remote disks without namespace scanner support to coordinator-driven workers"
             );
         }
-        let disk_scan_limit = scanner_budgeted_concurrency_limit(
-            scanner_max_concurrent_disk_scans(workers.len()),
-            budget.requires_serial_progress_accounting(),
-        );
+        let dispatch_limit = scanner_max_concurrent_disk_scans(workers.len());
+        let serial_progress_accounting = budget.requires_serial_progress_accounting();
+        let disk_scan_limit = scanner_budgeted_concurrency_limit(dispatch_limit, serial_progress_accounting);
         record_disk_scan_concurrency_limit(&pool_label, &set_label, disk_scan_limit);
-        debug!(
+        tracing::info!(
             target: "rustfs::scanner::io",
             event = EVENT_SCANNER_SET_STATE,
             component = LOG_COMPONENT_SCANNER,
             subsystem = LOG_SUBSYSTEM_IO,
+            state = "disk_concurrency_budget",
+            reason = if serial_progress_accounting { "count_budget" } else { "configured_or_available_work" },
             pool = self.pool_index,
             set = self.set_index,
             online_disks = workers.len(),
+            dispatch_limit,
             concurrency_limit = disk_scan_limit,
-            state = "disk_concurrency_budget",
             "Scanner disk concurrency budget resolved"
         );
         let disk_scan_semaphore = Arc::new(Semaphore::new(disk_scan_limit));

crates/scanner/src/scanner_io/io_cycle.rs

@@ -506,22 +506,23 @@ where
         return Ok(ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_publication_epoch(publication_epoch));
     }
 
-    let set_scan_limit = scanner_budgeted_concurrency_limit(
-        scanner_max_concurrent_set_scans(total_results),
-        budget.requires_serial_progress_accounting(),
-    );
+    let dispatch_limit = scanner_max_concurrent_set_scans(total_results);
+    let serial_progress_accounting = budget.requires_serial_progress_accounting();
+    let set_scan_limit = scanner_budgeted_concurrency_limit(dispatch_limit, serial_progress_accounting);
     let bucket_failures = ScannerBucketFailureState::default();
     let pending_maintenance_work = Arc::new(AtomicBool::new(false));
     let cold_zero_walk_reuse_observed = Arc::new(AtomicBool::new(false));
     record_set_scan_concurrency_limit(set_scan_limit);
-    debug!(
+    tracing::info!(
         target: "rustfs::scanner::io",
         event = EVENT_SCANNER_SET_STATE,
         component = LOG_COMPONENT_SCANNER,
         subsystem = LOG_SUBSYSTEM_IO,
-        total_sets = total_results,
-        concurrency_limit = set_scan_limit,
         state = "concurrency_budget",
+        reason = if serial_progress_accounting { "count_budget" } else { "configured_or_available_work" },
+        total_sets = total_results,
+        dispatch_limit,
+        concurrency_limit = set_scan_limit,
         "Scanner set concurrency budget resolved"
     );
     let set_scan_semaphore = Arc::new(Semaphore::new(set_scan_limit));

crates/scanner/src/scanner_io/tests.rs

@@ -3959,39 +3959,63 @@ fn scanner_concurrency_limit_handles_no_available_work() {
 
 #[test]
 #[serial]
-fn scanner_concurrency_limit_yields_to_foreground_reads() {
+fn scanner_concurrency_limit_is_stable_across_foreground_read_transitions() {
     crate::reset_foreground_read_activity_for_test();
     crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test();
-    crate::set_foreground_read_activity(8);
-    assert_eq!(scanner_concurrency_limit(0, 4), 1);
-    assert_eq!(scanner_concurrency_limit(3, 4), 1);
+    for foreground_activity in [0, 1, 8, 0] {
+        crate::set_foreground_read_activity(foreground_activity);
+        assert_eq!(scanner_concurrency_limit(0, 4), 4);
+        assert_eq!(scanner_concurrency_limit(3, 4), 3);
+        assert_eq!(scanner_concurrency_limit(1, 4), 1);
+        assert_eq!(scanner_concurrency_limit(8, 0), 0);
+        assert_eq!(scanner_budgeted_concurrency_limit(scanner_concurrency_limit(3, 4), true), 1);
+        assert_eq!(scanner_budgeted_concurrency_limit(scanner_concurrency_limit(3, 4), false), 3);
+    }
     crate::reset_foreground_read_activity_for_test();
 }
 
 #[test]
 #[serial]
-fn scanner_concurrency_limit_yields_to_shared_foreground_pressure() {
+fn scanner_concurrency_limit_is_stable_across_shared_foreground_pressure() {
     crate::reset_foreground_read_activity_for_test();
     crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test();
-    install_scanner_workload_provider(WorkloadAdmissionRegistrySnapshot::new(vec![
-        WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, AdmissionState::Open).with_counts(Some(2), None, Some(16)),
-    ]));
-
-    assert_eq!(scanner_concurrency_limit(0, 4), 1);
-    assert_eq!(scanner_concurrency_limit(3, 4), 1);
-
+    for (state, active) in [
+        (AdmissionState::Open, Some(0)),
+        (AdmissionState::Open, Some(2)),
+        (AdmissionState::Saturated, None),
+        (AdmissionState::Open, Some(0)),
+    ] {
+        install_scanner_workload_provider(WorkloadAdmissionRegistrySnapshot::new(vec![
+            WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(active, None, Some(16)),
+        ]));
+        assert_eq!(scanner_concurrency_limit(0, 4), 4);
+        assert_eq!(scanner_concurrency_limit(3, 4), 3);
+    }
     crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test();
 }
 
 #[test]
 #[serial]
-fn scanner_concurrency_limit_yields_to_streaming_reads() {
+fn scanner_dispatch_limits_preserve_configured_caps_during_streaming_reads() {
     crate::reset_foreground_read_activity_for_test();
     crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test();
-    let _guard = crate::ForegroundReadGuard::new();
-
-    assert_eq!(scanner_concurrency_limit(0, 4), 1);
-    assert_eq!(scanner_concurrency_limit(3, 4), 1);
+    temp_env::with_vars(
+        [
+            (ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, Some("3")),
+            (ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, Some("2")),
+        ],
+        || {
+            crate::runtime_config::refresh_scanner_runtime_config_for_tests();
+            for streaming in [false, true, false] {
+                let _guard = streaming.then(crate::ForegroundReadGuard::new);
+                assert_eq!(scanner_max_concurrent_set_scans(4), 3);
+                assert_eq!(scanner_max_concurrent_disk_scans(4), 2);
+                assert_eq!(scanner_budgeted_concurrency_limit(scanner_max_concurrent_set_scans(4), true), 1);
+                assert_eq!(scanner_budgeted_concurrency_limit(scanner_max_concurrent_disk_scans(4), true), 1);
+            }
+        },
+    );
+    crate::runtime_config::refresh_scanner_runtime_config_for_tests();
 }
 
 #[test]