fix(scanner): protect single-disk foreground latency (#8022)

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

crates/scanner/src/remote_scanner/stream.rs

@@ -16,8 +16,8 @@
 use crate::RUSTFS_META_BUCKET;
 use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig};
 use crate::scanner_io::{
-    DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistResult, ScannerDiskScanOptions,
-    ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, cache_root_entry_info,
+    DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistContext, ScannerCheckpointPersistResult,
+    ScannerDiskScanOptions, ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, cache_root_entry_info,
     current_cache_root_or_prepare_with_generation, persist_scanner_checkpoint, scanner_set_disk_inventory,
 };
 use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION;
@@ -840,12 +840,15 @@ async fn scan_and_persist_local_bucket(
                 }
                 match persist_scanner_checkpoint(
                     set.clone(),
+                    ScannerCheckpointPersistContext {
+                        ctx: &scan_ctx,
+                        expected_publication_epoch,
+                        cycle: next_cycle,
+                        leader_epoch,
+                    },
                     &cache_name,
                     &checkpoint,
                     &mut revisions,
-                    expected_publication_epoch,
-                    next_cycle,
-                    leader_epoch,
                 )
                 .await
                 {

crates/scanner/src/scanner.rs

@@ -1778,9 +1778,7 @@ where
             mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
             return ScannerCycleOutcome::Failed;
         }
-        BackgroundHealInfoReadStatus::ErasureSd
-        | BackgroundHealInfoReadStatus::Loaded
-        | BackgroundHealInfoReadStatus::Missing => {}
+        BackgroundHealInfoReadStatus::Loaded | BackgroundHealInfoReadStatus::Missing => {}
     }
     let mut background_heal_info = background_heal_read.info;
     let background_heal_epoch = background_heal_read.expected_epoch;

crates/scanner/src/scanner/heal_info.rs

