diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs
index c0c97231e..1bef2760e 100644
--- a/crates/ecstore/src/cache_value/metacache_set.rs
+++ b/crates/ecstore/src/cache_value/metacache_set.rs
@@ -12,9 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(test)]
+use crate::disk::DiskIoClass;
 use crate::disk::disk_store::{get_drive_walkdir_peek_timeout, get_drive_walkdir_stall_timeout};
 use crate::disk::error::DiskError;
-use crate::disk::{self, DiskAPI, DiskStore, WalkDirOptions};
+use crate::disk::{self, DiskAPI, DiskStore, WalkDirOptions, current_disk_io_context, with_optional_disk_io_context};
 use futures::future::join_all;
 use metrics::counter;
 use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetacacheReader, is_io_eof};
@@ -245,6 +247,8 @@ pub struct ListPathRawOptions {
     pub(crate) test_fallback_reader_behaviors: Vec<TestReaderBehavior>,
     #[cfg(test)]
     pub(crate) peek_timeout: Option<Duration>,
+    #[cfg(test)]
+    pub(crate) test_disk_io_context_observed: Option<Arc<AtomicBool>>,
     // pub agreed: Option<Arc<dyn Fn(MetaCacheEntry) + Send + Sync>>,
     // pub partial: Option<Arc<dyn Fn(MetaCacheEntries, &[Option<Error>]) + Send + Sync>>,
     // pub finished: Option<Arc<dyn Fn(&[Option<Error>]) + Send + Sync>>,
@@ -276,6 +280,8 @@ impl Clone for ListPathRawOptions {
             test_fallback_reader_behaviors: self.test_fallback_reader_behaviors.clone(),
             #[cfg(test)]
             peek_timeout: self.peek_timeout,
+            #[cfg(test)]
+            test_disk_io_context_observed: self.test_disk_io_context_observed.clone(),
             ..Default::default()
         }
     }
@@ -344,6 +350,7 @@ async fn list_path_raw_inner(
     let log_path = opts.path.clone();
 
     let mut jobs = JoinSet::new();
+    let disk_io_context = current_disk_io_context();
     let mut readers = Vec::with_capacity(opts.disks.len());
     let fds = Arc::new(TokioMutex::new(opts.fallback_disks.iter().flatten().cloned().collect::<VecDeque<_>>()));
     #[cfg(test)]
@@ -364,11 +371,19 @@ async fn list_path_raw_inner(
         let test_fallbacks_clone = test_fallbacks.clone();
         let cancel_rx_clone = cancel_rx.clone();
         let producer_errs_clone = producer_errs.clone();
+        let disk_io_context_clone = disk_io_context.clone();
         let (rd, wr) = tokio::io::duplex(64);
         readers.push(MetacacheReader::new(rd));
-        jobs.spawn(async move {
+        jobs.spawn(with_optional_disk_io_context(disk_io_context_clone, async move {
             let mut wr = PublishedBytesWriter::new(wr);
             #[cfg(test)]
+            if let Some(observer) = &opts_clone.test_disk_io_context_observed {
+                observer.store(
+                    crate::disk::current_disk_io_context().is_some_and(|context| context.class == DiskIoClass::Background),
+                    std::sync::atomic::Ordering::Release,
+                );
+            }
+            #[cfg(test)]
             let test_primary_error = if let Some(behavior) = opts_clone.test_reader_behaviors.get(disk_idx).cloned() {
                 match behavior {
                     TestReaderBehavior::Eof => return Ok(()),
@@ -652,7 +667,7 @@ async fn list_path_raw_inner(
 
             // warn!("list_path_raw: while need_fallback done");
             Ok(())
-        });
+        }));
     }
 
     let revjob_rx = rx.clone();
@@ -1168,6 +1183,68 @@ mod tests {
         assert_eq!(err, DiskError::ErasureReadQuorum);
     }
 
+    #[tokio::test]
+    async fn list_path_raw_joinset_producers_inherit_scanner_disk_io_class() {
+        let observed = Arc::new(AtomicBool::new(false));
+        crate::disk::with_background_disk_io(
+            CancellationToken::new(),
+            list_path_raw(
+                CancellationToken::new(),
+                ListPathRawOptions {
+                    disks: vec![None],
+                    min_disks: 1,
+                    test_reader_behaviors: vec![TestReaderBehavior::Eof],
+                    test_disk_io_context_observed: Some(observed.clone()),
+                    ..Default::default()
+                },
+            ),
+        )
+        .await
+        .expect("empty test producer should complete the listing");
+
+        assert!(
+            observed.load(std::sync::atomic::Ordering::Acquire),
+            "the per-disk JoinSet producer must retain the scanner background class"
+        );
+    }
+
+    #[tokio::test]
+    async fn scanner_joinset_walk_uses_the_local_disk_background_gate() {
+        let root = tempfile::tempdir().expect("local disk root should be created");
+        let endpoint = crate::disk::endpoint::Endpoint::try_from(root.path().to_str().expect("disk path should be UTF-8"))
+            .expect("endpoint should parse");
+        let local = Arc::new(
+            crate::disk::local::LocalDisk::new(&endpoint, false)
+                .await
+                .expect("local disk should open"),
+        );
+        let wrapper = crate::disk::disk_store::LocalDiskWrapper::new(local, false);
+        wrapper.make_volume("bucket").await.expect("bucket volume should be created");
+        let disk = Arc::new(crate::disk::Disk::Local(Box::new(wrapper.clone())));
+        let cancellation = CancellationToken::new();
+
+        crate::disk::with_background_disk_io(
+            cancellation.clone(),
+            list_path_raw(
+                cancellation,
+                ListPathRawOptions {
+                    disks: vec![Some(disk)],
+                    bucket: "bucket".to_string(),
+                    min_disks: 1,
+                    recursive: true,
+                    ..Default::default()
+                },
+            ),
+        )
+        .await
+        .expect("scanner metacache walk should complete");
+
+        assert!(
+            wrapper.background_io_admissions_for_tests() > 0,
+            "the LocalDiskWrapper walk should acquire background admission inside its JoinSet producer"
+        );
+    }
+
     #[tokio::test]
     async fn list_path_raw_returns_typed_quorum_error_for_multiple_drive_failures() {
         let err = list_path_raw(
diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs
index 0e057eabe..425aa2a74 100644
--- a/crates/ecstore/src/disk/disk_store.rs
+++ b/crates/ecstore/src/disk/disk_store.rs
@@ -12,6 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::disk::io_admission::{
+    AdmissionReader, BoxedChunkReader, DiskIoAdmission, DiskIoClass, DiskIoContext, current_disk_io_context,
+    with_disk_io_context, with_disk_io_permit,
+};
 use crate::disk::{
     CheckPartsResp, DataDirDeleteStatus, DeleteOptions, DiskAPI, DiskError, DiskInfo, DiskInfoOptions, DiskLocation, Endpoint,
     Error, FileInfoVersions, MmapCopyStageMetrics, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, Result,
@@ -813,6 +817,7 @@ impl Drop for DiskMetricWaitingGuard<'_> {
 pub(crate) struct ReconnectDiskHealthState {
     pub(crate) health: Arc<DiskHealthTracker>,
     pub(crate) metrics: Arc<DiskHealthMetricEpoch>,
+    pub(crate) io_admission: Arc<DiskIoAdmission>,
 }
 
 #[derive(Debug)]
@@ -1329,6 +1334,8 @@ pub struct LocalDiskWrapper {
     health: Arc<DiskHealthTracker>,
     /// Internal metrics epoch preserved across local disk reconnects.
     metrics: Arc<DiskHealthMetricEpoch>,
+    /// Per-drive foreground/background admission shared across wrapper reconnects.
+    io_admission: Arc<DiskIoAdmission>,
     /// Whether health checking is enabled
     health_check: bool,
     /// Cancellation token for monitoring tasks
@@ -1342,11 +1349,13 @@ pub struct LocalDiskWrapper {
 impl LocalDiskWrapper {
     /// Create a new LocalDiskWrapper
     pub fn new(disk: Arc<LocalDisk>, health_check: bool) -> Self {
+        let admission = DiskIoAdmission::new(disk.endpoint().to_string());
         Self::new_with_health_and_metrics(
             disk,
             health_check,
             Arc::new(DiskHealthTracker::new()),
             Arc::new(DiskHealthMetricEpoch::default()),
+            admission,
         )
     }
 
@@ -1386,11 +1395,13 @@ impl LocalDiskWrapper {
         health_check: bool,
         reconnect: Option<ReconnectDiskHealthState>,
     ) -> Self {
+        let endpoint = disk.endpoint().to_string();
         let reconnect = reconnect.unwrap_or_else(|| ReconnectDiskHealthState {
             health: Arc::new(DiskHealthTracker::new()),
             metrics: Arc::new(DiskHealthMetricEpoch::default()),
+            io_admission: DiskIoAdmission::new(endpoint),
         });
-        Self::new_with_health_and_metrics(disk, health_check, reconnect.health, reconnect.metrics)
+        Self::new_with_health_and_metrics(disk, health_check, reconnect.health, reconnect.metrics, reconnect.io_admission)
     }
 
     fn new_with_health_and_metrics(
@@ -1398,6 +1409,7 @@ impl LocalDiskWrapper {
         health_check: bool,
         health: Arc<DiskHealthTracker>,
         metrics: Arc<DiskHealthMetricEpoch>,
+        io_admission: Arc<DiskIoAdmission>,
     ) -> Self {
         // Check environment variable for health check override.
         // Only enable if both param and env are true.
@@ -1408,6 +1420,7 @@ impl LocalDiskWrapper {
             disk,
             health,
             metrics,
+            io_admission,
             health_check: health_check && env_health_check,
             cancel_token: CancellationToken::new(),
             disk_id: Arc::new(RwLock::new(None)),
@@ -1421,6 +1434,7 @@ impl LocalDiskWrapper {
         ReconnectDiskHealthState {
             health: Arc::new(self.health.metric_epoch_for_reconnect()),
             metrics: self.metrics.clone(),
+            io_admission: self.io_admission.clone(),
         }
     }
 
@@ -1428,6 +1442,58 @@ impl LocalDiskWrapper {
         self.disk.clone()
     }
 
+    #[cfg(test)]
+    pub(crate) fn background_io_admissions_for_tests(&self) -> usize {
+        self.io_admission.background_admissions_for_tests()
+    }
+
+    #[cfg(test)]
+    pub(crate) fn background_io_active_for_tests(&self) -> bool {
+        self.io_admission.background_active_for_tests()
+    }
+
+    fn disk_io_context(&self) -> DiskIoContext {
+        current_disk_io_context().unwrap_or_else(|| DiskIoContext {
+            class: DiskIoClass::Foreground,
+            cancellation: CancellationToken::new(),
+            admission: None,
+            permit: None,
+        })
+    }
+
+    fn disk_walk_io_context(&self) -> DiskIoContext {
+        let mut context = self.disk_io_context();
+        context.admission = Some(self.io_admission.clone());
+        context.permit = None;
+        context
+    }
+
+    async fn acquire_disk_io_permit(&self) -> Result<Option<crate::disk::io_admission::DiskIoPermit>> {
+        if let Some(context) = current_disk_io_context() {
+            if let Some(permit) = context.permit
+                && permit.is_for(&self.io_admission)
+            {
+                return Ok(Some(permit));
+            }
+            if context
+                .admission
+                .as_ref()
+                .is_some_and(|admission| Arc::ptr_eq(admission, &self.io_admission))
+            {
+                return Ok(None);
+            }
+            if context.class == DiskIoClass::Background {
+                return self
+                    .io_admission
+                    .background(&context.cancellation)
+                    .await
+                    .map(Some)
+                    .map_err(DiskError::Io);
+            }
+        }
+        Ok(Some(self.io_admission.foreground()))
+    }
+
     pub fn get_object_path_if_local(&self, volume: &str, path: &str) -> crate::disk::error::Result<std::path::PathBuf> {
         self.disk.get_object_path(volume, path)
     }
@@ -1899,6 +1965,31 @@ impl LocalDiskWrapper {
     where
         F: FnOnce() -> Fut,
         Fut: std::future::Future<Output = Result<T>>,
+    {
+        self.track_disk_health_with_op_timeout_action_and_mutation_with_io_permit(
+            op,
+            move |io_permit| async move {
+                let _io_permit = io_permit;
+                operation().await
+            },
+            timeout_duration,
+            timeout_health_action,
+            mutation,
+        )
+        .await
+    }
+
+    async fn track_disk_health_with_op_timeout_action_and_mutation_with_io_permit<T, F, Fut>(
+        &self,
+        op: &'static str,
+        operation: F,
+        timeout_duration: Duration,
+        timeout_health_action: TimeoutHealthAction,
+        mutation: DiskMetricMutation,
+    ) -> Result<T>
+    where
+        F: FnOnce(Option<crate::disk::io_admission::DiskIoPermit>) -> Fut,
+        Fut: std::future::Future<Output = Result<T>>,
     {
         self.metrics.record_operation_call(op);
         // Check if disk is faulty
@@ -1921,6 +2012,8 @@ impl LocalDiskWrapper {
             return Err(err);
         }
 
+        let io_permit = self.acquire_disk_io_permit().await?;
+
         // Record operation start
         self.health.last_started.store(current_unix_nanos(), Ordering::Relaxed);
         let _waiting_guard = self.health.waiting_guard();
@@ -1928,7 +2021,7 @@ impl LocalDiskWrapper {
         let started = Instant::now();
 
         if timeout_duration == Duration::ZERO {
-            let result = operation().await;
+            let result = operation(io_permit).await;
             self.metrics.record_operation_latency(op, started.elapsed());
             self.record_result_error_metrics(&result);
             if result.is_ok() {
@@ -1938,7 +2031,7 @@ impl LocalDiskWrapper {
             return result;
         }
         // Execute the operation with timeout
-        let result = tokio::time::timeout(timeout_duration, operation()).await;
+        let result = tokio::time::timeout(timeout_duration, operation(io_permit)).await;
 
         match result {
             Ok(operation_result) => {
@@ -1989,11 +2082,12 @@ impl DiskAPI for LocalDiskWrapper {
     }
 
     async fn read_metadata(&self, volume: &str, path: &str) -> Result<Bytes> {
-        self.track_disk_health_with_op_and_timeout_action(
+        self.track_disk_health_with_op_timeout_action_and_mutation_with_io_permit(
             "read_metadata",
-            || async { self.disk.read_metadata(volume, path).await },
+            |io_permit| async { self.disk.read_metadata_with_permit(volume, path, io_permit).await },
             get_drive_metadata_timeout(),
             self.scanner_timeout_health_action(),
+            DiskMetricMutation::None,
         )
         .await
     }
@@ -2165,12 +2259,15 @@ impl DiskAPI for LocalDiskWrapper {
             opts.timeout_duration().unwrap_or_else(get_drive_walkdir_timeout)
         };
 
-        self.track_disk_health_with_op_and_timeout_action(
-            "walk_dir",
-            || async { self.disk.walk_dir(opts, wr).await },
-            timeout_duration,
-            // Listing/scanner backpressure should fail only the current walk, not poison drive health.
-            TimeoutHealthAction::IgnoreFailure,
+        with_disk_io_context(
+            self.disk_walk_io_context(),
+            self.track_disk_health_with_op_and_timeout_action(
+                "walk_dir",
+                || async { self.disk.walk_dir(opts, wr).await },
+                timeout_duration,
+                // Listing/scanner backpressure should fail only the current walk, not poison drive health.
+                TimeoutHealthAction::IgnoreFailure,
+            ),
         )
         .await
     }
@@ -2193,6 +2290,7 @@ impl DiskAPI for LocalDiskWrapper {
     }
 
     async fn delete_versions(&self, volume: &str, versions: Vec<FileInfoVersions>, opts: DeleteOptions) -> Vec<Option<Error>> {
+        let version_count = versions.len();
         self.metrics.record_operation_call("delete_versions");
         // Check if disk is faulty before proceeding
         if self.health.is_faulty() {
@@ -2206,6 +2304,11 @@ impl DiskAPI for LocalDiskWrapper {
             return vec![Some(e); versions.len()];
         }
 
+        let _io_permit = match self.acquire_disk_io_permit().await {
+            Ok(permit) => permit,
+            Err(error) => return vec![Some(error); version_count],
+        };
+
         // Record operation start
         self.health.last_started.store(current_unix_nanos(), Ordering::Relaxed);
         self.health.increment_waiting();
@@ -2322,19 +2425,25 @@ impl DiskAPI for LocalDiskWrapper {
         version_id: &str,
         opts: &ReadOptions,
     ) -> Result<FileInfo> {
-        self.track_disk_health_with_op(
+        self.track_disk_health_with_op_timeout_action_and_mutation_with_io_permit(
             "read_version",
-            || async { self.disk.read_version(org_volume, volume, path, version_id, opts).await },
+            |io_permit| async move {
+                with_disk_io_permit(io_permit, self.disk.read_version(org_volume, volume, path, version_id, opts)).await
+            },
             get_max_timeout_duration(),
+            TimeoutHealthAction::MarkFailure,
+            DiskMetricMutation::None,
         )
         .await
     }
 
     async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
-        self.track_disk_health_with_op(
+        self.track_disk_health_with_op_timeout_action_and_mutation_with_io_permit(
             "read_xl",
-            || async { self.disk.read_xl(volume, path, read_data).await },
+            |io_permit| async move { with_disk_io_permit(io_permit, self.disk.read_xl(volume, path, read_data)).await },
             get_max_timeout_duration(),
+            TimeoutHealthAction::MarkFailure,
+            DiskMetricMutation::None,
         )
         .await
     }
@@ -2362,21 +2471,25 @@ impl DiskAPI for LocalDiskWrapper {
     }
 
     async fn read_file(&self, volume: &str, path: &str) -> Result<crate::disk::FileReader> {
-        self.track_disk_health_with_op(
-            "read_file",
-            || async { self.disk.read_file(volume, path).await },
-            get_max_timeout_duration(),
-        )
-        .await
+        let reader = self
+            .track_disk_health_with_op(
+                "read_file",
+                || async { self.disk.read_file(volume, path).await },
+                get_max_timeout_duration(),
+            )
+            .await?;
+        Ok(Box::new(AdmissionReader::new(reader, self.io_admission.clone(), self.disk_io_context())))
     }
 
     async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<crate::disk::FileReader> {
-        self.track_disk_health_with_op(
-            "read_file_stream",
-            || async { self.disk.read_file_stream(volume, path, offset, length).await },
-            get_max_timeout_duration(),
-        )
-        .await
+        let reader = self
+            .track_disk_health_with_op(
+                "read_file_stream",
+                || async { self.disk.read_file_stream(volume, path, offset, length).await },
+                get_max_timeout_duration(),
+            )
+            .await?;
+        Ok(Box::new(AdmissionReader::new(reader, self.io_admission.clone(), self.disk_io_context())))
     }
 
     async fn read_file_stream_chunks(
@@ -2386,12 +2499,20 @@ impl DiskAPI for LocalDiskWrapper {
         offset: usize,
         length: usize,
     ) -> Result<Option<rustfs_rio::ChunkReaderBox>> {
-        self.track_disk_health_with_op(
-            "read_file_stream_chunks",
-            || async { self.disk.read_file_stream_chunks(volume, path, offset, length).await },
-            get_max_timeout_duration(),
-        )
-        .await
+        let reader = self
+            .track_disk_health_with_op(
+                "read_file_stream_chunks",
+                || async { self.disk.read_file_stream_chunks(volume, path, offset, length).await },
+                get_max_timeout_duration(),
+            )
+            .await?;
+        Ok(reader.map(|reader| {
+            Box::new(AdmissionReader::new(
+                BoxedChunkReader(reader),
+                self.io_admission.clone(),
+                self.disk_io_context(),
+            )) as rustfs_rio::ChunkReaderBox
+        }))
     }
 
     async fn read_file_mmap_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<bytes::Bytes> {
@@ -2524,8 +2645,14 @@ impl DiskAPI for LocalDiskWrapper {
     }
 
     async fn read_parts(&self, bucket: &str, paths: &[String]) -> Result<Vec<ObjectPartInfo>> {
-        self.track_disk_health_with_op("read_parts", || async { self.disk.read_parts(bucket, paths).await }, Duration::ZERO)
-            .await
+        self.track_disk_health_with_op_timeout_action_and_mutation_with_io_permit(
+            "read_parts",
+            |io_permit| async move { with_disk_io_permit(io_permit, self.disk.read_parts(bucket, paths)).await },
+            Duration::ZERO,
+            TimeoutHealthAction::MarkFailure,
+            DiskMetricMutation::None,
+        )
+        .await
     }
 
     async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>> {
@@ -2592,6 +2719,84 @@ mod tests {
         }
     }
 
+    struct StalledMetacacheWriter {
+        entered: Arc<tokio::sync::Notify>,
+    }
+
+    impl AsyncWrite for StalledMetacacheWriter {
+        fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll<io::Result<usize>> {
+            self.entered.notify_one();
+            Poll::Pending
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    #[tokio::test]
+    async fn scanner_walk_releases_disk_admission_while_output_is_backpressured() {
+        let root = tempfile::tempdir().expect("disk root should be created");
+        let endpoint =
+            Endpoint::try_from(root.path().to_str().expect("disk root should be UTF-8")).expect("endpoint should parse");
+        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should open"));
+        let wrapper = Arc::new(LocalDiskWrapper::new(disk, false));
+        wrapper.make_volume("bucket").await.expect("bucket should be created");
+        let object_path = root.path().join("bucket/object");
+        tokio::fs::create_dir_all(&object_path)
+            .await
+            .expect("object directory should be created");
+        tokio::fs::write(object_path.join("xl.meta"), b"scanner metadata")
+            .await
+            .expect("object metadata should be created");
+
+        let entered = Arc::new(tokio::sync::Notify::new());
+        let writer = StalledMetacacheWriter {
+            entered: entered.clone(),
+        };
+        let walk_wrapper = wrapper.clone();
+        let walk = tokio::spawn(async move {
+            crate::disk::with_background_disk_io(CancellationToken::new(), async move {
+                let mut writer = writer;
+                walk_wrapper
+                    .walk_dir(
+                        WalkDirOptions {
+                            bucket: "bucket".to_string(),
+                            recursive: true,
+                            skip_total_timeout: true,
+                            ..Default::default()
+                        },
+                        &mut writer,
+                    )
+                    .await
+            })
+            .await
+        });
+
+        tokio::time::timeout(Duration::from_secs(2), entered.notified())
+            .await
+            .expect("walk should reach the blocked output writer");
+        assert!(
+            !wrapper.background_io_active_for_tests(),
+            "filesystem read permits must be released before metacache output backpressure"
+        );
+
+        let checkpoint_permit = tokio::time::timeout(
+            Duration::from_secs(1),
+            crate::disk::with_background_disk_io(CancellationToken::new(), wrapper.acquire_disk_io_permit()),
+        )
+        .await
+        .expect("checkpoint admission should progress while the scanner consumer is paused")
+        .expect("checkpoint should acquire the disk gate");
+        drop(checkpoint_permit);
+        walk.abort();
+        let _ = walk.await;
+    }
+
     #[tokio::test]
     async fn owned_mutation_keeps_publication_guard_after_waiter_cancellation() {
         let drops = Arc::new(std::sync::atomic::AtomicUsize::new(0));
@@ -2742,6 +2947,7 @@ mod tests {
         let reconnect = ReconnectDiskHealthState {
             health: Arc::new(health.metric_epoch_for_reconnect()),
             metrics: metrics.clone(),
+            io_admission: DiskIoAdmission::new("test-disk"),
         };
         metrics.record_operation_call("read_all");
         reconnect.metrics.record_timeout_error();
diff --git a/crates/ecstore/src/disk/io_admission.rs b/crates/ecstore/src/disk/io_admission.rs
new file mode 100644
index 000000000..3a9b63d04
--- /dev/null
+++ b/crates/ecstore/src/disk/io_admission.rs
@@ -0,0 +1,706 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Per-local-disk foreground-priority admission for scanner I/O.
+
+use std::collections::VecDeque;
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll};
+use std::time::Duration;
+
+use metrics::{counter, gauge, histogram};
+use rustfs_rio::ChunkReader;
+use tokio::io::{AsyncRead, ReadBuf};
+use tokio::sync::Notify;
+use tokio::time::Instant;
+use tokio_util::sync::CancellationToken;
+
+const BACKGROUND_FOREGROUND_BURST: u32 = 8;
+const BACKGROUND_MAX_WAIT: Duration = Duration::from_millis(100);
+
+tokio::task_local! {
+    static DISK_IO_CONTEXT: DiskIoContext;
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum DiskIoClass {
+    Foreground,
+    Background,
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct DiskIoContext {
+    pub(crate) class: DiskIoClass,
+    pub(crate) cancellation: CancellationToken,
+    pub(crate) admission: Option<Arc<DiskIoAdmission>>,
+    pub(crate) permit: Option<DiskIoPermit>,
+}
+
+/// Mark disk I/O issued by `future` as background work for local-disk admission.
+#[doc(hidden)]
+pub async fn with_background_disk_io<F: Future>(cancellation: CancellationToken, future: F) -> F::Output {
+    with_disk_io_context(
+        DiskIoContext {
+            class: DiskIoClass::Background,
+            cancellation,
+            admission: None,
+            permit: None,
+        },
+        future,
+    )
+    .await
+}
+
+pub(crate) async fn with_disk_io_context<F: Future>(context: DiskIoContext, future: F) -> F::Output {
+    DISK_IO_CONTEXT.scope(context, future).await
+}
+
+pub(crate) async fn with_disk_io_permit<F: Future>(permit: Option<DiskIoPermit>, future: F) -> F::Output {
+    let Some(permit) = permit else {
+        return future.await;
+    };
+    let mut context = current_disk_io_context().unwrap_or_else(|| DiskIoContext {
+        class: DiskIoClass::Foreground,
+        cancellation: CancellationToken::new(),
+        admission: None,
+        permit: None,
+    });
+    context.admission = Some(permit.admission());
+    with_disk_io_context(
+        DiskIoContext {
+            permit: Some(permit),
+            ..context
+        },
+        future,
+    )
+    .await
+}
+
+pub(crate) async fn with_optional_disk_io_context<F: Future>(context: Option<DiskIoContext>, future: F) -> F::Output {
+    match context {
+        Some(context) => with_disk_io_context(context, future).await,
+        None => future.await,
+    }
+}
+
+pub(crate) fn current_disk_io_context() -> Option<DiskIoContext> {
+    DISK_IO_CONTEXT.try_with(Clone::clone).ok()
+}
+
+/// Acquire one low-level permit only when the current walker explicitly owns a gate.
+/// Ordinary wrapper calls use their outer operation permit instead.
+pub(crate) async fn acquire_context_disk_io_permit() -> io::Result<Option<DiskIoPermit>> {
+    let Some(context) = current_disk_io_context() else {
+        return Ok(None);
+    };
+    let Some(admission) = context.admission else {
+        return Ok(None);
+    };
+    if let Some(permit) = context.permit
+        && permit.is_for(&admission)
+    {
+        return Ok(Some(permit));
+    }
+    match context.class {
+        DiskIoClass::Foreground => Ok(Some(admission.foreground())),
+        DiskIoClass::Background => Ok(Some(admission.background(&context.cancellation).await?)),
+    }
+}
+
+#[derive(Debug, Default)]
+struct AdmissionState {
+    foreground_completions: u32,
+    next_background_ticket: u64,
+    background_queue: VecDeque<u64>,
+    background_front_since: Option<Instant>,
+}
+
+/// One gate is shared by all handles to a local disk, including reconnects.
+/// Foreground I/O never waits on this gate. Background I/O runs exclusively
+/// while the disk is idle, or after a bounded foreground burst/wait.
+#[derive(Debug)]
+pub(crate) struct DiskIoAdmission {
+    state: Mutex<AdmissionState>,
+    changed: Notify,
+    disk: String,
+    foreground_active: AtomicUsize,
+    background_active: AtomicBool,
+    background_waiters: AtomicUsize,
+    #[cfg(test)]
+    background_admissions: AtomicUsize,
+}
+
+impl DiskIoAdmission {
+    pub(crate) fn new(disk: impl Into<String>) -> Arc<Self> {
+        Arc::new(Self {
+            state: Mutex::new(AdmissionState::default()),
+            changed: Notify::new(),
+            disk: disk.into(),
+            foreground_active: AtomicUsize::new(0),
+            background_active: AtomicBool::new(false),
+            background_waiters: AtomicUsize::new(0),
+            #[cfg(test)]
+            background_admissions: AtomicUsize::new(0),
+        })
+    }
+
+    #[cfg(test)]
+    pub(crate) fn background_admissions_for_tests(&self) -> usize {
+        self.background_admissions.load(Ordering::Relaxed)
+    }
+
+    #[cfg(test)]
+    pub(crate) fn background_active_for_tests(&self) -> bool {
+        self.background_active.load(Ordering::Acquire)
+    }
+
+    #[cfg(test)]
+    fn foreground_active_for_tests(&self) -> usize {
+        self.foreground_active.load(Ordering::Acquire)
+    }
+
+    pub(crate) fn foreground(self: &Arc<Self>) -> DiskIoPermit {
+        self.foreground_active.fetch_add(1, Ordering::AcqRel);
+        if self.background_waiters.load(Ordering::Acquire) > 0 {
+            self.record_state();
+        }
+        DiskIoPermit {
+            lease: Arc::new(DiskIoPermitLease {
+                admission: self.clone(),
+                class: DiskIoClass::Foreground,
+            }),
+        }
+    }
+
+    pub(crate) async fn background(self: &Arc<Self>, cancellation: &CancellationToken) -> std::io::Result<DiskIoPermit> {
+        let started_at = Instant::now();
+        let ticket = {
+            let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+            let ticket = state.next_background_ticket;
+            state.next_background_ticket = state.next_background_ticket.wrapping_add(1);
+            if state.background_queue.is_empty() && !self.background_active.load(Ordering::Acquire) {
+                state.background_front_since = Some(Instant::now());
+            }
+            state.background_queue.push_back(ticket);
+            self.background_waiters.fetch_add(1, Ordering::Release);
+            ticket
+        };
+        self.record_state();
+
+        let mut waiter = BackgroundWaiter {
+            admission: self.clone(),
+            ticket,
+            waiting: true,
+        };
+        loop {
+            let notified = self.changed.notified();
+            tokio::pin!(notified);
+            notified.as_mut().enable();
+
+            let granted = {
+                let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+                let is_front = state.background_queue.front() == Some(&ticket);
+                let bounded_wait_elapsed = is_front
+                    && state
+                        .background_front_since
+                        .is_some_and(|front_since| front_since.elapsed() >= BACKGROUND_MAX_WAIT);
+                if is_front
+                    && !self.background_active.load(Ordering::Acquire)
+                    && (self.foreground_active.load(Ordering::Acquire) == 0
+                        || state.foreground_completions >= BACKGROUND_FOREGROUND_BURST
+                        || bounded_wait_elapsed)
+                {
+                    self.background_active.store(true, Ordering::Release);
+                    state.background_queue.pop_front();
+                    state.background_front_since = None;
+                    self.background_waiters.fetch_sub(1, Ordering::AcqRel);
+                    state.foreground_completions = 0;
+                    true
+                } else {
+                    false
+                }
+            };
+
+            if granted {
+                waiter.waiting = false;
+                #[cfg(test)]
+                self.background_admissions.fetch_add(1, Ordering::Relaxed);
+                histogram!("rustfs_disk_io_admission_wait_seconds", "disk" => self.disk.clone(), "class" => "background")
+                    .record(started_at.elapsed().as_secs_f64());
+                counter!("rustfs_disk_io_admission_total", "disk" => self.disk.clone(), "class" => "background").increment(1);
+                self.record_state();
+                return Ok(DiskIoPermit {
+                    lease: Arc::new(DiskIoPermitLease {
+                        admission: self.clone(),
+                        class: DiskIoClass::Background,
+                    }),
+                });
+            }
+
+            let remaining = {
+                let state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+                if state.background_queue.front() == Some(&ticket) {
+                    state
+                        .background_front_since
+                        .map(|front_since| BACKGROUND_MAX_WAIT.saturating_sub(front_since.elapsed()))
+                } else {
+                    None
+                }
+            };
+            if let Some(remaining) = remaining.filter(|remaining| !remaining.is_zero()) {
+                tokio::select! {
+                    _ = cancellation.cancelled() => {
+                        return Err(std::io::Error::new(std::io::ErrorKind::Interrupted, "background disk I/O admission cancelled"));
+                    }
+                    _ = tokio::time::sleep(remaining) => {
+                        self.changed.notify_waiters();
+                    }
+                    _ = &mut notified => {}
+                }
+            } else {
+                tokio::select! {
+                    _ = cancellation.cancelled() => {
+                        return Err(std::io::Error::new(std::io::ErrorKind::Interrupted, "background disk I/O admission cancelled"));
+                    }
+                    _ = &mut notified => {}
+                }
+            }
+        }
+    }
+
+    fn record_state(&self) {
+        let (foreground_active, background_waiters) = {
+            let state = self.state.lock().unwrap_or_else(|error| error.into_inner());
+            (
+                if state.background_queue.is_empty() {
+                    0
+                } else {
+                    self.foreground_active.load(Ordering::Acquire)
+                },
+                state.background_queue.len(),
+            )
+        };
+        let foreground_active = u32::try_from(foreground_active).unwrap_or(u32::MAX);
+        let background_waiters = u32::try_from(background_waiters).unwrap_or(u32::MAX);
+        gauge!("rustfs_disk_io_admission_active", "disk" => self.disk.clone(), "class" => "foreground")
+            .set(f64::from(foreground_active));
+        gauge!("rustfs_disk_io_admission_active", "disk" => self.disk.clone(), "class" => "background").set(
+            if self.background_active.load(Ordering::Acquire) {
+                1.0
+            } else {
+                0.0
+            },
+        );
+        gauge!("rustfs_disk_io_admission_waiters", "disk" => self.disk.clone(), "class" => "background")
+            .set(f64::from(background_waiters));
+    }
+}
+
+struct BackgroundWaiter {
+    admission: Arc<DiskIoAdmission>,
+    ticket: u64,
+    waiting: bool,
+}
+
+impl Drop for BackgroundWaiter {
+    fn drop(&mut self) {
+        if !self.waiting {
+            return;
+        }
+        let mut state = self.admission.state.lock().unwrap_or_else(|error| error.into_inner());
+        if let Some(index) = state.background_queue.iter().position(|ticket| *ticket == self.ticket) {
+            state.background_queue.remove(index);
+            if index == 0 {
+                state.background_front_since =
+                    if state.background_queue.is_empty() || self.admission.background_active.load(Ordering::Acquire) {
+                        None
+                    } else {
+                        Some(Instant::now())
+                    };
+            }
+            self.admission.background_waiters.fetch_sub(1, Ordering::AcqRel);
+        }
+        drop(state);
+        self.admission.record_state();
+        self.admission.changed.notify_waiters();
+    }
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct DiskIoPermit {
+    lease: Arc<DiskIoPermitLease>,
+}
+
+impl DiskIoPermit {
+    pub(crate) fn is_for(&self, admission: &Arc<DiskIoAdmission>) -> bool {
+        Arc::ptr_eq(&self.lease.admission, admission)
+    }
+
+    fn admission(&self) -> Arc<DiskIoAdmission> {
+        self.lease.admission.clone()
+    }
+}
+
+#[derive(Debug)]
+struct DiskIoPermitLease {
+    admission: Arc<DiskIoAdmission>,
+    class: DiskIoClass,
+}
+
+pub(crate) fn spawn_blocking_with_disk_io_permit<T, F>(
+    io_permit: Option<DiskIoPermit>,
+    operation: F,
+) -> tokio::task::JoinHandle<T>
+where
+    T: Send + 'static,
+    F: FnOnce() -> T + Send + 'static,
+{
+    tokio::task::spawn_blocking(move || {
+        let _io_permit = io_permit;
+        operation()
+    })
+}
+
+type AdmissionWait = Pin<Box<dyn Future<Output = io::Result<DiskIoPermit>> + Send>>;
+
+impl Drop for DiskIoPermitLease {
+    fn drop(&mut self) {
+        match self.class {
+            DiskIoClass::Foreground => {
+                self.admission.foreground_active.fetch_sub(1, Ordering::AcqRel);
+                let record_state = self.admission.background_waiters.load(Ordering::Acquire) > 0;
+                if record_state {
+                    let mut state = self.admission.state.lock().unwrap_or_else(|error| error.into_inner());
+                    if !state.background_queue.is_empty() {
+                        state.foreground_completions = state.foreground_completions.saturating_add(1);
+                    }
+                }
+                if record_state {
+                    self.admission.record_state();
+                    self.admission.changed.notify_waiters();
+                }
+            }
+            DiskIoClass::Background => {
+                let mut state = self.admission.state.lock().unwrap_or_else(|error| error.into_inner());
+                self.admission.background_active.store(false, Ordering::Release);
+                state.background_front_since = if state.background_queue.is_empty() {
+                    None
+                } else {
+                    Some(Instant::now())
+                };
+                drop(state);
+                self.admission.record_state();
+                self.admission.changed.notify_waiters();
+            }
+        }
+    }
+}
+
+pub(crate) struct AdmissionReader<R> {
+    inner: R,
+    admission: Arc<DiskIoAdmission>,
+    context: DiskIoContext,
+    wait: Mutex<Option<AdmissionWait>>,
+    permit: Option<DiskIoPermit>,
+}
+
+impl<R> AdmissionReader<R> {
+    pub(crate) fn new(inner: R, admission: Arc<DiskIoAdmission>, context: DiskIoContext) -> Self {
+        Self {
+            inner,
+            admission,
+            context,
+            wait: Mutex::new(None),
+            permit: None,
+        }
+    }
+
+    fn poll_permit(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        if self.permit.is_some() {
+            return Poll::Ready(Ok(()));
+        }
+
+        if self.context.class == DiskIoClass::Foreground {
+            self.permit = Some(self.admission.foreground());
+            return Poll::Ready(Ok(()));
+        }
+
+        let mut wait = self.wait.lock().unwrap_or_else(|error| error.into_inner());
+        if wait.is_none() {
+            let admission = self.admission.clone();
+            let cancellation = self.context.cancellation.clone();
+            *wait = Some(Box::pin(async move { admission.background(&cancellation).await }));
+        }
+        let Some(wait_future) = wait.as_mut() else {
+            return Poll::Ready(Err(io::Error::other("background disk I/O admission waiter is missing")));
+        };
+        let result = wait_future.as_mut().poll(cx);
+        match result {
+            Poll::Ready(Ok(permit)) => {
+                wait.take();
+                drop(wait);
+                self.permit = Some(permit);
+                Poll::Ready(Ok(()))
+            }
+            Poll::Ready(Err(error)) => {
+                wait.take();
+                Poll::Ready(Err(error))
+            }
+            Poll::Pending => Poll::Pending,
+        }
+    }
+
+    fn finish_poll<T>(&mut self, result: Poll<io::Result<T>>) -> Poll<io::Result<T>> {
+        if result.is_ready() {
+            self.permit = None;
+        }
+        result
+    }
+}
+
+impl<R: AsyncRead + Unpin> AsyncRead for AdmissionReader<R> {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
+        match self.poll_permit(cx) {
+            Poll::Ready(Ok(())) => {}
+            Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
+            Poll::Pending => return Poll::Pending,
+        }
+        let result = Pin::new(&mut self.inner).poll_read(cx, buf);
+        self.finish_poll(result)
+    }
+}
+
+impl<R: ChunkReader + Unpin> ChunkReader for AdmissionReader<R> {
+    fn poll_read_chunk(mut self: Pin<&mut Self>, cx: &mut Context<'_>, max: usize) -> Poll<io::Result<Option<bytes::Bytes>>> {
+        match self.poll_permit(cx) {
+            Poll::Ready(Ok(())) => {}
+            Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
+            Poll::Pending => return Poll::Pending,
+        }
+        let result = Pin::new(&mut self.inner).poll_read_chunk(cx, max);
+        self.finish_poll(result)
+    }
+}
+
+pub(crate) struct BoxedChunkReader(pub(crate) rustfs_rio::ChunkReaderBox);
+
+impl AsyncRead for BoxedChunkReader {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.0).poll_read(cx, buf)
+    }
+}
+
+impl ChunkReader for BoxedChunkReader {
+    fn poll_read_chunk(self: Pin<&mut Self>, cx: &mut Context<'_>, max: usize) -> Poll<io::Result<Option<bytes::Bytes>>> {
+        Pin::new(self.get_mut().0.as_mut()).poll_read_chunk(cx, max)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    struct BlockingWorker(tokio::task::JoinHandle<()>);
+
+    async fn wait_for_background_waiters(admission: &DiskIoAdmission, expected: usize) {
+        tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                let waiters = admission
+                    .state
+                    .lock()
+                    .unwrap_or_else(|error| error.into_inner())
+                    .background_queue
+                    .len();
+                if waiters == expected {
+                    return;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("background admission waiter should reach expected count");
+    }
+
+    async fn wait_for_foreground_active(admission: &DiskIoAdmission, expected: usize) {
+        tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                let active = admission.foreground_active.load(Ordering::Acquire);
+                if active == expected {
+                    return;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("foreground disk read should retain its permit while pending");
+    }
+
+    #[tokio::test]
+    async fn foreground_bypasses_background_and_background_progress_is_bounded() {
+        let admission = DiskIoAdmission::new("disk-a");
+        let foreground = admission.foreground();
+        let background_admission = admission.clone();
+        let cancellation = CancellationToken::new();
+        let background = tokio::spawn(async move { background_admission.background(&cancellation).await });
+
+        wait_for_background_waiters(&admission, 1).await;
+        for _ in 0..BACKGROUND_FOREGROUND_BURST {
+            drop(admission.foreground());
+        }
+
+        let background_permit = tokio::time::timeout(Duration::from_secs(1), background)
+            .await
+            .expect("background admission should progress after the foreground burst")
+            .expect("background waiter task should complete")
+            .expect("background admission should not be cancelled");
+        assert_eq!(
+            admission.foreground_active.load(Ordering::Acquire),
+            1,
+            "foreground operations remain admitted while one background operation runs"
+        );
+        drop(background_permit);
+        drop(foreground);
+    }
+
+    #[tokio::test]
+    async fn cancellation_removes_a_queued_background_waiter() {
+        let admission = DiskIoAdmission::new("disk-a");
+        let background = admission
+            .background(&CancellationToken::new())
+            .await
+            .expect("first background operation should be admitted");
+        let cancellation = CancellationToken::new();
+        let waiter_admission = admission.clone();
+        let waiter_cancellation = cancellation.clone();
+        let waiter = tokio::spawn(async move { waiter_admission.background(&waiter_cancellation).await });
+        wait_for_background_waiters(&admission, 1).await;
+
+        cancellation.cancel();
+        let error = tokio::time::timeout(Duration::from_secs(1), waiter)
+            .await
+            .expect("cancelled background waiter should stop")
+            .expect("background waiter task should complete")
+            .expect_err("cancelled admission must not start disk I/O");
+        assert_eq!(error.kind(), io::ErrorKind::Interrupted);
+        assert_eq!(
+            admission
+                .state
+                .lock()
+                .unwrap_or_else(|error| error.into_inner())
+                .background_queue
+                .len(),
+            0
+        );
+        drop(background);
+    }
+
+    #[tokio::test]
+    async fn background_admission_is_isolated_per_disk() {
+        let disk_a = DiskIoAdmission::new("disk-a");
+        let disk_b = DiskIoAdmission::new("disk-b");
+        let _disk_a_permit = disk_a
+            .background(&CancellationToken::new())
+            .await
+            .expect("disk A background operation should be admitted");
+        let _disk_b_permit = tokio::time::timeout(Duration::from_secs(1), disk_b.background(&CancellationToken::new()))
+            .await
+            .expect("disk A must not block disk B")
+            .expect("disk B background operation should be admitted");
+    }
+
+    #[tokio::test]
+    async fn streamed_read_holds_admission_until_disk_read_resolves() {
+        let admission = DiskIoAdmission::new("disk-a");
+        let (mut writer, reader) = tokio::io::duplex(8);
+        let mut reader = AdmissionReader::new(
+            reader,
+            admission.clone(),
+            DiskIoContext {
+                class: DiskIoClass::Foreground,
+                cancellation: CancellationToken::new(),
+                admission: None,
+                permit: None,
+            },
+        );
+        let read = tokio::spawn(async move {
+            let mut bytes = [0; 4];
+            reader
+                .read_exact(&mut bytes)
+                .await
+                .expect("streamed disk read should complete");
+            bytes
+        });
+
+        wait_for_foreground_active(&admission, 1).await;
+        writer.write_all(b"rust").await.expect("test bytes should be written");
+        assert_eq!(read.await.expect("reader task should finish"), *b"rust");
+        wait_for_foreground_active(&admission, 0).await;
+    }
+
+    #[tokio::test]
+    async fn blocking_worker_keeps_permit_after_caller_timeout() {
+        let admission = DiskIoAdmission::new("disk-a");
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let worker = spawn_blocking_with_disk_io_permit(Some(admission.foreground()), move || {
+            started_tx.send(()).expect("worker start should be observed");
+            std::thread::sleep(Duration::from_millis(100));
+        });
+
+        started_rx.await.expect("blocking worker should start");
+        assert!(tokio::time::timeout(Duration::from_millis(10), worker).await.is_err());
+        assert_eq!(admission.foreground_active_for_tests(), 1);
+        wait_for_foreground_active(&admission, 0).await;
+    }
+
+    #[tokio::test]
+    async fn cloned_permit_remains_active_until_all_workers_finish() {
+        let admission = DiskIoAdmission::new("disk-a");
+        let permit = admission.foreground();
+        let worker_permit = permit.clone();
+        drop(permit);
+        assert_eq!(admission.foreground_active_for_tests(), 1);
+
+        drop(worker_permit);
+        assert_eq!(admission.foreground_active_for_tests(), 0);
+    }
+
+    #[tokio::test]
+    async fn scoped_foreground_permit_is_inherited_by_blocking_worker() {
+        let admission = DiskIoAdmission::new("disk-a");
+        let permit = admission.foreground();
+        let worker = with_disk_io_permit(Some(permit), async {
+            let worker_permit = acquire_context_disk_io_permit()
+                .await
+                .expect("context permit should be available")
+                .expect("scoped permit should be inherited");
+            let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+            let worker = spawn_blocking_with_disk_io_permit(Some(worker_permit), move || {
+                started_tx.send(()).expect("worker start should be observed");
+                std::thread::sleep(Duration::from_millis(100));
+            });
+            started_rx.await.expect("blocking worker should start");
+            BlockingWorker(worker)
+        })
+        .await;
+
+        assert_eq!(admission.foreground_active_for_tests(), 1);
+        assert!(tokio::time::timeout(Duration::from_millis(10), worker.0).await.is_err());
+        wait_for_foreground_active(&admission, 0).await;
+    }
+}
diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs
index 7480ddaae..b567a62c8 100644
--- a/crates/ecstore/src/disk/local.rs
+++ b/crates/ecstore/src/disk/local.rs
@@ -48,6 +48,7 @@ use crate::diagnostics::get::{
 #[cfg(test)]
 use crate::disk::HEALING_MARKER_PATH;
 use crate::disk::disk_store::{get_drive_walkdir_stall_timeout, get_object_disk_read_timeout};
+use crate::disk::io_admission::{DiskIoPermit, spawn_blocking_with_disk_io_permit};
 use crate::disk::{
     BUCKET_META_PREFIX, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN,
     CHECK_PART_VOLUME_NOT_FOUND, CheckPartsResp, ConditionalFileUpdate, DataDirDeleteStatus, DeleteOptions, DiskAPI, DiskInfo,
@@ -55,7 +56,7 @@ use crate::disk::{
     PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction, RUSTFS_META_BUCKET,
     RUSTFS_META_TMP_BUCKET, RUSTFS_META_TMP_DELETED_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
     STORAGE_FORMAT_FILE, STORAGE_FORMAT_FILE_BACKUP, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
-    conv_part_err_to_int,
+    acquire_context_disk_io_permit, conv_part_err_to_int,
     endpoint::Endpoint,
     error::{DiskError, Error, FileAccessDeniedWithContext, Result},
     error_conv::{to_access_error, to_file_error, to_unformatted_disk_error, to_volume_error},
@@ -2809,39 +2810,44 @@ where
     with_walk_stall_deadline(stall, fut).await?
 }
 
+async fn await_walk_stall_progress<T>(
+    stall: Option<Duration>,
+    mut task: tokio::task::JoinHandle<Result<T>>,
+    mut progress: tokio::sync::mpsc::UnboundedReceiver<()>,
+) -> Result<T> {
+    let Some(stall) = stall.filter(|stall| !stall.is_zero()) else {
+        return task.await.map_err(DiskError::from)?;
+    };
+
+    let mut deadline = Box::pin(tokio::time::sleep(stall));
+    loop {
+        tokio::select! {
+            result = &mut task => return result.map_err(DiskError::from)?,
+            progress_update = progress.recv() => match progress_update {
+                Some(()) => deadline.as_mut().reset(tokio::time::Instant::now() + stall),
+                None => return task.await.map_err(DiskError::from)?,
+            },
+            _ = &mut deadline => return Err(DiskError::Timeout),
+        }
+    }
+}
+
 async fn read_dir_entries_with_walk_stall(path: &Path, count: i32, stall: Option<Duration>) -> Result<Vec<String>> {
-    let mut entries = with_walk_stall_deadline(stall, fs::read_dir(path))
-        .await?
-        .map_err(to_file_error)?;
+    let mut entries = read_dir_with_walk_stall(path.to_path_buf(), stall).await?;
     let mut names = Vec::new();
     let mut remaining = count;
 
     loop {
-        let Some(entry) = with_walk_stall_deadline(stall, entries.next_entry())
-            .await?
-            .map_err(to_file_error)?
-        else {
-            break;
-        };
-        let name = entry.file_name().to_string_lossy().to_string();
-
-        if name.is_empty() || name == "." || name == ".." {
-            continue;
-        }
-
-        let file_type = with_walk_stall_deadline(stall, entry.file_type())
-            .await?
-            .map_err(to_file_error)?;
-        if file_type.is_file() {
-            names.push(name);
-        } else if file_type.is_dir() {
-            names.push(format!("{name}{SLASH_SEPARATOR}"));
+        let result_limit = if remaining > 0 {
+            (remaining as usize).min(WALK_DIR_BATCH_SIZE)
         } else {
-            continue;
-        }
-
-        remaining -= 1;
-        if remaining == 0 {
+            WALK_DIR_BATCH_SIZE
+        };
+        let (next_entries, next_names, reached_eof) = read_dir_entry_batch_with_walk_stall(entries, result_limit, stall).await?;
+        entries = next_entries;
+        remaining -= next_names.len() as i32;
+        names.extend(next_names);
+        if (count > 0 && remaining == 0) || reached_eof {
             break;
         }
     }
@@ -2849,6 +2855,52 @@ async fn read_dir_entries_with_walk_stall(path: &Path, count: i32, stall: Option
     Ok(names)
 }
 
+const WALK_DIR_BATCH_SIZE: usize = 32;
+
+async fn read_dir_with_walk_stall(path: PathBuf, stall: Option<Duration>) -> Result<std::fs::ReadDir> {
+    let io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;
+    let read_dir = spawn_blocking_with_disk_io_permit(io_permit, move || std::fs::read_dir(path).map_err(to_file_error));
+    let read_dir = with_walk_stall_deadline(stall, read_dir).await?.map_err(DiskError::from)?;
+    read_dir.map_err(|error| DiskError::from(to_file_error(error)))
+}
+
+async fn read_dir_entry_batch_with_walk_stall(
+    mut entries: std::fs::ReadDir,
+    result_limit: usize,
+    stall: Option<Duration>,
+) -> Result<(std::fs::ReadDir, Vec<String>, bool)> {
+    let io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;
+    let (progress_tx, progress_rx) = tokio::sync::mpsc::unbounded_channel();
+    let batch = spawn_blocking_with_disk_io_permit(io_permit, move || {
+        let mut names = Vec::with_capacity(result_limit);
+        let mut reached_eof = false;
+        for _ in 0..WALK_DIR_BATCH_SIZE {
+            if names.len() == result_limit {
+                break;
+            }
+            let Some(entry) = entries.next() else {
+                reached_eof = true;
+                break;
+            };
+            let _ = progress_tx.send(());
+            let entry = entry.map_err(to_file_error)?;
+            let name = entry.file_name().to_string_lossy().to_string();
+            if name.is_empty() || name == "." || name == ".." {
+                continue;
+            }
+            let file_type = entry.file_type().map_err(to_file_error)?;
+            let _ = progress_tx.send(());
+            if file_type.is_file() {
+                names.push(name);
+            } else if file_type.is_dir() {
+                names.push(format!("{name}{SLASH_SEPARATOR}"));
+            }
+        }
+        Ok::<_, DiskError>((entries, names, reached_eof))
+    });
+    await_walk_stall_progress(stall, batch, progress_rx).await
+}
+
 impl FileCacheReclaimReader {
     fn new(inner: File, reclaim_offset: u64, reclaim_len: usize, reclaim_on_drop: bool) -> Self {
         #[cfg(target_os = "macos")]
@@ -6899,7 +6951,8 @@ impl LocalDisk {
 
         let res = {
             if read_data {
-                self.read_all_data_with_dmtime(bucket, volume_dir, meta_path).await
+                let io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;
+                self.read_all_data_with_dmtime(bucket, volume_dir, meta_path, io_permit).await
             } else {
                 match self.read_metadata_with_dmtime(meta_path).await {
                     Ok(res) => Ok(res),
@@ -6932,6 +6985,7 @@ impl LocalDisk {
     #[hotpath::measure(impl_type = "LocalDisk")]
     async fn read_metadata_with_dmtime(&self, file_path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
         check_path_length(file_path.as_ref().to_string_lossy().as_ref())?;
+        let io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;
 
         // HP-12 item 1 (sub-change A): fold the open + fstat + bounded xl.meta
         // read into a single spawn_blocking dispatch instead of three separate
@@ -6944,30 +6998,31 @@ impl LocalDisk {
         //  - metadata failure -> to_file_error
         //  - parse failure    -> propagated verbatim from read_xl_meta_no_data_sync (`?`)
         let path = file_path.as_ref().to_path_buf();
-        let (data, modtime) = tokio::task::spawn_blocking(move || -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
-            // Read-only open, equivalent to O_RDONLY (get_readonly_options only sets read(true)).
-            let mut f = std::fs::File::open(&path).map_err(to_file_error)?;
+        let (data, modtime) =
+            spawn_blocking_with_disk_io_permit(io_permit, move || -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
+                // Read-only open, equivalent to O_RDONLY (get_readonly_options only sets read(true)).
+                let mut f = std::fs::File::open(&path).map_err(to_file_error)?;
 
-            let meta = f.metadata().map_err(to_file_error)?;
+                let meta = f.metadata().map_err(to_file_error)?;
 
-            if meta.is_dir() {
-                // fix use io::Error
-                return Err(Error::FileNotFound);
-            }
+                if meta.is_dir() {
+                    // fix use io::Error
+                    return Err(Error::FileNotFound);
+                }
 
-            let size = meta.len() as usize;
+                let size = meta.len() as usize;
 
-            let data = read_xl_meta_no_data_sync(&mut f, size)?;
+                let data = read_xl_meta_no_data_sync(&mut f, size)?;
 
-            let modtime = match meta.modified() {
-                Ok(md) => Some(OffsetDateTime::from(md)),
-                Err(_) => None,
-            };
+                let modtime = match meta.modified() {
+                    Ok(md) => Some(OffsetDateTime::from(md)),
+                    Err(_) => None,
+                };
 
-            Ok((data, modtime))
-        })
-        .await
-        .map_err(DiskError::from)??;
+                Ok((data, modtime))
+            })
+            .await
+            .map_err(DiskError::from)??;
 
         Ok((data, modtime))
     }
@@ -6975,11 +7030,12 @@ impl LocalDisk {
     #[hotpath::measure(impl_type = "LocalDisk")]
     async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef<Path>, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
         // TODO(backlog): add configurable timeout for read_all_data operations
-        let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?;
+        let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path, None).await?;
         Ok(data)
     }
 
     async fn read_part_metadata(&self, bucket: &str, volume_dir: &Path, path_str: &str) -> Result<ObjectPartInfo> {
+        let io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;
         let path = Path::new(path_str);
         let num = path
             .file_name()
@@ -6999,7 +7055,7 @@ impl LocalDisk {
         let metadata_path = self.io_get_object_path(bucket, path.to_string_lossy().as_ref());
         // A part's existence check, metadata read and decode share one dispatch.
         // Keep open errors unmapped for the existing missing-volume fallback.
-        let result = tokio::task::spawn_blocking(move || -> Result<_> {
+        let result = spawn_blocking_with_disk_io_permit(io_permit, move || -> Result<_> {
             let part_error = |error: String| ObjectPartInfo {
                 number: num,
                 error: Some(error),
@@ -7027,10 +7083,11 @@ impl LocalDisk {
     }
 
     async fn read_listing_metadata(&self, volume: &str, object_name: &str) -> Result<ListingMetadataRead> {
+        let io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;
         let object_dir = self.io_get_object_path(volume, object_name)?;
         let metadata_path = object_dir.join(STORAGE_FORMAT_FILE);
         let volume_dir = self.io_get_bucket_path(volume)?;
-        let result = tokio::task::spawn_blocking(move || {
+        let result = spawn_blocking_with_disk_io_permit(io_permit, move || {
             let (bytes, _) = read_all_data_std(&metadata_path)?;
             let file_meta = FileMeta::load(&bytes).ok();
             let data_dirs: HashSet<String> = file_meta
@@ -7113,6 +7170,7 @@ impl LocalDisk {
         volume: &str,
         volume_dir: impl AsRef<Path>,
         file_path: impl AsRef<Path>,
+        io_permit: Option<DiskIoPermit>,
     ) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
         // HP-12 item 1 (sub-change A): fold open + fstat + is_dir + try_reserve +
         // read_to_end into a single spawn_blocking dispatch. The closure mirrors
@@ -7128,7 +7186,7 @@ impl LocalDisk {
         // gating the volume fallback on the open error alone is equivalent to
         // the original code, where the fallback lived solely in the open match arm.
         let path = file_path.as_ref().to_path_buf();
-        let res = tokio::task::spawn_blocking(move || read_all_data_std(&path))
+        let res = spawn_blocking_with_disk_io_permit(io_permit, move || read_all_data_std(&path))
             .await
             .map_err(DiskError::from)?;
 
@@ -7215,7 +7273,10 @@ impl LocalDisk {
             .await;
         }
 
-        let (data, _) = match self.read_all_data_with_dmtime(volume, volume_dir.as_path(), &xlpath).await {
+        let (data, _) = match self
+            .read_all_data_with_dmtime(volume, volume_dir.as_path(), &xlpath, None)
+            .await
+        {
             Ok(data) => data,
             Err(DiskError::FileNotFound) => {
                 // `deleted` alone can be an explicit marker purge; only
@@ -8352,13 +8413,11 @@ impl LocalDisk {
                         if err == Error::FileNotFound || err == Error::IsNotRegular {
                             // NOT an object, append to stack (with slash)
                             // If dirObject, but no metadata (which is unexpected) we skip it.
-                            if !is_dir_obj
-                                && !with_walk_stall_deadline(
-                                    stall,
-                                    is_empty_dir(self.io_get_object_path(&opts.bucket, &meta.name)?),
-                                )
-                                .await?
-                            {
+                            if !is_dir_obj && {
+                                let _io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;
+                                !with_walk_stall_deadline(stall, is_empty_dir(self.io_get_object_path(&opts.bucket, &meta.name)?))
+                                    .await?
+                            } {
                                 meta.name.push_str(SLASH_SEPARATOR);
                                 // Conservative listings verify physical prefixes. Never-versioned
                                 // buckets use the bounded fast path, which only has to rule out
@@ -10393,6 +10452,7 @@ impl DiskAPI for LocalDisk {
     /// listing resilient while preserving existing full-enumeration behavior.
     #[tracing::instrument(level = "trace", skip_all)]
     async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result<Vec<String>> {
+        let _io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;
         if !origvolume.is_empty() {
             let origvolume_dir = self.io_get_bucket_path(origvolume)?;
             if !skip_access_checks(origvolume)
@@ -10405,21 +10465,16 @@ impl DiskAPI for LocalDisk {
         let volume_dir = self.io_get_bucket_path(volume)?;
         let dir_path_abs = self.io_get_object_path(volume, dir_path.trim_start_matches(SLASH_SEPARATOR))?;
 
-        // Whole-directory enumeration in one syscall path (see the wide-directory
-        // stall hazard on this fn): with `count < 0` this reads every entry, and
-        // the caller's stall budget bounds the entire call as a unit.
-        let entries = match os::read_dir(&dir_path_abs, count).await {
-            Ok(res) => res,
-            Err(e) => {
-                if e.kind() == ErrorKind::NotFound
+        let entries = match read_dir_entries_with_walk_stall(&dir_path_abs, count, None).await {
+            Ok(entries) => entries,
+            Err(err)
+                if err == DiskError::FileNotFound
                     && !skip_access_checks(volume)
-                    && let Err(e) = cached_access(&volume_dir).await
-                {
-                    return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
-                }
-
-                return Err(to_file_error(e).into());
+                    && let Err(err) = cached_access(&volume_dir).await =>
+            {
+                return Err(to_access_error(err, DiskError::VolumeAccessDenied).into());
             }
+            Err(err) => return Err(err),
         };
 
         Ok(entries)
@@ -10441,10 +10496,11 @@ impl DiskAPI for LocalDisk {
 
         let volume_dir = self.io_get_bucket_path(&opts.bucket)?;
 
-        if !skip_access_checks(&opts.bucket)
-            && let Err(e) = with_walk_stall_deadline(stall, cached_access(&volume_dir)).await?
-        {
-            return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
+        if !skip_access_checks(&opts.bucket) {
+            let _io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;
+            if let Err(e) = with_walk_stall_deadline(stall, cached_access(&volume_dir)).await? {
+                return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
+            }
         }
 
         let mut wr = wr;
@@ -10483,8 +10539,10 @@ impl DiskAPI for LocalDisk {
             let fpath =
                 self.io_get_object_path(&opts.bucket, path_join_buf(&[opts.base_dir.as_str(), STORAGE_FORMAT_FILE]).as_str())?;
 
-            if let Ok(meta) = with_walk_stall_deadline(stall, tokio::fs::metadata(&fpath)).await?
-                && meta.is_file()
+            if let Ok(meta) = {
+                let _io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;
+                with_walk_stall_deadline(stall, tokio::fs::metadata(&fpath)).await?
+            } && meta.is_file()
             {
                 skip_current_dir_object = true;
                 if let Ok(meta_bytes) = with_walk_stall_deadline(
@@ -11209,11 +11267,7 @@ impl DiskAPI for LocalDisk {
 
     #[tracing::instrument(level = "trace", skip_all)]
     async fn read_metadata(&self, volume: &str, path: &str) -> Result<Bytes> {
-        crate::hp_guard!("LocalDisk::read_metadata");
-        let file_path = self.io_get_object_path(volume, path)?;
-        let volume_dir = self.io_get_bucket_path(volume)?;
-        let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?;
-        Ok(data.into())
+        self.read_metadata_with_permit(volume, path, None).await
     }
 }
 
@@ -11233,6 +11287,25 @@ fn should_read_legacy_inline_part(fi: &FileInfo, storage_class_config: &crate::c
 }
 
 impl LocalDisk {
+    pub(crate) async fn read_metadata_with_permit(
+        &self,
+        volume: &str,
+        path: &str,
+        io_permit: Option<DiskIoPermit>,
+    ) -> Result<Bytes> {
+        crate::hp_guard!("LocalDisk::read_metadata");
+        let io_permit = match io_permit {
+            Some(io_permit) => Some(io_permit),
+            None => acquire_context_disk_io_permit().await.map_err(DiskError::Io)?,
+        };
+        let file_path = self.io_get_object_path(volume, path)?;
+        let volume_dir = self.io_get_bucket_path(volume)?;
+        let (data, _) = self
+            .read_all_data_with_dmtime(volume, volume_dir, file_path, io_permit)
+            .await?;
+        Ok(data.into())
+    }
+
     pub(crate) async fn rename_data_borrowed(
         &self,
         src_volume: &str,
@@ -11303,6 +11376,42 @@ mod test {
     use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
     use tracing_subscriber::fmt::MakeWriter;
 
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn read_dir_entries_skip_symlinks_without_truncating_listing() {
+        use std::os::unix::fs::symlink;
+
+        let dir = tempfile::tempdir().expect("directory should be created");
+        std::fs::write(dir.path().join("before"), b"file").expect("file should be created");
+        symlink(dir.path().join("before"), dir.path().join("link")).expect("symlink should be created");
+        std::fs::write(dir.path().join("after"), b"file").expect("file should be created");
+
+        let names = read_dir_entries_with_walk_stall(dir.path(), -1, None)
+            .await
+            .expect("directory listing should complete");
+
+        assert_eq!(names.len(), 2);
+        assert!(names.iter().any(|name| name == "before"));
+        assert!(names.iter().any(|name| name == "after"));
+        assert!(!names.iter().any(|name| name == "link"));
+    }
+
+    #[tokio::test]
+    async fn stalled_walk_batch_reports_timeout_at_one_stall_interval() {
+        let (release_tx, release_rx) = std::sync::mpsc::channel();
+        let task = tokio::task::spawn_blocking(move || {
+            let _ = release_rx.recv();
+            Ok::<_, DiskError>(())
+        });
+        let (_progress_tx, progress_rx) = tokio::sync::mpsc::unbounded_channel();
+
+        let started = Instant::now();
+        let result = await_walk_stall_progress(Some(Duration::from_millis(50)), task, progress_rx).await;
+        assert_eq!(result, Err(DiskError::Timeout));
+        assert!(started.elapsed() < Duration::from_millis(500));
+        release_tx.send(()).expect("blocking worker should be released");
+    }
+
     #[derive(Clone, Default)]
     struct CapturedLogs {
         buffer: Arc<Mutex<Vec<u8>>>,
diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs
index 87b51dd73..89fa4c76e 100644
--- a/crates/ecstore/src/disk/mod.rs
+++ b/crates/ecstore/src/disk/mod.rs
@@ -22,9 +22,15 @@ pub mod error_reduce;
 pub mod format;
 pub mod fs;
 pub mod health_state;
+mod io_admission;
 pub mod local;
 pub mod os;
 
+#[cfg(test)]
+pub(crate) use io_admission::DiskIoClass;
+pub use io_admission::with_background_disk_io;
+pub(crate) use io_admission::{acquire_context_disk_io_permit, current_disk_io_context, with_optional_disk_io_context};
+
 pub const RUSTFS_META_BUCKET: &str = ".rustfs.sys";
 pub const MIGRATING_META_BUCKET: &str = ".minio.sys";
 pub const RUSTFS_META_MULTIPART_BUCKET: &str = ".rustfs.sys/multipart";
diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs
index c2076e3fd..ff2844d6d 100644
--- a/crates/ecstore/src/lib.rs
+++ b/crates/ecstore/src/lib.rs
@@ -55,6 +55,7 @@ mod set_disk;
 mod storage_api_contracts;
 mod store;
 
+pub use disk::with_background_disk_io;
 pub use store::PoolMetaWriteGateStatus;
 
 // pub mod checksum;
diff --git a/crates/scanner/src/data_usage_define/persistence.rs b/crates/scanner/src/data_usage_define/persistence.rs
index c31c8f2a4..00851af64 100644
--- a/crates/scanner/src/data_usage_define/persistence.rs
+++ b/crates/scanner/src/data_usage_define/persistence.rs
@@ -420,6 +420,9 @@ impl DataUsageCache {
         if is_cache_publication_epoch_changed(err) {
             return false;
         }
+        if matches!(err, StorageError::Io(error) if error.kind() == std::io::ErrorKind::Interrupted) {
+            return false;
+        }
         !matches!(
             err,
             StorageError::Lock(_)
@@ -490,6 +493,7 @@ impl DataUsageCache {
         let sha256hex = (revision.is_some() && !buf.is_empty())
             .then(|| hex_simd::encode_to_string(Sha256::digest(buf), hex_simd::AsciiCase::Lower));
 
+        let persist_started = Instant::now();
         let save_result = Self::retry_save_op(path_type, timeout_duration, max_retries, || {
             let store_clone = store.clone();
             let path_clone = path.clone();
@@ -526,6 +530,8 @@ impl DataUsageCache {
             }
         })
         .await;
+        histogram!("rustfs_scanner_cache_persist_attempt_seconds", "path_type" => path_type, "retry_result" => if save_result.is_ok() { "success" } else { "error" })
+            .record(persist_started.elapsed().as_secs_f64());
         let Err(save_err) = save_result else {
             return Ok(());
         };
@@ -601,10 +607,13 @@ impl DataUsageCache {
         revisions: Option<&DataUsageCacheRevisions>,
         expected_epoch: Option<u64>,
     ) -> StorageResult<()> {
+        let serialize_started = Instant::now();
         let mut buf = Vec::new();
         self.serialize(&mut rmp_serde::Serializer::new(&mut buf))?;
         // Primary, backup, and retries read the same immutable encoded body.
         let buf = Bytes::from(buf);
+        counter!("rustfs_scanner_cache_serialized_bytes_total").increment(u64::try_from(buf.len()).unwrap_or(u64::MAX));
+        histogram!("rustfs_scanner_cache_serialize_seconds").record(serialize_started.elapsed().as_secs_f64());
         let timeout_duration = Self::cache_save_timeout();
 
         let path = path_join_buf(&[BUCKET_META_PREFIX, name]);
@@ -655,3 +664,17 @@ impl DataUsageCache {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn interrupted_checkpoint_admission_is_not_retried() {
+        let error = StorageError::Io(std::io::Error::new(std::io::ErrorKind::Interrupted, "cancelled while queued"));
+        assert!(
+            !DataUsageCache::should_retry_save_error(&error),
+            "a cancelled background admission must stop before retry backoff"
+        );
+    }
+}
diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs
index 8569f2180..fdc7a7f41 100644
--- a/crates/scanner/src/scanner_folder.rs
+++ b/crates/scanner/src/scanner_folder.rs
@@ -1806,14 +1806,6 @@ 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(
@@ -2416,7 +2408,12 @@ impl FolderScanner {
                             options.walkdir_stall_timeout,
                         ));
                     }
-                    if let Err(e) = list_path_raw(child_ctx_clone.clone(), options).await {
+                    if let Err(e) = crate::storage_api::ecstore_with_background_disk_io(
+                        child_ctx_clone.clone(),
+                        list_path_raw(child_ctx_clone.clone(), options),
+                    )
+                    .await
+                    {
                         if is_missing_path_disk_error(&e) {
                             debug!(
                                 target: "rustfs::scanner::folder",
diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs
index ea1c5fc48..4ae95d6a0 100644
--- a/crates/scanner/src/scanner_io/cache.rs
+++ b/crates/scanner/src/scanner_io/cache.rs
@@ -38,34 +38,6 @@ pub(crate) struct ScannerCheckpointPersistContext<'a> {
     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
@@ -85,22 +57,6 @@ where
     S: ScannerObjectIO + ScannerConfigObjectDelete,
     F: ScannerObjectIO,
 {
-    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,
@@ -118,10 +74,12 @@ where
         return ScannerCheckpointPersistResult::FenceChanged;
     }
 
-    if let Err(error) = checkpoint
-        .save_with_revisions_for_epoch(store.clone(), cache_name, revisions, context.expected_publication_epoch)
-        .await
-    {
+    let checkpoint_save = crate::storage_api::ecstore_with_background_disk_io(
+        context.ctx.clone(),
+        checkpoint.save_with_revisions_for_epoch(store.clone(), cache_name, revisions, context.expected_publication_epoch),
+    )
+    .await;
+    if let Err(error) = checkpoint_save {
         return ScannerCheckpointPersistResult::Failed(error);
     }
 
@@ -135,7 +93,12 @@ where
         return ScannerCheckpointPersistResult::FenceChanged;
     }
 
-    match DataUsageCache::read_revisions(store, cache_name).await {
+    match crate::storage_api::ecstore_with_background_disk_io(
+        context.ctx.clone(),
+        DataUsageCache::read_revisions(store, cache_name),
+    )
+    .await
+    {
         Ok(next_revisions) => {
             *revisions = next_revisions;
             ScannerCheckpointPersistResult::Saved
diff --git a/crates/scanner/src/scanner_io/io_disk.rs b/crates/scanner/src/scanner_io/io_disk.rs
index 40409ea7a..a77639c49 100644
--- a/crates/scanner/src/scanner_io/io_disk.rs
+++ b/crates/scanner/src/scanner_io/io_disk.rs
@@ -175,21 +175,25 @@ impl ScannerIODisk for Disk {
 
         let mut cache = cache;
 
-        let (lifecycle_config, _) = get_lifecycle_config(&cache.info.name)
-            .await
-            .unwrap_or_else(|_| (BucketLifecycleConfiguration::default(), OffsetDateTime::now_utc()));
+        let (lifecycle_config, _) =
+            crate::storage_api::ecstore_with_background_disk_io(ctx.clone(), get_lifecycle_config(&cache.info.name))
+                .await
+                .unwrap_or_else(|_| (BucketLifecycleConfiguration::default(), OffsetDateTime::now_utc()));
 
         if lifecycle_config.has_active_rules("") {
             cache.info.lifecycle = Some(Arc::new(lifecycle_config));
         }
 
-        let (replication_config, _) = get_replication_config(&cache.info.name).await.unwrap_or((
-            ReplicationConfiguration {
-                role: "".to_string(),
-                rules: vec![],
-            },
-            OffsetDateTime::now_utc(),
-        ));
+        let (replication_config, _) =
+            crate::storage_api::ecstore_with_background_disk_io(ctx.clone(), get_replication_config(&cache.info.name))
+                .await
+                .unwrap_or((
+                    ReplicationConfiguration {
+                        role: "".to_string(),
+                        rules: vec![],
+                    },
+                    OffsetDateTime::now_utc(),
+                ));
 
         if replication_config.has_active_rules("", true)
             && let Ok(targets) = BucketTargetSys::get().list_bucket_targets(&cache.info.name).await
@@ -197,7 +201,8 @@ impl ScannerIODisk for Disk {
             cache.info.replication = Some(Arc::new(ReplicationConfig::new(Some(replication_config), Some(targets))));
         }
 
-        if let Ok((object_lock_config, _)) = get_object_lock_config(&cache.info.name).await
+        if let Ok((object_lock_config, _)) =
+            crate::storage_api::ecstore_with_background_disk_io(ctx.clone(), get_object_lock_config(&cache.info.name)).await
             && object_lock_config_enabled(&object_lock_config)
         {
             cache.info.object_lock = Some(Arc::new(object_lock_config));
@@ -215,17 +220,20 @@ impl ScannerIODisk for Disk {
         .then_some(prefix_scan_scope)
         .flatten();
 
-        let result = scan_data_folder_scoped(
+        let result = crate::storage_api::ecstore_with_background_disk_io(
             ctx.clone(),
-            budget,
-            set_disks,
-            self.clone(),
-            cache,
-            updates,
-            scan_mode,
-            SCANNER_SLEEPER.clone(),
-            prefix_scan_scope,
-            checkpoint_tx,
+            scan_data_folder_scoped(
+                ctx.clone(),
+                budget,
+                set_disks,
+                self.clone(),
+                cache,
+                updates,
+                scan_mode,
+                SCANNER_SLEEPER.clone(),
+                prefix_scan_scope,
+                checkpoint_tx,
+            ),
         )
         .await;
 
diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs
index cf715f089..295a9ef62 100644
--- a/crates/scanner/src/storage_api.rs
+++ b/crates/scanner/src/storage_api.rs
@@ -21,6 +21,8 @@ use rustfs_lock::NamespaceLockWrapper;
 use serde::{Deserialize, Serialize};
 use tokio::sync::Notify;
 
+pub(crate) use rustfs_ecstore::with_background_disk_io as ecstore_with_background_disk_io;
+
 pub(crate) use s3s::dto::{
     BucketLifecycleConfiguration as EcstoreBucketLifecycleConfiguration, LifecycleRuleFilter as EcstoreLifecycleRuleFilter,
     ObjectLockConfiguration as EcstoreObjectLockConfiguration, VersioningConfiguration as EcstoreVersioningConfiguration,
diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md
index d60fe817f..2f497cffe 100644
--- a/docs/operations/scanner-runtime-controls.md
+++ b/docs/operations/scanner-runtime-controls.md
@@ -40,7 +40,7 @@ Bitrot cycle resolution differs because the canonical persistent key belongs to
 | `scanner.cycle_max_objects` | `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` | objects | `0` (`DEFAULT_SCANNER_CYCLE_MAX_OBJECTS`) | Caps objects processed by one cycle. `0` disables this budget. |
 | `scanner.cycle_max_directories` | `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` | directories | `0` (`DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES`) | Caps directories entered by one cycle. `0` disables this budget. |
 | `heal.bitrot_cycle` | `RUSTFS_SCANNER_BITROT_CYCLE_SECS` | seconds | `2592000` (`DEFAULT_HEAL_BITROT_CYCLE_SECS`, 30 days) | Controls periodic deep bitrot scans. `false`, `off`, `no`, or `disabled` disables periodic deep scans; `0`, `true`, `on`, or `yes` runs deep mode every scanner cycle. |
-| `scanner.idle_mode` | `RUSTFS_SCANNER_IDLE_MODE` | boolean | `true` (`DEFAULT_SCANNER_IDLE_MODE`) | Master switch for scanner throttling: preset sleeps plus the foreground-read backoff floor. `false` disables both and the scanner runs at full speed. |
+| `scanner.idle_mode` | `RUSTFS_SCANNER_IDLE_MODE` | boolean | `true` (`DEFAULT_SCANNER_IDLE_MODE`) | Enables preset scanner sleeps and foreground-read backoff. `false` disables those sleeps; per-disk foreground-priority admission remains active. |
 | `scanner.cache_save_timeout` | `RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS` | seconds | `14` (`DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS`) | Timeout for saving scanner cache; runtime enforces a minimum of `1` and keeps the default persistence budget within the distributed publication lease. |
 | `scanner.max_concurrent_set_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS`) | Caps concurrent set-level scanner tasks. `0` keeps topology-derived concurrency. |
 | `scanner.max_concurrent_disk_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS`) | Caps concurrent disk bucket walks per set. `0` keeps disk-count-derived concurrency. |
@@ -59,7 +59,7 @@ Speed presets (`crates/config/src/constants/scanner.rs`) set the base sleep mult
 | `slow` | 10x | 15s | 60s |
 | `slowest` | 100x | 15s | 30m |
 
-Use `scanner.delay`, `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis needs a precise override. With `idle_mode=true`, directory-level sleep is `1ms x factor` and object-level sleep is `time spent on the object x factor`, both capped at `max_wait`; a foreground-read floor of `FOREGROUND_READ_BACKOFF_PER_REQUEST_MS` (10ms) per concurrent GetObject/streaming read, capped at `FOREGROUND_READ_BACKOFF_MAX_MS` (250ms), is applied on top and can exceed the preset's `max_wait` (`crates/scanner/src/sleeper.rs`).
+Use `scanner.delay`, `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis needs a precise override. With `idle_mode=true`, directory-level sleep is `1ms x factor` and object-level sleep is `time spent on the object x factor`, both capped at `max_wait`; a foreground-read floor of `FOREGROUND_READ_BACKOFF_PER_REQUEST_MS` (10ms) per concurrent GetObject/streaming read, capped at `FOREGROUND_READ_BACKOFF_MAX_MS` (250ms), is applied on top and can exceed the preset's `max_wait` (`crates/scanner/src/sleeper.rs`). The per-disk foreground-priority gate is independent of `idle_mode` and remains active when sleeps are disabled.
 
 ### Environment-only scanner knobs
 
@@ -67,7 +67,7 @@ These have no persistent key and are read from the environment only.
 
 | Environment variable | Default (constant) | Effect |
 |---|---|---|
-| `RUSTFS_SCANNER_ENABLED` (deprecated alias `RUSTFS_ENABLE_SCANNER`) | `true` (`scanner_enabled_from_env`, `rustfs/src/module_switches.rs`) | Starts the data scanner at all. The heal manager is initialized whenever heal or scanner is enabled, because scanner-produced heal candidates need a consumer. |
+| `RUSTFS_SCANNER_ENABLED` (deprecated alias `RUSTFS_ENABLE_SCANNER`) | `true` (`scanner_enabled_from_env`, `rustfs/src/module_switches.rs`) | Starts the data scanner. Scanner-produced heal candidates require the heal manager, so the manager and MRF consumer also start when scanner is on, even if `RUSTFS_HEAL_ENABLED=false`. |
 | `RUSTFS_SCANNER_ALERT_COOLDOWN_SECS` | `86400` (`DEFAULT_SCANNER_ALERT_COOLDOWN_SECS`, `scanner_folder.rs`) | Per-(kind, bucket, object) cooldown between S3 excess-alert events; `0` emits every cycle. See [Scanner Excess Alerts](scanner-excess-alerts.md). |
 | `RUSTFS_SCANNER_DEEP_VERIFY_COOLDOWN_SECS` | `60` (`DEFAULT_SCANNER_DEEP_VERIFY_COOLDOWN_SECS`, `scanner_folder.rs`) | Objects modified within this window are skipped by deep (bitrot) verification in the current cycle. |
 | `RUSTFS_HEAL_OBJECT_SELECT_PROB` | `1024` (`DEFAULT_HEAL_OBJECT_SELECT_PROB`, `scanner_folder.rs`) | Sampling divisor for scanner-originated heal checks: roughly one object in N per cycle is selected for a low-priority heal check. |
@@ -75,6 +75,19 @@ These have no persistent key and are read from the environment only.
 | `RUSTFS_DATA_USAGE_FAILED_OBJECT_TTL_SECS` | `86400` (`DEFAULT_FAILED_OBJECT_TTL_SECS`, `scanner_folder.rs`) | Retention of per-bucket failed-object entries in the usage cache. |
 | `RUSTFS_DATA_USAGE_FAILED_OBJECTS_MAX` | `10000` (`DEFAULT_FAILED_OBJECTS_MAX`, `scanner_folder.rs`) | Cap on retained failed-object entries per bucket. |
 
+### Single-node, single-drive startup
+
+RustFS does not turn off background services based on drive count. This follows MinIO's service-level startup model: MinIO starts auto-heal/MRF from its server-pool initialization and starts the data scanner separately when its scanner switch is on; its single-drive branch changes background-heal state handling rather than silently disabling both service initializers ([server startup](https://github.com/minio/minio/blob/master/cmd/server-main.go#L1050-L1055), [heal startup](https://github.com/minio/minio/blob/master/cmd/background-newdisks-heal-ops.go#L373-L387), [scanner single-drive path](https://github.com/minio/minio/blob/master/cmd/data-scanner.go#L75-L87)). RustFS starts the scanner by default and starts the heal manager/MRF consumer whenever either `RUSTFS_SCANNER_ENABLED` or `RUSTFS_HEAL_ENABLED` is true. Startup logs report the configured switches and effective scanner, heal-manager, and MRF-consumer states.
+
+| `RUSTFS_SCANNER_ENABLED` | `RUSTFS_HEAL_ENABLED` | Effective behavior |
+|---|---|---|
+| `true` | `true` | Scanner usage, lifecycle/transition and replication discovery run; heal manager and MRF consumer run. |
+| `true` | `false` | Scanner remains on; heal manager and MRF consumer still start because scanner work can submit heal requests. |
+| `false` | `true` | Scanner-owned usage/lifecycle/replication discovery pauses; heal manager and MRF consumer remain available. |
+| `false` | `false` | Scanner and manager-backed heal/MRF consumer do not start. Usage freshness and scanner-driven lifecycle/replication discovery pause; manager-backed admin heal is unavailable, and durable MRF work waits for the consumer to return. |
+
+For a single slow drive, keep the default services enabled and use the per-disk foreground-priority gate and scanner pacing controls first. To opt into a maintenance-off mode explicitly, set both `RUSTFS_SCANNER_ENABLED=false` and `RUSTFS_HEAL_ENABLED=false`; changing disk count alone never selects that mode. `RUSTFS_SCANNER_BITROT_CYCLE_SECS=off` only disables periodic deep bitrot work and does not stop regular scanner cycles or checkpoints. `RUSTFS_HEAL_AUTO_HEAL_ENABLE=false` disables automatic heal policy but does not prevent the heal manager from starting when the scanner needs its request consumer.
+
 ### Cycle budgets and cadence
 
 When the cycle duration control is unset, RustFS uses the finite 1800-second default. An explicit `0` preserves the compatibility behavior of an unbounded cycle; object and directory budgets likewise remain unbounded when explicitly set to `0`. Invalid or overflowing duration environment values are configuration errors rather than silent fallback values.
@@ -93,6 +106,14 @@ Lifecycle and replication configuration inspection is bounded so a slow metadata
 
 With the default 30-day bitrot cycle, the clean-idle interval is capped at the bitrot cycle divided by the object selection window (about 42 minutes with the default `RUSTFS_HEAL_OBJECT_SELECT_PROB`), which preserves the intended wall-clock bitrot coverage. If periodic bitrot is disabled, the clean-idle cap is 24 hours. The effective interval is jittered by up to 10 percent to avoid synchronized scanner starts.
 
+## Per-disk background I/O admission
+
+Local disk wrapper operations share an admission gate across foreground requests and scanner work. Foreground I/O never waits for a scanner permit. A queued scanner operation may start when its wrapper is idle or after eight foreground operation completions. If it is at the head of the background queue and no background operation is active, it starts after at most 100 ms; time queued behind an active background operation is additional. Only one background operation per wrapper runs at a time. Scanner directory reads use bounded batches, and scanner metadata reads release their permit before sending data to the listing consumer. GET streaming reads hold permits only while a disk read is pending, not for the lifetime of a client response. Cancellation removes queued waiters; a permit is released after its last operation or blocking worker holder exits. Tokio filesystem work already dispatched before cancellation may continue beyond the caller's future lifetime.
+
+The `rustfs_disk_io_admission_active` foreground gauge is sampled while background waiters are queued; `rustfs_disk_io_admission_active{class="background"}` reports the active background permit, while `rustfs_disk_io_admission_waiters`, `rustfs_disk_io_admission_total`, and `rustfs_disk_io_admission_wait_seconds` report queued work, background admission counts, and wait time by disk endpoint. `rustfs_scanner_cache_serialized_bytes_total` counts logical encoded cache bytes; `rustfs_scanner_cache_serialize_seconds` records encoding time; `rustfs_scanner_cache_persist_attempt_seconds` records save-attempt duration and retry result, before ambiguous-write reconciliation.
+
+This is logical operation admission, not an OS disk scheduler: foreground and background operations may overlap after a background permit is granted, and the gate does not promise a bandwidth or IOPS limit. It is scoped to each `LocalDiskWrapper`; separate paths to one physical device and remote disk RPCs do not share this gate. It also does not observe or schedule unrelated host processes that share the physical device.
+
 ## Status Endpoint
 
 ```text
@@ -324,7 +345,7 @@ Heal knobs are environment-only and read by `HealConfig::default` (`crates/heal/
 
 | Environment variable | Default (constant) | Effect |
 |---|---|---|
-| `RUSTFS_HEAL_ENABLED` (deprecated alias `RUSTFS_ENABLE_HEAL`) | `true` (`heal_enabled_from_env`, `rustfs/src/module_switches.rs`) | Master switch for the background heal manager. |
+| `RUSTFS_HEAL_ENABLED` (deprecated alias `RUSTFS_ENABLE_HEAL`) | `true` (`heal_enabled_from_env`, `rustfs/src/module_switches.rs`) | Starts the background heal manager when enabled. The manager also starts whenever the scanner is enabled; disable both switches to keep the manager and MRF consumer from starting. |
 | `RUSTFS_HEAL_AUTO_HEAL_ENABLE` | `true` (`DEFAULT_HEAL_AUTO_HEAL_ENABLE`) | Enables automatic healing of detected issues; `false` leaves healing to manual admin requests. |
 | `RUSTFS_HEAL_QUEUE_SIZE` | `10000` (`DEFAULT_HEAL_QUEUE_SIZE`) | Heal request queue capacity. The manager reserves at least one slot per Scanner/ReadRepair/AutoHeal source, or 10% for larger queues, so sustained normal-priority MRF or internal background work cannot consume every slot. |
 | `RUSTFS_HEAL_INTERVAL_SECS` | `10` (`DEFAULT_HEAL_INTERVAL_SECS`) | Heal manager polling interval. |
@@ -438,7 +459,7 @@ Only two MinIO scanner variables are recognized. `apply_external_env_compat` (`c
 |---|---|---|
 | `MINIO_SCANNER_SPEED` / `scanner speed` | `RUSTFS_SCANNER_SPEED` / `scanner.speed` | Env alias mapped at startup; preset names and the preset table are identical. |
 | `MINIO_SCANNER_CYCLE` / `scanner cycle` | `RUSTFS_SCANNER_CYCLE` / `scanner.cycle` | Env alias mapped at startup. |
-| `MINIO_SCANNER_IDLE_SPEED` / `scanner idle_speed` (`on` default, `off`) | `RUSTFS_SCANNER_IDLE_MODE` / `scanner.idle_mode` (`true` default, `false`) | Not mapped; must be rewritten. Direction matches (`on` and `true` both mean throttled). Both RustFS channels also accept `on`/`off` as booleans (`parse_config_bool`, `parse_bool_str`). RustFS `false` additionally disables the foreground-read backoff floor that MinIO does not have, so the scanner competes with foreground reads at full speed; use it only for benchmarks or exclusive-I/O windows. |
+| `MINIO_SCANNER_IDLE_SPEED` / `scanner idle_speed` (`on` default, `off`) | `RUSTFS_SCANNER_IDLE_MODE` / `scanner.idle_mode` (`true` default, `false`) | Not mapped; must be rewritten. Direction matches (`on` and `true` both mean throttled). Both RustFS channels also accept `on`/`off` as booleans (`parse_config_bool`, `parse_bool_str`). RustFS `false` disables sleeper delays and foreground-read backoff, but per-disk foreground-priority admission remains active. |
 | `MINIO_HEAL_BITROTSCAN` / `heal bitrotscan` (default `off`) | `RUSTFS_SCANNER_BITROT_CYCLE_SECS` / `heal.bitrot_cycle` (default 30 days) | Not mapped. RustFS deep-scans periodically by default; set `off` or `disabled` to reproduce MinIO's default. |
 | `MINIO_API_STALE_UPLOADS_EXPIRY` (24h) | `RUSTFS_API_STALE_UPLOADS_EXPIRY` (`DEFAULT_STALE_UPLOADS_EXPIRY`, 24h, `crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs`) | Not mapped; same default. |
 | `MINIO_API_STALE_UPLOADS_CLEANUP_INTERVAL` (6h) | `RUSTFS_API_STALE_UPLOADS_CLEANUP_INTERVAL` (`DEFAULT_STALE_UPLOADS_CLEANUP_INTERVAL`, 6h) | Not mapped; same default. |
diff --git a/rustfs/src/startup_background.rs b/rustfs/src/startup_background.rs
index 393b7ad3d..a7fde64d2 100644
--- a/rustfs/src/startup_background.rs
+++ b/rustfs/src/startup_background.rs
@@ -35,6 +35,21 @@ const LOG_SUBSYSTEM_STARTUP: &str = "startup";
 const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured";
 const EVENT_ODM_BACKFILL_RECOVERY_CONFIGURED: &str = "odm_backfill_recovery_configured";
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+struct BackgroundServicePlan {
+    scanner_enabled: bool,
+    heal_enabled: bool,
+    heal_manager_enabled: bool,
+}
+
+fn resolve_background_service_plan(scanner_enabled: bool, heal_enabled: bool) -> BackgroundServicePlan {
+    BackgroundServicePlan {
+        scanner_enabled,
+        heal_enabled,
+        heal_manager_enabled: scanner_enabled || heal_enabled,
+    }
+}
+
 pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Result<bool> {
     // Pin the bitrot algorithms before anything can write or verify a shard:
     // the check costs well under a millisecond, and in strict mode a drifted
@@ -44,46 +59,43 @@ pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Resu
 
     let _ = create_ahm_services_cancel_token();
 
-    let enable_scanner = scanner_enabled_from_env();
-    let enable_heal = heal_enabled_from_env();
-
-    info!(
-        target: "rustfs::main::run",
-        event = EVENT_BACKGROUND_SERVICES_CONFIGURED,
-        component = LOG_COMPONENT_MAIN,
-        subsystem = LOG_SUBSYSTEM_STARTUP,
-        enable_scanner = enable_scanner,
-        enable_heal = enable_heal,
-        "Background services configured"
-    );
+    let plan = resolve_background_service_plan(scanner_enabled_from_env(), heal_enabled_from_env());
 
     let workload_provider: Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync> =
         Arc::new(RustFsWorkloadAdmissionSnapshotProvider);
     let _ = set_workload_admission_snapshot_provider(workload_provider.clone());
     rustfs_scanner::set_scanner_workload_admission_snapshot_provider(workload_provider.clone());
 
-    if enable_heal || enable_scanner {
+    if plan.heal_manager_enabled {
         let heal_storage = Arc::new(ECStoreHealStorage::new(store.clone()));
         init_heal_manager_with_workload_provider(heal_storage, None, Some(workload_provider)).await?;
     }
 
-    if !enable_heal && !enable_scanner {
-        debug!(
-            target: "rustfs::main::run",
-            event = EVENT_BACKGROUND_SERVICES_CONFIGURED,
-            component = LOG_COMPONENT_MAIN,
-            subsystem = LOG_SUBSYSTEM_STARTUP,
-            enable_scanner = false,
-            enable_heal = false,
-            ahm_state = "skipped",
-            reason = "disabled",
-            "Background services disabled"
-        );
-    }
+    let heal_manager_reason = if plan.heal_enabled {
+        "heal_switch"
+    } else if plan.scanner_enabled {
+        "scanner_dependency"
+    } else {
+        "disabled"
+    };
+    info!(
+        target: "rustfs::main::run",
+        event = EVENT_BACKGROUND_SERVICES_CONFIGURED,
+        component = LOG_COMPONENT_MAIN,
+        subsystem = LOG_SUBSYSTEM_STARTUP,
+        configured_scanner = plan.scanner_enabled,
+        configured_heal = plan.heal_enabled,
+        scanner = if plan.scanner_enabled { "enabled" } else { "disabled" },
+        scanner_usage_lifecycle_replication = if plan.scanner_enabled { "enabled" } else { "paused" },
+        heal_manager = if plan.heal_manager_enabled { "started" } else { "not_started" },
+        mrf_consumer = if plan.heal_manager_enabled { "started" } else { "not_started" },
+        heal_manager_reason,
+        "Background service startup resolved"
+    );
 
     init_on_demand_migration_backfill_runtime(store).await;
 
-    Ok(enable_scanner)
+    Ok(plan.scanner_enabled)
 }
 
 /// Installs the backfill runner (admin start/cancel/status need it even
@@ -123,3 +135,56 @@ async fn init_on_demand_migration_backfill_runtime(store: Arc<ECStore>) {
         "On-demand migration backfill recovery configured"
     );
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{BackgroundServicePlan, resolve_background_service_plan};
+
+    #[test]
+    fn enabling_both_switches_starts_scanner_and_heal_manager() {
+        assert_eq!(
+            resolve_background_service_plan(true, true),
+            BackgroundServicePlan {
+                scanner_enabled: true,
+                heal_enabled: true,
+                heal_manager_enabled: true,
+            }
+        );
+    }
+
+    #[test]
+    fn scanner_requires_the_heal_manager_and_mrf_consumer() {
+        assert_eq!(
+            resolve_background_service_plan(true, false),
+            BackgroundServicePlan {
+                scanner_enabled: true,
+                heal_enabled: false,
+                heal_manager_enabled: true,
+            }
+        );
+    }
+
+    #[test]
+    fn disabling_scanner_keeps_heal_manager_when_explicitly_enabled() {
+        assert_eq!(
+            resolve_background_service_plan(false, true),
+            BackgroundServicePlan {
+                scanner_enabled: false,
+                heal_enabled: true,
+                heal_manager_enabled: true,
+            }
+        );
+    }
+
+    #[test]
+    fn disabling_both_background_switches_skips_heal_manager() {
+        assert_eq!(
+            resolve_background_service_plan(false, false),
+            BackgroundServicePlan {
+                scanner_enabled: false,
+                heal_enabled: false,
+                heal_manager_enabled: false,
+            }
+        );
+    }
+}