@@ -28,7 +28,6 @@ pub struct BackgroundHealInfo {
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(super) enum BackgroundHealInfoReadStatus {
-    ErasureSd,
     Loaded,
     Missing,
     Blocked,
@@ -66,15 +65,6 @@ pub(super) async fn read_background_heal_info_with_epoch<S>(storeapi: Arc<S>) ->
 where
     S: ScannerStorage,
 {
-    // Skip for ErasureSD setup
-    if storeapi.setup_is_erasure_sd().await {
-        return BackgroundHealInfoRead {
-            info: BackgroundHealInfo::default(),
-            expected_epoch: None,
-            status: BackgroundHealInfoReadStatus::ErasureSd,
-        };
-    }
-
     let expected_epoch = scanner_publication_epoch(storeapi.clone()).await;
     if expected_epoch.is_none() {
         return BackgroundHealInfoRead {
@@ -146,11 +136,6 @@ pub(super) async fn save_background_heal_info_for_epoch<S>(
 ) where
     S: ScannerStorage,
 {
-    // Skip for ErasureSD setup
-    if storeapi.setup_is_erasure_sd().await {
-        return;
-    }
-
     // Serialize to JSON
     let data = match serde_json::to_vec(&info) {
         Ok(data) => data,

crates/scanner/src/scanner_folder.rs

@@ -1806,6 +1806,14 @@ impl FolderScanner {
                     continue;
                 }
 
+                // Do not start another metadata read while foreground work is
+                // active. The post-object timer protects the next request only
+                // after the read has already been dispatched; this admission
+                // point keeps the scanner from extending a single-disk I/O
+                // burst across foreground requests.
+                if crate::workload_admission::foreground_workload_activity() > 0 {
+                    self.sleeper.sleep_folder().await;
+                }
                 let timer = self.sleeper.timer();
 
                 let heal_enabled = this_hash.mod_alt(

crates/scanner/src/scanner_io.rs

@@ -1609,8 +1609,8 @@ use dirty_usage::*;
 use guards::*;
 
 pub(crate) use cache::{
-    DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistResult, acquire_scanner_cache_locks,
-    current_cache_root_or_prepare_with_generation, persist_scanner_checkpoint,
+    DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerCheckpointPersistContext, ScannerCheckpointPersistResult,
+    acquire_scanner_cache_locks, current_cache_root_or_prepare_with_generation, persist_scanner_checkpoint,
 };
 pub use dirty_usage::{
     ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageMutationObserver,

crates/scanner/src/scanner_io/cache.rs

@@ -31,6 +31,41 @@ pub(crate) enum ScannerCheckpointPersistResult {
     Failed(StorageError),
 }
 
+pub(crate) struct ScannerCheckpointPersistContext<'a> {
+    pub(crate) ctx: &'a CancellationToken,
+    pub(crate) expected_publication_epoch: u64,
+    pub(crate) cycle: u64,
+    pub(crate) leader_epoch: u64,
+}
+
+const CHECKPOINT_FOREGROUND_QUIET_WAIT: Duration = Duration::from_secs(1);
+
+async fn wait_for_checkpoint_foreground_quiet(ctx: &CancellationToken) -> bool {
+    let deadline = tokio::time::Instant::now() + CHECKPOINT_FOREGROUND_QUIET_WAIT;
+    loop {
+        if crate::workload_admission::foreground_workload_activity() == 0 {
+            return true;
+        }
+
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        if remaining.is_zero() {
+            return false;
+        }
+
+        let backoff = Duration::from_millis(
+            crate::workload_admission::foreground_workload_activity()
+                .saturating_mul(10)
+                .min(250),
+        )
+        .max(Duration::from_millis(10))
+        .min(remaining);
+        tokio::select! {
+            _ = ctx.cancelled() => return false,
+            _ = tokio::time::sleep(backoff) => {}
+        }
+    }
+}
+
 /// Persist one bounded checkpoint and refresh its CAS revisions.
 ///
 /// Local and remote workers share the same publication/leader fencing and
@@ -38,23 +73,37 @@ pub(crate) enum ScannerCheckpointPersistResult {
 /// at the caller because those guards have different concrete types.
 pub(crate) async fn persist_scanner_checkpoint<S>(
     store: Arc<S>,
+    context: ScannerCheckpointPersistContext<'_>,
     cache_name: &str,
     checkpoint: &DataUsageCache,
     revisions: &mut DataUsageCacheRevisions,
-    expected_publication_epoch: u64,
-    cycle: u64,
-    leader_epoch: u64,
 ) -> ScannerCheckpointPersistResult
 where
     S: ScannerObjectIO + ScannerConfigObjectDelete,
 {
-    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(cycle, leader_epoch, store.clone())
+    let foreground_quiet = wait_for_checkpoint_foreground_quiet(context.ctx).await;
+    if !foreground_quiet && context.ctx.is_cancelled() {
+        return ScannerCheckpointPersistResult::FenceChanged;
+    }
+    if !foreground_quiet {
+        debug!(
+            target: "rustfs::scanner::io",
+            event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+            component = LOG_COMPONENT_SCANNER,
+            subsystem = LOG_SUBSYSTEM_IO,
+            cache_name,
+            state = "checkpoint_foreground_wait_expired",
+            "Scanner checkpoint foreground quiet wait expired; preserving bounded progress"
+        );
+    }
+
+    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(context.cycle, context.leader_epoch, store.clone())
         .await
         .is_err()
     {
         return ScannerCheckpointPersistResult::FenceChanged;
     }
-    if scanner_publication_admission_for_epoch(store.clone(), expected_publication_epoch)
+    if scanner_publication_admission_for_epoch(store.clone(), context.expected_publication_epoch)
         .await
         .is_none()
     {
@@ -62,16 +111,16 @@ where
     }
 
     if let Err(error) = checkpoint
-        .save_with_revisions_for_epoch(store.clone(), cache_name, revisions, expected_publication_epoch)
+        .save_with_revisions_for_epoch(store.clone(), cache_name, revisions, context.expected_publication_epoch)
         .await
     {
         return ScannerCheckpointPersistResult::Failed(error);
     }
 
-    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(cycle, leader_epoch, store.clone())
+    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(context.cycle, context.leader_epoch, store.clone())
         .await
         .is_err()
-        || scanner_publication_admission_for_epoch(store.clone(), expected_publication_epoch)
+        || scanner_publication_admission_for_epoch(store.clone(), context.expected_publication_epoch)
             .await
             .is_none()
     {

crates/scanner/src/scanner_io/io_cache.rs

@@ -1191,12 +1191,15 @@ impl ScannerIOCache for SetDisks {
                                 }
                                 match persist_scanner_checkpoint(
                                     store_clone_clone.clone(),
+                                    ScannerCheckpointPersistContext {
+                                        ctx: &ctx_clone,
+                                        expected_publication_epoch: expected_publication_epoch_clone,
+                                        cycle: want_cycle,
+                                        leader_epoch,
+                                    },
                                     cache_name.as_str(),
                                     &checkpoint,
                                     &mut revisions,
-                                    expected_publication_epoch_clone,
-                                    want_cycle,
-                                    leader_epoch,
                                 )
                                 .await
                                 {